Remove relevant metrics series on pod deletion (#23162).

Documentation/observability/metrics.rst

@@ -609,6 +609,16 @@ All labels listed are included in the metric, even if empty. For example, a metr
 ``http:labelsContext=source_namespace,source_pod`` will add the ``source_namespace`` and ``source_pod``
 labels to all Hubble HTTP metrics.
 
+.. note::
+
+    To limit metrics cardinality hubble will remove data series bound to specific pod after one minute from pod deletion.
+    Metric is considered to be bound to a specific pod when at least one of the following conditions is met:
+
+    * ``sourceContext`` is set to ``pod`` and metric series has ``source`` label matching ``<pod_namespace>/<pod_name>``
+    * ``destinationContext`` is set to ``pod`` and metric series has ``destination`` label matching ``<pod_namespace>/<pod_name>``
+    * ``labelsContext`` contains both ``source_namespace`` and ``source_pod`` and metric series labels match namespace and name of deleted pod
+    * ``labelsContext`` contains both ``destination_namespace`` and ``destination_pod`` and metric series labels match namespace and name of deleted pod
+
 .. _hubble_exported_metrics:
 
 Exported Metrics

pkg/hubble/metrics/api/api.go

@@ -13,6 +13,7 @@ import (
 	"go.uber.org/multierr"
 
 	pb "github.com/cilium/cilium/api/v1/flow"
+	slim_corev1 "github.com/cilium/cilium/pkg/k8s/slim/k8s/api/core/v1"
 	"github.com/cilium/cilium/pkg/logging"
 	"github.com/cilium/cilium/pkg/logging/logfields"
 )
@@ -73,6 +74,12 @@ type Handler interface {
 	// specifies Prometheus registry
 	Init(registry *prometheus.Registry, options Options) error
 
+	// ListMetricVec returns an array of MetricVec used by a handler
+	ListMetricVec() []*prometheus.MetricVec
+
+	// Context used by this metrics handler
+	Context() *ContextOptions
+
 	// Status returns the configuration status of the metric handler
 	Status() string
 }
@@ -120,6 +127,18 @@ func (h Handlers) ProcessFlow(ctx context.Context, flow *pb.Flow) error {
 	return processingErr
 }
 
+// ProcessPodDeletion queries all handlers for a list of MetricVec and removes
+// metrics directly associated to deleted pod.
+func (h Handlers) ProcessPodDeletion(pod *slim_corev1.Pod) {
+	for _, h := range h.handlers {
+		for _, mv := range h.ListMetricVec() {
+			if ctx := h.Context(); ctx != nil {
+				ctx.DeleteMetricsAssociatedWithPod(pod.GetName(), pod.GetNamespace(), mv)
+			}
+		}
+	}
+}
+
 var registry = NewRegistry(
 	logging.DefaultLogger.WithField(logfields.LogSubsys, "hubble"),
 )

pkg/hubble/metrics/api/context.go

@@ -9,6 +9,7 @@ import (
 	"sort"
 	"strings"
 
+	"github.com/prometheus/client_golang/prometheus"
 	"k8s.io/utils/strings/slices"
 
 	pb "github.com/cilium/cilium/api/v1/flow"
@@ -137,13 +138,17 @@ type ContextOptions struct {
 	// Destination is the destination context to include in metrics for ingress traffic (overrides Destination)
 	DestinationIngress ContextIdentifierList
 
+	allDestinationCtx ContextIdentifierList
+
 	// Source is the source context to include in metrics for both egress and ingress traffic
 	Source ContextIdentifierList
 	// Source is the source context to include in metrics for egress traffic (overrides Source)
 	SourceEgress ContextIdentifierList
 	// Source is the source context to include in metrics for ingress traffic (overrides Source)
 	SourceIngress ContextIdentifierList
 
+	allSourceCtx ContextIdentifierList
+
 	// Labels is the full set of labels that have been allowlisted when using the
 	// ContextLabels ContextIdentifier.
 	Labels labelsSet
@@ -206,31 +211,37 @@ func ParseContextOptions(options Options) (*ContextOptions, error) {
 		switch strings.ToLower(key) {
 		case "destinationcontext":
 			o.Destination, err = parseContext(value)
+			o.allDestinationCtx = append(o.allDestinationCtx, o.Destination...)
 			if err != nil {
 				return nil, err
 			}
 		case "destinationegresscontext":
 			o.DestinationEgress, err = parseContext(value)
+			o.allDestinationCtx = append(o.allDestinationCtx, o.DestinationEgress...)
 			if err != nil {
 				return nil, err
 			}
 		case "destinationingresscontext":
 			o.DestinationIngress, err = parseContext(value)
+			o.allDestinationCtx = append(o.allDestinationCtx, o.DestinationIngress...)
 			if err != nil {
 				return nil, err
 			}
 		case "sourcecontext":
 			o.Source, err = parseContext(value)
+			o.allSourceCtx = append(o.allSourceCtx, o.Source...)
 			if err != nil {
 				return nil, err
 			}
 		case "sourceegresscontext":
 			o.SourceEgress, err = parseContext(value)
+			o.allSourceCtx = append(o.allSourceCtx, o.SourceEgress...)
 			if err != nil {
 				return nil, err
 			}
 		case "sourceingresscontext":
 			o.SourceIngress, err = parseContext(value)
+			o.allSourceCtx = append(o.allSourceCtx, o.SourceIngress...)
 			if err != nil {
 				return nil, err
 			}
@@ -524,3 +535,33 @@ func (o *ContextOptions) Status() string {
 
 	return strings.Join(status, ",")
 }
+
+func (o *ContextOptions) DeleteMetricsAssociatedWithPod(name string, namespace string, vec *prometheus.MetricVec) {
+	for _, contextID := range o.allSourceCtx {
+		if contextID == ContextPod {
+			vec.DeletePartialMatch(prometheus.Labels{
+				"source": namespace + "/" + name,
+			})
+		}
+	}
+	for _, contextID := range o.allDestinationCtx {
+		if contextID == ContextPod {
+			vec.DeletePartialMatch(prometheus.Labels{
+				"destination": namespace + "/" + name,
+			})
+		}
+	}
+
+	if o.Labels.HasLabel("source_pod") && o.Labels.HasLabel("source_namespace") {
+		vec.DeletePartialMatch(prometheus.Labels{
+			"source_namespace": namespace,
+			"source_pod":       name,
+		})
+	}
+	if o.Labels.HasLabel("destination_pod") && o.Labels.HasLabel("destination_namespace") {
+		vec.DeletePartialMatch(prometheus.Labels{
+			"destination_namespace": namespace,
+			"destination_pod":       name,
+		})
+	}
+}