Allow a metric defined as a counter to match all lines, useful for cr…

…eating line count metrics which include all your labels.

Found a bug in the the test runner where it didn't fail if the actual error was nil but the test expected an error
Added some tests to the counters to test valid configs

Signed-off-by: Edward Welch <edward.welch@grafana.com>

docs/clients/promtail/stages/metrics.md

@@ -29,11 +29,16 @@ type: Counter
 # Defines custom prefix name for the metric. If undefined, default name "promtail_custom_" will be prefixed.
 [prefix: <string>]
 
-# Key from the extracted data map to use for the mtric,
+# Key from the extracted data map to use for the metric,
 # defaulting to the metric's name if not present.
 [source: <string>]
 
 config:
+  # If present and true all log lines will be counted without
+  # attempting to match the source to the extract map.
+  # It is an error to specify `match_all: true` and also specify a `value`
+  [match_all: <bool>]
+
   # Filters down source data and only changes the metric
   # if the targeted value exactly matches the provided string.
   # If not present, all data will match.

pkg/logentry/metric/counters.go

@@ -13,13 +13,15 @@ const (
 	CounterInc = "inc"
 	CounterAdd = "add"
 
-	ErrCounterActionRequired = "counter action must be defined as either `inc` or `add`"
-	ErrCounterInvalidAction  = "action %s is not valid, action must be either `inc` or `add`"
+	ErrCounterActionRequired  = "counter action must be defined as either `inc` or `add`"
+	ErrCounterInvalidAction   = "action %s is not valid, action must be either `inc` or `add`"
+	ErrCounterInvalidMatchAll = "`match_all: true` cannot be combined with `value`, please remove `match_all` or `value`"
 )
 
 type CounterConfig struct {
-	Value  *string `mapstructure:"value"`
-	Action string  `mapstructure:"action"`
+	MatchAll *bool   `mapstructure:"match_all"`
+	Value    *string `mapstructure:"value"`
+	Action   string  `mapstructure:"action"`
 }
 
 func validateCounterConfig(config *CounterConfig) error {
@@ -30,6 +32,9 @@ func validateCounterConfig(config *CounterConfig) error {
 	if config.Action != CounterInc && config.Action != CounterAdd {
 		return errors.Errorf(ErrCounterInvalidAction, config.Action)
 	}
+	if config.MatchAll != nil && *config.MatchAll && config.Value != nil {
+		return errors.Errorf(ErrCounterInvalidMatchAll)
+	}
 	return nil
 }
 

pkg/logentry/metric/counters_test.go

@@ -0,0 +1,62 @@
+package metric
+
+import (
+	"testing"
+
+	"github.com/pkg/errors"
+)
+
+var (
+	counterTestTrue  = true
+	counterTestFalse = false
+	counterTestVal   = "some val"
+)
+
+func Test_validateCounterConfig(t *testing.T) {
+	tests := []struct {
+		name   string
+		config CounterConfig
+		err    error
+	}{
+		{"invalid action",
+			CounterConfig{
+				Action: "del",
+			},
+			errors.Errorf(ErrCounterInvalidAction, "del"),
+		},
+		{"invalid counter match all",
+			CounterConfig{
+				MatchAll: &counterTestTrue,
+				Value:    &counterTestVal,
+				Action:   "inc",
+			},
+			errors.New(ErrCounterInvalidMatchAll),
+		},
+		{"valid",
+			CounterConfig{
+				Value:  &counterTestVal,
+				Action: "inc",
+			},
+			nil,
+		},
+		{"valid match all is false",
+			CounterConfig{
+				MatchAll: &counterTestFalse,
+				Value:    &counterTestVal,
+				Action:   "inc",
+			},
+			nil,
+		},
+	}
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+			err := validateCounterConfig(&tt.config)
+			if ((err != nil) && (err.Error() != tt.err.Error())) || (err == nil && tt.err != nil) {
+				t.Errorf("Metrics stage validation error, expected error = %v, actual error = %v", tt.err, err)
+				return
+			}
+		})
+	}
+}

pkg/logentry/stages/metrics.go

@@ -123,6 +123,13 @@ type metricStage struct {
 // Process implements Stage
 func (m *metricStage) Process(labels model.LabelSet, extracted map[string]interface{}, t *time.Time, entry *string) {
 	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 {
+			if c != nil && c.Cfg.MatchAll != nil && *c.Cfg.MatchAll {
+				m.recordCounter(name, c, labels, nil)
+				continue
+			}
+		}
 		if v, ok := extracted[*m.cfg[name].Source]; ok {
 			switch vec := collector.(type) {
 			case *metric.Counters:

pkg/logentry/stages/metrics_test.go

@@ -36,6 +36,12 @@ pipeline_stages:
       config:
         value: bloki
         action: dec
+    total_lines_count:
+      type: Counter
+      description: nothing to see here...
+      config:
+        match_all: true
+        action: inc
     payload_size_bytes:
       type: Histogram
       description: grrrragh
@@ -63,19 +69,22 @@ var testMetricLogLine2 = `
 }
 `
 
-const expectedMetrics = `# HELP promtail_custom_bloki_count blerrrgh
-# TYPE promtail_custom_bloki_count gauge
-promtail_custom_bloki_count -1.0
-# HELP my_promtail_custom_loki_count uhhhhhhh
+const expectedMetrics = `# HELP my_promtail_custom_loki_count uhhhhhhh
 # TYPE my_promtail_custom_loki_count counter
-my_promtail_custom_loki_count 1.0
+my_promtail_custom_loki_count{test="app"} 1
+# HELP promtail_custom_bloki_count blerrrgh
+# TYPE promtail_custom_bloki_count gauge
+promtail_custom_bloki_count{test="app"} -1
 # HELP promtail_custom_payload_size_bytes grrrragh
 # TYPE promtail_custom_payload_size_bytes histogram
-promtail_custom_payload_size_bytes_bucket{le="10.0"} 1.0
-promtail_custom_payload_size_bytes_bucket{le="20.0"} 2.0
-promtail_custom_payload_size_bytes_bucket{le="+Inf"} 2.0
-promtail_custom_payload_size_bytes_sum 30.0
-promtail_custom_payload_size_bytes_count 2.0
+promtail_custom_payload_size_bytes_bucket{test="app",le="10"} 1
+promtail_custom_payload_size_bytes_bucket{test="app",le="20"} 2
+promtail_custom_payload_size_bytes_bucket{test="app",le="+Inf"} 2
+promtail_custom_payload_size_bytes_sum{test="app"} 30
+promtail_custom_payload_size_bytes_count{test="app"} 2
+# HELP promtail_custom_total_lines_count nothing to see here...
+# TYPE promtail_custom_total_lines_count counter
+promtail_custom_total_lines_count{test="app"} 2
 `
 
 func TestMetricsPipeline(t *testing.T) {
@@ -85,6 +94,7 @@ func TestMetricsPipeline(t *testing.T) {
 		t.Fatal(err)
 	}
 	lbls := model.LabelSet{}
+	lbls["test"] = "app"
 	ts := time.Now()
 	extracted := map[string]interface{}{}
 	entry := testMetricLogLine1
@@ -134,7 +144,7 @@ func Test(t *testing.T) {
 		t.Run(name, func(t *testing.T) {
 			t.Parallel()
 			err := validateMetricsConfig(test.config)
-			if (err != nil) && (err.Error() != test.err.Error()) {
+			if ((err != nil) && (err.Error() != test.err.Error())) || (err == nil && test.err != nil) {
 				t.Errorf("Metrics stage validation error, expected error = %v, actual error = %v", test.err, err)
 				return
 			}