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

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,