feat(attribution): index facts in memory and publish them per type

.coverage-baseline

@@ -1 +1 @@
-78.2
+78.4

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())

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.

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)))
+}

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)
+}