Promtail: Ignore dropped entries in subsequent metric-stages in pipelines. (#1743)

* Adds testcase for promtail that records metrics after a "match-drop".

* Makes subsequent metrics stages in promtail ignore dropped entries.

Before this change a "metric" stage after a "match" stage with `action: drop` would result in an error.
This specifies the behavior in the docs and adds logic to ignore dropped entries.

docs/clients/promtail/stages/match.md

@@ -17,7 +17,8 @@ match:
   [pipeline_name: <string>]
 
   # Determines what action is taken when the selector matches the log 
-  # line. Defaults to keep. When set to drop, entries will be dropped. 
+  # line. Defaults to keep. When set to drop, entries will be dropped
+  # and no later metrics will be recorded.
   # Stages must be not defined when dropping entries. 
   [action: <string> | default = "keep"]
 

pkg/logentry/stages/metrics.go

@@ -144,6 +144,9 @@ type metricStage struct {
 
 // Process implements Stage
 func (m *metricStage) Process(labels model.LabelSet, extracted map[string]interface{}, t *time.Time, entry *string) {
+	if _, ok := labels[dropLabel]; ok {
+		return
+	}
 	for name, collector := range m.metrics {
 		// There is a special case for counters where we count even if there is no match in the extracted map.
 		if c, ok := collector.(*metric.Counters); ok {

pkg/logentry/stages/metrics_test.go

@@ -139,6 +139,53 @@ func TestPipelineWithMissingKey_Metrics(t *testing.T) {
 	}
 }
 
+var testMetricWithDropYaml = `
+pipeline_stages:
+- json:
+    expressions:
+      app: app
+      drop: drop
+- match:
+    selector: '{drop="true"}'
+    action: drop
+- metrics:
+    loki_count:
+      type: Counter
+      source: app
+      description: "should only inc on non dropped labels"
+      config:
+        action: inc
+`
+
+const expectedDropMetrics = `# HELP promtail_custom_loki_count should only inc on non dropped labels
+# TYPE promtail_custom_loki_count counter
+promtail_custom_loki_count 1.0
+`
+
+func TestMetricsWithDropInPipeline(t *testing.T) {
+	registry := prometheus.NewRegistry()
+	pl, err := NewPipeline(util.Logger, loadConfig(testMetricWithDropYaml), nil, registry)
+	if err != nil {
+		t.Fatal(err)
+	}
+	lbls := model.LabelSet{}
+	droppingLabels := model.LabelSet{
+		"drop": "true",
+	}
+
+	ts := time.Now()
+	extracted := map[string]interface{}{}
+	entry := testMetricLogLine1
+	pl.Process(lbls, extracted, &ts, &entry)
+	entry = testMetricLogLine2
+	pl.Process(droppingLabels, extracted, &ts, &entry)
+
+	if err := testutil.GatherAndCompare(registry,
+		strings.NewReader(expectedDropMetrics)); err != nil {
+		t.Fatalf("missmatch metrics: %v", err)
+	}
+}
+
 var metricTestInvalidIdle = "10f"
 
 func Test(t *testing.T) {