Add conditional artificial delay (#11139)

Signed-off-by: Marco Pracucci <[email protected]>

Author: pracucci

pkg/util/validation/limits.go

@@ -13,6 +13,7 @@ import (
 	"fmt"
 	"math"
 	"reflect"
+	"strconv"
 	"strings"
 	"time"
 
@@ -129,7 +130,13 @@ type Limits struct {
 	MetricRelabelConfigs                        []*relabel.Config   `yaml:"metric_relabel_configs,omitempty" json:"metric_relabel_configs,omitempty" doc:"nocli|description=List of metric relabel configurations. Note that in most situations, it is more effective to use metrics relabeling directly in the Prometheus server, e.g. remote_write.write_relabel_configs. Labels available during the relabeling phase and cleaned afterwards: __meta_tenant_id" category:"experimental"`
 	MetricRelabelingEnabled                     bool                `yaml:"metric_relabeling_enabled" json:"metric_relabeling_enabled" category:"experimental"`
 	ServiceOverloadStatusCodeOnRateLimitEnabled bool                `yaml:"service_overload_status_code_on_rate_limit_enabled" json:"service_overload_status_code_on_rate_limit_enabled" category:"experimental"`
-	IngestionArtificialDelay                    model.Duration      `yaml:"ingestion_artificial_delay" json:"ingestion_artificial_delay" category:"experimental" doc:"hidden"`
+
+	IngestionArtificialDelay                                         model.Duration `yaml:"ingestion_artificial_delay" json:"ingestion_artificial_delay" category:"experimental" doc:"hidden"`
+	IngestionArtificialDelayConditionForTenantsWithLessThanMaxSeries int            `yaml:"ingestion_artificial_delay_condition_for_tenants_with_less_than_max_series" json:"ingestion_artificial_delay_condition_for_tenants_with_less_than_max_series" category:"experimental" doc:"hidden"`
+	IngestionArtificialDelayDurationForTenantsWithLessThanMaxSeries  model.Duration `yaml:"ingestion_artificial_delay_duration_for_tenants_with_less_than_max_series" json:"ingestion_artificial_delay_duration_for_tenants_with_less_than_max_series" category:"experimental" doc:"hidden"`
+	IngestionArtificialDelayConditionForTenantsWithIDGreaterThan     int            `yaml:"ingestion_artificial_delay_condition_for_tenants_with_id_greater_than" json:"ingestion_artificial_delay_condition_for_tenants_with_id_greater_than" category:"experimental" doc:"hidden"`
+	IngestionArtificialDelayDurationForTenantsWithIDGreaterThan      model.Duration `yaml:"ingestion_artificial_delay_duration_for_tenants_with_id_greater_than" json:"ingestion_artificial_delay_duration_for_tenants_with_id_greater_than" category:"experimental" doc:"hidden"`
+
 	// Ingester enforced limits.
 	// Series
 	MaxGlobalSeriesPerUser   int `yaml:"max_global_series_per_user" json:"max_global_series_per_user"`
@@ -302,7 +309,12 @@ func (l *Limits) RegisterFlags(f *flag.FlagSet) {
 	f.BoolVar(&l.OTelCreatedTimestampZeroIngestionEnabled, "distributor.otel-created-timestamp-zero-ingestion-enabled", false, "Whether to enable translation of OTel start timestamps to Prometheus zero samples in the OTLP endpoint.")
 	f.Var(&l.PromoteOTelResourceAttributes, "distributor.otel-promote-resource-attributes", "Optionally specify OTel resource attributes to promote to labels.")
 	f.BoolVar(&l.OTelKeepIdentifyingResourceAttributes, "distributor.otel-keep-identifying-resource-attributes", false, "Whether to keep identifying OTel resource attributes in the target_info metric on top of converting to job and instance labels.")
-	f.Var(&l.IngestionArtificialDelay, "distributor.ingestion-artificial-delay", "Target ingestion delay. If set to a non-zero value, the distributor will artificially delay ingestion time-frame by the specified duration by computing the difference between actual ingestion and the target. There is no delay on actual ingestion of samples, it is only the response back to the client.")
+
+	f.Var(&l.IngestionArtificialDelay, "distributor.ingestion-artificial-delay", "Target ingestion delay to apply to all tenants. If set to a non-zero value, the distributor will artificially delay ingestion time-frame by the specified duration by computing the difference between actual ingestion and the target. There is no delay on actual ingestion of samples, it is only the response back to the client.")
+	f.IntVar(&l.IngestionArtificialDelayConditionForTenantsWithLessThanMaxSeries, "distributor.ingestion-artificial-delay-condition-for-tenants-with-less-than-max-series", 0, "Condition to select tenants for which -distributor.ingestion-artificial-delay-duration-for-tenants-with-less-than-max-series should be applied.")
+	f.Var(&l.IngestionArtificialDelayDurationForTenantsWithLessThanMaxSeries, "distributor.ingestion-artificial-delay-duration-for-tenants-with-less-than-max-series", "Target ingestion delay to apply to tenants with configured max global series to a value lower than -distributor.ingestion-artificial-delay-condition-for-tenants-with-less-than-max-series.")
+	f.IntVar(&l.IngestionArtificialDelayConditionForTenantsWithIDGreaterThan, "distributor.ingestion-artificial-delay-condition-for-tenants-with-id-greater-than", 0, "Condition to select tenants for which -distributor.ingestion-artificial-delay-duration-for-tenants-with-id-greater-than should be applied.")
+	f.Var(&l.IngestionArtificialDelayDurationForTenantsWithIDGreaterThan, "distributor.ingestion-artificial-delay-duration-for-tenants-with-id-greater-than", "Target ingestion delay to apply to tenants with a numeric ID whose value is greater than -distributor.ingestion-artificial-delay-condition-for-tenants-with-id-greater-than.")
 
 	f.IntVar(&l.MaxGlobalSeriesPerUser, MaxSeriesPerUserFlag, 150000, "The maximum number of in-memory series per tenant, across the cluster before replication. 0 to disable.")
 	f.IntVar(&l.MaxGlobalSeriesPerMetric, MaxSeriesPerMetricFlag, 0, "The maximum number of in-memory series per metric name, across the cluster before replication. 0 to disable.")
@@ -1224,9 +1236,29 @@ func (o *Overrides) OTelKeepIdentifyingResourceAttributes(tenantID string) bool
 	return o.getOverridesForUser(tenantID).OTelKeepIdentifyingResourceAttributes
 }
 
-// DistributorIngestionArtificialDelay returns the artificial ingestion latency for a given use.
+// DistributorIngestionArtificialDelay returns the artificial ingestion latency for a given user.
 func (o *Overrides) DistributorIngestionArtificialDelay(tenantID string) time.Duration {
-	return time.Duration(o.getOverridesForUser(tenantID).IngestionArtificialDelay)
+	overrides := o.getOverridesForUser(tenantID)
+
+	// Default delay to apply to all tenants.
+	delay := overrides.IngestionArtificialDelay
+
+	// Check if the "max series" condition applies to this tenant.
+	maxSeriesCondition := overrides.IngestionArtificialDelayConditionForTenantsWithLessThanMaxSeries
+	maxSeriesDelay := overrides.IngestionArtificialDelayDurationForTenantsWithLessThanMaxSeries
+	if maxSeriesCondition > 0 && maxSeriesDelay > delay && o.MaxGlobalSeriesPerUser(tenantID) < maxSeriesCondition {
+		delay = maxSeriesDelay
+	}
+
+	// Check if the "tenant ID" condition applies to this tenant.
+	idCondition := overrides.IngestionArtificialDelayConditionForTenantsWithIDGreaterThan
+	idDelay := overrides.IngestionArtificialDelayDurationForTenantsWithIDGreaterThan
+	idNumber, idNumberErr := strconv.ParseInt(tenantID, 10, 32)
+	if idCondition > 0 && idDelay > delay && idNumberErr == nil && int(idNumber) > idCondition {
+		delay = idDelay
+	}
+
+	return time.Duration(delay)
 }
 
 func (o *Overrides) AlignQueriesWithStep(userID string) bool {

pkg/util/validation/limits_test.go

@@ -399,6 +399,94 @@ func TestMaxPartialQueryLengthWithoutDefault(t *testing.T) {
 	assert.Equal(t, time.Duration(0), ov.MaxPartialQueryLength("tenant-b"))
 }
 
+func TestDistributorIngestionArtificialDelay(t *testing.T) {
+	tests := map[string]struct {
+		tenantID      string
+		tenantLimits  func(*Limits)
+		expectedDelay time.Duration
+	}{
+		"should not apply delay by default": {
+			tenantID:      "tenant-a",
+			tenantLimits:  func(*Limits) {},
+			expectedDelay: 0,
+		},
+		"should apply delay if a plain delay has been configured for the tenant": {
+			tenantID: "tenant-a",
+			tenantLimits: func(l *Limits) {
+				l.IngestionArtificialDelay = model.Duration(time.Second)
+			},
+			expectedDelay: time.Second,
+		},
+		"should apply delay based on 'max series less than' condition if tenant max series is < the threshold": {
+			tenantID: "tenant-a",
+			tenantLimits: func(l *Limits) {
+				l.IngestionArtificialDelayConditionForTenantsWithLessThanMaxSeries = 15001
+				l.IngestionArtificialDelayDurationForTenantsWithLessThanMaxSeries = model.Duration(time.Second)
+				l.MaxGlobalSeriesPerUser = 15000
+			},
+			expectedDelay: time.Second,
+		},
+		"should not apply delay based on 'max series less than' condition if tenant max series is >= the threshold": {
+			tenantID: "tenant-a",
+			tenantLimits: func(l *Limits) {
+				l.IngestionArtificialDelayConditionForTenantsWithLessThanMaxSeries = 15001
+				l.IngestionArtificialDelayDurationForTenantsWithLessThanMaxSeries = model.Duration(time.Second)
+				l.MaxGlobalSeriesPerUser = 15001
+			},
+			expectedDelay: 0,
+		},
+		"should apply delay based on 'tenant ID greater than' condition if tenant ID is numeric and > the condition": {
+			tenantID: "12346",
+			tenantLimits: func(l *Limits) {
+				l.IngestionArtificialDelayConditionForTenantsWithIDGreaterThan = 12345
+				l.IngestionArtificialDelayDurationForTenantsWithIDGreaterThan = model.Duration(time.Second)
+			},
+			expectedDelay: time.Second,
+		},
+		"should not apply delay based on 'tenant ID greater than' condition if tenant ID is numeric and <= the condition": {
+			tenantID: "12345",
+			tenantLimits: func(l *Limits) {
+				l.IngestionArtificialDelayConditionForTenantsWithIDGreaterThan = 12345
+				l.IngestionArtificialDelayDurationForTenantsWithIDGreaterThan = model.Duration(time.Second)
+			},
+			expectedDelay: 0,
+		},
+		"should not apply delay based on 'tenant ID greater than' condition if tenant ID is not numeric": {
+			tenantID: "tenant-123456",
+			tenantLimits: func(l *Limits) {
+				l.IngestionArtificialDelayConditionForTenantsWithIDGreaterThan = 12345
+				l.IngestionArtificialDelayDurationForTenantsWithIDGreaterThan = model.Duration(time.Second)
+			},
+			expectedDelay: 0,
+		},
+		"should apply the highest delay among matching conditions": {
+			tenantID: "12346",
+			tenantLimits: func(l *Limits) {
+				l.IngestionArtificialDelay = model.Duration(300 * time.Millisecond)
+
+				l.IngestionArtificialDelayConditionForTenantsWithLessThanMaxSeries = 15001
+				l.IngestionArtificialDelayDurationForTenantsWithLessThanMaxSeries = model.Duration(200 * time.Millisecond)
+				l.MaxGlobalSeriesPerUser = 15000
+
+				l.IngestionArtificialDelayConditionForTenantsWithIDGreaterThan = 12345
+				l.IngestionArtificialDelayDurationForTenantsWithIDGreaterThan = model.Duration(100 * time.Millisecond)
+			},
+			expectedDelay: 300 * time.Millisecond,
+		},
+	}
+
+	for testName, testData := range tests {
+		t.Run(testName, func(t *testing.T) {
+			tenantLimits := &Limits{}
+			flagext.DefaultValues(tenantLimits)
+			testData.tenantLimits(tenantLimits)
+
+			ov := NewOverrides(Limits{}, NewMockTenantLimits(map[string]*Limits{testData.tenantID: tenantLimits}))
+			require.Equal(t, testData.expectedDelay, ov.DistributorIngestionArtificialDelay(testData.tenantID))
+		})
+	}
+}
+
 func TestAlertmanagerNotificationLimits(t *testing.T) {
 	for name, tc := range map[string]struct {
 		inputYAML         string