Route activity policy failures to their owners

config/components/observability/alerts/dlq-alerts.yaml

@@ -9,23 +9,49 @@ metadata:
     monitoring: "true"
 spec:
   groups:
+    # Select one current owner per policy across processor replicas. Newest
+    # scrape wins during metadata propagation; normalize the join value to 1.
+    - name: activity-policy-ownership
+      interval: 30s
+      rules:
+        - record: activity_processor:policy_owner:info
+          expr: |
+            max by (cluster, namespace, policy_name, policy_service, policy_team) (
+              topk by (cluster, namespace, policy_name) (
+                1, timestamp(activity_processor_policy_info)
+              )
+            ) ^ 0
     # =========================================================================
-    # DLQ Health Alerts - SRE Ownership
+    # DLQ Health Alerts
     # =========================================================================
     - name: dlq-health
       interval: 30s
       rules:
         # DLQ Growing - Queue Not Draining
         - alert: ActivityPolicyFailuresHigh
           expr: |
-            rate(activity_processor_dlq_events_published_total[5m]) > 0.1
-            AND
-            increase(activity_processor_dlq_events_published_total[15m]) > 100
+            label_replace(
+              (
+                rate(activity_processor_dlq_events_published_total[5m]) > 0.1
+                AND
+                increase(activity_processor_dlq_events_published_total[15m]) > 100
+              )
+              * on (cluster, namespace, policy_name) group_left (policy_service, policy_team)
+                activity_processor:policy_owner:info
+              or
+              (
+                rate(activity_processor_dlq_events_published_total[5m]) > 0.1
+                AND
+                increase(activity_processor_dlq_events_published_total[15m]) > 100
+              )
+              unless on (cluster, namespace, policy_name) activity_processor:policy_owner:info,
+              "service", "$1", "policy_service", "(.*)"
+            )
           for: 15m
           labels:
             severity: warning
             component: activity-processor
-            team: sre
+            team: '{{ if $labels.policy_team }}{{ $labels.policy_team }}{{ else }}sre{{ end }}'
           annotations:
             summary: "Activity policy{{ if $labels.policy_name }} {{ $labels.policy_name }}{{ end }} is failing to generate activities"
             description: "Activity processing{{ if $labels.cluster }} on {{ $labels.cluster }}{{ end }} is sending {{ $value }} {{ $labels.event_type }} events/sec for {{ $labels.api_group }}/{{ $labels.kind }} to the dead-letter queue after {{ $labels.error_type }} failures. This rate has persisted for 15 minutes. Check the activity policy and processor logs for the failing rule and event payload."
@@ -49,12 +75,24 @@ spec:
         # High Retry Count Events
         - alert: ActivityPolicyHighRetryCount
           expr: |
-            increase(activity_processor_dlq_retry_events_high_retry_total[1h]) > 10
+            label_replace(
+              (
+                increase(activity_processor_dlq_retry_events_high_retry_total[1h]) > 10
+              )
+              * on (cluster, namespace, policy_name) group_left (policy_service, policy_team)
+                activity_processor:policy_owner:info
+              or
+              (
+                increase(activity_processor_dlq_retry_events_high_retry_total[1h]) > 10
+              )
+              unless on (cluster, namespace, policy_name) activity_processor:policy_owner:info,
+              "service", "$1", "policy_service", "(.*)"
+            )
           for: 1h
           labels:
             severity: warning
             component: activity-processor
-            team: sre
+            team: '{{ if $labels.policy_team }}{{ $labels.policy_team }}{{ else }}sre{{ end }}'
           annotations:
             summary: "Activity policy{{ if $labels.policy_name }} {{ $labels.policy_name }}{{ end }} events keep failing retries"
             description: "Activity policy retries{{ if $labels.cluster }} on {{ $labels.cluster }}{{ end }} for {{ $labels.api_group }}/{{ $labels.kind }} recorded {{ $value }} failures at or above the high retry threshold in the last hour. The count can include repeated attempts for the same event. Check the policy and processor logs for persistent evaluation or delivery failures."
@@ -64,12 +102,24 @@ spec:
         # DLQ slow-leak - low-rate but persistent failures
         - alert: ActivityPolicyFailuresRecurring
           expr: |
-            increase(activity_processor_dlq_events_published_total[6h]) > 10
+            label_replace(
+              (
+                increase(activity_processor_dlq_events_published_total[6h]) > 10
+              )
+              * on (cluster, namespace, policy_name) group_left (policy_service, policy_team)
+                activity_processor:policy_owner:info
+              or
+              (
+                increase(activity_processor_dlq_events_published_total[6h]) > 10
+              )
+              unless on (cluster, namespace, policy_name) activity_processor:policy_owner:info,
+              "service", "$1", "policy_service", "(.*)"
+            )
           for: 1h
           labels:
             severity: warning
             component: activity-processor
