From 0aa2beca57db34dd4f533c064111a7cffd63eee1 Mon Sep 17 00:00:00 2001 From: Simon Koudijs Date: Tue, 28 Jul 2026 12:26:32 +0000 Subject: [PATCH 1/3] feat(queue): add the transport-agnostic attribution fact index MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The layer that sits above the fact transport seam and never learns which transport it has: the four match structures the join reads, the reference-counted subscription set, the waiter registry a blocked resolver parks on, and the follower loop that fills all three. The route leads every key. 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. Entries apply in delivery order, because the latest tier is last-writer-wins. Each carries its insertion time, so expiry is decided on read and the sweep is memory reclaim rather than correctness. Each type is capped, with oldest-first eviction that is COUNTED: an attribution dropped because the index was full has to look different in the metrics from one that was never published. The waiter registry registers THEN checks, never the reverse. A fact applied in the gap between the two must signal a waiter that is already listening; registering after the check loses exactly that fact, which is the race the poll loop papers over by looking again. AuthorFact gains the collection fields the index reads — the selector the request URI expressed, and the optional uid set — so this layer is complete on its own. Nothing above it changes: the resolver still polls, the audit handler is untouched, and the v1 fact keys are still the path in use. Numbers picked here: 4096 facts per type as the primary cap, 65536 in total, a 30s collection window, and a 30s sweep. All configurable on FactIndexConfig. --- .coverage-baseline | 2 +- internal/queue/attribution_index.go | 21 ++ internal/queue/fact_index.go | 481 +++++++++++++++++++++++++++ internal/queue/fact_index_store.go | 387 ++++++++++++++++++++++ internal/queue/fact_index_test.go | 483 ++++++++++++++++++++++++++++ internal/queue/fact_streams.go | 114 +++++++ internal/queue/fact_waiters.go | 92 ++++++ internal/telemetry/exporter.go | 10 + 8 files changed, 1589 insertions(+), 1 deletion(-) create mode 100644 internal/queue/fact_index.go create mode 100644 internal/queue/fact_index_store.go create mode 100644 internal/queue/fact_index_test.go create mode 100644 internal/queue/fact_streams.go create mode 100644 internal/queue/fact_waiters.go diff --git a/.coverage-baseline b/.coverage-baseline index e59d5c490..6c2b500ec 100644 --- a/.coverage-baseline +++ b/.coverage-baseline @@ -1 +1 @@ -78.2 +78.3 diff --git a/internal/queue/attribution_index.go b/internal/queue/attribution_index.go index 8f9d4108d..c5d7fe855 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/fact_index.go b/internal/queue/fact_index.go new file mode 100644 index 000000000..90fdc6230 --- /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 000000000..e8047dbe6 --- /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 000000000..5abcd4de3 --- /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 000000000..10dd1b0a6 --- /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 000000000..e0a079f0c --- /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 2d36a7434..8be6aebaa 100644 --- a/internal/telemetry/exporter.go +++ b/internal/telemetry/exporter.go @@ -108,6 +108,14 @@ 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 + // 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 +232,8 @@ 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_api_catalog_refresh_total", &APICatalogRefreshTotal}, {"gitopsreverser_secret_encryption_attempts_total", &SecretEncryptionAttemptsTotal}, {"gitopsreverser_secret_encryption_success_total", &SecretEncryptionSuccessTotal}, From 4079d5cae6dfda2386c54f17ae9c708f619b9152 Mon Sep 17 00:00:00 2001 From: Simon Koudijs Date: Tue, 28 Jul 2026 12:31:51 +0000 Subject: [PATCH 2/3] feat(webhook): publish attribution facts per type per request MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The audit receiver stops writing per event and accumulates instead: each accepted event is reduced to an AuthorFact, the facts are grouped by (audit route, group/resource) across the whole request, and each group is appended once. An apiserver batch of 400 events over three types becomes three appends. The batching that causes the delivery delay attribution waits out is the same batching that makes publishing nearly free. A deletecollection is published as ONE fact describing the collection rather than expanded into one fact per object, so the "no resolvable name" rule becomes "no name and not a collection verb". The fact carries what the audit event actually said — actor, verb, type, namespace, stage timestamp, the selector from the request URI — plus the uids the response body listed, reduced at the receiver and dropped past a cap with the degradation counted. The selector is the better evidence: it is the intent the actor expressed, and it is there even when a truncating production cluster sends no body at all. Only facts that would have been stored are published. An event with no objectRef or no user produces nothing, or waiters wake for facts that can name nobody. The v1 keys are still written and still read. The resolver is not switched over here, so behaviour is unchanged and e2e attribution keeps passing on the path it always used. The uid-set cap is 10000 — a few hundred kilobytes of uids against a response body for the same request that runs to tens of megabytes. --- .coverage-baseline | 2 +- internal/queue/author_fact.go | 153 ++++++++++++++++ internal/queue/author_fact_test.go | 120 +++++++++++++ internal/telemetry/exporter.go | 6 + internal/webhook/audit_fact_publish_test.go | 187 ++++++++++++++++++++ internal/webhook/audit_handler.go | 111 ++++++++++-- 6 files changed, 567 insertions(+), 12 deletions(-) create mode 100644 internal/queue/author_fact.go create mode 100644 internal/queue/author_fact_test.go create mode 100644 internal/webhook/audit_fact_publish_test.go diff --git a/.coverage-baseline b/.coverage-baseline index 6c2b500ec..2ce0a0b65 100644 --- a/.coverage-baseline +++ b/.coverage-baseline @@ -1 +1 @@ -78.3 +78.4 diff --git a/internal/queue/author_fact.go b/internal/queue/author_fact.go new file mode 100644 index 000000000..73c74d8c3 --- /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 000000000..18663d94a --- /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/telemetry/exporter.go b/internal/telemetry/exporter.go index 8be6aebaa..6e34950ec 100644 --- a/internal/telemetry/exporter.go +++ b/internal/telemetry/exporter.go @@ -112,6 +112,11 @@ var ( // 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. @@ -234,6 +239,7 @@ func registerCounters() error { {"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 000000000..661a17d5f --- /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, "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/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, "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/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, "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/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, "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/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, "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/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 7087d7341..7d4dcaa58 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 From 091052e71892bfaa4875c75f49a501a9b00a3ec5 Mon Sep 17 00:00:00 2001 From: Simon Koudijs Date: Tue, 28 Jul 2026 13:44:47 +0000 Subject: [PATCH 3/3] test(gittarget): give the age-key recreation wait a spare requeue tick MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The recreate-on-deletion spec waits for a recovery only the periodic requeue can deliver — this controller runs no control-plane Secret watch — and its budget was exactly two RequeueStreamSettleInterval ticks with no slack. A CI runner busy enough to delay one tick by a second failed the spec on timing alone, twice in a row, on a branch whose diff cannot reach the age-key path. Eventually returns as soon as the Secret is back, so the third tick costs nothing on any run that was going to pass. --- internal/controller/gittarget_controller_test.go | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/internal/controller/gittarget_controller_test.go b/internal/controller/gittarget_controller_test.go index 0bb5c191d..2aa4f6b32 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())