fix(limits): Adapt defaults and expose evict interval (#17808)

Author: periklis

docs/sources/shared/configuration.md

@@ -911,7 +911,7 @@ ingest_limits:
   # have not been updated within this window are considered inactive and not
   # counted towards limits.
   # CLI flag: -ingest-limits.active-window
-  [active_window: <duration> | default = 1h]
+  [active_window: <duration> | default = 2h]
 
   # The time window for rate calculation. This should match the window used in
   # Prometheus rate() queries for consistency.
@@ -923,6 +923,10 @@ ingest_limits:
   # CLI flag: -ingest-limits.bucket-size
   [bucket_size: <duration> | default = 1m]
 
+  # The interval at which old streams are evicted.
+  # CLI flag: -ingest-limits.eviction-interval
+  [eviction_interval: <duration> | default = 30m]
+
   # The number of partitions for the Kafka topic used to read and write stream
   # metadata. It is fixed, not a maximum.
   # CLI flag: -ingest-limits.num-partitions

pkg/limits/config.go

@@ -12,9 +12,10 @@ import (
 )
 
 const (
-	DefaultActiveWindow  = 1 * time.Hour
+	DefaultActiveWindow  = 2 * time.Hour
 	DefaultRateWindow    = 5 * time.Minute
 	DefaultBucketSize    = 1 * time.Minute
+	DefaultEvictInterval = 30 * time.Minute
 	DefaultNumPartitions = 64
 )
 
@@ -38,6 +39,9 @@ type Config struct {
 	// memory.
 	BucketSize time.Duration `yaml:"bucket_size"`
 
+	// EvictionInterval defines the interval at which old streams are evicted.
+	EvictionInterval time.Duration `yaml:"eviction_interval"`
+
 	// The number of partitions for the Kafka topic used to read and write stream metadata.
 	// It is fixed, not a maximum.
 	NumPartitions int `yaml:"num_partitions"`
@@ -77,6 +81,12 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 		DefaultBucketSize,
 		"The size of the buckets used to calculate stream rates. Smaller buckets provide more precise rates but require more memory.",
 	)
+	f.DurationVar(
+		&cfg.EvictionInterval,
+		"ingest-limits.eviction-interval",
+		DefaultEvictInterval,
+		"The interval at which old streams are evicted.",
+	)
 	f.IntVar(
 		&cfg.NumPartitions,
 		"ingest-limits.num-partitions",
@@ -98,6 +108,9 @@ func (cfg *Config) Validate() error {
 	if cfg.RateWindow < cfg.BucketSize {
 		return errors.New("rate-window must be greater than or equal to bucket-size")
 	}
+	if cfg.EvictionInterval <= 0 {
+		return errors.New("eviction-interval must be greater than 0")
+	}
 	if cfg.NumPartitions <= 0 {
 		return errors.New("num-partitions must be greater than 0")
 	}

pkg/limits/service.go

@@ -321,7 +321,7 @@ func (s *Service) running(ctx context.Context) error {
 // evictOldStreamsPeriodic runs a periodic job that evicts old streams.
 // It runs two evictions per window size.
 func (s *Service) evictOldStreamsPeriodic(ctx context.Context) {
-	ticker := time.NewTicker(s.cfg.ActiveWindow / 2)
+	ticker := time.NewTicker(s.cfg.EvictionInterval)
 	defer ticker.Stop()
 	for {
 		select {

pkg/limits/store_test.go

@@ -179,7 +179,7 @@ func TestUsageStore_Evict(t *testing.T) {
 	s.clock = clock
 	s1 := streamUsage{hash: 0x1, lastSeenAt: clock.Now().UnixNano()}
 	s.set("tenant1", s1)
-	s2 := streamUsage{hash: 0x2, lastSeenAt: clock.Now().Add(-61 * time.Minute).UnixNano()}
+	s2 := streamUsage{hash: 0x2, lastSeenAt: clock.Now().Add(-121 * time.Minute).UnixNano()}
 	s.set("tenant1", s2)
 	s3 := streamUsage{hash: 0x3, lastSeenAt: clock.Now().UnixNano()}
 	s.set("tenant2", s3)