Add service-route observability

cmd/galactic-router/root.go

@@ -23,6 +23,7 @@ import (
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/log/zap"
+	ctrlmetrics "sigs.k8s.io/controller-runtime/pkg/metrics"
 	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
 	"sigs.k8s.io/controller-runtime/pkg/webhook"
 
@@ -106,6 +107,8 @@ func runCmd(cfg *config.RouterConfig) error {
 	if err != nil {
 		return fmt.Errorf("create manager: %w", err)
 	}
+	serviceRouteMetrics := serviceroute.NewMetrics()
+	serviceRouteMetrics.MustRegister(ctrlmetrics.Registry)
 
 	// ctx carries a cause so an ordinary signal-triggered shutdown can be told
 	// apart from the health server's own Serve failure below. See the cause
@@ -278,7 +281,7 @@ func runCmd(cfg *config.RouterConfig) error {
 	// controller process before informer reconciliation starts. A missing map is
 	// tolerated because the datapath loader may start after galactic-router;
 	// the first Apply retries initialization.
-	if err := setupServiceRouteController(mgr, nodeName, cfg.GCNamespace); err != nil {
+	if err := setupServiceRouteController(mgr, nodeName, cfg.GCNamespace, serviceRouteMetrics); err != nil {
 		return fmt.Errorf("setup ServiceRoutePolicy controller: %w", err)
 	}
 
@@ -346,7 +349,11 @@ func runCmd(cfg *config.RouterConfig) error {
 	return nil
 }
 
-func setupServiceRouteController(mgr ctrl.Manager, nodeName, bgpNamespace string) error {
+func setupServiceRouteController(
+	mgr ctrl.Manager,
+	nodeName, bgpNamespace string,
+	serviceRouteMetrics *serviceroute.Metrics,
+) error {
 	programmer := &serviceroute.EBPFRouteProgrammer{}
 	if err := programmer.Initialize(); err != nil {
 		ctrl.Log.Error(err, "service route eBPF maps are not available yet; initialization will retry on reconcile")
@@ -358,6 +365,7 @@ func setupServiceRouteController(mgr ctrl.Manager, nodeName, bgpNamespace string
 		NodeName:     nodeName,
 		BGPNamespace: bgpNamespace,
 		Programmer:   programmer,
+		Metrics:      serviceRouteMetrics,
 		Applied:      make(map[types.NamespacedName]map[types.NamespacedName]serviceroute.RouteIntent),
 	}).SetupWithManager(mgr)
 }

config/monitoring/kustomization.yaml

@@ -14,6 +14,7 @@
 namespace: galactic-system
 resources:
   - podmonitor-fabric-router.yaml
+  - podmonitor-galactic-cni.yaml
   - podmonitor-galactic-nat.yaml
   - podmonitor-galactic-router.yaml
   - prometheusrule.yaml

config/monitoring/podmonitor-galactic-cni.yaml

@@ -0,0 +1,19 @@
+# Scrapes galactic-cni's per-node metrics port (metrics, 9180), including the
+# TC-eBPF service policy and drop counters.
+apiVersion: monitoring.coreos.com/v1
+kind: PodMonitor
+metadata:
+  name: galactic-cni
+  labels:
+    app.kubernetes.io/name: galactic-cni
+spec:
+  jobLabel: app.kubernetes.io/name
+  selector:
+    matchLabels:
+      app.kubernetes.io/name: galactic-cni
+  podMetricsEndpoints:
+    - port: metrics
+      interval: 30s
+      relabelings:
+        - sourceLabels: [__meta_kubernetes_pod_node_name]
+          targetLabel: node

internal/controller/serviceroute_controller.go

@@ -33,6 +33,8 @@ import (
 	networkv1alpha1 "go.datum.net/network/api/v1alpha1"
 )
 
+const reconcileResultError = "error"
+
 // ServiceRoutePolicyReconciler resolves service policies against local Cloud
 // API attachments and programs the resulting stateful eBPF forwarding policy
 // directly on this node. It intentionally does not create child Kubernetes
@@ -43,6 +45,7 @@ type ServiceRoutePolicyReconciler struct {
 	NodeName     string
 	BGPNamespace string
 	Programmer   serviceroute.RouteProgrammer
+	Metrics      *serviceroute.Metrics
 
 	mu      sync.Mutex
 	Applied map[types.NamespacedName]map[types.NamespacedName]serviceroute.RouteIntent
@@ -83,29 +86,50 @@ func startupSyncNeedsRetry(err error) bool {
 // Reconcile resolves one policy and replaces the local routes previously
 // programmed for that policy.
 func (r *ServiceRoutePolicyReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
+	start := time.Now()
+	resultLabel := "success"
+	defer func() { r.Metrics.ObserveReconcile(resultLabel, time.Since(start)) }()
+	logger := log.FromContext(ctx).WithValues("policy", req.String(), "node", r.NodeName)
+
 	r.mu.Lock()
 	defer r.mu.Unlock()
 
 	policy := &networkv1alpha1.ServiceRoutePolicy{}
 	if err := r.Get(ctx, req.NamespacedName, policy); err != nil {
 		if apierrors.IsNotFound(err) {
-			return ctrl.Result{}, r.removePolicyLocked(req.NamespacedName)
+			err := r.removePolicyLocked(req.NamespacedName)
+			if err != nil {
+				resultLabel = reconcileResultError
+				logger.Error(err, "remove eBPF policy for deleted ServiceRoutePolicy")
+			}
+			return ctrl.Result{}, err
 		}
+		resultLabel = reconcileResultError
+		logger.Error(err, "get ServiceRoutePolicy")
 		return ctrl.Result{}, err
 	}
 	endpoint := &networkv1alpha1.ServiceEndpoint{}
 	endpointKey := types.NamespacedName{Namespace: policy.Namespace, Name: policy.Spec.ServiceRef.Name}
 	if err := r.Get(ctx, endpointKey, endpoint); err != nil {
 		if apierrors.IsNotFound(err) {
-			return ctrl.Result{}, errors.Join(
+			joined := errors.Join(
 				r.replacePolicyLocked(req.NamespacedName, nil),
 				r.setAccepted(ctx, policy, metav1.ConditionFalse, "EndpointNotFound", "referenced ServiceEndpoint does not exist"),
 			)
+			if joined != nil {
+				resultLabel = reconcileResultError
+				logger.Error(joined, "revoke eBPF policy for missing ServiceEndpoint")
+			}
+			return ctrl.Result{}, joined
 		}
+		resultLabel = reconcileResultError
+		logger.Error(err, "get ServiceEndpoint", "endpoint", policy.Spec.ServiceRef.Name)
 		return ctrl.Result{}, fmt.Errorf("get ServiceEndpoint %s/%s: %w", policy.Namespace, policy.Spec.ServiceRef.Name, err)
 	}
 	attachments := &cloudv1alpha1.VPCAttachmentList{}
 	if err := r.List(ctx, attachments); err != nil {
+		resultLabel = reconcileResultError
+		logger.Error(err, "list VPCAttachments")
 		return ctrl.Result{}, fmt.Errorf("list VPCAttachments: %w", err)
 	}
 	all := make([]*cloudv1alpha1.VPCAttachment, 0, len(attachments.Items))
@@ -121,35 +145,51 @@ func (r *ServiceRoutePolicyReconciler) Reconcile(ctx context.Context, req ctrl.R
 			// uninvolved nodes that correctly compile no local intents. Revoke
 			// this node's stale state and return the error for controller-runtime
 			// to retry while leaving the policy accepted cluster-wide.
-			return ctrl.Result{}, errors.Join(
+			joined := errors.Join(
 				err,
 				r.removePolicyLocked(req.NamespacedName),
 				r.setAccepted(ctx, policy, metav1.ConditionTrue, "Accepted", "policy is valid"),
 			)
+			resultLabel = reconcileResultError
+			logger.Error(joined, "resolve service routing dependency", "endpoint", endpoint.Name)
+			return ctrl.Result{}, joined
 		}
-		return ctrl.Result{}, errors.Join(
+		joined := errors.Join(
 			err,
 			r.removePolicyLocked(req.NamespacedName),
 			r.setAccepted(ctx, policy, metav1.ConditionFalse, "Invalid", err.Error()),
 		)
+		resultLabel = reconcileResultError
+		logger.Error(joined, "compile service eBPF policy", "endpoint", endpoint.Name)
+		return ctrl.Result{}, joined
 	}
 	if len(intents) != 0 {
 		// The datapath loader may replace an incompatible pinned map while this
 		// controller process stays alive. Initialize detects the map identity
 		// change and reconstructs all desired state before an unchanged intent
 		// is otherwise skipped by replacePolicy.
 		if err := r.Programmer.Initialize(); err != nil {
+			resultLabel = reconcileResultError
+			logger.Error(err, "refresh service route maps", "endpoint", endpoint.Name)
 			return ctrl.Result{}, fmt.Errorf("refresh service route maps: %w", err)
 		}
 	}
+	logger.Info("compiled service eBPF policy", "endpoint", endpoint.Name, "attachments", len(intents))
 	if err := r.replacePolicyLocked(req.NamespacedName, intents); err != nil {
+		resultLabel = reconcileResultError
+		logger.Error(err, "program service eBPF policy", "endpoint", endpoint.Name)
 		return ctrl.Result{}, err
 	}
-	result := ctrl.Result{}
+	if err := r.setAccepted(ctx, policy, metav1.ConditionTrue, "Accepted", "policy is valid"); err != nil {
+		resultLabel = reconcileResultError
+		logger.Error(err, "update ServiceRoutePolicy acceptance status")
+		return ctrl.Result{}, err
+	}
+	requeue := ctrl.Result{}
 	if len(intents) != 0 {
-		result.RequeueAfter = serviceRouteMapResyncInterval
+		requeue.RequeueAfter = serviceRouteMapResyncInterval
 	}
-	return result, r.setAccepted(ctx, policy, metav1.ConditionTrue, "Accepted", "policy is valid")
+	return requeue, nil
 }
 
 func (r *ServiceRoutePolicyReconciler) setAccepted(
@@ -427,6 +467,7 @@ func (r *ServiceRoutePolicyReconciler) replacePolicyLocked(
 	key types.NamespacedName,
 	intents []serviceroute.RouteIntent,
 ) error {
+	defer r.updateProgrammedGauge()
 	if r.Applied == nil {
 		r.Applied = make(map[types.NamespacedName]map[types.NamespacedName]serviceroute.RouteIntent)
 	}
@@ -449,20 +490,24 @@ func (r *ServiceRoutePolicyReconciler) replacePolicyLocked(
 			continue
 		}
 		if err := r.Programmer.Remove(existing); err != nil {
+			r.Metrics.ObserveOperation("remove", err)
 			return &serviceRoutePolicyCleanupError{err: err}
 		}
+		r.Metrics.ObserveOperation("remove", nil)
 		delete(current, attachment)
 	}
 	for attachment, intent := range desired {
 		if existing, ok := current[attachment]; ok && reflect.DeepEqual(existing, intent) {
 			continue
 		}
 		if err := r.Programmer.Apply(intent); err != nil {
+			r.Metrics.ObserveOperation("apply", err)
 			if cleanupErr := r.Programmer.Cleanup(intent); cleanupErr != nil {
 				return errors.Join(err, &serviceRoutePolicyCleanupError{err: cleanupErr})
 			}
 			return err
 		}
+		r.Metrics.ObserveOperation("apply", nil)
 		current[attachment] = intent
 	}
 	if len(current) == 0 {
@@ -478,13 +523,24 @@ func (r *ServiceRoutePolicyReconciler) removePolicy(key types.NamespacedName) er
 }
 
 func (r *ServiceRoutePolicyReconciler) removePolicyLocked(key types.NamespacedName) error {
+	defer r.updateProgrammedGauge()
 	current := r.Applied[key]
 	for attachment, intent := range current {
 		if err := r.Programmer.Remove(intent); err != nil {
+			r.Metrics.ObserveOperation("remove", err)
 			return &serviceRoutePolicyCleanupError{err: err}
 		}
+		r.Metrics.ObserveOperation("remove", nil)
 		delete(current, attachment)
 	}
 	delete(r.Applied, key)
 	return nil
 }
+
+func (r *ServiceRoutePolicyReconciler) updateProgrammedGauge() {
+	count := 0
+	for _, intents := range r.Applied {
+		count += len(intents)
+	}
+	r.Metrics.SetProgrammedRoutes(count)
+}

internal/plumbing/ebpf/metrics/collector.go

@@ -324,7 +324,7 @@ func (c *Collector) collectDrops(ch chan<- prometheus.Metric) {
 	if c.dropReasons == nil {
 		return
 	}
-	for i := range prog.DropReasonCount {
+	for _, i := range prog.DropReasonMetricIndices {
 		var perCPU []uint64
 		if err := c.dropReasons.Lookup(i, &perCPU); err != nil {
 			ch <- prometheus.NewInvalidMetric(dropsDesc, fmt.Errorf("lookup drop_reasons[%d]: %w", i, err))

internal/plumbing/ebpf/metrics/collector_test.go

@@ -286,8 +286,9 @@ func findBlockGauges(metrics []*dto.Metric, block uint64) (used, ratio float64,
 
 func TestCollector_Drops(t *testing.T) {
 	drops := fakeDropReasons{
-		prog.DropReasonUnknownArgument: 42,
-		prog.DropReasonFibLookupFailed: 7,
+		prog.DropReasonUnknownArgument:    42,
+		prog.DropReasonFibLookupFailed:    7,
+		prog.DropReasonServiceRouteDenied: 3,
 	}
 	c := NewCollector(usidmap.NewVRFTable(newFakeTable()), usidmap.NewLocatorTable(newFakeTable()), drops, nil)
 
@@ -308,20 +309,21 @@ func TestCollector_Drops(t *testing.T) {
 		"strip_failed":          0,
 		"fib_lookup_failed":     7,
 		"redirect_failed":       0,
+		"service_route_denied":  3,
 	}
 	for reason, wantVal := range want {
 		got, ok := seen[reason]
 		if !ok {
-			t.Errorf("reason %q not emitted at all (all %d drop reasons must always be emitted, even at zero)",
-				reason, prog.DropReasonCount)
+			t.Errorf("reason %q not emitted at all (all %d metric reasons must always be emitted, even at zero)",
+				reason, len(prog.DropReasonMetricIndices))
 			continue
 		}
 		if got != wantVal {
 			t.Errorf("reason %q = %v, want %v", reason, got, wantVal)
 		}
 	}
-	if len(seen) != int(prog.DropReasonCount) {
-		t.Errorf("emitted %d distinct drop reasons, want %d", len(seen), prog.DropReasonCount)
+	if len(seen) != len(prog.DropReasonMetricIndices) {
+		t.Errorf("emitted %d distinct drop reasons, want %d", len(seen), len(prog.DropReasonMetricIndices))
 	}
 }
 

internal/plumbing/ebpf/prog/dropreason.go

@@ -30,8 +30,21 @@ const (
 	DropReasonEgressRouteRedirectFailed  uint32 = 14
 	DropReasonPublicUplinkRedirectFailed uint32 = 15
 	DropReasonCount                      uint32 = 16
+	DropReasonServiceRouteDenied         uint32 = 40
+	DropReasonServiceRouteRedirectFailed uint32 = 41
 )
 
+// DropReasonMetricIndices lists stable production counters. Diagnostic trace
+// slots between DropReasonCount and the service-route counters are deliberately
+// excluded from metrics.
+var DropReasonMetricIndices = func() []uint32 {
+	indices := make([]uint32, 0, DropReasonCount+2)
+	for index := range DropReasonCount {
+		indices = append(indices, index)
+	}
+	return append(indices, DropReasonServiceRouteDenied, DropReasonServiceRouteRedirectFailed)
+}()
+
 // DropReasonNames maps each index to a short, stable, metrics-friendly name,
 // decoupling label values and any other external representation from the C
 // identifier spelling.
@@ -52,4 +65,6 @@ var DropReasonNames = map[uint32]string{
 	DropReasonEgressRouteFibLookupFailed: "egress_route_fib_lookup_failed",
 	DropReasonEgressRouteRedirectFailed:  "egress_route_redirect_failed",
 	DropReasonPublicUplinkRedirectFailed: "public_uplink_redirect_failed",
+	DropReasonServiceRouteDenied:         "service_route_denied",
+	DropReasonServiceRouteRedirectFailed: "service_route_redirect_failed",
 }