feat(healthchecks): Adding Agent Health Checks for legacy & hubble control planes

cmd/hubble/daemon_linux.go

@@ -7,6 +7,7 @@ package hubble
 import (
 	"context"
 	"fmt"
+	"os"
 
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
@@ -32,6 +33,7 @@ import (
 	"k8s.io/client-go/rest"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/healthz"
 	logf "sigs.k8s.io/controller-runtime/pkg/log"
 	zapf "sigs.k8s.io/controller-runtime/pkg/log/zap"
 	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
@@ -66,6 +68,16 @@ var (
 				return nil, nil, fmt.Errorf("creating new controller-runtime manager: %w", err)
 			}
 
+			if err := ctrlManager.AddHealthzCheck("healthz", healthz.Ping); err != nil {
+				logger.Error("unable to set up healthz check", err)
+				os.Exit(1)
+			}
+
+			if err := ctrlManager.AddReadyzCheck("readyz", healthz.Ping); err != nil {
+				logger.Error("unable to set up readyz check", err)
+				os.Exit(1)
+			}
+
 			return ctrlManager, ctrlManager.GetClient(), nil
 		}),
 

cmd/legacy/daemon.go

@@ -4,6 +4,7 @@ package legacy
 
 import (
 	"fmt"
+	"net/http"
 	"os"
 	"strings"
 	"time"
@@ -55,6 +56,10 @@ const (
 	nodeIPEnvKey   = "NODE_IP"
 )
 
+var (
+	healthzChecker healthz.Checker
+)
+
 var scheme = k8sruntime.NewScheme()
 
 func init() {
@@ -215,14 +220,6 @@ func (d *Daemon) Start() error {
 	}
 
 	//+kubebuilder:scaffold:builder
-
-	if healthCheckErr := mgr.AddHealthzCheck("healthz", healthz.Ping); healthCheckErr != nil {
-		mainLogger.Fatal("Unable to set up health check", zap.Error(healthCheckErr))
-	}
-	if addReadyCheckErr := mgr.AddReadyzCheck("readyz", healthz.Ping); addReadyCheckErr != nil {
-		mainLogger.Fatal("Unable to set up ready check", zap.Error(addReadyCheckErr))
-	}
-
 	// k8s Client used for informers
 	cl := kubernetes.NewForConfigOrDie(mgr.GetConfig())
 
@@ -294,6 +291,21 @@ func (d *Daemon) Start() error {
 				mainLogger.Fatal("unable to create metricsConfigController", zap.Error(err))
 			}
 		}
+
+		// Define a custom health check for advanced metrics
+		healthzChecker = healthz.CheckHandler{
+			Checker: healthz.Checker(func(req *http.Request) error {
+				_, err := metricsModule.Status()
+				if err != nil {
+					mainLogger.Error("failed to get metrics module status fr advanced metrics", zap.Error(err))
+					return err
+				}
+				return nil
+			}),
+		}.Checker
+	} else {
+		// Advanced Metric not enabled, Ping healthcheck
+		healthzChecker = healthz.Ping
 	}
 
 	controllerMgr, err := cm.NewControllerManager(daemonConfig, cl, tel)
