Merge pull request prometheus-operator#8478 from simonpasquier/update-shard-retention

Update shard retention

Author: simonpasquier

CHANGELOG.md

@@ -1,3 +1,7 @@
+## UNRELEASED
+
+* [FEATURE] Implement shard retention based on Prometheus data retention (it requires the `PrometheusShardRetentionPolicy` feature gate). #8478
+
 ## 0.90.1 / 2026-03-25
 
 * [BUGFIX] Fix Probe ignoring HTTP client settings in scrape configuration. #8461

pkg/k8s/k8s.go

@@ -309,13 +309,15 @@ func AddTypeInformationToObject(obj runtime.Object) error {
 
 // mergeMetadata takes labels and annotations from the old resource and merges
 // them into the new resource. If a key is present in both resources, the new
-// resource wins. It also copies the ResourceVersion from the old resource to
-// the new resource to prevent update conflicts.
+// resource wins. All keys starting with the "operator.prometheus.io/" prefix
+// in the old resource are dropped before merging.
+// It also copies the ResourceVersion from the old resource to the new resource
+// to prevent update conflicts.
 func mergeMetadata(newObj *metav1.ObjectMeta, oldObj metav1.ObjectMeta) {
 	newObj.ResourceVersion = oldObj.ResourceVersion
 
-	newObj.SetLabels(mergeMap(oldObj.Labels, maps.All(newObj.Labels)))
-	newObj.SetAnnotations(mergeMap(oldObj.Annotations, maps.All(newObj.Annotations)))
+	newObj.SetLabels(mergeMap(maps.Collect(excludeOperatorPrefixSeq(oldObj.Labels)), maps.All(newObj.Labels)))
+	newObj.SetAnnotations(mergeMap(maps.Collect(excludeOperatorPrefixSeq(oldObj.Annotations)), maps.All(newObj.Annotations)))
 }
 
 // copyKubectlAnnotations copies the kubectl's annotations into the object
@@ -324,12 +326,12 @@ func copyKubectlAnnotations(objMeta *metav1.ObjectMeta, annotations map[string]s
 	objMeta.SetAnnotations(mergeMap(objMeta.Annotations, filterByPrefixSeq(annotations, "kubectl.kubernetes.io/")))
 }
 
-// filterByPrefixSeq returns a iterator over m for all keys matching the
-// prefix.
-func filterByPrefixSeq(m map[string]string, prefix string) iter.Seq2[string, string] {
+// excludeOperatorPrefixSeq returns a iterator over m excluding all keys
+// which start by the reserved operator's prefix.
+func excludeOperatorPrefixSeq(m map[string]string) iter.Seq2[string, string] {
 	return func(yield func(k, v string) bool) {
 		for k, v := range m {
-			if !strings.HasPrefix(k, prefix) {
+			if strings.HasPrefix(k, "operator.prometheus.io/") {
 				continue
 			}
 			if !yield(k, v) {
@@ -339,6 +341,20 @@ func filterByPrefixSeq(m map[string]string, prefix string) iter.Seq2[string, str
 	}
 }
 
+// filterByPrefixSeq returns a iterator over m for all keys matching the
+// prefix.
+func filterByPrefixSeq(m map[string]string, prefix string) iter.Seq2[string, string] {
+	return func(yield func(k, v string) bool) {
+		for k, v := range m {
+			if strings.HasPrefix(k, prefix) {
+				if !yield(k, v) {
+					return
+				}
+			}
+		}
+	}
+}
+
 // mergeMap returns a clone of m for which key-value pairs from seq have been added.
 func mergeMap(m map[string]string, seq iter.Seq2[string, string]) map[string]string {
 	// Don't mutate the input maps.
@@ -348,6 +364,5 @@ func mergeMap(m map[string]string, seq iter.Seq2[string, string]) map[string]str
 	}
 
 	maps.Insert(m, seq)
-
 	return m
 }

pkg/k8s/k8s_test.go

@@ -61,7 +61,7 @@ func TestMergeMetadata_CreateOrUpdateSecret(t *testing.T) {
 			},
 		},
 		{
-			name: "overridden label amd annotation",
+			name: "overridden label and annotation",
 			expectedLabels: map[string]string{
 				"app.kubernetes.io/name": "kube-state-metrics",
 			},
@@ -75,6 +75,21 @@ func TestMergeMetadata_CreateOrUpdateSecret(t *testing.T) {
 				"app.kubernetes.io/name": "overridden-value",
 			},
 		},
+		{
+			name: "labels and annotations with reserved prefix are dropped",
+			expectedLabels: map[string]string{
+				"app.kubernetes.io/name": "kube-state-metrics",
+			},
+			modifiedLabels: map[string]string{
+				"operator.prometheus.io/foo": "some value",
+			},
+			expectedAnnotations: map[string]string{
+				"app.kubernetes.io/name": "kube-state-metrics",
+			},
+			modifiedAnnotations: map[string]string{
+				"operator.prometheus.io/bar": "some value",
+			},
+		},
 	}
 
 	namespace := "ns-1"

pkg/k8s/statefulset.go

@@ -72,7 +72,7 @@ func CreateStatefulSetOrPatchLabels(ctx context.Context, ssetClient clientappsv1
 func ForceUpdateStatefulSet(ctx context.Context, ssetClient clientappsv1.StatefulSetInterface, sset *appsv1.StatefulSet, onDeleteFunc func(string)) error {
 	err := updateStatefulSet(ctx, ssetClient, sset)
 	if err == nil {
-		return err
+		return nil
 	}
 
 	// When trying to update immutable fields, the API server returns a 422 status code.

pkg/operator/resource_reconciler.go

@@ -371,7 +371,9 @@ func (rr *ResourceReconciler) objectKey(obj any) (string, bool) {
 	return k, true
 }
 
-func (rr *ResourceReconciler) resolve(obj metav1.Object) metav1.Object {
+// FindOwner returns the resource owning the given object.
+// For example it can return the Prometheus resource owning a StatefulSet.
+func (rr *ResourceReconciler) FindOwner(obj metav1.Object) metav1.Object {
 	for _, or := range obj.GetOwnerReferences() {
 		if !ptr.Deref(or.Controller, false) {
 			continue
@@ -513,7 +515,7 @@ func (rr *ResourceReconciler) OnDelete(obj any) {
 }
 
 func (rr *ResourceReconciler) onStatefulSetAdd(ss *appsv1.StatefulSet) {
-	obj := rr.resolve(ss)
+	obj := rr.FindOwner(ss)
 	if obj == nil {
 		return
 	}
@@ -525,7 +527,7 @@ func (rr *ResourceReconciler) onStatefulSetAdd(ss *appsv1.StatefulSet) {
 }
 
 func (rr *ResourceReconciler) onDaemonSetAdd(ds *appsv1.DaemonSet) {
-	obj := rr.resolve(ds)
+	obj := rr.FindOwner(ds)
 	if obj == nil {
 		return
 	}
@@ -546,7 +548,7 @@ func (rr *ResourceReconciler) onStatefulSetUpdate(old, cur *appsv1.StatefulSet)
 		return
 	}
 
-	obj := rr.resolve(cur)
+	obj := rr.FindOwner(cur)
 	if obj == nil {
 		return
 	}
@@ -576,7 +578,7 @@ func (rr *ResourceReconciler) onDaemonSetUpdate(old, cur *appsv1.DaemonSet) {
 		return
 	}
 
-	obj := rr.resolve(cur)
+	obj := rr.FindOwner(cur)
 	if obj == nil {
 		return
 	}
@@ -595,7 +597,7 @@ func (rr *ResourceReconciler) onDaemonSetUpdate(old, cur *appsv1.DaemonSet) {
 }
 
 func (rr *ResourceReconciler) onStatefulSetDelete(ss *appsv1.StatefulSet) {
-	obj := rr.resolve(ss)
+	obj := rr.FindOwner(ss)
 	if obj == nil {
 		return
 	}
@@ -607,7 +609,7 @@ func (rr *ResourceReconciler) onStatefulSetDelete(ss *appsv1.StatefulSet) {
 }
 
 func (rr *ResourceReconciler) onDaemonSetDelete(ds *appsv1.DaemonSet) {
-	obj := rr.resolve(ds)
+	obj := rr.FindOwner(ds)
 	if obj == nil {
 		return
 	}

pkg/prometheus/server/operator.go

@@ -16,6 +16,7 @@ package prometheus
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"log/slog"
@@ -25,12 +26,15 @@ import (
 
 	"github.com/mitchellh/hashstructure"
 	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/common/model"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/intstr"
+	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/dynamic"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/metadata"
@@ -60,6 +64,9 @@ const (
 
 	unmanagedConfigurationReason  = "ConfigurationUnmanaged"
 	unmanagedConfigurationMessage = "the operator doesn't manage the Prometheus configuration secret because neither serviceMonitorSelector nor podMonitorSelector, nor probeSelector, nor scrapeConfigSelector is specified. Unmanaged Prometheus configuration is deprecated, use additionalScrapeConfigs or the ScrapeConfig Custom Resource Definition instead. Unmanaged Prometheus configuration can also be disabled from the operator's command-line (check './operator --help')."
+
+	deletionDeadlineAnnotation = "operator.prometheus.io/deletion-deadline"
+	annotationTimeFormat       = time.RFC3339
 )
 
 // Operator manages the life cycle of Prometheus deployments and
@@ -623,6 +630,45 @@ func (c *Operator) Run(ctx context.Context) error {
 	// TODO(simonpasquier): watch for Prometheus pods instead of polling.
 	go operator.StatusPoller(ctx, c)
 
+	if c.retentionPoliciesEnabled {
+		// Periodically scan statefulsets for expired shards.
+		go func() {
+			_ = wait.PollUntilContextCancel(ctx, time.Minute, true, func(context.Context) (bool, error) {
+				_ = c.ssetInfs.ListAll(labels.Everything(), func(o any) {
+					sset := o.(*appsv1.StatefulSet)
+
+					deadline, found := sset.Annotations[deletionDeadlineAnnotation]
+					if !found {
+						return
+					}
+
+					logger := c.logger.With("statefulset", fmt.Sprintf("%s/%s", sset.Namespace, sset.Name))
+					expired, err := deadlineExpired(deadline)
+					if err != nil {
+						logger.Warn("failed to parse deletion deadline annotation", "err", err)
+						return
+					}
+
+					if !expired {
+						logger.Debug("deletion deadline not expired", "deadline", deadline)
+						return
+					}
+
+					logger.Info("deletion deadline expired", "deadline", deadline)
+					owner := c.rr.FindOwner(sset)
+					if owner == nil {
+						logger.Warn("owner not found")
+						return
+					}
+
+					c.rr.EnqueueForReconciliation(owner)
+				})
+
+				return false, nil
+			})
+		}()
+	}
+
 	c.metrics.Ready().Set(1)
 	<-ctx.Done()
 	return nil
@@ -962,7 +1008,7 @@ func (c *Operator) sync(ctx context.Context, key string) (func(context.Context)
 
 	ssetClient := c.kclient.AppsV1().StatefulSets(p.Namespace)
 
-	// Ensure we have a StatefulSet running Prometheus deployed and that StatefulSet names are created correctly.
+	// Reconcile all active statefulset shards.
 	expected := prompkg.ExpectedStatefulSetShardNames(p)
 	for shard, ssetName := range expected {
 		logger := logger.With("statefulset", ssetName, "shard", fmt.Sprintf("%d", shard))
@@ -1060,18 +1106,19 @@ func (c *Operator) sync(ctx context.Context, key string) (func(context.Context)
 			return
 		}
 
-		shouldRetain, retainErr := c.shouldRetain(p)
-		if retainErr != nil {
-			deleteErrs = append(deleteErrs, fmt.Errorf("failed to determine if StatefulSet %s should be retained: %w", s.GetName(), retainErr))
+		shouldDelete, err := c.processShardRetention(ctx, p, s)
+		if err != nil {
+			logger.Warn("failed to process shard retention, not deleting the shard", "err", err, "statefulset", fmt.Sprintf("%s/%s", s.Namespace, s.Name))
 			return
 		}
-		if shouldRetain {
+
+		if !shouldDelete {
 			return
 		}
 
-		if delErr := ssetClient.Delete(ctx, s.GetName(), metav1.DeleteOptions{PropagationPolicy: ptr.To(metav1.DeletePropagationForeground)}); delErr != nil {
-			if !apierrors.IsNotFound(delErr) {
-				deleteErrs = append(deleteErrs, fmt.Errorf("failed to delete StatefulSet %s: %w", s.GetName(), delErr))
+		if err := ssetClient.Delete(ctx, s.GetName(), metav1.DeleteOptions{PropagationPolicy: ptr.To(metav1.DeletePropagationForeground)}); err != nil {
+			if !apierrors.IsNotFound(err) {
+				deleteErrs = append(deleteErrs, fmt.Errorf("failed to delete StatefulSet %s: %w", s.GetName(), err))
 			}
 		}
 	})
@@ -1202,23 +1249,95 @@ func (c *Operator) configResStatusCleanup(ctx context.Context, p *monitoringv1.P
 	return nil
 }
 
-// As the ShardRetentionPolicy feature evolves, should retain will evolve accordingly.
-// For now, shouldRetain just returns the appropriate boolean based on the retention type.
-func (c *Operator) shouldRetain(p *monitoringv1.Prometheus) (bool, error) {
+// processShardRetention returns true if the associated statefulset should be
+// deleted.
+func (c *Operator) processShardRetention(ctx context.Context, p *monitoringv1.Prometheus, sset *appsv1.StatefulSet) (bool, error) {
 	if !c.retentionPoliciesEnabled {
-		// Feature-gate is disabled, default behavior is always to delete.
-		return false, nil
+		// When the feature gate is disabled, the default behavior is always to delete.
+		return true, nil
 	}
+
 	if p.Spec.ShardRetentionPolicy == nil {
-		// ShardRetentionPolicy not configured, default behavior is to delete.
-		return false, nil
+		// ShardRetentionPolicy not configured, the default behavior is to delete.
+		return true, nil
 	}
-	if ptr.Deref(p.Spec.ShardRetentionPolicy.WhenScaled,
-		monitoringv1.DeleteWhenScaledRetentionType) == monitoringv1.RetainWhenScaledRetentionType {
+
+	if ptr.Deref(p.Spec.ShardRetentionPolicy.WhenScaled, monitoringv1.DeleteWhenScaledRetentionType) != monitoringv1.RetainWhenScaledRetentionType {
 		return true, nil
 	}
 
-	return false, nil
+	if deadline, found := sset.Annotations[deletionDeadlineAnnotation]; found {
+		expired, err := deadlineExpired(deadline)
+		if err != nil {
+			return false, err
+		}
+
+		return expired, nil
+	}
+
+	// Set the annotation on the statefulset. If the deadline never expires,
+	// the annotation is set to the zero time value.
+	gracePeriod, err := gracePeriodForPrometheusStorage(p)
+	if err != nil {
+		return false, err
+	}
+
+	var deadline time.Time
+	if gracePeriod > 0 {
+		deadline = time.Now().UTC().Add(gracePeriod)
+	}
+
+	patchData, err := json.Marshal(map[string]any{
+		"metadata": map[string]any{
+			"annotations": map[string]string{
+				deletionDeadlineAnnotation: deadline.Format(annotationTimeFormat),
+			},
+		},
+	})
+	if err != nil {
+		return false, err
+	}
+
+	_, err = c.kclient.AppsV1().StatefulSets(sset.Namespace).Patch(
+		ctx,
+		sset.Name,
+		types.StrategicMergePatchType,
+		patchData,
+		metav1.PatchOptions{FieldManager: k8s.PrometheusOperatorFieldManager})
+
+	return false, err
+}
+
+func deadlineExpired(deadline string) (bool, error) {
+	t, err := time.Parse(annotationTimeFormat, deadline)
+	if err != nil {
+		return false, err
+	}
+
+	// A zero time means that the deadline never expires.
+	return !t.IsZero() && time.Now().UTC().After(t), nil
+}
+
+// gracePeriodForPrometheusStorage returns how long the Prometheus data can be available based
+// on the retention settings.
+// If Prometheus is configured with size-based retention only, it returns a
+// zero value.
+func gracePeriodForPrometheusStorage(p *monitoringv1.Prometheus) (time.Duration, error) {
+	if p.Spec.RetentionSize != "" && p.Spec.Retention == "" {
+		return time.Duration(0), nil
+	}
+
+	retention := p.Spec.Retention
+	if retention == "" {
+		retention = defaultRetention
+	}
+
+	d, err := model.ParseDuration(string(retention))
+	if err != nil {
+		return time.Duration(0), err
+	}
+
+	return time.Duration(d), nil
 }
 
 // UpdateStatus updates the status subresource of the object identified by the given