fix(loki.process): Registration of stage.metric when used inside stage.match [backport]

internal/component/loki/process/process_test.go

@@ -734,26 +734,41 @@ func TestMetricsStageRefresh(t *testing.T) {
 
 	numLogsToSend := 3
 
-	cfgWithMetric := `
+	cfg := `
+	stage.match {
+		selector = "{ test = \"\" }"
+
+		stage.metrics {
+			metric.counter {
+			  name = "inner_counter"
+			  action = "inc"
+			  match_all = true
+			}
+		}
+	}
+
 	stage.metrics {
         metric.counter {
-          name = "paulin_test"
+          name = "outer_counter"
           action = "inc"
           match_all = true
         }
 	}` + forwardArgs
 
-	cfgWithMetric_Metrics := `
-	# HELP loki_process_custom_paulin_test
-	# TYPE loki_process_custom_paulin_test counter
-	loki_process_custom_paulin_test{filename="/var/log/pods/agent/agent/1.log",foo="bar"} %d
+	expectedMetrics := `
+	# HELP loki_process_custom_inner_counter
+	# TYPE loki_process_custom_inner_counter counter
+	loki_process_custom_inner_counter{filename="/var/log/pods/agent/agent/1.log",foo="bar"} %d
+	# HELP loki_process_custom_outer_counter
+	# TYPE loki_process_custom_outer_counter counter
+	loki_process_custom_outer_counter{filename="/var/log/pods/agent/agent/1.log",foo="bar"} %d
 	`
 
 	// The component will be reconfigured so that it has a metric.
 	t.Run("config with a metric", func(t *testing.T) {
-		tester.updateAndTest(numLogsToSend, cfgWithMetric,
+		tester.updateAndTest(numLogsToSend, cfg,
 			"",
-			fmt.Sprintf(cfgWithMetric_Metrics, numLogsToSend))
+			fmt.Sprintf(expectedMetrics, numLogsToSend, numLogsToSend))
 	})
 
 	// The component will be "updated" with the same config.
@@ -763,9 +778,9 @@ func TestMetricsStageRefresh(t *testing.T) {
 	// which reloads the collector config every X seconds.
 	// Those users wouldn't expect their metrics to be reset every time the config is reloaded.
 	t.Run("config with the same metric", func(t *testing.T) {
-		tester.updateAndTest(numLogsToSend, cfgWithMetric,
-			fmt.Sprintf(cfgWithMetric_Metrics, numLogsToSend),
-			fmt.Sprintf(cfgWithMetric_Metrics, 2*numLogsToSend))
+		tester.updateAndTest(numLogsToSend, cfg,
+			fmt.Sprintf(expectedMetrics, numLogsToSend, numLogsToSend),
+			fmt.Sprintf(expectedMetrics, 2*numLogsToSend, 2*numLogsToSend))
 	})
 
 	// Use a config which has no metrics stage.
@@ -779,33 +794,49 @@ func TestMetricsStageRefresh(t *testing.T) {
 	// as well as a metric with the same name as the one in the previous config.
 	// We try having a metric with the same name as before so that we can see if there
 	// is some sort of double registration error for that metric.
-	cfgWithTwoMetrics := `
+	updatedCfg := `
+	stage.match {
+		selector = "{ test = \"\" }"
+
+		stage.metrics {
+			metric.counter {
+			  name = "inner_counter"
+			  action = "inc"
+			  match_all = true
+			}
+		}
+	}
+
 	stage.metrics {
 		metric.counter {
-		  name = "paulin_test_3"
+		  name = "outer_counter"
 		  action = "inc"
 		  match_all = true
 		}
+
         metric.counter {
-          name = "paulin_test"
+          name = "outer_counter_2"
           action = "inc"
           match_all = true
         }
 	}` + forwardArgs
 
 	expectedMetrics3 := `
-	# HELP loki_process_custom_paulin_test_3
-	# TYPE loki_process_custom_paulin_test_3 counter
-	loki_process_custom_paulin_test_3{filename="/var/log/pods/agent/agent/1.log",foo="bar"} %d
-	# HELP loki_process_custom_paulin_test
-	# TYPE loki_process_custom_paulin_test counter
-	loki_process_custom_paulin_test{filename="/var/log/pods/agent/agent/1.log",foo="bar"} %d
+	# HELP loki_process_custom_inner_counter
+	# TYPE loki_process_custom_inner_counter counter
+	loki_process_custom_inner_counter{filename="/var/log/pods/agent/agent/1.log",foo="bar"} %d
+	# HELP loki_process_custom_outer_counter
+	# TYPE loki_process_custom_outer_counter counter
+	loki_process_custom_outer_counter{filename="/var/log/pods/agent/agent/1.log",foo="bar"} %d
+	# HELP loki_process_custom_outer_counter_2
+	# TYPE loki_process_custom_outer_counter_2 counter
+	loki_process_custom_outer_counter_2{filename="/var/log/pods/agent/agent/1.log",foo="bar"} %d
 	`
 
 	t.Run("config with a new and old metric", func(t *testing.T) {
-		tester.updateAndTest(numLogsToSend, cfgWithTwoMetrics,
+		tester.updateAndTest(numLogsToSend, updatedCfg,
 			"",
-			fmt.Sprintf(expectedMetrics3, numLogsToSend, numLogsToSend))
+			fmt.Sprintf(expectedMetrics3, numLogsToSend, numLogsToSend, numLogsToSend))
 	})
 }
 

internal/component/loki/process/stages/match.go

@@ -97,7 +97,7 @@ func newMatcherStage(logger log.Logger, jobName *string, config MatchConfig, reg
 		dropReason: dropReason,
 		dropCount:  getDropCountMetric(registerer),
 		matchers:   selector.Matchers(),
-		stage:      pl,
+		pipeline:   pl,
 		action:     config.Action,
 		filter:     filter,
 	}, nil
@@ -126,7 +126,7 @@ type matcherStage struct {
 	dropCount  *prometheus.CounterVec
 	matchers   []*labels.Matcher
 	filter     logql.Filter
-	stage      Stage
+	pipeline   *Pipeline
 	action     string
 }
 
@@ -143,7 +143,7 @@ func (m *matcherStage) Run(in chan Entry) chan Entry {
 func (m *matcherStage) runKeep(in chan Entry) chan Entry {
 	next := make(chan Entry)
 	out := make(chan Entry)
-	outNext := m.stage.Run(next)
+	outNext := m.pipeline.Run(next)
 	go func() {
 		defer close(out)
 		for e := range outNext {
@@ -192,12 +192,12 @@ func (m *matcherStage) processLogQL(e Entry) (Entry, bool) {
 	return e, false
 }
 
-// Name implements Stage
 func (m *matcherStage) Name() string {
 	return StageTypeMatch
 }
 
-// Cleanup implements Stage.
-func (*matcherStage) Cleanup() {
-	// no-op
+func (m *matcherStage) Cleanup() {
+	if m.pipeline != nil {
+		m.pipeline.Cleanup()
+	}
 }