Runtime override of tenant-specific active series custom trackers

cmd/mimir/help-all.txt.tmpl

@@ -909,8 +909,6 @@ Usage of ./cmd/mimir/mimir:
     	Expected number of samples per timeseries, used for preallocations. (default 10)
   -ingester-client.expected-timeseries int
     	Expected number of timeseries per request, used for preallocations. (default 100)
-  -ingester.active-series-custom-trackers value
-    	Additional active series metrics, matching the provided matchers. Matchers should be in form <name>:<matcher>, like 'foobar:{foo="bar"}'. Multiple matchers can be provided either providing the flag multiple times or providing multiple semicolon-separated values to a single flag.
   -ingester.active-series-metrics-enabled
     	Enable tracking of active series and export them as metrics. (default true)
   -ingester.active-series-metrics-idle-timeout duration

cmd/mimir/help.txt.tmpl

@@ -371,8 +371,6 @@ Usage of ./cmd/mimir/mimir:
     	Expected number of samples per timeseries, used for preallocations. (default 10)
   -ingester-client.expected-timeseries int
     	Expected number of timeseries per request, used for preallocations. (default 100)
-  -ingester.active-series-custom-trackers value
-    	Additional active series metrics, matching the provided matchers. Matchers should be in form <name>:<matcher>, like 'foobar:{foo="bar"}'. Multiple matchers can be provided either providing the flag multiple times or providing multiple semicolon-separated values to a single flag.
   -ingester.active-series-metrics-enabled
     	Enable tracking of active series and export them as metrics. (default true)
   -ingester.availability-zone string

pkg/ingester/active_series.go