-            team: sre
+            team: '{{ if $labels.policy_team }}{{ $labels.policy_team }}{{ else }}sre{{ end }}'
           annotations:
             summary: "Activity policy{{ if $labels.policy_name }} {{ $labels.policy_name }}{{ end }} has recurring activity generation failures"
             description: "Activity processing{{ if $labels.cluster }} on {{ $labels.cluster }}{{ end }} sent {{ $value }} {{ $labels.event_type }} events for {{ $labels.api_group }}/{{ $labels.kind }} to the dead-letter queue in the last 6 hours with {{ $labels.error_type }} errors. Inspect the activity policy and failed payloads for event shapes its rules do not handle."
@@ -86,14 +136,28 @@ spec:
         # policy_name label, so it cannot provide a per-policy success ratio.
         - alert: ActivityPolicyRetriesFailing
           expr: |
-            sum by (cluster, namespace, policy_name, api_group, kind, error_type) (
-              increase(activity_processor_dlq_retry_failed_total{policy_name!=""}[1h])
-            ) >= 3
+            label_replace(
+              (
+                sum by (cluster, namespace, policy_name, api_group, kind, error_type) (
+                  increase(activity_processor_dlq_retry_failed_total{policy_name!=""}[1h])
+                ) >= 3
+              )
+              * on (cluster, namespace, policy_name) group_left (policy_service, policy_team)
+                activity_processor:policy_owner:info
+              or
+              (
+                sum by (cluster, namespace, policy_name, api_group, kind, error_type) (
+                  increase(activity_processor_dlq_retry_failed_total{policy_name!=""}[1h])
+                ) >= 3
+              )
+              unless on (cluster, namespace, policy_name) activity_processor:policy_owner:info,
+              "service", "$1", "policy_service", "(.*)"
+            )
           for: 30m
           labels:
             severity: warning
             component: activity-processor
-            team: sre
+            team: '{{ if $labels.policy_team }}{{ $labels.policy_team }}{{ else }}sre{{ end }}'
             subject: "{{ $labels.policy_name }}"
           annotations:
             summary: "Activity policy {{ $labels.policy_name }} keeps failing during retries"
@@ -110,14 +174,28 @@ spec:
         # Policy Sending Events to DLQ
         - alert: ActivityPolicyEventProcessingErrors
           expr: |
-            sum by (policy_name, api_group, kind, error_type) (
-              rate(activity_processor_dlq_events_published_total{policy_name!=""}[10m])
-            ) > 0.1
+            label_replace(
+              (
+                sum by (cluster, namespace, policy_name, api_group, kind, error_type) (
+                  rate(activity_processor_dlq_events_published_total{policy_name!=""}[10m])
+                ) > 0.1
+              )
+              * on (cluster, namespace, policy_name) group_left (policy_service, policy_team)
+                activity_processor:policy_owner:info
+              or
+              (
+                sum by (cluster, namespace, policy_name, api_group, kind, error_type) (
+                  rate(activity_processor_dlq_events_published_total{policy_name!=""}[10m])
+                ) > 0.1
+              )
+              unless on (cluster, namespace, policy_name) activity_processor:policy_owner:info,
+              "service", "$1", "policy_service", "(.*)"
+            )
           for: 15m
           labels:
             severity: warning
             component: activity-processor
-            team: sre
+            team: '{{ if $labels.policy_team }}{{ $labels.policy_team }}{{ else }}sre{{ end }}'
           annotations:
             summary: "Activity policy {{ $labels.policy_name }} is failing to generate activities"
             description: "Activity policy {{ $labels.policy_name }} for {{ $labels.api_group }}/{{ $labels.kind }} is sending {{ $value }} events/sec to the dead-letter queue with {{ $labels.error_type }} errors. Check the policy's rules against failed event payloads in the activity-processor logs."

docs/guides/activity-policies.md

@@ -42,6 +42,9 @@ apiVersion: activity.miloapis.com/v1alpha1
 kind: ActivityPolicy
 metadata:
   name: networking-httpproxy        # Unique name, applies to the whole control plane
+  labels:
+    meta.datumapis.com/service: network-services
+    meta.datumapis.com/team: delivery
 spec:
   resource:
     apiGroup: networking.datumapis.com
@@ -84,6 +87,26 @@ rules correctly when you update the policy with `kubectl apply` — two rules
 with the same name will silently collide. Choose names that describe the rule's
 intent: `create`, `delete`, `status`, `fallback`.
 
