fix(pattern): Fix feature flag for enabling pattern persistence (#18215)

PR #17737 introduced a feature to persist detected patterns on the pattern ingesters. And while there is a dedicated global config `PatternPersistence` and a per-tenant config `PatternPersistenceEnabled`, none of these were honoured, because for writing the patterns to the distributor, the same writer (and its config) as for persisting aggregated metrics is used. So, when aggregated metrics is enabled, also patterns were automatically written/persisted as well.

Signed-off-by: Christian Haudum <christian.haudum@gmail.com>

Author: chaudum

pkg/pattern/ingester.go

@@ -398,12 +398,12 @@ func (i *Ingester) GetOrCreateInstance(instanceID string) (*instance, error) { /
 	inst, ok = i.instances[instanceID]
 	if !ok {
 		var err error
-		var writer aggregation.EntryWriter
+		metricAggregationMetrics := aggregation.NewMetrics(i.registerer)
 
+		var metricWriter aggregation.EntryWriter
 		aggCfg := i.cfg.MetricAggregation
 		if i.limits.MetricAggregationEnabled(instanceID) {
-			metricAggregationMetrics := aggregation.NewMetrics(i.registerer)
-			writer, err = aggregation.NewPush(
+			metricWriter, err = aggregation.NewPush(
 				aggCfg.LokiAddr,
 				instanceID,
 				aggCfg.WriteTimeout,
@@ -413,13 +413,35 @@ func (i *Ingester) GetOrCreateInstance(instanceID string) (*instance, error) { /
 				string(aggCfg.BasicAuth.Password),
 				aggCfg.UseTLS,
 				&aggCfg.BackoffConfig,
-				i.logger,
+				log.With(i.logger, "writer", "metric-aggregation"),
 				metricAggregationMetrics,
 			)
 			if err != nil {
 				return nil, err
 			}
 		}
+
+		var patternWriter aggregation.EntryWriter
+		patternCfg := i.cfg.PatternPersistence
+		if i.limits.PatternPersistenceEnabled(instanceID) {
+			metricWriter, err = aggregation.NewPush(
+				patternCfg.LokiAddr,
+				instanceID,
+				patternCfg.WriteTimeout,
+				patternCfg.PushPeriod,
+				patternCfg.HTTPClientConfig,
+				patternCfg.BasicAuth.Username,
+				string(patternCfg.BasicAuth.Password),
+				patternCfg.UseTLS,
+				&patternCfg.BackoffConfig,
+				log.With(i.logger, "writer", "pattern"),
+				metricAggregationMetrics,
+			)
+			if err != nil {
+				return nil, err
+			}
+		}
+
 		inst, err = newInstance(
 			instanceID,
 			i.logger,
@@ -428,7 +450,8 @@ func (i *Ingester) GetOrCreateInstance(instanceID string) (*instance, error) { /
 			i.limits,
 			i.ringClient,
 			i.lifecycler.ID,
-			writer,
+			metricWriter,
+			patternWriter,
 		)
 		if err != nil {
 			return nil, err
@@ -461,8 +484,8 @@ func (i *Ingester) stopWriters() {
 	instances := i.getInstances()
 
 	for _, instance := range instances {
-		if instance.writer != nil {
-			instance.writer.Stop()
+		if instance.metricWriter != nil {
+			instance.metricWriter.Stop()
 		}
 	}
 }

pkg/pattern/ingester_test.go

@@ -59,6 +59,7 @@ func TestInstancePushQuery(t *testing.T) {
 		ringClient,
 		ingesterID,
 		mockWriter,
+		mockWriter,
 	)
 	require.NoError(t, err)
 
@@ -245,6 +246,7 @@ func TestInstancePushAggregateMetrics(t *testing.T) {
 			ringClient,
 			ingesterID,
 			mockWriter,
+			mockWriter,
 		)
 		require.NoError(t, err)
 
@@ -442,14 +444,21 @@ func (m *mockEntryWriter) Stop() {
 }
 
 type fakeLimits struct {
-	Limits
-	metricAggregationEnabled bool
+	metricAggregationEnabled  bool
+	patternPersistenceEnabled bool
 }
 
+var _ drain.Limits = &fakeLimits{}
+var _ Limits = &fakeLimits{}
+
 func (f *fakeLimits) PatternIngesterTokenizableJSONFields(_ string) []string {
 	return []string{"log", "message", "msg", "msg_", "_msg", "content"}
 }
 
 func (f *fakeLimits) MetricAggregationEnabled(_ string) bool {
 	return f.metricAggregationEnabled
 }
+
+func (f *fakeLimits) PatternPersistenceEnabled(_ string) bool {
+	return f.patternPersistenceEnabled
+}

pkg/pattern/instance.go

@@ -53,7 +53,8 @@ type instance struct {
 	aggMetricsLock             sync.Mutex
 	aggMetricsByStreamAndLevel map[string]map[string]*aggregatedMetrics
 
-	writer aggregation.EntryWriter
+	metricWriter  aggregation.EntryWriter
+	patternWriter aggregation.EntryWriter
 }
 
 type aggregatedMetrics struct {
@@ -69,7 +70,8 @@ func newInstance(
 	drainLimits drain.Limits,
 	ringClient RingClient,
 	ingesterID string,
-	writer aggregation.EntryWriter,
+	metricWriter aggregation.EntryWriter,
+	patternWriter aggregation.EntryWriter,
 ) (*instance, error) {
 	index, err := index.NewBitPrefixWithShards(indexShards)
 	if err != nil {
@@ -87,7 +89,8 @@ func newInstance(
 		ringClient:                 ringClient,
 		ingesterID:                 ingesterID,
 		aggMetricsByStreamAndLevel: make(map[string]map[string]*aggregatedMetrics),
-		writer:                     writer,
+		metricWriter:               metricWriter,
+		patternWriter:              patternWriter,
 	}
 	i.mapper = ingester.NewFPMapper(i.getLabelsFromFingerprint)
 	return i, nil
@@ -233,7 +236,7 @@ func (i *instance) createStream(_ context.Context, pushReqStream logproto.Stream
 	fp := i.getHashForLabels(labels)
 	sortedLabels := i.index.Add(logproto.FromLabelsToLabelAdapters(labels), fp)
 	firstEntryLine := pushReqStream.Entries[0].Line
-	s, err := newStream(fp, sortedLabels, i.metrics, i.logger, drain.DetectLogFormat(firstEntryLine), i.instanceID, i.drainCfg, i.drainLimits, i.writer)
+	s, err := newStream(fp, sortedLabels, i.metrics, i.logger, drain.DetectLogFormat(firstEntryLine), i.instanceID, i.drainCfg, i.drainLimits, i.patternWriter)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create stream: %w", err)
 	}
@@ -339,8 +342,8 @@ func (i *instance) writeAggregatedMetrics(
 		{Name: constants.LevelLabel, Value: level},
 	}
 
-	if i.writer != nil {
-		i.writer.WriteEntry(
+	if i.metricWriter != nil {
+		i.metricWriter.WriteEntry(
 			now.Time(),
 			aggregation.AggregatedMetricEntry(now, totalBytes, totalCount, streamLbls),
 			newLbls,