@@ -22,8 +22,9 @@ const (
 
 // ActiveSeries is keeping track of recently active series for a single tenant.
 type ActiveSeries struct {
-	asm     *ActiveSeriesMatchers
-	stripes [numActiveSeriesStripes]activeSeriesStripe
+	asm        *ActiveSeriesMatchers
+	stripes    [numActiveSeriesStripes]activeSeriesStripe
+	lastUpdate time.Time
 }
 
 // activeSeriesStripe holds a subset of the series timestamps for a single tenant.
@@ -63,6 +64,20 @@ func NewActiveSeries(asm *ActiveSeriesMatchers) *ActiveSeries {
 	return c
 }
 
+func (c *ActiveSeries) ReloadSeriesMatchers(asm *ActiveSeriesMatchers) {
+	c.asm = asm
+	for i := 0; i < numActiveSeriesStripes; i++ {
+		c.stripes[i].mu.Lock()
+		defer c.stripes[i].mu.Unlock()
+	}
+
+	for i := 0; i < numActiveSeriesStripes; i++ {
+		c.stripes[i].asm = asm
+		c.stripes[i].activeMatching = makeIntSliceIfNotEmpty(len(asm.MatcherNames()))
+	}
+	c.lastUpdate = time.Now()
+}
+
 // Updates series timestamp to 'now'. Function is called to make a copy of labels if entry doesn't exist yet.
 func (c *ActiveSeries) UpdateSeries(series labels.Labels, now time.Time, labelsCopy func(labels.Labels) labels.Labels) {
 	fp := fingerprint(series)

pkg/ingester/ingester.go

@@ -121,10 +121,10 @@ type Config struct {
 
 	RateUpdatePeriod time.Duration `yaml:"rate_update_period" category:"advanced"`
 
-	ActiveSeriesMetricsEnabled      bool                             `yaml:"active_series_metrics_enabled"`
-	ActiveSeriesMetricsUpdatePeriod time.Duration                    `yaml:"active_series_metrics_update_period" category:"advanced"`
-	ActiveSeriesMetricsIdleTimeout  time.Duration                    `yaml:"active_series_metrics_idle_timeout" category:"advanced"`
-	ActiveSeriesCustomTrackers      ActiveSeriesCustomTrackersConfig `yaml:"active_series_custom_trackers" doc:"description=Additional custom trackers for active metrics. If there are active series matching a provided matcher (map value), the count will be exposed in the custom trackers metric labeled using the tracker name (map key). Zero valued counts are not exposed (and removed when they go back to zero)."`
+	ActiveSeriesMetricsEnabled      bool                          `yaml:"active_series_metrics_enabled"`
+	ActiveSeriesMetricsUpdatePeriod time.Duration                 `yaml:"active_series_metrics_update_period" category:"advanced"`
+	ActiveSeriesMetricsIdleTimeout  time.Duration                 `yaml:"active_series_metrics_idle_timeout" category:"advanced"`
+	RuntimeMatchersConfigFn         func() *RuntimeMatchersConfig `yaml:"-"`
 
 	ExemplarsUpdatePeriod time.Duration `yaml:"exemplars_update_period" category:"experimental"`
 
@@ -153,7 +153,6 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 	f.BoolVar(&cfg.ActiveSeriesMetricsEnabled, "ingester.active-series-metrics-enabled", true, "Enable tracking of active series and export them as metrics.")
 	f.DurationVar(&cfg.ActiveSeriesMetricsUpdatePeriod, "ingester.active-series-metrics-update-period", 1*time.Minute, "How often to update active series metrics.")
 	f.DurationVar(&cfg.ActiveSeriesMetricsIdleTimeout, "ingester.active-series-metrics-idle-timeout", 10*time.Minute, "After what time a series is considered to be inactive.")
-	f.Var(&cfg.ActiveSeriesCustomTrackers, "ingester.active-series-custom-trackers", "Additional active series metrics, matching the provided matchers. Matchers should be in form <name>:<matcher>, like 'foobar:{foo=\"bar\"}'. Multiple matchers can be provided either providing the flag multiple times or providing multiple semicolon-separated values to a single flag.")
 
 	f.BoolVar(&cfg.StreamChunksWhenUsingBlocks, "ingester.stream-chunks-when-using-blocks", true, "Stream chunks from ingesters to queriers.")
 	f.DurationVar(&cfg.ExemplarsUpdatePeriod, "ingester.exemplars-update-period", 15*time.Second, "Period with which to update per-user max exemplars.")
@@ -198,8 +197,6 @@ type Ingester struct {
 	metrics *ingesterMetrics
 	logger  log.Logger
 
-	activeSeriesMatcher *ActiveSeriesMatchers
-
 	lifecycler         *ring.Lifecycler
 	limits             *validation.Overrides
 	limiter            *Limiter
@@ -237,6 +234,9 @@ type Ingester struct {
 	// Rate of pushed samples. Used to limit global samples push rate.
 	ingestionRate        *util_math.EwmaRate
 	inflightPushRequests atomic.Int64
+
+	// Last seen activeSeriesConfiguration. Needed for diff based updating.
+	runtimeMatchersConfig *RuntimeMatchersConfig
 }
 
 func newIngester(cfg Config, limits *validation.Overrides, registerer prometheus.Registerer, logger log.Logger) (*Ingester, error) {
@@ -250,18 +250,18 @@ func newIngester(cfg Config, limits *validation.Overrides, registerer prometheus
 	}
 
 	return &Ingester{
-		cfg:                 cfg,
-		limits:              limits,
-		logger:              logger,
-		activeSeriesMatcher: &ActiveSeriesMatchers{},
-
-		tsdbs:               make(map[string]*userTSDB),
-		usersMetadata:       make(map[string]*userMetricsMetadata),
-		bucket:              bucketClient,
-		tsdbMetrics:         newTSDBMetrics(registerer),
-		forceCompactTrigger: make(chan requestWithUsersAndCallback),
-		shipTrigger:         make(chan requestWithUsersAndCallback),
-		seriesHashCache:     hashcache.NewSeriesHashCache(cfg.BlocksStorageConfig.TSDB.SeriesHashCacheMaxBytes),
+		cfg:    cfg,
+		limits: limits,
+		logger: logger,
+
+		tsdbs:                 make(map[string]*userTSDB),
+		usersMetadata:         make(map[string]*userMetricsMetadata),
+		bucket:                bucketClient,
+		tsdbMetrics:           newTSDBMetrics(registerer),
+		forceCompactTrigger:   make(chan requestWithUsersAndCallback),
+		shipTrigger:           make(chan requestWithUsersAndCallback),
+		seriesHashCache:       hashcache.NewSeriesHashCache(cfg.BlocksStorageConfig.TSDB.SeriesHashCacheMaxBytes),
+		runtimeMatchersConfig: &RuntimeMatchersConfig{},
 	}, nil
 }
 
@@ -279,13 +279,7 @@ func New(cfg Config, clientConfig client.Config, limits *validation.Overrides, r
 	}
 	i.clientConfig = clientConfig
 	i.ingestionRate = util_math.NewEWMARate(0.2, instanceIngestionRateTickInterval)
-	i.metrics = newIngesterMetrics(registerer, cfg.ActiveSeriesMetricsEnabled, i.activeSeriesMatcher.MatcherNames(), i.getInstanceLimits, i.ingestionRate, &i.inflightPushRequests)
-
-	asm, err := NewActiveSeriesMatchers(cfg.ActiveSeriesCustomTrackers)
-	if err != nil {
-		return nil, err
-	}
-	i.activeSeriesMatcher = asm
+	i.metrics = newIngesterMetrics(registerer, cfg.ActiveSeriesMetricsEnabled, i.getInstanceLimits, i.ingestionRate, &i.inflightPushRequests)
 
 	// Replace specific metrics which we can't directly track but we need to read
 	// them from the underlying system (ie. TSDB).
@@ -333,7 +327,7 @@ func NewForFlusher(cfg Config, limits *validation.Overrides, registerer promethe
 	if err != nil {
 		return nil, err
 	}
-	i.metrics = newIngesterMetrics(registerer, false, nil, i.getInstanceLimits, nil, &i.inflightPushRequests)
+	i.metrics = newIngesterMetrics(registerer, false, i.getInstanceLimits, nil, &i.inflightPushRequests)
 
 	i.shipperIngesterID = "flusher"
 
@@ -465,6 +459,7 @@ func (i *Ingester) updateLoop(ctx context.Context) error {
 			i.applyExemplarsSettings()
 
 		case <-activeSeriesTickerChan:
+			i.reloadConfig(time.Now())
 			i.updateActiveSeries(time.Now())
 
 		case <-ctx.Done():
@@ -475,6 +470,66 @@ func (i *Ingester) updateLoop(ctx context.Context) error {
 	}
 }
 
+func (i *Ingester) reloadConfig(now time.Time) {
+	newConfig := i.getRuntimeMatchersConfig()
+	defaultMatchersEquals := i.runtimeMatchersConfig.DefaultMatchers.String() == newConfig.DefaultMatchers.String()
+
+	// it is crucial to only reload matchers which have been changed, as this function runs even if there is no change in config
+	for _, userID := range i.getTSDBUsers() {
+		userDB := i.getTSDB(userID)
+		if userDB == nil {
+			continue
+		}
+		oldValue, oldOk := i.runtimeMatchersConfig.TenantSpecificMatchers[userID]
+		newValue, newOk := newConfig.TenantSpecificMatchers[userID]
+		var replaceValue ActiveSeriesCustomTrackersConfig = nil
+		if !newOk && !oldOk && !defaultMatchersEquals {
+			// default changed and no overwrite
+			replaceValue = newConfig.DefaultMatchers
+		} else if newOk && !oldOk {
+			// tenant specific added
+			replaceValue = newValue
+		} else if !newOk && oldOk {
+			// tenant specific removed
+			replaceValue = newConfig.DefaultMatchers
+		} else if oldValue.String() != newValue.String() {
+			// tenant specific changed
+			replaceValue = newValue
+		}
+		if replaceValue != nil {
+			err := i.ReplaceMatchers(replaceValue, userDB)
+			if err != nil {
+				level.Error(i.logger).Log("msg", "failed to update config", "user", userID, "err", err)
+			}
+		}
+	}
+
+	i.runtimeMatchersConfig = newConfig
+}
+
+func (i *Ingester) ReplaceMatchers(config ActiveSeriesCustomTrackersConfig, userDB *userTSDB) error {
+	newMatcher, err := NewActiveSeriesMatchers(config)
+	if err != nil {
+		return err
+	}
+	i.metrics.deletePerUserCustomTrackerMetrics(userDB.userID, userDB.activeSeries.asm.names)
+	userDB.activeSeries.ReloadSeriesMatchers(newMatcher)
+	return nil
+}
+
+func (i *Ingester) getRuntimeMatchersConfig() *RuntimeMatchersConfig {
+	if i.cfg.RuntimeMatchersConfigFn == nil {
+		return &RuntimeMatchersConfig{}
+	}
+
+	r := i.cfg.RuntimeMatchersConfigFn()
+	if r == nil {
+		return &RuntimeMatchersConfig{}
+	}
+
+	return r
+}
+
 func (i *Ingester) updateActiveSeries(now time.Time) {
 	purgeTime := now.Add(-i.cfg.ActiveSeriesMetricsIdleTimeout)
 	for _, userID := range i.getTSDBUsers() {
@@ -490,14 +545,19 @@ func (i *Ingester) updateActiveSeries(now time.Time) {
 		} else {
 			i.metrics.activeSeriesPerUser.DeleteLabelValues(userID)
 		}
-		for idx, name := range i.activeSeriesMatcher.MatcherNames() {
-			// We only set the metrics for matchers that actually exist, to avoid increasing cardinality with zero valued metrics.
-			if activeMatching[idx] > 0 {
-				i.metrics.activeSeriesCustomTrackersPerUser.WithLabelValues(userID, name).Set(float64(activeMatching[idx]))
-			} else {
-				i.metrics.activeSeriesCustomTrackersPerUser.DeleteLabelValues(userID, name)
+		if userDB.activeSeries.lastUpdate.IsZero() || userDB.activeSeries.lastUpdate.After(now.Add(i.cfg.ActiveSeriesMetricsIdleTimeout)) {
+			// Do not publish metrics until the new matcher setup had time to catch up
+			// LastUpdate is Zero when it never get updated
+			for idx, name := range userDB.activeSeries.asm.names {
+				// We only set the metrics for matchers that actually exist, to avoid increasing cardinality with zero valued metrics.
+				if activeMatching[idx] > 0 {
+					i.metrics.activeSeriesCustomTrackersPerUser.WithLabelValues(userID, name).Set(float64(activeMatching[idx]))
+				} else {
+					i.metrics.activeSeriesCustomTrackersPerUser.DeleteLabelValues(userID, name)
+				}
 			}
 		}
+
 	}
 }
 
@@ -1444,10 +1504,24 @@ func (i *Ingester) createTSDB(userID string) (*userTSDB, error) {
 	userLogger := util_log.WithUserID(userID, i.logger)
 
 	blockRanges := i.cfg.BlocksStorageConfig.TSDB.BlockRanges.ToMilliseconds()
+	var activeSeriesMathers *ActiveSeriesMatchers
+	matchersCfg := i.getRuntimeMatchersConfig()
+	val, ok := matchersCfg.TenantSpecificMatchers[userID]
+	var matchers ActiveSeriesCustomTrackersConfig = nil
+	if ok {
+		matchers = val
+	} else {
+		matchers = matchersCfg.DefaultMatchers
+	}
+	activeSeriesMathers, err := NewActiveSeriesMatchers(matchers)
+	if err != nil {
+		level.Error(i.logger).Log("msg", "failed to apply runtime matchers", "user", userID, "err", err)
+		activeSeriesMathers = &ActiveSeriesMatchers{}
+	}
 
 	userDB := &userTSDB{
 		userID:              userID,
-		activeSeries:        NewActiveSeries(i.activeSeriesMatcher),
+		activeSeries:        NewActiveSeries(activeSeriesMathers),
 		seriesInMetric:      newMetricCounter(i.limiter, i.cfg.getIgnoreSeriesLimitForMetricNamesMap()),
 		ingestedAPISamples:  util_math.NewEWMARate(0.2, i.cfg.RateUpdatePeriod),
 		ingestedRuleSamples: util_math.NewEWMARate(0.2, i.cfg.RateUpdatePeriod),
@@ -1570,9 +1644,7 @@ func (i *Ingester) closeAllTSDB() {
 
 			i.metrics.memUsers.Dec()
 			i.metrics.activeSeriesPerUser.DeleteLabelValues(userID)
-			for _, name := range i.metrics.activeSeriesCustomTrackerNames {
-				i.metrics.activeSeriesCustomTrackersPerUser.DeleteLabelValues(userID, name)
-			}
+			i.metrics.deletePerUserCustomTrackerMetrics(userID, db.activeSeries.asm.names)
 		}(userDB)
 	}
 
@@ -1969,6 +2041,7 @@ func (i *Ingester) closeAndDeleteUserTSDBIfIdle(userID string) tsdbCloseCheckRes
 
 	i.deleteUserMetadata(userID)
 	i.metrics.deletePerUserMetrics(userID)
+	i.metrics.deletePerUserCustomTrackerMetrics(userID, userDB.activeSeries.asm.names)
 
 	validation.DeletePerUserValidationMetrics(userID, i.logger)