diff --git a/.coverage-baseline b/.coverage-baseline index e59d5c49..2ce0a0b6 100644 --- a/.coverage-baseline +++ b/.coverage-baseline @@ -1 +1 @@ -78.2 +78.4 diff --git a/internal/controller/gittarget_controller_test.go b/internal/controller/gittarget_controller_test.go index 0bb5c191..2aa4f6b3 100644 --- a/internal/controller/gittarget_controller_test.go +++ b/internal/controller/gittarget_controller_test.go @@ -1119,6 +1119,11 @@ var _ = Describe("GitTarget Controller Security", func() { // GitTarget when its age-key Secret disappears. The wait must therefore exceed a full // RequeueStreamSettleInterval — with the shared 10s `timeout` it equalled one, so a // deletion landing just after a reconcile lost the race by milliseconds. + // + // Two intervals, not one: the previous budget covered exactly two ticks with no slack, so + // a CI runner busy enough to delay one of them by a second failed the spec on timing + // alone. Eventually returns as soon as the Secret is back, so the extra room is free on + // every run that was going to pass anyway. Eventually(func(g Gomega) { var recreated corev1.Secret err := k8sClient.Get(ctx, secretKey, &recreated) @@ -1127,7 +1132,7 @@ var _ = Describe("GitTarget Controller Security", func() { g.Expect(ageKeyName).NotTo(BeEmpty()) g.Expect(string(ageKeyValue)).To(ContainSubstring("AGE-SECRET-KEY-")) g.Expect(recreated.Annotations).To(HaveKey(encryptionSecretRecipientAnnoKey)) - }, RequeueStreamSettleInterval+timeout, interval).Should(Succeed()) + }, 2*RequeueStreamSettleInterval+timeout, interval).Should(Succeed()) Expect(k8sClient.Delete(ctx, target)).Should(Succeed()) Expect(k8sClient.Delete(ctx, gitProvider)).Should(Succeed()) diff --git a/internal/queue/attribution_index.go b/internal/queue/attribution_index.go index 8f9d4108..c5d7fe85 100644 --- a/internal/queue/attribution_index.go +++ b/internal/queue/attribution_index.go @@ -78,6 +78,15 @@ const ( // RV and never matches the removal event's RV). The reason is driven by the value's // verb, not by which key matched. AttributionExactDeleteCollectionItem AttributionResult = "exact_deletecollection_item" + // AttributionCollectionUID is a removal matched to a deletecollection fact whose uid set + // contains this object. There is no over-attribution risk in it: either the API server said it + // deleted this object, or it did not. + AttributionCollectionUID AttributionResult = "collection_uid" + // AttributionCollectionScope is a removal matched to a deletecollection fact by scope alone — + // same type and namespace, selector accepting the object's labels, within the collection window. + // It is the weakest evidence the join has, which is why it is reached only when every more + // specific tier missed. + AttributionCollectionScope AttributionResult = "collection_scope" // AttributionAbsent means no usable author fact matched before the grace elapsed. AttributionAbsent AttributionResult = "absent" ) @@ -101,6 +110,18 @@ type AuthorFact struct { ResourceVersion string `json:"resourceVersion,omitempty"` StageTimestamp string `json:"stageTimestamp,omitempty"` IsServiceAccount bool `json:"isServiceAccount,omitempty"` + + // LabelSelector is the selector the request URI expressed, carried on a COLLECTION fact only. + // It is the intent the actor stated, and evaluating it against the object a watch event carries + // is a better test of membership than reading back a list the API server may not have sent. + // Empty means the collection covered everything of its type in its namespace, which is what + // --all means. + LabelSelector string `json:"labelSelector,omitempty"` + // UIDs is the set of objects a collection delete covered, reduced from the response body at the + // receiver, on a COLLECTION fact only. It is absent when the API server sent no body — a + // truncated, aggregated, or metadata-only response — and when the set was larger than the cap, + // in which case the join falls back to scope matching, which is already correct. + UIDs []string `json:"uids,omitempty"` } // AuthorResolution is the structured result of an attribution lookup. diff --git a/internal/queue/author_fact.go b/internal/queue/author_fact.go new file mode 100644 index 00000000..73c74d8c --- /dev/null +++ b/internal/queue/author_fact.go @@ -0,0 +1,153 @@ +// SPDX-License-Identifier: Apache-2.0 + +package queue + +import ( + "context" + "net/url" + "strings" + "time" + + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/metric" + "k8s.io/apimachinery/pkg/runtime/schema" + auditv1 "k8s.io/apiserver/pkg/apis/audit/v1" + + "github.com/ConfigButler/gitops-reverser/internal/auditutil" + "github.com/ConfigButler/gitops-reverser/internal/telemetry" +) + +// DefaultCollectionUIDCap is how many uids a collection fact may carry before the set is dropped +// and the join falls back to scope matching. +// +// It bounds one entry's size, the broadcast to every subscriber of the type, and the replay on +// restart. A uid is 36 bytes, so this cap is a few hundred kilobytes at worst — against a response +// body for the same request that runs to tens of megabytes. It is a tuning number rather than a +// correctness one: the fallback is already correct, so the cap only decides how often the precise +// path is taken, and a collection delete large enough to exceed it is exactly the one whose body a +// production cluster with audit truncation enabled would not have sent in the first place. +const DefaultCollectionUIDCap = 10000 + +// labelSelectorQueryParam is where a collection request states which objects it meant. +const labelSelectorQueryParam = "labelSelector" + +// AuthorFactFromEvent reduces one accepted, mutating audit event to the fact the stream carries, +// reporting false when the event can never name an author. Only facts that WOULD have been stored +// may be published: an event with no objectRef or no user produces nothing, or waiters are woken by +// facts that can name nobody. +// +// The one rule that changes from the per-key write path is the name check. A deletecollection is +// name-less by nature and is now exactly the case that produces a fact — one fact describing the +// COLLECTION, which every removal in its scope joins — so "no resolvable name" becomes "no name and +// not a collection verb". +// +// The caller has already applied the intrinsic accept gate: reads, failures, dry runs, and +// non-ResponseComplete stages never reach here. +func AuthorFactFromEvent(ctx context.Context, event auditv1.Event) (AuthorFact, schema.GroupResource, bool) { + if event.ObjectRef == nil || event.ObjectRef.Resource == "" { + return AuthorFact{}, schema.GroupResource{}, false + } + user := resolveUserInfo(event) + if user.Username == "" { + return AuthorFact{}, schema.GroupResource{}, false + } + + collection := strings.EqualFold(event.Verb, deleteCollectionVerb) + op, _ := auditutil.VerbToOperation(event.Verb) + identity := auditutil.IdentityFromAuditEvent(event, op) + if identity.Name == "" && !collection { + return AuthorFact{}, schema.GroupResource{}, false + } + + groupResource := schema.GroupResource{Group: event.ObjectRef.APIGroup, Resource: event.ObjectRef.Resource} + fact := AuthorFact{ + GroupResource: groupResourceKey(groupResource.Group, groupResource.Resource), + Namespace: identity.Namespace, + Name: identity.Name, + UID: string(identity.UID), + Author: user.Username, + DisplayName: user.DisplayName, + Email: user.Email, + Verb: event.Verb, + Subresource: event.ObjectRef.Subresource, + AuditID: string(event.AuditID), + ResourceVersion: resourceVersionFromEvent(event), + IsServiceAccount: strings.HasPrefix(user.Username, serviceAccountUserPrefix), + } + if !event.StageTimestamp.IsZero() { + fact.StageTimestamp = event.StageTimestamp.UTC().Format(time.RFC3339Nano) + } + if collection { + describeCollection(ctx, &fact, event) + } + return fact, groupResource, true +} + +// describeCollection turns an object-shaped fact into a fact about a COLLECTION: what the actor +// asked for — the type, the namespace, and the selector from the request URI — plus the uids the +// API server said it covered, when it sent them. +// +// The per-object identity goes away, because a collection request names no object. That is the +// asymmetry the expander used to fight: audit reports the ONE request that was made, the watch +// reports each of the N objects that changed, and the join belongs at the point where both are in +// hand rather than in a receiver rebuilding N from one. +func describeCollection(ctx context.Context, fact *AuthorFact, event auditv1.Event) { + fact.Name = "" + fact.UID = "" + fact.ResourceVersion = "" + fact.LabelSelector = labelSelectorFromRequestURI(event.RequestURI) + fact.UIDs = collectionUIDs(ctx, event) +} + +// labelSelectorFromRequestURI reads the selector a collection request expressed. It is better +// evidence than the response body: the selector is the INTENT the actor stated, evaluating it +// against the object a watch event carries tests membership directly, and it is there even when the +// body is not. An empty selector means the request covered everything of its type in its namespace, +// which is what --all means. +func labelSelectorFromRequestURI(requestURI string) string { + if requestURI == "" { + return "" + } + parsed, err := url.Parse(requestURI) + if err != nil { + return "" + } + return parsed.Query().Get(labelSelectorQueryParam) +} + +// collectionUIDs reduces a deletecollection response body to the set of uids it covered, at the +// receiver, on the goroutine that already has the body decoded. It returns nil when the body was +// absent, hollow, or larger than the cap — all of which degrade the join to scope matching, which +// is the floor that must work on its own. Crossing the cap is COUNTED, so "we fell back to scope" +// is visible rather than inferred. +func collectionUIDs(ctx context.Context, event auditv1.Event) []string { + items := deleteCollectionItems(event.ResponseObject) + if len(items) == 0 { + return nil + } + if len(items) > DefaultCollectionUIDCap { + recordCollectionDegraded(ctx, "uid_cap") + return nil + } + uids := make([]string, 0, len(items)) + for _, item := range items { + if item.UID != "" { + uids = append(uids, string(item.UID)) + } + } + if len(uids) == 0 { + recordCollectionDegraded(ctx, "no_uids") + return nil + } + return uids +} + +// recordCollectionDegraded counts one collection fact that lost its uid set, under the bounded +// reason it lost it. +func recordCollectionDegraded(ctx context.Context, reason string) { + if telemetry.AttributionCollectionDegradedTotal == nil { + return + } + telemetry.AttributionCollectionDegradedTotal.Add(ctx, 1, + metric.WithAttributes(attribute.String("reason", reason))) +} diff --git a/internal/queue/author_fact_test.go b/internal/queue/author_fact_test.go new file mode 100644 index 00000000..18663d94 --- /dev/null +++ b/internal/queue/author_fact_test.go @@ -0,0 +1,120 @@ +// SPDX-License-Identifier: Apache-2.0 + +package queue + +import ( + "fmt" + "strings" + "testing" + + "github.com/stretchr/testify/require" + authnv1 "k8s.io/api/authentication/v1" + "k8s.io/apimachinery/pkg/runtime" + auditv1 "k8s.io/apiserver/pkg/apis/audit/v1" + + "github.com/ConfigButler/gitops-reverser/internal/telemetry" +) + +// collectionDeleteEvent is one name-less collection delete over count objects. +func collectionDeleteEvent(selector string, count int) auditv1.Event { + items := make([]string, 0, count) + for i := range count { + items = append(items, fmt.Sprintf(`{"metadata":{"name":"cm-%d","namespace":"team-a","uid":"uid-%d"}}`, i, i)) + } + uri := "/api/v1/namespaces/team-a/configmaps" + if selector != "" { + uri += "?labelSelector=" + selector + } + return auditv1.Event{ + AuditID: "dc-1", + Verb: "deletecollection", + Stage: auditv1.StageResponseComplete, + User: authnv1.UserInfo{Username: "alice"}, + RequestURI: uri, + ObjectRef: &auditv1.ObjectReference{ + Resource: "configmaps", Namespace: "team-a", APIVersion: "v1", + }, + ResponseObject: &runtime.Unknown{ + Raw: []byte(`{"apiVersion":"v1","kind":"ConfigMapList","items":[` + strings.Join(items, ",") + `]}`), + }, + } +} + +func TestAuthorFactFromEvent_CollectionCarriesScopeSelectorAndUIDs(t *testing.T) { + fact, groupResource, ok := AuthorFactFromEvent(t.Context(), collectionDeleteEvent("app%3Dweb", 2)) + require.True(t, ok, "a name-less deletecollection is the case that DOES produce a fact") + require.Equal(t, "configmaps", groupResource.Resource) + require.Equal(t, "team-a", fact.Namespace) + require.Equal(t, "app=web", fact.LabelSelector) + require.Equal(t, []string{"uid-0", "uid-1"}, fact.UIDs) + require.Empty(t, fact.Name) + require.Empty(t, fact.UID) +} + +func TestAuthorFactFromEvent_UIDSetIsDroppedPastTheCapAndCounted(t *testing.T) { + reader, err := telemetry.InitTestExporter() + require.NoError(t, err) + + fact, _, ok := AuthorFactFromEvent(t.Context(), collectionDeleteEvent("", DefaultCollectionUIDCap+1)) + require.True(t, ok) + // The fact degrades to scope matching, which is already correct — and says so in the metrics, + // so "we fell back to scope" is visible rather than inferred. + require.Nil(t, fact.UIDs) + require.Empty(t, fact.LabelSelector) + + degraded, found := telemetry.CollectInt64Sum(reader, "gitopsreverser_attribution_collection_degraded_total", + map[string]string{"reason": "uid_cap"}) + require.True(t, found) + require.Equal(t, int64(1), degraded) +} + +func TestAuthorFactFromEvent_BodylessCollectionStillProducesAFact(t *testing.T) { + event := collectionDeleteEvent("", 0) + event.ResponseObject = nil + + // The shape a production cluster with --audit-webhook-truncate-enabled actually sends, and the + // one the old expander gave up on entirely. + fact, _, ok := AuthorFactFromEvent(t.Context(), event) + require.True(t, ok) + require.Nil(t, fact.UIDs) + require.Equal(t, "alice", fact.Author) +} + +func TestAuthorFactFromEvent_EventsThatCanNameNobody(t *testing.T) { + cases := map[string]auditv1.Event{ + "no objectRef": {Verb: "create", User: authnv1.UserInfo{Username: "alice"}}, + "no resource": { + Verb: "create", User: authnv1.UserInfo{Username: "alice"}, + ObjectRef: &auditv1.ObjectReference{APIGroup: "apps"}, + }, + "no user": { + Verb: "create", + ObjectRef: &auditv1.ObjectReference{Resource: "configmaps", Name: "cm"}, + }, + "no resolvable name on an object verb": { + Verb: "create", User: authnv1.UserInfo{Username: "alice"}, + ObjectRef: &auditv1.ObjectReference{Resource: "configmaps"}, + }, + } + for name, event := range cases { + t.Run(name, func(t *testing.T) { + _, _, ok := AuthorFactFromEvent(t.Context(), event) + require.False(t, ok) + }) + } +} + +func TestAuthorFactFromEvent_ObjectWriteCarriesTheIdentityTheJoinNeeds(t *testing.T) { + fact, groupResource, ok := AuthorFactFromEvent(t.Context(), mutationEvent("update", "uid-1", "101", "alice")) + require.True(t, ok) + require.Equal(t, "apps", groupResource.Group) + require.Equal(t, "deployments", groupResource.Resource) + require.Equal(t, "apps/deployments", fact.GroupResource) + require.Equal(t, "uid-1", fact.UID) + require.Equal(t, "101", fact.ResourceVersion) + require.Equal(t, "alice", fact.Author) + require.NotEmpty(t, fact.StageTimestamp) + // An ordinary write is about one object, so it carries no collection fields. + require.Empty(t, fact.LabelSelector) + require.Nil(t, fact.UIDs) +} diff --git a/internal/queue/fact_index.go b/internal/queue/fact_index.go new file mode 100644 index 00000000..90fdc623 --- /dev/null +++ b/internal/queue/fact_index.go @@ -0,0 +1,481 @@ +// SPDX-License-Identifier: Apache-2.0 + +package queue + +import ( + "context" + "strings" + "sync" + "time" + + "github.com/go-logr/logr" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/metric" + "k8s.io/apimachinery/pkg/runtime/schema" + + "github.com/ConfigButler/gitops-reverser/internal/telemetry" +) + +// Defaults for the in-memory fact index. Redis used to enforce the TTL and the memory ceiling for +// free; holding the facts in process moves both jobs here, which is why every one of these numbers +// is a bound rather than a hint. +const ( + // DefaultFactIndexMaxFactsPerType caps one (audit route, group/resource)'s entries across all + // four match structures. It is the primary cap because it is the fair one: a burst on one noisy + // type — a deletecollection over ten thousand objects, a large rollout — must not evict every + // other type's facts. One entry is a fact plus its bookkeeping, a few hundred bytes, so a type + // at its cap costs low single-digit megabytes. + DefaultFactIndexMaxFactsPerType = 4096 + + // DefaultFactIndexMaxFactsTotal caps the whole index, so the pod's memory is bounded by a number + // that does not scale with how many types happen to be watched. It sits well above the per-type + // cap: reaching it takes many types simultaneously busy, and eviction then falls on the type + // holding the most. + DefaultFactIndexMaxFactsTotal = 65536 + + // DefaultFactCollectionWindow is how long after a deletecollection's stageTimestamp a removal in + // its scope may still be credited to it. It is far shorter than the fact TTL and can afford to + // be: under the deletion-as-intent rule the removal being attributed happens at delete-REQUEST + // time, so finalizers do not stretch it, and the window only has to cover audit batching plus + // clock skew. Ten times the default grace window leaves room for a slow batch without letting an + // unrelated delete a minute later be claimed. + DefaultFactCollectionWindow = 30 * time.Second + + // DefaultFactIndexSweepInterval is how often aged-out entries are reclaimed. It only bounds + // MEMORY, never correctness: a lookup checks the TTL itself, so an entry past its horizon is + // never joined merely because the sweep has not run yet. + DefaultFactIndexSweepInterval = 30 * time.Second + + // factFollowErrorBackoff paces retries when the transport fails. The follower does not give up + // on a transport error: a follower that returned would leave attribution silently dead for the + // life of the process. + factFollowErrorBackoff = time.Second + + // deleteCollectionVerb is the one verb published as a fact about a COLLECTION rather than about + // an object. + deleteCollectionVerb = "deletecollection" +) + +// evictionReasonPerType and evictionReasonTotal are the bounded reasons on the eviction counter. +// They are separate because they mean different things to an operator: per-type says one type is +// hotter than its share, total says the whole index is under pressure. +const ( + evictionReasonPerType = "per_type" + evictionReasonTotal = "total" +) + +// FactQuery is one watch event's identity, as the join reads it. It is everything the index needs +// to try all five tiers, so a caller assembles it once rather than threading five arguments. +type FactQuery struct { + // AuditRoute partitions the index. It leads every key for the same reason the streams are named + // per route: a fact from cluster A must never name the author of an object watched on cluster B. + AuditRoute string + GroupResource schema.GroupResource + UID string + ResourceVersion string + // Namespace and Labels serve the collection tier only: they are how a removal finds the + // deletecollection whose scope covered it. + Namespace string + Labels map[string]string + // ExactCapable is true for ADDED and MODIFIED, whose resourceVersion is the one the write + // produced. A removal's is not, so it consults the weaker tiers the exact-capable events skip. + ExactCapable bool +} + +// FactIndexConfig configures the index. Every zero field falls back to its Default… constant, so +// the zero value is the supported configuration. +type FactIndexConfig struct { + // TTL is how long a fact stays joinable, and doubles as the follower's replay horizon so a + // restart warms the index with exactly the window that is still usable. + TTL time.Duration + // MaxFactsPerType caps one (route, group/resource); MaxFactsTotal caps the whole index. + MaxFactsPerType int + MaxFactsTotal int + // CollectionWindow bounds the scope-matching tier. + CollectionWindow time.Duration + // SweepInterval is how often aged-out entries are reclaimed. + SweepInterval time.Duration + Log logr.Logger +} + +// FactIndex is the transport-agnostic half of attribution: the four match structures the join reads, +// the waiter registry a blocked resolver parks on, and the loop that fills both from whatever +// transport it was handed. +// +// There is exactly ONE index per process, not one per GitTarget. A fact names a write that happened +// in Kubernetes, not a consumer interested in it, so one fact already serves every GitTarget that +// needs it; five GitTargets mirroring one Deployment would otherwise hold five copies of every fact +// and bill memory against a number that has nothing to do with how much is happening in the cluster. +// The fan-out that does do useful work is the SUBSCRIPTION set, which is per type. +type FactIndex struct { + ttl time.Duration + maxPerType int + maxTotal int + collectionWindow time.Duration + sweepInterval time.Duration + log logr.Logger + + streams *FactStreamSet + waiters *factWaiterRegistry + + mu sync.Mutex + scopes map[factScope]*scopeFacts + total int + seq uint64 +} + +// NewFactIndex builds an empty index. +func NewFactIndex(cfg FactIndexConfig) *FactIndex { + index := &FactIndex{ + ttl: cfg.TTL, + maxPerType: cfg.MaxFactsPerType, + maxTotal: cfg.MaxFactsTotal, + collectionWindow: cfg.CollectionWindow, + sweepInterval: cfg.SweepInterval, + log: cfg.Log, + streams: NewFactStreamSet(), + waiters: newFactWaiterRegistry(), + scopes: map[factScope]*scopeFacts{}, + } + if index.ttl <= 0 { + index.ttl = DefaultAttributionFactTTL + } + if index.maxPerType <= 0 { + index.maxPerType = DefaultFactIndexMaxFactsPerType + } + if index.maxTotal <= 0 { + index.maxTotal = DefaultFactIndexMaxFactsTotal + } + if index.collectionWindow <= 0 { + index.collectionWindow = DefaultFactCollectionWindow + } + if index.sweepInterval <= 0 { + index.sweepInterval = DefaultFactIndexSweepInterval + } + return index +} + +// Apply stores one delivered entry's facts and wakes whoever was waiting for them. Facts are +// applied in the order they were delivered, which is what makes the latest tier last-writer-wins +// mean the last fact APPENDED rather than whichever goroutine reached the map first. +func (i *FactIndex) Apply(ctx context.Context, entry FactEntry) { + scope := factScope{route: entry.Key.AuditRoute, groupResource: entry.Key.groupResource()} + now := time.Now() + for _, fact := range entry.Facts { + i.waiters.wake(i.store(ctx, scope, fact, now)) + } +} + +// Await resolves a watch event, waiting up to grace for a fact that has not been delivered yet. It +// returns an AttributionAbsent resolution when nothing matched in time; it never blocks longer than +// the grace and never returns an error path. +// +// The order of the first two statements is the design, not a detail. The waiter is registered +// BEFORE the index is read, so a fact applied in the gap between the two signals a waiter that is +// already listening. Checking first and registering after loses exactly that fact — the race the +// poll loop used to paper over by looking again. +func (i *FactIndex) Await(ctx context.Context, query FactQuery, grace time.Duration) AuthorResolution { + waiter := i.waiters.register(query.waiterKeys()) + defer i.waiters.unregister(waiter) + + if resolution := i.Lookup(query); resolution.Result != AttributionAbsent { + return resolution + } + if grace <= 0 { + return AuthorResolution{Result: AttributionAbsent} + } + + timer := time.NewTimer(grace) + defer timer.Stop() + for { + select { + case <-ctx.Done(): + return AuthorResolution{Result: AttributionAbsent} + case <-timer.C: + return AuthorResolution{Result: AttributionAbsent} + case <-waiter.ch: + if resolution := i.Lookup(query); resolution.Result != AttributionAbsent { + return resolution + } + } + } +} + +// Lookup reads the index once, trying the tiers strongest-first: +// +// 1. the exact (uid, rv) fact, the only exact-capable join; +// 2. the last-writer-wins fact for that uid, for a removal whose rv never matches; +// 3. a collection fact whose uid set contains this object; +// 4. a collection fact whose scope, selector, and window cover it; +// 5. the rv-only escape hatch. +// +// Precedence is the correctness argument for the collection tiers. A scope match is the weakest +// evidence here and can name the wrong human, so it is only ever reached when nothing more specific +// applies: an unrelated delete by another actor during the same window is claimed by its own fact +// at tier 2 and never reaches tier 4. +func (i *FactIndex) Lookup(query FactQuery) AuthorResolution { + now := time.Now() + cutoff := now.Add(-i.ttl) + + i.mu.Lock() + defer i.mu.Unlock() + facts, ok := i.scopes[query.scope()] + if !ok { + return AuthorResolution{Result: AttributionAbsent} + } + + if query.UID != "" && query.ResourceVersion != "" { + if fact, found := facts.lookupExact(query.UID, query.ResourceVersion, cutoff); found { + return AuthorResolution{Fact: fact, Result: attributionResultForFact(fact, false)} + } + } + if !query.ExactCapable { + if query.UID != "" { + if fact, found := facts.lookupLatest(query.UID, cutoff); found { + return AuthorResolution{Fact: fact, Result: attributionResultForFact(fact, true)} + } + } + resolution := facts.matchCollection(query, now, cutoff, i.collectionWindow) + if resolution.Result != AttributionAbsent { + return resolution + } + } + if query.ResourceVersion != "" { + if fact, found := facts.lookupRV(query.ResourceVersion, cutoff); found { + return AuthorResolution{Fact: fact, Result: attributionResultForFact(fact, true)} + } + } + return AuthorResolution{Result: AttributionAbsent} +} + +// Run follows the subscription set until the context ends, applying what it reads and reporting +// what it lost. It returns only when the context ends: a transport failure is retried, because a +// follower that gave up would leave attribution silently dead for the life of the process. +func (i *FactIndex) Run(ctx context.Context, follower FactFollower) error { + subscription := follower.FollowFacts(i.streams.Keys(), i.ttl) + i.streams.Observe(subscription.SetStreams) + defer i.streams.Observe(nil) + + lastSweep := time.Now() + for { + delivery, err := subscription.Next(ctx) + if err != nil { + if ctx.Err() != nil { + return nil + } + i.log.Error(err, "attribution fact follower failed; retrying") + if waitErr := waitBlock(ctx, factFollowErrorBackoff); waitErr != nil { + return nil + } + continue + } + for _, entry := range delivery.Entries { + i.Apply(ctx, entry) + } + i.reportGaps(ctx, delivery.Gaps) + if now := time.Now(); now.Sub(lastSweep) >= i.sweepInterval { + i.Sweep(now) + lastSweep = now + } + } +} + +// Streams is the reference-counted set of (route, group/resource) pairs this process follows. The +// watch side acquires a reference when it starts covering a type and releases it when the last +// watch on that type goes away; Run makes the follower track it. +func (i *FactIndex) Streams() *FactStreamSet { + return i.streams +} + +// Sweep reclaims every entry past the TTL horizon. Lookups already ignore aged-out entries, so this +// bounds memory rather than deciding what may be joined. +func (i *FactIndex) Sweep(now time.Time) int { + cutoff := now.Add(-i.ttl) + i.mu.Lock() + defer i.mu.Unlock() + removed := 0 + for scope, facts := range i.scopes { + removed += facts.sweep(cutoff) + if facts.empty() { + delete(i.scopes, scope) + } + } + i.total -= removed + return removed +} + +// Len reports how many entries the index holds across every scope and structure. +func (i *FactIndex) Len() int { + i.mu.Lock() + defer i.mu.Unlock() + return i.total +} + +// store files one fact under every structure it can serve and returns the waiter keys it filled. +// The policy mirrors the tiers Lookup reads: a collection fact is about a collection and lands only +// there, a uid-bearing fact takes the exact and latest tiers, and the rv-only hatch exists for the +// fact that has an rv and no uid — a uid-bearing fact's rv-only entry would be dead, since the watch +// side always carries a uid. +func (i *FactIndex) store(ctx context.Context, scope factScope, fact AuthorFact, now time.Time) []factWaiterKey { + i.mu.Lock() + defer i.mu.Unlock() + + facts := i.scopeFor(scope) + before := facts.count + keys := i.file(facts, scope, fact, now) + i.total += facts.count - before + i.enforceCaps(ctx, scope, facts) + return keys +} + +// file writes one fact into the structures it can serve. +func (i *FactIndex) file(facts *scopeFacts, scope factScope, fact AuthorFact, now time.Time) []factWaiterKey { + switch { + case strings.EqualFold(fact.Verb, deleteCollectionVerb): + facts.putCollection(newIndexedCollection(fact, now, i.nextSeq())) + return []factWaiterKey{{scope: scope, kind: factKindCollection, value: fact.Namespace}} + case fact.UID != "": + var keys []factWaiterKey + if fact.ResourceVersion != "" { + facts.putExact(fact.UID, fact.ResourceVersion, &indexedFact{fact: fact, at: now, seq: i.nextSeq()}) + keys = append( + keys, + factWaiterKey{ + scope: scope, + kind: factKindExact, + value: exactWaiterValue(fact.UID, fact.ResourceVersion), + }, + ) + } + facts.putLatest(fact.UID, &indexedFact{fact: fact, at: now, seq: i.nextSeq()}) + return append(keys, factWaiterKey{scope: scope, kind: factKindLatest, value: fact.UID}) + case fact.ResourceVersion != "": + facts.putRV(fact.ResourceVersion, &indexedFact{fact: fact, at: now, seq: i.nextSeq()}) + return []factWaiterKey{{scope: scope, kind: factKindRV, value: fact.ResourceVersion}} + default: + // A fact with neither a uid nor a resourceVersion can never be joined. The publish side does + // not produce one; storing it anyway would only fill the index with entries no query reaches. + return nil + } +} + +// scopeFor returns the scope's structures, creating them on first use. +func (i *FactIndex) scopeFor(scope factScope) *scopeFacts { + facts, ok := i.scopes[scope] + if !ok { + facts = newScopeFacts() + i.scopes[scope] = facts + } + return facts +} + +// nextSeq hands out the sequence number that distinguishes a live entry from a stale eviction +// reference to the entry it replaced. +func (i *FactIndex) nextSeq() uint64 { + i.seq++ + return i.seq +} + +// enforceCaps evicts oldest-first until the scope and the index are both back within their caps. +// Every eviction is COUNTED: an attribution that was dropped because the index was full has to look +// different in the metrics from one that was never published, or the index silently absorbs the +// bursts it was bounded for. +func (i *FactIndex) enforceCaps(ctx context.Context, scope factScope, facts *scopeFacts) { + for facts.count > i.maxPerType { + if !facts.evictOldest() { + break + } + i.total-- + recordFactIndexEviction(ctx, evictionReasonPerType) + } + if facts.empty() { + delete(i.scopes, scope) + } + for i.total > i.maxTotal { + if !i.evictFromLargestScope(ctx) { + break + } + } +} + +// evictFromLargestScope takes one entry from whichever type is holding the most, which puts the +// pressure of a global overflow on whatever caused it. +func (i *FactIndex) evictFromLargestScope(ctx context.Context) bool { + var largest *scopeFacts + var largestScope factScope + for scope, facts := range i.scopes { + if largest == nil || facts.count > largest.count { + largest, largestScope = facts, scope + } + } + if largest == nil || !largest.evictOldest() { + return false + } + i.total-- + recordFactIndexEviction(ctx, evictionReasonTotal) + if largest.empty() { + delete(i.scopes, largestScope) + } + return true +} + +// reportGaps counts and names every stream this follower was trimmed past. A gap is the one loss +// this transport can see, and it is a real degradation: the facts it covers are gone, so the commits +// that needed them will be authored unresolved. Reporting it is why the transport is a log with +// positions rather than fire-and-forget publish and subscribe. +func (i *FactIndex) reportGaps(ctx context.Context, gaps []FactStreamGap) { + for _, gap := range gaps { + if telemetry.AttributionFactStreamGapsTotal != nil { + telemetry.AttributionFactStreamGapsTotal.Add(ctx, 1, + metric.WithAttributes(attribute.String("stream", gap.Key.String()))) + } + i.log.Info("attribution fact stream was trimmed past this follower; the facts in the gap are "+ + "lost and the commits that needed them are authored unresolved", + "stream", gap.Key.String(), "cursor", gap.Cursor, "firstSurviving", gap.FirstSurviving) + } +} + +// scope is the partition this query resolves within. +func (q FactQuery) scope() factScope { + return factScope{ + route: q.AuditRoute, + groupResource: groupResourceKey(q.GroupResource.Group, q.GroupResource.Resource), + } +} + +// waiterKeys are the candidates this query could resolve through — exactly the tiers Lookup tries, +// so a fact that would satisfy the lookup always wakes the waiter, and no fact that would not +// wakes it for nothing. +func (q FactQuery) waiterKeys() []factWaiterKey { + scope := q.scope() + var keys []factWaiterKey + if q.UID != "" && q.ResourceVersion != "" { + keys = append(keys, factWaiterKey{scope: scope, kind: factKindExact, + value: exactWaiterValue(q.UID, q.ResourceVersion)}) + } + if !q.ExactCapable { + if q.UID != "" { + keys = append(keys, factWaiterKey{scope: scope, kind: factKindLatest, value: q.UID}) + } + keys = append(keys, factWaiterKey{scope: scope, kind: factKindCollection, value: q.Namespace}) + } + if q.ResourceVersion != "" { + keys = append(keys, factWaiterKey{scope: scope, kind: factKindRV, value: q.ResourceVersion}) + } + return keys +} + +// exactWaiterValue renders the exact tier's (uid, rv) pair as one waiter value. The separator is a +// byte neither half can contain. +func exactWaiterValue(uid, rv string) string { + return uid + "\x00" + rv +} + +// recordFactIndexEviction counts one evicted entry under its bounded reason. +func recordFactIndexEviction(ctx context.Context, reason string) { + if telemetry.AttributionFactIndexEvictionsTotal == nil { + return + } + telemetry.AttributionFactIndexEvictionsTotal.Add(ctx, 1, + metric.WithAttributes(attribute.String("reason", reason))) +} diff --git a/internal/queue/fact_index_store.go b/internal/queue/fact_index_store.go new file mode 100644 index 00000000..e8047dbe --- /dev/null +++ b/internal/queue/fact_index_store.go @@ -0,0 +1,387 @@ +// SPDX-License-Identifier: Apache-2.0 + +package queue + +import ( + "time" + + "k8s.io/apimachinery/pkg/labels" +) + +// factKind names one of the four match structures a fact can land in. The set is closed: a fact +// that fits none of them is not stored, because a fact nothing can ever join is only memory. +type factKind uint8 + +const ( + // factKindExact is (uid, rv), the only exact-capable join, serving ADDED and MODIFIED. + factKindExact factKind = iota + // factKindLatest is (uid), last-writer-wins, serving removals whose rv never matches. + factKindLatest + // factKindRV is (rv), the escape hatch for a fact with an rv but no uid. + factKindRV + // factKindCollection is (namespace), time-bounded, serving removals caused by a + // deletecollection. + factKindCollection +) + +// factScope is the partition every key in the index leads with: the audit route the facts arrived +// under and the group/resource they are about. +// +// The route is not decoration. The index is one per process while the streams are one per (route, +// group/resource), so an index keyed on the type alone would pool two clusters' facts in one map +// and hand a watch event on cluster B an author from cluster A. The rv-only tier is where that +// bites hardest, because a resourceVersion is opaque and not unique across clusters, and the +// collection tier is where it bites most quietly, because a namespace name says nothing about which +// cluster it is in. +type factScope struct { + route string + groupResource string +} + +// exactFactKey is the (uid, rv) pair of the exact tier. +type exactFactKey struct { + uid string + rv string +} + +// indexedFact is one stored fact plus the insertion time the TTL sweep reads and the sequence +// number that tells a stale eviction reference from a live entry. +type indexedFact struct { + fact AuthorFact + at time.Time + seq uint64 +} + +// indexedCollection is one deletecollection fact: the actor, the scope it named, the selector it +// expressed, and the set of uids it covered when the API server sent a body. +// +// The selector is parsed here, on the applying goroutine, rather than at lookup: a lookup runs once +// per removal event, so parsing there would parse one collection's selector N times on the watch +// shard's blocking path. What stays at lookup time is the DECISION — try uid membership, fall back +// to scope — not the parsing. +type indexedCollection struct { + fact AuthorFact + at time.Time + seq uint64 + stage time.Time + // selector is nil when the fact carried none, which matches every object of the type in the + // namespace, because that is what --all means. invalidSelector is set when the fact carried one + // that would not parse: such a fact is never scope-matched, since treating it as match-all would + // name an author over a wider scope than the actor asked for. + selector labels.Selector + invalidSelector bool + // uids is the set the collection covered, or nil when the body was absent or the set was + // dropped past its cap. Nil is not empty: it means fall back to scope matching, which is the + // floor that must work on its own. + uids map[string]struct{} +} + +// newIndexedCollection reduces one collection fact to what the join needs: the parsed selector, the +// uid set as a set, and the delete-request time the window is measured from. +func newIndexedCollection(fact AuthorFact, now time.Time, seq uint64) *indexedCollection { + entry := &indexedCollection{fact: fact, at: now, seq: seq, stage: parseStageTimestamp(fact.StageTimestamp)} + if fact.LabelSelector != "" { + selector, err := labels.Parse(fact.LabelSelector) + if err != nil { + entry.invalidSelector = true + } else { + entry.selector = selector + } + } + if len(fact.UIDs) > 0 { + entry.uids = make(map[string]struct{}, len(fact.UIDs)) + for _, uid := range fact.UIDs { + entry.uids[uid] = struct{}{} + } + } + return entry +} + +// parseStageTimestamp reads a fact's stage timestamp, returning the zero time when it carries none +// or carries one that will not parse. The caller then falls back to the insertion time, which is +// later and therefore only ever narrows the window. +func parseStageTimestamp(stamp string) time.Time { + if stamp == "" { + return time.Time{} + } + at, err := time.Parse(time.RFC3339Nano, stamp) + if err != nil { + return time.Time{} + } + return at +} + +// factRef locates one stored entry for eviction without holding a pointer into a map. The sequence +// number is what makes it safe to keep a stale reference: an entry overwritten since the reference +// was taken has a different sequence, so the reference is skipped rather than removing the newer +// entry that took its place. +type factRef struct { + kind factKind + uid string + rv string + seq uint64 +} + +// scopeFacts is one (route, group/resource)'s four match structures, its oldest-first insertion +// order, and its entry count. Bounding per scope rather than globally is what keeps a burst on one +// noisy type — a deletecollection over ten thousand objects, a large rollout — from evicting every +// other type's facts. +type scopeFacts struct { + exact map[exactFactKey]*indexedFact + latest map[string]*indexedFact + rvOnly map[string]*indexedFact + collections []*indexedCollection + // order is every live entry's reference in insertion order, oldest first. It may hold stale + // references, which remove skips. + order []factRef + count int +} + +func newScopeFacts() *scopeFacts { + return &scopeFacts{ + exact: map[exactFactKey]*indexedFact{}, + latest: map[string]*indexedFact{}, + rvOnly: map[string]*indexedFact{}, + } +} + +// putExact stores the immutable (uid, rv) fact. +func (s *scopeFacts) putExact(uid, rv string, entry *indexedFact) { + key := exactFactKey{uid: uid, rv: rv} + if _, ok := s.exact[key]; !ok { + s.count++ + } + s.exact[key] = entry + s.order = append(s.order, factRef{kind: factKindExact, uid: uid, rv: rv, seq: entry.seq}) +} + +// putLatest stores the last-writer-wins pointer for an object. Entries are applied in delivery +// order, which is what makes "last writer" mean the last fact appended rather than whichever +// goroutine got there first. +func (s *scopeFacts) putLatest(uid string, entry *indexedFact) { + if _, ok := s.latest[uid]; !ok { + s.count++ + } + s.latest[uid] = entry + s.order = append(s.order, factRef{kind: factKindLatest, uid: uid, seq: entry.seq}) +} + +// putRV stores the rv-only escape hatch. +func (s *scopeFacts) putRV(rv string, entry *indexedFact) { + if _, ok := s.rvOnly[rv]; !ok { + s.count++ + } + s.rvOnly[rv] = entry + s.order = append(s.order, factRef{kind: factKindRV, rv: rv, seq: entry.seq}) +} + +// putCollection appends one collection fact. Collection facts are not keyed one per namespace: +// two actors may delete collections in one namespace within the same window, and each removal must +// be able to find the one that covered it. +func (s *scopeFacts) putCollection(entry *indexedCollection) { + s.collections = append(s.collections, entry) + s.order = append(s.order, factRef{kind: factKindCollection, seq: entry.seq}) + s.count++ +} + +// lookupExact reads the exact tier. +func (s *scopeFacts) lookupExact(uid, rv string, cutoff time.Time) (AuthorFact, bool) { + return liveFact(s.exact[exactFactKey{uid: uid, rv: rv}], cutoff) +} + +// lookupLatest reads the last-writer-wins tier. +func (s *scopeFacts) lookupLatest(uid string, cutoff time.Time) (AuthorFact, bool) { + return liveFact(s.latest[uid], cutoff) +} + +// lookupRV reads the rv-only escape hatch. +func (s *scopeFacts) lookupRV(rv string, cutoff time.Time) (AuthorFact, bool) { + return liveFact(s.rvOnly[rv], cutoff) +} + +// matchCollection resolves a removal against the collection tier, in the two passes the design +// orders: uid membership first, because either the API server said it deleted this object or it did +// not, and scope matching second, because it accepts a bounded risk of naming the wrong human and +// so must only ever be reached when nothing more precise applies. +func (s *scopeFacts) matchCollection(q FactQuery, now, cutoff time.Time, window time.Duration) AuthorResolution { + for i := len(s.collections) - 1; i >= 0; i-- { + entry := s.collections[i] + if !entry.covers(q, cutoff) || entry.uids == nil { + continue + } + if _, ok := entry.uids[q.UID]; ok { + return AuthorResolution{Fact: entry.fact, Result: AttributionCollectionUID} + } + } + for i := len(s.collections) - 1; i >= 0; i-- { + entry := s.collections[i] + if !entry.covers(q, cutoff) || !entry.inWindow(now, window) || !entry.selects(q.Labels) { + continue + } + return AuthorResolution{Fact: entry.fact, Result: AttributionCollectionScope} + } + return AuthorResolution{Result: AttributionAbsent} +} + +// sweep drops every entry inserted before the cutoff and reports how many went. It also compacts +// the eviction order, so a scope that has aged out entirely leaves nothing behind. +func (s *scopeFacts) sweep(cutoff time.Time) int { + removed := 0 + for key, entry := range s.exact { + if entry.at.Before(cutoff) { + delete(s.exact, key) + removed++ + } + } + for key, entry := range s.latest { + if entry.at.Before(cutoff) { + delete(s.latest, key) + removed++ + } + } + for key, entry := range s.rvOnly { + if entry.at.Before(cutoff) { + delete(s.rvOnly, key) + removed++ + } + } + kept := s.collections[:0] + for _, entry := range s.collections { + if entry.at.Before(cutoff) { + removed++ + continue + } + kept = append(kept, entry) + } + s.collections = kept + s.count -= removed + s.compact() + return removed +} + +// empty reports whether the scope holds nothing, so the index can forget it. +func (s *scopeFacts) empty() bool { + return s.count == 0 +} + +// evictOldest removes the oldest live entry, reporting whether it found one. Eviction is +// oldest-first across all four structures of the scope: the eviction order is insertion order, and +// which structure an entry landed in says nothing about how likely it still is to be joined. +func (s *scopeFacts) evictOldest() bool { + for len(s.order) > 0 { + ref := s.order[0] + s.order = s.order[1:] + if s.remove(ref) { + s.count-- + return true + } + } + return false +} + +// remove deletes the entry a reference names, reporting whether it was still live. A reference +// whose sequence no longer matches is stale — the entry was overwritten by a later fact, which has +// its own reference further along the order — and removing the newer entry for it would evict a +// fact that had only just arrived. +func (s *scopeFacts) remove(ref factRef) bool { + switch ref.kind { + case factKindExact: + key := exactFactKey{uid: ref.uid, rv: ref.rv} + if entry, ok := s.exact[key]; ok && entry.seq == ref.seq { + delete(s.exact, key) + return true + } + case factKindLatest: + if entry, ok := s.latest[ref.uid]; ok && entry.seq == ref.seq { + delete(s.latest, ref.uid) + return true + } + case factKindRV: + if entry, ok := s.rvOnly[ref.rv]; ok && entry.seq == ref.seq { + delete(s.rvOnly, ref.rv) + return true + } + case factKindCollection: + for i, entry := range s.collections { + if entry.seq == ref.seq { + s.collections = append(s.collections[:i], s.collections[i+1:]...) + return true + } + } + } + return false +} + +// compact drops eviction references whose entries are gone, so the order does not grow with every +// swept entry. +func (s *scopeFacts) compact() { + kept := s.order[:0] + for _, ref := range s.order { + if s.live(ref) { + kept = append(kept, ref) + } + } + s.order = kept +} + +// live reports whether a reference still names the entry it was taken for. +func (s *scopeFacts) live(ref factRef) bool { + switch ref.kind { + case factKindExact: + entry, ok := s.exact[exactFactKey{uid: ref.uid, rv: ref.rv}] + return ok && entry.seq == ref.seq + case factKindLatest: + entry, ok := s.latest[ref.uid] + return ok && entry.seq == ref.seq + case factKindRV: + entry, ok := s.rvOnly[ref.rv] + return ok && entry.seq == ref.seq + case factKindCollection: + for _, entry := range s.collections { + if entry.seq == ref.seq { + return true + } + } + } + return false +} + +// covers reports whether a collection fact is in scope for a query at all: same namespace, and not +// yet aged out. The namespace is part of the collection key, so it binds in both passes — uid +// membership included, because the uid set of a collection in another namespace has no business +// naming an author here. +func (c *indexedCollection) covers(q FactQuery, cutoff time.Time) bool { + return !c.at.Before(cutoff) && c.fact.Namespace == q.Namespace +} + +// inWindow reports whether the collection was requested recently enough for a removal to be +// credited to it. The clock that matters is the audit event's own stageTimestamp, i.e. delete +// REQUEST time: under the deletion-as-intent rule the removal being attributed happens when +// deletionTimestamp is set, so finalizers do not stretch this window. +func (c *indexedCollection) inWindow(now time.Time, window time.Duration) bool { + at := c.stage + if at.IsZero() { + at = c.at + } + return !now.After(at.Add(window)) +} + +// selects reports whether the collection's selector accepts an object's labels. +func (c *indexedCollection) selects(objectLabels map[string]string) bool { + if c.invalidSelector { + return false + } + if c.selector == nil { + return true + } + return c.selector.Matches(labels.Set(objectLabels)) +} + +// liveFact returns a stored fact when it is present and has not aged out. Expiry is checked on read +// as well as swept, so a fact is never joined past its TTL just because the sweep has not run. +func liveFact(entry *indexedFact, cutoff time.Time) (AuthorFact, bool) { + if entry == nil || entry.at.Before(cutoff) { + return AuthorFact{}, false + } + return entry.fact, true +} diff --git a/internal/queue/fact_index_test.go b/internal/queue/fact_index_test.go new file mode 100644 index 00000000..5abcd4de --- /dev/null +++ b/internal/queue/fact_index_test.go @@ -0,0 +1,483 @@ +// SPDX-License-Identifier: Apache-2.0 + +package queue + +import ( + "testing" + "time" + + "github.com/stretchr/testify/require" + "k8s.io/apimachinery/pkg/runtime/schema" + + "github.com/ConfigButler/gitops-reverser/internal/telemetry" +) + +// The index is exercised through the in-memory transport rather than by calling Apply, because +// "publish a fact and have a watch event resolve it" is the whole behaviour. Driving it end to end +// also keeps the two halves honest about delivery ORDER, which the latest tier depends on. + +const ( + // factIndexTestBlock keeps the follower's idle round short so a test that waits for a fact does + // not wait out a whole second. + factIndexTestBlock = 10 * time.Millisecond + // factIndexTestGrace bounds a wait that is meant to succeed; a passing test never spends it. + factIndexTestGrace = 5 * time.Second + // factIndexTestSettle is how long a negative assertion gives a fact it does not want to see. + factIndexTestSettle = 200 * time.Millisecond +) + +// factIndexHarness is one index following one in-memory transport, with the follower running for +// the test's lifetime. +type factIndexHarness struct { + t *testing.T + index *FactIndex + transport *MemoryFactStream +} + +func newFactIndexHarness(t *testing.T, cfg FactIndexConfig) *factIndexHarness { + t.Helper() + if cfg.SweepInterval == 0 { + cfg.SweepInterval = time.Hour + } + // The transport keeps everything for the test's lifetime: what a case with a short TTL is + // exercising is the INDEX's horizon, and letting retention race delivery would only make it flaky. + transport := NewMemoryFactStream(MemoryFactStreamConfig{TTL: time.Hour, Block: factIndexTestBlock}) + index := NewFactIndex(cfg) + + // The follower runs for the test's lifetime; its outcome is judged on the test's own goroutine. + done := make(chan struct{}) + var runErr error + go func() { + defer close(done) + runErr = index.Run(t.Context(), transport) + }() + t.Cleanup(func() { + <-done + require.NoError(t, runErr) + }) + + return &factIndexHarness{t: t, index: index, transport: transport} +} + +// publish follows the stream if it is not followed already, then appends one batch to it. +func (h *factIndexHarness) publish(key FactStreamKey, facts ...AuthorFact) { + h.t.Helper() + h.index.Streams().Acquire(key) + require.NoError(h.t, h.transport.PublishFacts(h.t.Context(), key, facts)) +} + +// waitForFacts blocks until the index holds at least n entries, so a NEGATIVE assertion can be made +// about a fact that has definitely been applied rather than about one still in flight. +func (h *factIndexHarness) waitForFacts(n int) { + h.t.Helper() + require.Eventually(h.t, func() bool { return h.index.Len() >= n }, + factIndexTestGrace, factIndexTestBlock, "the follower never applied %d facts", n) +} + +// resolve waits out the grace for a query it expects to succeed. +func (h *factIndexHarness) resolve(query FactQuery) AuthorResolution { + h.t.Helper() + return h.index.Await(h.t.Context(), query, factIndexTestGrace) +} + +// absent asserts a query does not resolve, giving a late fact time to arrive and prove it wrong. +func (h *factIndexHarness) absent(query FactQuery) { + h.t.Helper() + resolution := h.index.Await(h.t.Context(), query, factIndexTestSettle) + require.Equal(h.t, AttributionAbsent, resolution.Result, "resolved to %q", resolution.Fact.Author) +} + +func deploymentsGroupResource() schema.GroupResource { + return schema.GroupResource{Group: "apps", Resource: "deployments"} +} + +func factIndexTestStream(route string) FactStreamKey { + return FactStreamKeyFor(route, deploymentsGroupResource()) +} + +// factIndexTestUID is the object every ordinary write in these tests is about. +const factIndexTestUID = "uid-1" + +// objectFact is one ordinary write's fact. +func objectFact(author, rv string) AuthorFact { + return AuthorFact{ + GroupResource: "apps/deployments", + Namespace: "team-a", + Name: "web", + UID: factIndexTestUID, + ResourceVersion: rv, + Author: author, + Verb: "update", + } +} + +// aliceCollectionFact is one deletecollection's fact: the actor, the scope, the selector, and the +// uids the API server said it covered. +func aliceCollectionFact(selector string, uids ...string) AuthorFact { + return AuthorFact{ + GroupResource: "apps/deployments", + Namespace: "team-a", + Author: "alice", + Verb: "deletecollection", + LabelSelector: selector, + UIDs: uids, + StageTimestamp: time.Now().UTC().Format(time.RFC3339Nano), + } +} + +// objectQuery is one watch event's identity. +func objectQuery(route, uid, rv string, exactCapable bool) FactQuery { + return FactQuery{ + AuditRoute: route, + GroupResource: deploymentsGroupResource(), + UID: uid, + ResourceVersion: rv, + Namespace: "team-a", + ExactCapable: exactCapable, + } +} + +func TestFactIndex_JoinPolicyDependsOnTheEventKind(t *testing.T) { + harness := newFactIndexHarness(t, FactIndexConfig{}) + harness.publish(factIndexTestStream("prod-eu-1"), objectFact("alice", "101")) + + // ADDED / MODIFIED present the resourceVersion the write produced, so they join exactly. + exact := harness.resolve(objectQuery("prod-eu-1", "uid-1", "101", true)) + require.Equal(t, AttributionExactUser, exact.Result) + require.Equal(t, "alice", exact.Fact.Author) + + harness.waitForFacts(2) + + // An exact-capable event whose rv does not match must NOT fall through to the latest tier: that + // pointer may name a different, older author than the create or update this event represents. + harness.absent(objectQuery("prod-eu-1", "uid-1", "999", true)) + + // A removal's rv never matches the write's, so it is the event kind that consults latest. + removal := harness.resolve(objectQuery("prod-eu-1", "uid-1", "999", false)) + require.Equal(t, AttributionWeak, removal.Result) + require.Equal(t, "alice", removal.Fact.Author) +} + +func TestFactIndex_RouteIsolatesOtherwiseIdenticalFacts(t *testing.T) { + harness := newFactIndexHarness(t, FactIndexConfig{}) + // The same type, uid, and resourceVersion on two clusters. A resourceVersion is opaque and not + // unique across clusters, so without the route leading every key one cluster's actor would name + // the author of the other cluster's object. + harness.publish(factIndexTestStream("prod-eu-1"), objectFact("alice", "101")) + harness.publish(factIndexTestStream("prod-us-1"), objectFact("bob", "101")) + // And the rv-only hatch, which is where it bites hardest because it carries no uid at all. + rvOnly := AuthorFact{GroupResource: "apps/deployments", Namespace: "team-a", ResourceVersion: "202", Verb: "update"} + euOnly, usOnly := rvOnly, rvOnly + euOnly.Author, usOnly.Author = "eu-rv", "us-rv" + harness.publish(factIndexTestStream("prod-eu-1"), euOnly) + harness.publish(factIndexTestStream("prod-us-1"), usOnly) + + require.Equal(t, "alice", harness.resolve(objectQuery("prod-eu-1", "uid-1", "101", true)).Fact.Author) + require.Equal(t, "bob", harness.resolve(objectQuery("prod-us-1", "uid-1", "101", true)).Fact.Author) + require.Equal(t, "eu-rv", harness.resolve(objectQuery("prod-eu-1", "", "202", true)).Fact.Author) + require.Equal(t, "us-rv", harness.resolve(objectQuery("prod-us-1", "", "202", true)).Fact.Author) + + // A route nobody published under resolves nothing, rather than borrowing a neighbour's fact. + harness.publish(factIndexTestStream("staging")) + harness.absent(objectQuery("staging", "uid-1", "101", true)) +} + +func TestFactIndex_LatestTierIsLastWriterWins(t *testing.T) { + harness := newFactIndexHarness(t, FactIndexConfig{}) + key := factIndexTestStream("prod-eu-1") + // Two writes to one object, in one batch, in the order they were appended. + harness.publish(key, objectFact("alice", "101"), objectFact("bob", "102")) + harness.waitForFacts(3) + + removal := harness.resolve(objectQuery("prod-eu-1", "uid-1", "999", false)) + require.Equal(t, "bob", removal.Fact.Author, "the latest tier must name the last fact applied") + + // The exact tier is immutable per (uid, rv), so the earlier write is still exactly joinable. + require.Equal(t, "alice", harness.resolve(objectQuery("prod-eu-1", "uid-1", "101", true)).Fact.Author) +} + +func TestFactIndex_FactsAreAbsentPastTheTTL(t *testing.T) { + const ttl = 500 * time.Millisecond + harness := newFactIndexHarness(t, FactIndexConfig{TTL: ttl}) + harness.publish(factIndexTestStream("prod-eu-1"), objectFact("alice", "101")) + require.Equal(t, "alice", harness.resolve(objectQuery("prod-eu-1", "uid-1", "101", true)).Fact.Author) + + time.Sleep(3 * ttl) + + // Expiry is decided on READ, so an aged-out fact is never joined merely because the sweep has + // not run yet; the sweep then reclaims it. + require.Equal(t, AttributionAbsent, harness.index.Lookup(objectQuery("prod-eu-1", "uid-1", "101", true)).Result) + require.Positive(t, harness.index.Sweep(time.Now())) + require.Zero(t, harness.index.Len()) +} + +func TestFactIndex_PerTypeCapEvictsOldestFirstAndCountsIt(t *testing.T) { + reader, err := telemetry.InitTestExporter() + require.NoError(t, err) + + // rv-only facts occupy one entry each, so the cap counts what the test publishes. + harness := newFactIndexHarness(t, FactIndexConfig{MaxFactsPerType: 2}) + rvFact := func(author, rv string) AuthorFact { + return AuthorFact{GroupResource: "apps/deployments", ResourceVersion: rv, Author: author, Verb: "update"} + } + key := factIndexTestStream("prod-eu-1") + harness.publish(key, rvFact("first", "1"), rvFact("second", "2"), rvFact("third", "3")) + // A second type must keep its own facts: the cap is per type precisely so a burst on one noisy + // type cannot evict everything else. + other := FactStreamKeyFor("prod-eu-1", schema.GroupResource{Resource: "configmaps"}) + harness.publish( + other, + AuthorFact{GroupResource: "configmaps", ResourceVersion: "9", Author: "quiet", Verb: "update"}, + ) + + quiet := FactQuery{AuditRoute: "prod-eu-1", GroupResource: schema.GroupResource{Resource: "configmaps"}, + ResourceVersion: "9", ExactCapable: true} + require.Equal(t, "quiet", harness.resolve(quiet).Fact.Author) + + require.Equal(t, "third", harness.resolve(objectQuery("prod-eu-1", "", "3", true)).Fact.Author) + require.Equal(t, "second", harness.resolve(objectQuery("prod-eu-1", "", "2", true)).Fact.Author) + harness.absent(objectQuery("prod-eu-1", "", "1", true)) + + evicted, ok := telemetry.CollectInt64Sum(reader, "gitopsreverser_attribution_fact_index_evictions_total", + map[string]string{"reason": evictionReasonPerType}) + require.True(t, ok, "an eviction must be counted, never silently absorbed") + require.Equal(t, int64(1), evicted) +} + +func TestFactIndex_TotalCapEvictsFromTheLargestType(t *testing.T) { + reader, err := telemetry.InitTestExporter() + require.NoError(t, err) + + harness := newFactIndexHarness(t, FactIndexConfig{MaxFactsTotal: 2}) + busy := factIndexTestStream("prod-eu-1") + harness.publish(busy, + AuthorFact{GroupResource: "apps/deployments", ResourceVersion: "1", Author: "first", Verb: "update"}, + AuthorFact{GroupResource: "apps/deployments", ResourceVersion: "2", Author: "second", Verb: "update"}, + ) + require.Equal(t, "second", harness.resolve(objectQuery("prod-eu-1", "", "2", true)).Fact.Author) + + quiet := FactStreamKeyFor("prod-eu-1", schema.GroupResource{Resource: "configmaps"}) + harness.publish(quiet, + AuthorFact{GroupResource: "configmaps", ResourceVersion: "9", Author: "quiet", Verb: "update"}) + + // The overflow falls on the type holding the most, so the pressure lands where it came from. + quietQuery := FactQuery{AuditRoute: "prod-eu-1", GroupResource: schema.GroupResource{Resource: "configmaps"}, + ResourceVersion: "9", ExactCapable: true} + require.Equal(t, "quiet", harness.resolve(quietQuery).Fact.Author) + harness.absent(objectQuery("prod-eu-1", "", "1", true)) + require.Equal(t, 2, harness.index.Len()) + + evicted, ok := telemetry.CollectInt64Sum(reader, "gitopsreverser_attribution_fact_index_evictions_total", + map[string]string{"reason": evictionReasonTotal}) + require.True(t, ok) + require.Equal(t, int64(1), evicted) +} + +func TestFactIndex_WaiterIsWokenByALateFact(t *testing.T) { + harness := newFactIndexHarness(t, FactIndexConfig{}) + key := factIndexTestStream("prod-eu-1") + harness.index.Streams().Acquire(key) + query := objectQuery("prod-eu-1", "uid-1", "101", true) + + resolved := make(chan AuthorResolution, 1) + go func() { resolved <- harness.index.Await(t.Context(), query, factIndexTestGrace) }() + + // The wait is what the whole design is for: the watch event arrives first, by roughly one audit + // batch window, and the fact that names its author has not been published yet. + require.Eventually(t, func() bool { return harness.index.waiters.len() > 0 }, + factIndexTestGrace, factIndexTestBlock, "the resolver never registered a waiter") + require.NoError(t, harness.transport.PublishFacts(t.Context(), key, []AuthorFact{objectFact("alice", "101")})) + + select { + case resolution := <-resolved: + require.Equal(t, "alice", resolution.Fact.Author) + case <-time.After(factIndexTestGrace): + t.Fatal("a waiter was never woken by the fact it was waiting for") + } + require.Zero(t, harness.index.waiters.len(), "a resolved waiter must leave nothing registered") +} + +func TestFactIndex_WaiterRegisteredBeforeTheCheckKeepsAFactAppliedInTheGap(t *testing.T) { + index := NewFactIndex(FactIndexConfig{}) + query := objectQuery("prod-eu-1", "uid-1", "101", true) + + // Await's step 1, verbatim: register the candidates BEFORE reading the index. + waiter := index.waiters.register(query.waiterKeys()) + defer index.waiters.unregister(waiter) + + // The gap between registering and checking. Registering after the check would lose exactly this + // fact, which is the race the poll loop used to paper over by looking again. + index.Apply(t.Context(), FactEntry{ + Key: factIndexTestStream("prod-eu-1"), + Facts: []AuthorFact{objectFact("alice", "101")}, + }) + + // The signal is buffered, so it is still there for a waiter that was not listening yet. + select { + case <-waiter.ch: + default: + t.Fatal("a fact applied in the register-then-check gap left no signal") + } + require.Equal(t, "alice", index.Lookup(query).Fact.Author) +} + +func TestFactIndex_CollectionMatchesByUIDMembership(t *testing.T) { + harness := newFactIndexHarness(t, FactIndexConfig{}) + key := factIndexTestStream("prod-eu-1") + harness.publish(key, aliceCollectionFact("", "uid-1", "uid-2")) + harness.waitForFacts(1) + + // One fact, N removals: uid membership carries no over-attribution risk at all, because either + // the API server said it deleted this object or it did not. + for _, uid := range []string{"uid-1", "uid-2"} { + resolution := harness.resolve(objectQuery("prod-eu-1", uid, "999", false)) + require.Equal(t, AttributionCollectionUID, resolution.Result) + require.Equal(t, "alice", resolution.Fact.Author) + } + + // An object outside the namespace the collection named is not covered by it. + outside := objectQuery("prod-eu-1", "uid-1", "999", false) + outside.Namespace = "team-b" + harness.absent(outside) +} + +func TestFactIndex_CollectionMatchesByScopeAndSelector(t *testing.T) { + harness := newFactIndexHarness(t, FactIndexConfig{}) + key := factIndexTestStream("prod-eu-1") + // A body-less deletecollection — the shape a production cluster with audit truncation enabled + // actually sends — so only the scope and the selector are left to join on. + harness.publish(key, aliceCollectionFact("app=web")) + harness.waitForFacts(1) + + matching := objectQuery("prod-eu-1", "uid-1", "999", false) + matching.Labels = map[string]string{"app": "web", "tier": "front"} + resolution := harness.resolve(matching) + require.Equal(t, AttributionCollectionScope, resolution.Result) + require.Equal(t, "alice", resolution.Fact.Author) + + // The selector is the intent the actor expressed: an object it does not select was not part of + // the collection, so naming that actor would name the wrong human. + unmatched := objectQuery("prod-eu-1", "uid-2", "999", false) + unmatched.Labels = map[string]string{"app": "api"} + harness.absent(unmatched) + + // An ADDED or MODIFIED event never reaches the collection tier: a collection delete produces + // removals only. + written := objectQuery("prod-eu-1", "uid-1", "999", true) + written.Labels = matching.Labels + harness.absent(written) +} + +func TestFactIndex_CollectionWithNoSelectorCoversTheWholeNamespace(t *testing.T) { + harness := newFactIndexHarness(t, FactIndexConfig{}) + harness.publish(factIndexTestStream("prod-eu-1"), aliceCollectionFact("")) + harness.waitForFacts(1) + + // No selector is what --all means. + unlabelled := objectQuery("prod-eu-1", "uid-9", "999", false) + require.Equal(t, AttributionCollectionScope, harness.resolve(unlabelled).Result) +} + +func TestFactIndex_CollectionScopeMatchStopsAtTheWindow(t *testing.T) { + const window = 500 * time.Millisecond + harness := newFactIndexHarness(t, FactIndexConfig{CollectionWindow: window}) + harness.publish(factIndexTestStream("prod-eu-1"), aliceCollectionFact("")) + harness.waitForFacts(1) + require.Equal(t, AttributionCollectionScope, harness.resolve(objectQuery("prod-eu-1", "uid-9", "9", false)).Result) + + // An unrelated delete later in the same namespace must not be claimed by a collection that has + // long since finished. + time.Sleep(3 * window) + harness.absent(objectQuery("prod-eu-1", "uid-9", "9", false)) +} + +func TestFactIndex_CollectionIsWeakerThanAnObjectsOwnFact(t *testing.T) { + harness := newFactIndexHarness(t, FactIndexConfig{}) + key := factIndexTestStream("prod-eu-1") + harness.publish(key, aliceCollectionFact(""), objectFact("bob", "101")) + harness.waitForFacts(3) + + // Precedence is the correctness argument: an unrelated delete by another actor during the same + // window is claimed by its own fact and never reaches the scope tier. + resolution := harness.resolve(objectQuery("prod-eu-1", "uid-1", "999", false)) + require.Equal(t, AttributionWeak, resolution.Result) + require.Equal(t, "bob", resolution.Fact.Author) +} + +func TestFactIndex_UnjoinableFactIsNotStored(t *testing.T) { + harness := newFactIndexHarness(t, FactIndexConfig{}) + key := factIndexTestStream("prod-eu-1") + harness.publish(key, + AuthorFact{GroupResource: "apps/deployments", Author: "nobody", Verb: "update"}, + objectFact("alice", "101"), + ) + require.Equal(t, "alice", harness.resolve(objectQuery("prod-eu-1", "uid-1", "101", true)).Fact.Author) + + // Two entries for the joinable fact, none for the one no query could ever reach. + require.Equal(t, 2, harness.index.Len()) +} + +func TestFactIndex_TrimGapIsCountedAndNamed(t *testing.T) { + reader, err := telemetry.InitTestExporter() + require.NoError(t, err) + index := NewFactIndex(FactIndexConfig{}) + key := factIndexTestStream("prod-eu-1") + + index.reportGaps(t.Context(), []FactStreamGap{{Key: key, Cursor: "1-0", FirstSurviving: "9-0"}}) + + gaps, ok := telemetry.CollectInt64Sum(reader, "gitopsreverser_attribution_fact_stream_gaps_total", + map[string]string{"stream": key.String()}) + require.True(t, ok) + require.Equal(t, int64(1), gaps) +} + +func TestFactStreamSet_UnionIsReferenceCounted(t *testing.T) { + set := NewFactStreamSet() + var observed [][]FactStreamKey + set.Observe(func(keys []FactStreamKey) { observed = append(observed, keys) }) + + deployments := factIndexTestStream("prod-eu-1") + configmaps := FactStreamKeyFor("prod-eu-1", schema.GroupResource{Resource: "configmaps"}) + + // Two watches covering one type, and one covering another. + firstWatch := set.Acquire(deployments) + secondWatch := set.Acquire(deployments) + releaseConfigMaps := set.Acquire(configmaps) + require.Equal(t, []FactStreamKey{deployments, configmaps}, set.Keys()) + + // The type stays followed while ANY watch covers it. + firstWatch() + require.Equal(t, []FactStreamKey{deployments, configmaps}, set.Keys()) + firstWatch() // Releasing twice must not unfollow a type another watch still needs. + require.Equal(t, []FactStreamKey{deployments, configmaps}, set.Keys()) + + secondWatch() + require.Equal(t, []FactStreamKey{configmaps}, set.Keys()) + releaseConfigMaps() + require.Empty(t, set.Keys()) + require.Zero(t, set.Len()) + + // The observer sees the set on install and on every change to its membership, never on a + // reference count that changed under an already-followed type. + require.Equal(t, [][]FactStreamKey{ + {}, + {deployments}, + {deployments, configmaps}, + {configmaps}, + {}, + }, observed) +} + +func TestFactIndex_FollowerPicksUpATypeWhenAWatchStartsCoveringIt(t *testing.T) { + harness := newFactIndexHarness(t, FactIndexConfig{}) + key := factIndexTestStream("prod-eu-1") + + // Written before anything watches the type: the stream exists and nobody follows it. + require.NoError(t, harness.transport.PublishFacts(t.Context(), key, []AuthorFact{objectFact("alice", "101")})) + require.Equal(t, AttributionAbsent, harness.index.Lookup(objectQuery("prod-eu-1", "uid-1", "101", true)).Result) + + // A new watch on that type replays the retention window rather than starting empty. + release := harness.index.Streams().Acquire(key) + defer release() + require.Equal(t, "alice", harness.resolve(objectQuery("prod-eu-1", "uid-1", "101", true)).Fact.Author) +} diff --git a/internal/queue/fact_streams.go b/internal/queue/fact_streams.go new file mode 100644 index 00000000..10dd1b0a --- /dev/null +++ b/internal/queue/fact_streams.go @@ -0,0 +1,114 @@ +// SPDX-License-Identifier: Apache-2.0 + +package queue + +import ( + "slices" + "strings" + "sync" +) + +// FactStreamSet is the reference-counted union of the (audit route, group/resource) pairs the +// watches running in this process cover. It is what makes the per-type fan-out mean anything: the +// process follows a type while at least one watch needs it and stops following it when the last one +// goes away, so facts for a type nobody watches are written and never received. +// +// Reference counting rather than a plain set is the point. Several WatchRules, and several +// GitTargets, routinely cover one type; the type must stay followed while ANY of them does, and a +// set that had forgotten how many watches added it would unfollow on the first one to stop. +// +// It is deliberately independent of the index and of the transport: the watch side acquires and +// releases, and whoever is following the streams observes the union. +type FactStreamSet struct { + mu sync.Mutex + counts map[FactStreamKey]int + observer func([]FactStreamKey) +} + +// NewFactStreamSet builds an empty subscription set. +func NewFactStreamSet() *FactStreamSet { + return &FactStreamSet{counts: map[FactStreamKey]int{}} +} + +// Acquire takes one reference on a stream and returns the release for it. The returned release is +// idempotent, so a caller that releases twice — a watch torn down on both an error path and its +// deferred cleanup — cannot unfollow a type another watch still needs. +func (s *FactStreamSet) Acquire(key FactStreamKey) func() { + s.mu.Lock() + s.counts[key]++ + changed := s.counts[key] == 1 + s.notifyLocked(changed) + s.mu.Unlock() + + var once sync.Once + return func() { + once.Do(func() { s.release(key) }) + } +} + +// Keys returns the followed set in a stable order, which is what a follower is given. +func (s *FactStreamSet) Keys() []FactStreamKey { + s.mu.Lock() + defer s.mu.Unlock() + return s.keysLocked() +} + +// Len reports how many distinct streams are followed. +func (s *FactStreamSet) Len() int { + s.mu.Lock() + defer s.mu.Unlock() + return len(s.counts) +} + +// Observe installs the callback that receives the followed set whenever it changes, and hands it +// the current set immediately so the follower and the set never start out disagreeing. A nil +// callback detaches. Only one observer is supported: one process follows one subscription. +func (s *FactStreamSet) Observe(observe func([]FactStreamKey)) { + s.mu.Lock() + defer s.mu.Unlock() + s.observer = observe + s.notifyLocked(observe != nil) +} + +// release drops one reference, unfollowing the stream when it was the last. +func (s *FactStreamSet) release(key FactStreamKey) { + s.mu.Lock() + defer s.mu.Unlock() + count, ok := s.counts[key] + if !ok { + return + } + count-- + if count > 0 { + s.counts[key] = count + s.notifyLocked(false) + return + } + delete(s.counts, key) + s.notifyLocked(true) +} + +// notifyLocked hands the observer the current set when the set itself changed. It runs under the +// lock so two concurrent changes cannot deliver out of order, which would leave the follower +// following a set that no longer exists. +func (s *FactStreamSet) notifyLocked(changed bool) { + if !changed || s.observer == nil { + return + } + s.observer(s.keysLocked()) +} + +// keysLocked renders the followed set in the order a follower reads it. +func (s *FactStreamSet) keysLocked() []FactStreamKey { + keys := make([]FactStreamKey, 0, len(s.counts)) + for key := range s.counts { + keys = append(keys, key) + } + slices.SortFunc(keys, func(a, b FactStreamKey) int { + if c := strings.Compare(a.AuditRoute, b.AuditRoute); c != 0 { + return c + } + return strings.Compare(a.groupResource(), b.groupResource()) + }) + return keys +} diff --git a/internal/queue/fact_waiters.go b/internal/queue/fact_waiters.go new file mode 100644 index 00000000..e0a079f0 --- /dev/null +++ b/internal/queue/fact_waiters.go @@ -0,0 +1,92 @@ +// SPDX-License-Identifier: Apache-2.0 + +package queue + +import "sync" + +// factWaiterKey is one candidate the join would match on: the scope every key leads with, which +// structure it belongs to, and the value within it. A waiter registers the keys its watch event +// could resolve through, and the goroutine applying a fact wakes the keys that fact filled. +type factWaiterKey struct { + scope factScope + kind factKind + value string +} + +// factWaiter is one blocked resolver. The channel is buffered and signalled without blocking, so +// the applying goroutine is never slowed by a waiter that has not looked yet, and a waiter that was +// signalled while it was re-checking finds the signal still there. +type factWaiter struct { + ch chan struct{} + keys []factWaiterKey +} + +// factWaiterRegistry maps candidate keys to the resolvers blocked on them. +// +// Register-then-check is the whole reason it exists, and the order is not an implementation detail: +// a resolver registers its keys BEFORE reading the index, so a fact applied in the gap between the +// two signals a waiter that is already listening. Checking first and registering after loses +// exactly that fact, which is the race the poll loop used to paper over by looking again. +type factWaiterRegistry struct { + mu sync.Mutex + byKey map[factWaiterKey]map[*factWaiter]struct{} +} + +func newFactWaiterRegistry() *factWaiterRegistry { + return &factWaiterRegistry{byKey: map[factWaiterKey]map[*factWaiter]struct{}{}} +} + +// register adds a waiter for every candidate key and returns it armed. The caller must unregister +// it, whatever it goes on to resolve to. +func (r *factWaiterRegistry) register(keys []factWaiterKey) *factWaiter { + waiter := &factWaiter{ch: make(chan struct{}, 1), keys: keys} + r.mu.Lock() + defer r.mu.Unlock() + for _, key := range keys { + waiters, ok := r.byKey[key] + if !ok { + waiters = map[*factWaiter]struct{}{} + r.byKey[key] = waiters + } + waiters[waiter] = struct{}{} + } + return waiter +} + +// unregister drops a waiter from every key it was registered under. +func (r *factWaiterRegistry) unregister(waiter *factWaiter) { + r.mu.Lock() + defer r.mu.Unlock() + for _, key := range waiter.keys { + waiters, ok := r.byKey[key] + if !ok { + continue + } + delete(waiters, waiter) + if len(waiters) == 0 { + delete(r.byKey, key) + } + } +} + +// wake signals every waiter registered under any of the keys a freshly applied fact filled. +func (r *factWaiterRegistry) wake(keys []factWaiterKey) { + r.mu.Lock() + defer r.mu.Unlock() + for _, key := range keys { + for waiter := range r.byKey[key] { + select { + case waiter.ch <- struct{}{}: + default: + } + } + } +} + +// len reports how many candidate keys currently have a waiter on them. It exists so a test can +// prove a resolver leaves nothing registered behind. +func (r *factWaiterRegistry) len() int { + r.mu.Lock() + defer r.mu.Unlock() + return len(r.byKey) +} diff --git a/internal/telemetry/exporter.go b/internal/telemetry/exporter.go index 2d36a743..6e34950e 100644 --- a/internal/telemetry/exporter.go +++ b/internal/telemetry/exporter.go @@ -108,6 +108,19 @@ var ( AttributionResolutionWaitSeconds metric.Float64Histogram // AttributionFactIndexSize gauges attribution fact keys currently held in Redis. AttributionFactIndexSize metric.Int64Gauge + // AttributionFactIndexEvictionsTotal counts facts dropped from the in-memory fact index because + // it was full, labelled by bounded reason (per_type/total). An attribution lost to a full index + // has to look different from one that was never published, or a burst is silently absorbed. + AttributionFactIndexEvictionsTotal metric.Int64Counter + // AttributionCollectionDegradedTotal counts collection facts published without the uid set the + // precise join would have used, labelled by bounded reason (uid_cap/no_uids). The scope fallback + // is already correct, so this says how often the precise path was available — not that anything + // broke. + AttributionCollectionDegradedTotal metric.Int64Counter + // AttributionFactStreamGapsTotal counts occasions a fact stream was trimmed past this process's + // follower, labelled by stream. Every gap is facts lost for good, and it is the one loss a log + // transport can see at all. + AttributionFactStreamGapsTotal metric.Int64Counter // APICatalogResources gauges the count of served top-level resources in the catalog, // split by the default-watch-policy allowed/excluded state. @@ -224,6 +237,9 @@ func registerCounters() error { {"gitopsreverser_audit_eventlist_events_total", &AuditEventListEventsTotal}, {"gitopsreverser_attribution_resolutions_total", &AttributionResolutionsTotal}, {"gitopsreverser_attribution_fact_events_total", &AttributionFactEventsTotal}, + {"gitopsreverser_attribution_fact_index_evictions_total", &AttributionFactIndexEvictionsTotal}, + {"gitopsreverser_attribution_fact_stream_gaps_total", &AttributionFactStreamGapsTotal}, + {"gitopsreverser_attribution_collection_degraded_total", &AttributionCollectionDegradedTotal}, {"gitopsreverser_api_catalog_refresh_total", &APICatalogRefreshTotal}, {"gitopsreverser_secret_encryption_attempts_total", &SecretEncryptionAttemptsTotal}, {"gitopsreverser_secret_encryption_success_total", &SecretEncryptionSuccessTotal}, diff --git a/internal/webhook/audit_fact_publish_test.go b/internal/webhook/audit_fact_publish_test.go new file mode 100644 index 00000000..661a17d5 --- /dev/null +++ b/internal/webhook/audit_fact_publish_test.go @@ -0,0 +1,187 @@ +// SPDX-License-Identifier: Apache-2.0 + +package webhook + +import ( + "context" + "errors" + "net/http" + "sync" + "testing" + + "github.com/stretchr/testify/require" + "k8s.io/apimachinery/pkg/runtime/schema" + + "github.com/ConfigButler/gitops-reverser/internal/queue" +) + +// factAppend is one PublishFacts call: one stream, one entry, however many facts the request +// produced for it. +type factAppend struct { + key queue.FactStreamKey + facts []queue.AuthorFact +} + +// fakeFactPublisher records every append, so a test can count them. The count is the point: an +// apiserver batch over three types must become three appends, not one per event. +type fakeFactPublisher struct { + mu sync.Mutex + err error + appends []factAppend +} + +func (p *fakeFactPublisher) PublishFacts(_ context.Context, key queue.FactStreamKey, facts []queue.AuthorFact) error { + p.mu.Lock() + defer p.mu.Unlock() + if p.err != nil { + return p.err + } + p.appends = append(p.appends, factAppend{key: key, facts: facts}) + return nil +} + +func (p *fakeFactPublisher) recorded() []factAppend { + p.mu.Lock() + defer p.mu.Unlock() + return append([]factAppend(nil), p.appends...) +} + +func newPublishingHandler(t *testing.T, publisher *fakeFactPublisher) *AuditHandler { + t.Helper() + handler, err := NewAuditHandler(AuditHandlerConfig{FactPublisher: publisher}) + require.NoError(t, err) + return handler +} + +// writeEventVersion is the version every type in these tests is served at. +const writeEventVersion = "v1" + +// writeEvent is one accepted write to a named object of a given type. +func writeEvent(auditID, group, resource, name, user string) string { + version := writeEventVersion + apiVersion := version + if group != "" { + apiVersion = group + "/" + version + } + return `{"kind":"Event","level":"RequestResponse","auditID":"` + auditID + `",` + + `"stage":"ResponseComplete","verb":"update","user":{"username":"` + user + `"},` + + `"requestURI":"/apis/` + group + `/` + version + `/namespaces/team-a/` + resource + `/` + name + `",` + + `"objectRef":{"apiGroup":"` + group + `","resource":"` + resource + `","namespace":"team-a",` + + `"name":"` + name + `","apiVersion":"` + apiVersion + `","uid":"uid-` + name + `"},` + + `"responseStatus":{"code":200},` + + `"responseObject":{"apiVersion":"` + apiVersion + `","metadata":{"name":"` + name + `",` + + `"namespace":"team-a","uid":"uid-` + name + `","resourceVersion":"101"}}}` +} + +func TestAuditHandler_OneRequestOverThreeTypesBecomesThreeAppends(t *testing.T) { + publisher := &fakeFactPublisher{} + handler := newPublishingHandler(t, publisher) + + // Two writes per type, interleaved exactly as the API server batches them. + body := eventListBody( + writeEvent("a", "apps", "deployments", "web", "alice"), + writeEvent("b", "", "configmaps", "config", "bob"), + writeEvent("c", "apps", "deployments", "api", "alice"), + writeEvent("d", "", "secrets", "creds", "carol"), + writeEvent("e", "", "configmaps", "other", "bob"), + ) + require.Equal(t, http.StatusOK, serveBody(t, handler, http.MethodPost, "/audit-webhook/prod-eu-1", body).Code) + + appends := publisher.recorded() + require.Len(t, appends, 3, "five events over three types must append once per type, not once per event") + + require.Equal(t, queue.FactStreamKeyFor("prod-eu-1", + schema.GroupResource{Group: "apps", Resource: "deployments"}), appends[0].key) + require.Equal(t, []string{"web", "api"}, factNames(appends[0].facts), + "a group keeps the order its events arrived in") + require.Equal(t, queue.FactStreamKeyFor("prod-eu-1", schema.GroupResource{Resource: "configmaps"}), appends[1].key) + require.Equal(t, []string{"config", "other"}, factNames(appends[1].facts)) + require.Equal(t, queue.FactStreamKeyFor("prod-eu-1", schema.GroupResource{Resource: "secrets"}), appends[2].key) + require.Equal(t, []string{"creds"}, factNames(appends[2].facts)) + + require.Equal(t, "alice", appends[0].facts[0].Author) + require.Equal(t, "101", appends[0].facts[0].ResourceVersion) + require.Equal(t, "uid-web", appends[0].facts[0].UID) +} + +func TestAuditHandler_EventsThatCanNameNobodyPublishNothing(t *testing.T) { + publisher := &fakeFactPublisher{} + handler := newPublishingHandler(t, publisher) + + // No objectRef: nothing to key a fact on. No user: nobody to name. Both pass the intrinsic + // accept gate, so it is the fact reduction that has to reject them — a waiter woken by a fact + // that can name nobody has been woken for nothing. + noObjectRef := `{"kind":"Event","level":"Metadata","auditID":"no-ref","stage":"ResponseComplete",` + + `"verb":"create","user":{"username":"alice"},"responseStatus":{"code":201}}` + noUser := `{"kind":"Event","level":"RequestResponse","auditID":"no-user","stage":"ResponseComplete",` + + `"verb":"update","user":{},"objectRef":{"resource":"configmaps","namespace":"team-a","name":"cm",` + + `"apiVersion":"v1","uid":"uid-cm"},"responseStatus":{"code":200}}` + + require.Equal(t, http.StatusOK, + serveBody(t, handler, http.MethodPost, "/audit-webhook/prod-eu-1", eventListBody(noObjectRef, noUser)).Code) + require.Empty(t, publisher.recorded()) +} + +func TestAuditHandler_NameLessDeleteCollectionPublishesOneFactWithItsSelector(t *testing.T) { + publisher := &fakeFactPublisher{} + handler := newPublishingHandler(t, publisher) + + // One name-less audit event for N deleted objects. It used to be expanded into one fact per + // object out of the response body; it is now one fact describing the collection, which every + // removal in its scope joins. + deleteCollection := `{"kind":"Event","level":"RequestResponse","auditID":"dc-1",` + + `"stage":"ResponseComplete","verb":"deletecollection","user":{"username":"alice"},` + + `"requestURI":"/api/v1/namespaces/team-a/configmaps?labelSelector=app%3Dweb",` + + `"objectRef":{"resource":"configmaps","namespace":"team-a","apiVersion":"v1"},` + + `"responseStatus":{"code":200},` + + `"responseObject":{"apiVersion":"v1","kind":"ConfigMapList","items":[` + + `{"metadata":{"name":"one","namespace":"team-a","uid":"uid-1"}},` + + `{"metadata":{"name":"two","namespace":"team-a","uid":"uid-2"}}]}}` + + require.Equal(t, http.StatusOK, + serveBody(t, handler, http.MethodPost, "/audit-webhook/prod-eu-1", eventListBody(deleteCollection)).Code) + + appends := publisher.recorded() + require.Len(t, appends, 1) + require.Len(t, appends[0].facts, 1, "a collection delete is ONE fact, not one per deleted object") + + fact := appends[0].facts[0] + require.Equal(t, "alice", fact.Author) + require.Equal(t, "deletecollection", fact.Verb) + require.Equal(t, "team-a", fact.Namespace) + require.Equal(t, "app=web", fact.LabelSelector, "the selector is the intent the actor expressed") + require.Equal(t, []string{"uid-1", "uid-2"}, fact.UIDs, "a body that was there upgrades the join to uid membership") + require.Empty(t, fact.Name, "a collection request names no object") +} + +func TestAuditHandler_PublishFailureIsRetryable(t *testing.T) { + publisher := &fakeFactPublisher{err: errors.New("transport down")} + handler := newPublishingHandler(t, publisher) + + body := eventListBody(writeEvent("a", "apps", "deployments", "web", "alice")) + // A 500 is how the API server is told to deliver the batch again. Appending it twice is safe: a + // fact is keyed data, so the second copy resolves to the same author. + require.Equal(t, http.StatusInternalServerError, + serveBody(t, handler, http.MethodPost, "/audit-webhook/prod-eu-1", body).Code) +} + +func TestAuditHandler_NoPublisherPublishesNothing(t *testing.T) { + recorder := &fakeFactRecorder{} + handler, err := NewAuditHandler(AuditHandlerConfig{FactRecorder: recorder}) + require.NoError(t, err) + + // Configured-author mode, and every install that has not wired the stream: the keys are still + // written and nothing else happens. + body := eventListBody(writeEvent("a", "apps", "deployments", "web", "alice")) + require.Equal(t, http.StatusOK, serveBody(t, handler, http.MethodPost, "/audit-webhook/prod-eu-1", body).Code) + require.Equal(t, 1, recorder.len()) +} + +// factNames flattens a batch to the object names it is about. +func factNames(facts []queue.AuthorFact) []string { + names := make([]string, 0, len(facts)) + for _, fact := range facts { + names = append(names, fact.Name) + } + return names +} diff --git a/internal/webhook/audit_handler.go b/internal/webhook/audit_handler.go index 7087d734..7d4dcaa5 100644 --- a/internal/webhook/audit_handler.go +++ b/internal/webhook/audit_handler.go @@ -27,6 +27,7 @@ import ( "github.com/ConfigButler/gitops-reverser/internal/audit/outcome" "github.com/ConfigButler/gitops-reverser/internal/auditutil" + "github.com/ConfigButler/gitops-reverser/internal/queue" "github.com/ConfigButler/gitops-reverser/internal/telemetry" ) @@ -52,6 +53,17 @@ type AuditFactRecorder interface { RecordFact(ctx context.Context, auditRoute string, event auditv1.Event) error } +// AuditFactPublisher appends one request's attribution facts to the stream for a +// (audit route, group/resource). A nil publisher means the fact stream is not wired. +// +// It is the batch shape that makes this cheap: the API server delivers audit events in batches, so +// one request's 400 events over three types become THREE appends rather than 400 writes. The +// batching that causes the delivery delay attribution has to wait out is the same batching that +// makes publishing nearly free. +type AuditFactPublisher interface { + PublishFacts(ctx context.Context, key queue.FactStreamKey, facts []queue.AuthorFact) error +} + // AuditHandlerConfig contains configuration for the audit handler. type AuditHandlerConfig struct { // MaxRequestBodyBytes is the maximum accepted HTTP request body size. @@ -60,6 +72,10 @@ type AuditHandlerConfig struct { // A write failure returns an audit-request error so the API server retries // delivery; mirrored-resource author attribution depends on these facts. FactRecorder AuditFactRecorder + // FactPublisher appends the facts one request produced, grouped by stream. It is additive to + // FactRecorder for now: the keys are still written and still read, and the stream is filled + // alongside them until the resolver reads from it instead. + FactPublisher AuditFactPublisher // AuditRouteAnnotationKey enables the bare /audit-webhook endpoint for a SHARED audit stream // that carries several logical clusters: the AUDIT ROUTE is read PER EVENT from this // audit-event annotation, so one batch may fan out to several routes. Empty (the default) means @@ -273,45 +289,56 @@ func (h *AuditHandler) decodeEventList(r *http.Request) (*auditv1.EventList, err // processEvents processes a list of audit events for one route. On the annotation-routed bare // endpoint the events in one list may belong to different source clusters, so each is resolved // independently and an unroutable event only drops itself. +// +// Facts are ACCUMULATED across the whole request and appended once per stream at the end, rather +// than written per event. That is what turns one apiserver batch over three types into three +// appends. func (h *AuditHandler) processEvents(ctx context.Context, route auditRoute, events []auditv1.Event) error { + batches := newFactStreamBatches() for i := range events { - if err := h.processEvent(ctx, route, events[i]); err != nil { + accepted, err := h.processEvent(ctx, route, events[i]) + if err != nil { return err } + batches.add(accepted) } - return nil + return h.publishFactBatches(ctx, batches) } -// processEvent applies the intrinsic accept gate, resolves the event's source cluster, and records -// the attribution fact for an accepted, mutating event. A rejected event is recorded with its -// terminal outcome and dropped; only a fact-store or provider-lookup failure returns an error (the -// API server then retries delivery). -func (h *AuditHandler) processEvent(ctx context.Context, route auditRoute, event auditv1.Event) error { +// processEvent applies the intrinsic accept gate, resolves the event's source cluster, records the +// attribution fact for an accepted, mutating event, and returns the fact the request will append. +// A rejected event is recorded with its terminal outcome and dropped; only a fact-store or +// provider-lookup failure returns an error (the API server then retries delivery). +func (h *AuditHandler) processEvent( + ctx context.Context, + route auditRoute, + event auditv1.Event, +) (acceptedFact, error) { log := logf.Log.WithName("audit-handler") h.logAuditEventReceived(event) if !shouldForwardSubresource(&event) { // A non-/scale subresource (or an unmapped-verb subresource): dropped before recording. outcome.Record(ctx, &event, outcome.NonScaleSubresource) - return nil + return acceptedFact{}, nil } if decision := classifyAuditIngress(&event); !decision.Process { outcome.Record(ctx, &event, outcome.Outcome(decision.Reason)) log.V(1).Info("Dropped audit event before recording", "reason", decision.Reason, "gvr", extractGVR(&event), "auditID", event.AuditID) - return nil + return acceptedFact{}, nil } eventRoute, routed := h.resolveEventRoute(ctx, route, &event) if !routed { - return nil + return acceptedFact{}, nil } if h.config.FactRecorder != nil { if err := h.config.FactRecorder.RecordFact(ctx, eventRoute, event); err != nil { outcome.Record(ctx, &event, outcome.WriteError) - return fmt.Errorf("record attribution fact %q: %w", event.AuditID, err) + return acceptedFact{}, fmt.Errorf("record attribution fact %q: %w", event.AuditID, err) } } outcome.Record(ctx, &event, outcome.Queued) @@ -322,9 +349,71 @@ func (h *AuditHandler) processEvent(ctx context.Context, route auditRoute, event log.V(1).Info("Recorded audit attribution fact", "gvr", extractGVR(&event), "verb", event.Verb, "auditID", event.AuditID, "user", effectiveAuditUsername(event)) + return h.factForStream(ctx, eventRoute, event), nil +} + +// factForStream reduces one accepted event to the fact its stream carries. It produces nothing when +// no publisher is wired, and nothing for an event that could never name an author — the same events +// the per-key write path skips, because a waiter woken by a fact that names nobody has been woken +// for nothing. +func (h *AuditHandler) factForStream(ctx context.Context, auditRoute string, event auditv1.Event) acceptedFact { + if h.config.FactPublisher == nil { + return acceptedFact{} + } + fact, groupResource, ok := queue.AuthorFactFromEvent(ctx, event) + if !ok { + return acceptedFact{} + } + return acceptedFact{key: queue.FactStreamKeyFor(auditRoute, groupResource), fact: fact, ok: true} +} + +// publishFactBatches appends each group once. A failure returns an audit-request error so the API +// server retries the delivery; a retried batch appends the same facts again under fresh stream IDs, +// which is safe because a fact is keyed data rather than a position in a sequence — the duplicate +// resolves to the same author and costs one entry's worth of retention. +func (h *AuditHandler) publishFactBatches(ctx context.Context, batches *factStreamBatches) error { + if h.config.FactPublisher == nil { + return nil + } + for _, key := range batches.order { + if err := h.config.FactPublisher.PublishFacts(ctx, key, batches.facts[key]); err != nil { + return fmt.Errorf("publish attribution facts for %s: %w", key, err) + } + } return nil } +// acceptedFact is one event's contribution to the request's appends, or the zero value when the +// event produced none. +type acceptedFact struct { + key queue.FactStreamKey + fact queue.AuthorFact + ok bool +} + +// factStreamBatches groups one request's facts by the stream they belong to, keeping the order the +// events arrived in — both across streams and within one, since the index applies a batch in the +// order it was published and its latest tier is last-writer-wins. +type factStreamBatches struct { + order []queue.FactStreamKey + facts map[queue.FactStreamKey][]queue.AuthorFact +} + +func newFactStreamBatches() *factStreamBatches { + return &factStreamBatches{facts: map[queue.FactStreamKey][]queue.AuthorFact{}} +} + +// add files one accepted event's fact under its stream. +func (b *factStreamBatches) add(accepted acceptedFact) { + if !accepted.ok { + return + } + if _, ok := b.facts[accepted.key]; !ok { + b.order = append(b.order, accepted.key) + } + b.facts[accepted.key] = append(b.facts[accepted.key], accepted.fact) +} + // resolveEventRoute returns the AUDIT ROUTE one accepted event's fact is filed under, and whether it // routed at all. On a named route that is the route's own value, unconditionally. On the shared, // annotation-routed bare endpoint it is read from the event's own annotations, and an event carrying