+## Policy ownership and alerts
+
+Set `meta.datumapis.com/service` and `meta.datumapis.com/team` to the service
+and team responsible for maintaining the policy. For example, NSO policies use
+`network-services` and `delivery`. These are policy owners, independent of the
+service running the activity processor.
+
+The processor exports these labels as `policy_service` and `policy_team` on
+`activity_processor_policy_info` for active policies. The policy failure alerts
+join this metadata by cluster, processor namespace, and policy name, then set
+alert `service` and `team` from the policy. The separate metric labels avoid
+collisions with the Kubernetes service label added during scraping.
+
+Ownership updates apply to existing retry failures after the policy cache and
+metrics refresh. Deleted or inactive policies stop exporting ownership. Failures
+without ownership telemetry still alert: their service is unset and their team
+falls back to `sre`. Queue publishing failures have no policy identity and remain
+processor infrastructure alerts. Configure Alertmanager routes for your service
+and team values; resource labels alone do not create routes.
+
 ## CEL match expressions
 
 The `match` field must be a CEL expression that returns `true` or `false`.

internal/activityprocessor/policy_ownership.go

@@ -0,0 +1,34 @@
+package activityprocessor
+
+import "github.com/prometheus/client_golang/prometheus"
+
+// Ownership labels are explicit policy metadata, independent of the service
+// that hosts the processor. Only these two labels are exposed as metrics.
+const (
+	policyServiceLabel = "meta.datumapis.com/service"
+	policyTeamLabel    = "meta.datumapis.com/team"
+)
+
+var policyInfoDesc = prometheus.NewDesc(
+	"activity_processor_policy_info",
+	"Ownership of an active ActivityPolicy (value 1).",
+	[]string{"policy_name", "policy_service", "policy_team"}, nil,
+)
+
+func (c *PolicyCache) Describe(ch chan<- *prometheus.Desc) {
+	ch <- policyInfoDesc
+}
+
+// Collect reads current metadata rather than copying owners into queued events.
+// Ownership updates and deletions therefore apply to existing retry failures.
+func (c *PolicyCache) Collect(ch chan<- prometheus.Metric) {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+	for _, policies := range c.policies {
+		for _, policy := range policies {
+			labels := policy.OriginalPolicy.Labels
+			ch <- prometheus.MustNewConstMetric(policyInfoDesc, prometheus.GaugeValue, 1,
+				policy.Name, labels[policyServiceLabel], labels[policyTeamLabel])
+		}
+	}
+}

internal/activityprocessor/policy_ownership_test.go

@@ -0,0 +1,51 @@
+package activityprocessor
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/testutil"
+	"go.miloapis.com/activity/pkg/apis/activity/v1alpha1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func TestPolicyOwnershipLifecycle(t *testing.T) {
+	cache := NewPolicyCache()
+	registry := prometheus.NewPedanticRegistry()
+	registry.MustRegister(cache)
+	policy := &v1alpha1.ActivityPolicy{ObjectMeta: metav1.ObjectMeta{Name: "httpproxy", Labels: map[string]string{
+		policyServiceLabel: "network-services", policyTeamLabel: "delivery", "unrelated": "not-exported",
+	}}}
+	check := func(service, team string) {
+		t.Helper()
+		expected := `# HELP activity_processor_policy_info Ownership of an active ActivityPolicy (value 1).
+# TYPE activity_processor_policy_info gauge
+activity_processor_policy_info{policy_name="httpproxy",policy_service="` + service + `",policy_team="` + team + `"} 1
+`
+		if err := testutil.GatherAndCompare(registry, strings.NewReader(expected), "activity_processor_policy_info"); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if err := cache.Add(policy, "httpproxies"); err != nil {
+		t.Fatal(err)
+	}
+	check("network-services", "delivery")
+	updated := policy.DeepCopy()
+	updated.Labels[policyServiceLabel] = "dns"
+	updated.Labels[policyTeamLabel] = "platform"
+	if err := cache.Update(policy, updated, "httpproxies", "httpproxies"); err != nil {
+		t.Fatal(err)
+	}
+	check("dns", "platform")
+	unlabeled := updated.DeepCopy()
+	unlabeled.Labels = nil
+	if err := cache.Update(updated, unlabeled, "httpproxies", "httpproxies"); err != nil {
+		t.Fatal(err)
+	}
+	check("", "")
+	cache.Remove(unlabeled, "httpproxies")
+	if err := testutil.GatherAndCompare(registry, strings.NewReader(""), "activity_processor_policy_info"); err != nil {
+		t.Fatal(err)
+	}
+}

internal/activityprocessor/processor.go

@@ -321,6 +321,10 @@ func New(config Config, restConfig *rest.Config) (*Processor, error) {
 
 // Start begins processing audit events.
 func (p *Processor) Start(ctx context.Context) error {
+	if err := metrics.Registry.Register(p.policyCache); err != nil {
+		return fmt.Errorf("failed to register policy ownership metrics: %w", err)
+	}
+
 	// Start health probe server early so Kubernetes can check liveness
 	if p.config.HealthProbeAddr != "" {
 		p.startHealthServer()
@@ -591,6 +595,7 @@ const drainTimeout = 30 * time.Second
 
 // Stop gracefully shuts down the processor.
 func (p *Processor) Stop() {
+	metrics.Registry.Unregister(p.policyCache)
 	klog.Info("Stopping activity processor")
 
 	// Mark as unhealthy immediately