@@ -315,6 +327,17 @@ func (d *Daemon) Start() error {
 	go controllerMgr.Start(ctx)
 	mainLogger.Info("Started controller manager")
 
+	//Set health checks according to retina confiuration
+	if err := mgr.AddHealthzCheck("healthz", healthzChecker); err != nil {
+		mainLogger.Error("unable to set up custom health check", zap.Error(err))
+		os.Exit(1)
+	}
+
+	if err := mgr.AddReadyzCheck("readyz", healthzChecker); err != nil {
+		mainLogger.Error("unable to set up custom ready check", zap.Error(err))
+		os.Exit(1)
+	}
+
 	// Start all registered controllers. This will block until container receives SIGTERM.
 	if err := mgr.Start(ctx); err != nil {
 		mainLogger.Fatal("unable to start manager", zap.Error(err))

deploy/hubble/manifests/controller/helm/retina/templates/agent/daemonset.yaml

@@ -77,10 +77,22 @@ spec:
               cpu: {{ .Values.resources.limits.cpu | quote }}
           readinessProbe:
             httpGet:
-              path: /metrics
-              port: {{ .Values.agent.container.retina.ports.containerPort }}
-            initialDelaySeconds: 10
-            periodSeconds: 30
+              path: /readyz
+              port: {{ .Values.agent.container.retina.ports.healthPort }}
+            initialDelaySeconds: {{ .Values.readinessProbe.initialDelaySeconds | default "30" }}
+            periodSeconds: {{ .Values.readinessProbe.periodSeconds | default "30" }}
+            timeoutSeconds: {{ .Values.readinessProbe.timeoutSeconds | default "1" }}
+            failureThreshold: {{ .Values.readinessProbe.failureThreshold | default "3" }}
+            successThreshold: {{ .Values.readinessProbe.successThreshold | default "1" }}
+          livenessProbe:
+            httpGet:
+              path: /healthz
+              port: {{ .Values.agent.container.retina.ports.healthPort }}
+            initialDelaySeconds: {{ .Values.livenessProbe.initialDelaySeconds | default "30" }}
+            periodSeconds: {{ .Values.livenessProbe.periodSeconds | default "30" }}
+            timeoutSeconds: {{ .Values.livenessProbe.timeoutSeconds | default "1" }}
+            failureThreshold: {{ .Values.livenessProbe.failureThreshold | default "3" }}
+            successThreshold: {{ .Values.livenessProbe.successThreshold | default "1" }}
           env:
           - name: POD_NAME
             valueFrom:

deploy/hubble/manifests/controller/helm/retina/values.yaml

@@ -55,6 +55,7 @@ agent:
       metricsBindAddress: ":18080"
       ports:
         containerPort: 10093
+        healthPort: 18081
 
 enablePodLevel: true
 remoteContext: false
@@ -933,3 +934,29 @@ tls:
     #   -----BEGIN CERTIFICATE-----
     #   ...
     #   -----END CERTIFICATE-----
+## @param readinessProbe.initialDelaySeconds [array] Initial delay seconds for readinessProbe
+## @param readinessProbe.periodSeconds [array] Period seconds for readinessProbe
+## @param readinessProbe.timeoutSeconds [array] Timeout seconds for readinessProbe
+## @param readinessProbe.failureThreshold [array] Failure threshold for readinessProbe
+## @param readinessProbe.successThreshold [array] Success threshold for readinessProbe
+## Ref: https://kubernetes.io/docs/tasks/configure-pod-container/configure-liveness-readiness-probes/#configure-probes
+##
+readinessProbe: {}
+  # initialDelaySeconds: 30
+  # periodSeconds: 10
+  # timeoutSeconds: 15
+  # failureThreshold: 5
+  # successThreshold: 1
+## @param livenessProbe.initialDelaySeconds [array] Initial delay seconds for livenessProbe
+## @param livenessProbe.periodSeconds [array] Period seconds for livenessProbe
+## @param livenessProbe.timeoutSeconds [array] Timeout seconds for livenessProbe
+## @param livenessProbe.failureThreshold [array] Failure threshold for livenessProbe
+## @param livenessProbe.successThreshold [array] Success threshold for livenessProbe
+## Ref: https://kubernetes.io/docs/tasks/configure-pod-container/configure-liveness-readiness-probes/#configure-probes
+##
+livenessProbe: {}
+  # initialDelaySeconds: 30
+  # periodSeconds: 10
+  # timeoutSeconds: 15
+  # failureThreshold: 5
+  # successThreshold: 1

deploy/legacy/manifests/controller/helm/retina/templates/daemonset.yaml

@@ -49,17 +49,17 @@ spec:
         - name: {{ include "retina.name" . }} 
           readinessProbe:
             httpGet:
-              path: /metrics
-              port: {{ .Values.retinaPort }}
+              path: /readyz
+              port: {{ .Values.daemonset.container.retina.ports.healthPort }}
             initialDelaySeconds: {{ .Values.readinessProbe.initialDelaySeconds | default "30" }}
             periodSeconds: {{ .Values.readinessProbe.periodSeconds | default "30" }}
             timeoutSeconds: {{ .Values.readinessProbe.timeoutSeconds | default "1" }}
             failureThreshold: {{ .Values.readinessProbe.failureThreshold | default "3" }}
             successThreshold: {{ .Values.readinessProbe.successThreshold | default "1" }}
           livenessProbe:
             httpGet:
-              path: /metrics
-              port: {{ .Values.retinaPort }}
+              path: /healthz
+              port: {{ .Values.daemonset.container.retina.ports.healthPort }}
             initialDelaySeconds: {{ .Values.livenessProbe.initialDelaySeconds | default "30" }}
             periodSeconds: {{ .Values.livenessProbe.periodSeconds | default "30" }}
             timeoutSeconds: {{ .Values.livenessProbe.timeoutSeconds | default "1" }}

deploy/legacy/manifests/controller/helm/retina/values.yaml

@@ -86,6 +86,7 @@ daemonset:
       metricsBindAddress: ":18080"
       ports:
         containerPort: 10093
+        healthPort: 18081
 
 # volume mounts with name and mountPath
 volumeMounts:

pkg/enricher/enricher.go

@@ -5,9 +5,13 @@ package enricher
 
 import (
 	"context"
+	"errors"
+	"io"
 	"reflect"
 	"sync"
 
+	"time"
+
 	"github.com/cilium/cilium/api/v1/flow"
 	v1 "github.com/cilium/cilium/pkg/hubble/api/v1"
 	"github.com/cilium/cilium/pkg/hubble/container"
@@ -189,3 +193,54 @@ func (e *Enricher) Write(ev *v1.Event) {
 func (e *Enricher) ExportReader() *container.RingReader {
 	return container.NewRingReader(e.outputRing, e.outputRing.OldestWrite())
 }
+
+func (e *Enricher) Status() (float64, error) {
+	rate, err := e.getFlowRate(e.outputRing, time.Now())
+	if err != nil {
+		e.l.Error("failed to get flow rate %w", zap.Error(err))
+		return 0, err
+	}
+	return rate, nil
+}
+
+// ref: "getFlowRate" "github.com/cilium/cilium/pkg/hubble/observer/local_observer.go"
+func (en *Enricher) getFlowRate(ring *container.Ring, at time.Time) (float64, error) {
+	reader := container.NewRingReader(ring, ring.LastWriteParallel())
+	count := 0
+	since := at.Add(-1 * time.Minute)
+	var lastSeenEvent *v1.Event
+	for {
+		e, err := reader.Previous()
+		lost := e.GetLostEvent()
+		if lost != nil {
+			// a lost event means we read the complete ring buffer
+			// if we read at least one flow, update `since` to calculate the rate over the available time range
+			if lastSeenEvent != nil {
+				since = lastSeenEvent.Timestamp.AsTime()
+			}
+			break
+		} else if errors.Is(err, io.EOF) {
+			// an EOF error means the ring buffer is empty, ignore error and continue
+			break
+		} else if err != nil {
+			// unexpected error
+			return 0, err
+		}
+		if _, isFlowEvent := e.Event.(*flow.Flow); !isFlowEvent {
+			// ignore non flow events
+			continue
+		}
+		if err := e.Timestamp.CheckValid(); err != nil {
+			return 0, err
+		}
+		ts := e.Timestamp.AsTime()
+		if ts.Before(since) {
+			// scanned the last minute, exit loop
+			break
+		}
+		lastSeenEvent = e
+		count++
+	}
+	fl := float64(count) / at.Sub(since).Seconds()
+	return fl, nil
+}

pkg/enricher/types.go

@@ -13,4 +13,5 @@ type EnricherInterface interface {
 	Run()
 	Write(ev *v1.Event)
 	ExportReader() *container.RingReader
+	Status() (float64, error)
 }

pkg/module/metrics/metrics_module.go

@@ -139,6 +139,10 @@ func InitModule(ctx context.Context,
 	return m
 }
 
+func (m *Module) Status() (float64, error) {
+	return m.enricher.Status()
+}
+
 func (m *Module) Reconcile(spec *api.MetricsSpec) error {
 	// If the new spec has not changed, then do nothing.
 	if m.currentSpec != nil && m.currentSpec.Equals(spec) {