Allow specifying multiple sets of filters. (#113)

This allows specifying multiple sets of filters that time series will be
checked against. It also switches to using PromQL native parser for
filters.

Existing `--filter` flag will keep working, but is now marked as
deprecated.

Author: knyar

.gitignore

@@ -1,2 +1,3 @@
 # Ignore checkpoint file.
 stackdriver_sidecar.json
+stackdriver-prometheus-sidecar

README.md

@@ -48,15 +48,17 @@ stackdriver-prometheus-sidecar --help
 
 #### Filters
 
-The `--filter` flag allows to provide filters which all series have to pass before being sent to Stackdriver. The flag may be repeated to provide several filters. Filters use the same syntax as the well-known PromQL label matchers, e.g.:
+The `--include` flag allows to provide filters which all series have to pass before being sent to Stackdriver. Filters use the same syntax as [Prometheus instant vector selectors](https://prometheus.io/docs/prometheus/latest/querying/basics/#instant-vector-selectors), e.g.:
 
 ```
-stackdriver-prometheus-sidecar --filter=job="k8s" --filter=__name__!~"cadvisor_.+" ...
+stackdriver-prometheus-sidecar --include='{__name__!~"cadvisor_.+",job="k8s"}' ...
 ```
 
 This drops all series which do not have a `job` label `k8s` and all metrics that have a name starting with `cadvisor_`.
 
-Note: On the command-line shell you may need to escape filters as `--filter='job="k8s"'`.
+For equality filter on metric name you can use the simpler notation, e.g. `--include='metric_name{label="foo"}'`.
+
+The flag may be repeated to provide several sets of filters, in which case the metric will be forwarded if it matches at least one of them.
 
 #### File
 

cmd/stackdriver-prometheus-sidecar/main.go

@@ -25,7 +25,6 @@ import (
 	"os/signal"
 	"path"
 	"path/filepath"
-	"regexp"
 	"runtime"
 	"strings"
 	"syscall"
@@ -52,6 +51,7 @@ import (
 	"github.com/prometheus/prometheus/config"
 	"github.com/prometheus/prometheus/pkg/labels"
 	"github.com/prometheus/prometheus/pkg/textparse"
+	"github.com/prometheus/prometheus/promql"
 	"github.com/prometheus/prometheus/scrape"
 	oc_prometheus "go.opencensus.io/exporter/prometheus"
 	"go.opencensus.io/plugin/ocgrpc"
@@ -183,6 +183,7 @@ func main() {
 		prometheusURL      *url.URL
 		listenAddress      string
 		filters            []string
+		filtersets         []string
 		metricRenames      map[string]string
 		staticMetadata     []scrape.MetricMetadata
 		monitoringBackends []string
@@ -236,7 +237,10 @@ func main() {
 	a.Flag("web.listen-address", "Address to listen on for UI, API, and telemetry.").
 		Default("0.0.0.0:9091").StringVar(&cfg.listenAddress)
 
-	a.Flag("filter", "PromQL-style label matcher which must pass for a series to be forwarded to Stackdriver. May be repeated.").
+	a.Flag("include", "PromQL metric and label matcher which must pass for a series to be forwarded to Stackdriver. If repeated, the series must pass any of the filter sets to be forwarded.").
+		StringsVar(&cfg.filtersets)
+
+	a.Flag("filter", "PromQL-style matcher for a single label which must pass for a series to be forwarded to Stackdriver. If repeated, the series must pass all filters to be forwarded. Deprecated, please use --include instead.").
 		StringsVar(&cfg.filters)
 
 	promlogflag.AddFlags(a, &cfg.logLevel)
@@ -309,9 +313,9 @@ func main() {
 		}
 	}
 
-	filters, err := parseFilters(cfg.filters...)
+	filtersets, err := parseFiltersets(logger, cfg.filtersets, cfg.filters)
 	if err != nil {
-		level.Error(logger).Log("msg", "Error parsing filters", "err", err)
+		level.Error(logger).Log("msg", "Error parsing --include (or --filter)", "err", err)
 		os.Exit(2)
 	}
 
@@ -368,7 +372,7 @@ func main() {
 		log.With(logger, "component", "Prometheus reader"),
 		cfg.walDirectory,
 		tailer,
-		filters,
+		filtersets,
 		cfg.metricRenames,
 		retrieval.TargetsWithDiscoveredLabels(targetCache, labels.FromMap(staticLabels)),
 		metadataCache,
@@ -550,32 +554,25 @@ func waitForPrometheus(ctx context.Context, logger log.Logger, promURL *url.URL)
 	}
 }
 
-// parseFilters parses a list of strings that contain PromQL-style label matchers and
+// parseFiltersets parses two flags that contain PromQL-style metric/label selectors and
 // returns a list of the resulting matchers.
-func parseFilters(strs ...string) (matchers []*labels.Matcher, err error) {
-	pattern := regexp.MustCompile(`^([a-zA-Z0-9_]+)(=|!=|=~|!~)"(.+)"$`)
-
-	for _, s := range strs {
-		parts := pattern.FindStringSubmatch(s)
-		if len(parts) != 4 {
-			return nil, fmt.Errorf("invalid filter %q", s)
-		}
-		var matcherType labels.MatchType
-		switch parts[2] {
-		case "=":
-			matcherType = labels.MatchEqual
-		case "!=":
-			matcherType = labels.MatchNotEqual
-		case "=~":
-			matcherType = labels.MatchRegexp
-		case "!~":
-			matcherType = labels.MatchNotRegexp
+func parseFiltersets(logger log.Logger, filtersets, filters []string) ([][]*labels.Matcher, error) {
+	var matchers [][]*labels.Matcher
+	if len(filters) > 0 {
+		level.Warn(logger).Log("msg", "--filter is deprecated; please use --include instead")
+		f := fmt.Sprintf("{%s}", strings.Join(filters, ","))
+		m, err := promql.ParseMetricSelector(f)
+		if err != nil {
+			return nil, errors.Errorf("cannot parse --filter flag (metric filter '%s'): %q", f, err)
 		}
-		matcher, err := labels.NewMatcher(matcherType, parts[1], parts[3])
+		matchers = append(matchers, m)
+	}
+	for _, f := range filtersets {
+		m, err := promql.ParseMetricSelector(f)
 		if err != nil {
-			return nil, fmt.Errorf("invalid filter %q: %s", s, err)
+			return nil, errors.Errorf("cannot parse --include flag '%s': %q", f, err)
 		}
-		matchers = append(matchers, matcher)
+		matchers = append(matchers, m)
 	}
 	return matchers, nil
 }

cmd/stackdriver-prometheus-sidecar/main_test.go

@@ -23,6 +23,8 @@ import (
 	"path/filepath"
 	"testing"
 	"time"
+
+	"github.com/go-kit/kit/log"
 )
 
 var promPath string
@@ -114,40 +116,58 @@ Loop:
 	}
 }
 
-func TestParseFilters(t *testing.T) {
-	input := []string{
-		`__name__="test1"`,
-		`a1=~"test2.+"`,
-		`a2!="test3"`,
-		`a3!~"test4.*"`,
-	}
-	// Test success cases.
-	filters, err := parseFilters(input...)
-	if err != nil {
-		t.Fatal(err)
-	}
-	// We verify by comparing the serializiation produced by the parsed matchers again.
-	// Deep equal comparison doesn't work since some fields deep down in the produced regexes differ
-	// even though everything being equal semantically.
-	if len(filters) != len(input) {
-		t.Fatalf("unexpected result length %d", len(filters))
-	}
-	for i, f := range filters {
-		if f.String() != input[i] {
-			t.Fatalf("unexpected parsed filter %v, want %v", f, input[i])
-		}
+func TestParseWhitelists(t *testing.T) {
+	logger := log.NewLogfmtLogger(log.NewSyncWriter(os.Stderr))
+	for _, tt := range []struct {
+		name         string
+		filtersets   []string
+		filters      []string
+		wantMatchers int
+	}{
+		{"both filters and filtersets defined",
+			[]string{
+				`metric_name`,
+				`metric_name{label="value"}`,
+			},
+			[]string{
+				`__name__="test1"`,
+				`a1=~"test2.+"`,
+				`a2!="test3"`,
+				`a3!~"test4.*"`,
+			}, 3},
+		{"just filtersets", []string{"metric_name"}, []string{}, 1},
+		{"just filters", []string{}, []string{`__name__="foo"`}, 1},
+		{"neither filtersets nor filters", []string{}, []string{}, 0},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			// Test success cases.
+			parsed, err := parseFiltersets(logger, tt.filtersets, tt.filters)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if len(parsed) != tt.wantMatchers {
+				t.Fatalf("expected %d matchers; got %d", tt.wantMatchers, len(parsed))
+			}
+		})
 	}
+
 	// Test failure cases.
-	cases := []string{
-		`a-b="1"`, // Invalid character in key.
-		`a="1`,    // Missing trailing quote.
-		`a=1"`,    // Missing leading quote.
-		`a!=="1"`, // Invalid operator.
-	}
-	for _, c := range cases {
-		if _, err := parseFilters(c); err == nil {
-			t.Fatalf("expected error for %q but got none", c)
-		}
+	for _, tt := range []struct {
+		name       string
+		filtersets []string
+		filters    []string
+	}{
+		{"Invalid character in key", []string{}, []string{`a-b="1"`}},
+		{"Missing trailing quote", []string{}, []string{`a="1`}},
+		{"Missing leading quote", []string{}, []string{`a=1"`}},
+		{"Invalid operator", []string{}, []string{`a!=="1"`}},
+		{"Invalid operator in filterset", []string{`{a!=="1"}`}, []string{}},
+		{"Empty filterset", []string{""}, []string{}},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			if _, err := parseFiltersets(logger, tt.filtersets, tt.filters); err == nil {
+				t.Fatalf("expected error, but got none")
+			}
+		})
 	}
-
 }

retrieval/manager.go

@@ -80,7 +80,7 @@ func NewPrometheusReader(
 	logger log.Logger,
 	walDirectory string,
 	tailer *tail.Tailer,
-	filters []*labels.Matcher,
+	filtersets [][]*labels.Matcher,
 	metricRenames map[string]string,
 	targetGetter TargetGetter,
 	metadataGetter MetadataGetter,
@@ -95,7 +95,7 @@ func NewPrometheusReader(
 		appender:             appender,
 		logger:               logger,
 		tailer:               tailer,
-		filters:              filters,
+		filtersets:           filtersets,
 		walDirectory:         walDirectory,
 		targetGetter:         targetGetter,
 		metadataGetter:       metadataGetter,
@@ -110,7 +110,7 @@ type PrometheusReader struct {
 	logger               log.Logger
 	walDirectory         string
 	tailer               *tail.Tailer
-	filters              []*labels.Matcher
+	filtersets           [][]*labels.Matcher
 	metricRenames        map[string]string
 	targetGetter         TargetGetter
 	metadataGetter       MetadataGetter
@@ -152,7 +152,7 @@ func (r *PrometheusReader) Run(ctx context.Context, startOffset int) error {
 	seriesCache := newSeriesCache(
 		r.logger,
 		r.walDirectory,
-		r.filters,
+		r.filtersets,
 		r.metricRenames,
 		r.targetGetter,
 		r.metadataGetter,

retrieval/series_cache.go

@@ -75,7 +75,7 @@ type seriesGetter interface {
 type seriesCache struct {
 	logger         log.Logger
 	dir            string
-	filters        []*promlabels.Matcher
+	filtersets     [][]*promlabels.Matcher
 	targets        TargetGetter
 	metadata       MetadataGetter
 	resourceMaps   []ResourceMap
@@ -129,7 +129,7 @@ func (e *seriesCacheEntry) shouldRefresh() bool {
 func newSeriesCache(
 	logger log.Logger,
 	dir string,
-	filters []*promlabels.Matcher,
+	filtersets [][]*promlabels.Matcher,
 	renames map[string]string,
 	targets TargetGetter,
 	metadata MetadataGetter,
@@ -143,7 +143,7 @@ func newSeriesCache(
 	return &seriesCache{
 		logger:         logger,
 		dir:            dir,
-		filters:        filters,
+		filtersets:     filtersets,
 		targets:        targets,
 		metadata:       metadata,
 		resourceMaps:   resourceMaps,
@@ -303,10 +303,8 @@ func (c *seriesCache) getResetAdjusted(ref uint64, t int64, v float64) (int64, f
 // set the label set for the given reference.
 // maxSegment indicates the the highest segment at which the series was possibly defined.
 func (c *seriesCache) set(ctx context.Context, ref uint64, lset labels.Labels, maxSegment int) error {
-	for _, f := range c.filters {
-		if v := lset.Get(f.Name); !f.Matches(v) {
-			return nil
-		}
+	if c.filtersets != nil && !matchFiltersets(lset, c.filtersets) {
+		return nil
 	}
 	c.mtx.Lock()
 	c.entries[ref] = &seriesCacheEntry{
@@ -470,3 +468,24 @@ func (c *seriesCache) getResource(discovered, final promlabels.Labels) (*monitor
 	}
 	return nil, false
 }
+
+// matchFiltersets checks whether any of the supplied filtersets passes.
+func matchFiltersets(lset labels.Labels, filtersets [][]*promlabels.Matcher) bool {
+	for _, fs := range filtersets {
+		if matchFilterset(lset, fs) {
+			return true
+		}
+	}
+	return false
+}
+
+// matchFilterset checks whether labels match a given list of label matchers.
+// All matchers need to match for the function to return true.
+func matchFilterset(lset labels.Labels, filterset []*promlabels.Matcher) bool {
+	for _, matcher := range filterset {
+		if !matcher.Matches(lset.Get(matcher.Name)) {
+			return false
+		}
+	}
+	return true
+}

retrieval/series_cache_test.go

@@ -416,30 +416,37 @@ func TestSeriesCache_Filter(t *testing.T) {
 		}
 	}()
 	logger := log.NewLogfmtLogger(logBuffer)
-	c := newSeriesCache(logger, "", []*promlabels.Matcher{
-		&promlabels.Matcher{Type: promlabels.MatchEqual, Name: "a", Value: "a1"},
-		&promlabels.Matcher{Type: promlabels.MatchEqual, Name: "b", Value: "b1"},
+	c := newSeriesCache(logger, "", [][]*promlabels.Matcher{
+		{
+			&promlabels.Matcher{Type: promlabels.MatchEqual, Name: "a", Value: "a1"},
+			&promlabels.Matcher{Type: promlabels.MatchEqual, Name: "b", Value: "b1"},
+		},
+		{&promlabels.Matcher{Type: promlabels.MatchEqual, Name: "c", Value: "c1"}},
 	}, nil, targetMap, metadataMap, resourceMaps, "", false)
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
-	// Test base case of metric that passes all filters. This primarily
-	// ensures that our setup is correct and metrics aren't dropped for reasons
-	// other than the filter.
-	err := c.set(ctx, 1, labels.FromStrings("__name__", "metric1", "job", "job1", "instance", "inst1", "a", "a1", "b", "b1"), 1)
-	if err != nil {
-		t.Fatal(err)
+	// Test that metrics that pass a single filterset do not get dropped.
+	lsets := []labels.Labels{
+		labels.FromStrings("__name__", "metric1", "job", "job1", "instance", "inst1", "a", "a1", "b", "b1"),
+		labels.FromStrings("__name__", "metric1", "job", "job1", "instance", "inst1", "c", "c1"),
 	}
-	if _, ok, err := c.get(ctx, 1); !ok || err != nil {
-		t.Fatalf("metric not found: %s", err)
+	for idx, lset := range lsets {
+		err := c.set(ctx, uint64(idx), lset, 1)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if _, ok, err := c.get(ctx, uint64(idx)); !ok || err != nil {
+			t.Fatalf("metric not found: %s", err)
+		}
 	}
 	// Test filtered metric.
-	err = c.set(ctx, 2, labels.FromStrings("__name__", "metric1", "job", "job1", "instance", "inst1", "a", "a1", "b", "b2"), 1)
+	err := c.set(ctx, 100, labels.FromStrings("__name__", "metric1", "job", "job1", "instance", "inst1", "a", "a1", "b", "b2"), 1)
 	if err != nil {
 		t.Fatal(err)
 	}
-	if _, ok, err := c.get(ctx, 2); err != nil {
+	if _, ok, err := c.get(ctx, 100); err != nil {
 		t.Fatalf("error retrieving metric: %s", err)
 	} else if ok {
 		t.Fatalf("metric was not filtered")