add match rule (#64)

Author: Gallardot

config.go

@@ -15,6 +15,10 @@
 package main
 
 import (
+	"fmt"
+	dto "github.com/prometheus/client_model/go"
+	"github.com/prometheus/common/expfmt"
+	"gopkg.in/yaml.v2"
 	"os"
 	"strings"
 	"text/template"
@@ -26,6 +30,7 @@ var (
 	kafkaBrokerList        = "kafka:9092"
 	kafkaTopic             = "metrics"
 	topicTemplate          *template.Template
+	match                  = make(map[string]*dto.MetricFamily, 0)
 	basicauth              = false
 	basicauthUsername      = ""
 	basicauthPassword      = ""
@@ -107,6 +112,14 @@ func init() {
 		kafkaSaslPassword = value
 	}
 
+	if value := os.Getenv("MATCH"); value != "" {
+		matchList, err := parseMatchList(value)
+		if err != nil {
+			logrus.WithError(err).Fatalln("couldn't parse the match rules")
+		}
+		match = matchList
+	}
+
 	var err error
 	serializer, err = parseSerializationFormat(os.Getenv("SERIALIZATION_FORMAT"))
 	if err != nil {
@@ -119,6 +132,27 @@ func init() {
 	}
 }
 
+func parseMatchList(text string) (map[string]*dto.MetricFamily, error) {
+	var matchRules []string
+	err := yaml.Unmarshal([]byte(text), &matchRules)
+	if err != nil {
+		return nil, err
+	}
+	var metricsList []string
+	for _, v := range matchRules {
+		metricsList = append(metricsList, fmt.Sprintf("%s 0\n", v))
+	}
+
+	metricsText := strings.Join(metricsList, "")
+
+	var parser expfmt.TextParser
+	metricFamilies, err := parser.TextToMetricFamilies(strings.NewReader(metricsText))
+	if err != nil {
+		return nil, fmt.Errorf("couldn't parse match rules: %s", err)
+	}
+	return metricFamilies, nil
+}
+
 func parseLogLevel(value string) logrus.Level {
 	level, err := logrus.ParseLevel(value)
 

go.mod

@@ -13,7 +13,7 @@ require (
 	github.com/linkedin/goavro v2.1.0+incompatible
 	github.com/matttproud/golang_protobuf_extensions v1.0.1 // indirect
 	github.com/prometheus/client_golang v0.8.0
-	github.com/prometheus/client_model v0.2.0 // indirect
+	github.com/prometheus/client_model v0.2.0
 	github.com/prometheus/common v0.0.0-20180801064454-c7de2306084e
 	github.com/prometheus/procfs v0.1.3 // indirect
 	github.com/prometheus/prometheus v2.4.2+incompatible
@@ -23,6 +23,7 @@ require (
 	google.golang.org/grpc v1.15.0 // indirect
 	gopkg.in/alecthomas/kingpin.v2 v2.2.6 // indirect
 	gopkg.in/linkedin/goavro.v1 v1.0.5 // indirect
+	gopkg.in/yaml.v2 v2.2.8
 )
 
 go 1.13

helm/prometheus-kafka-adapter/templates/deployment.yaml

@@ -64,6 +64,10 @@ spec:
                 name: {{ include "prometheus-kafka-adapter.fullname" . }}
                 key: BASIC_AUTH_PASSWORD
           {{- end }}
+          {{- if .Values.environment.MATCH }}
+          - name: MATCH
+            value: {{ .Values.environment.MATCH | quote }}
+          {{- end }}
           - name: LOG_LEVEL
             value: {{ .Values.environment.LOG_LEVEL | quote }}
           - name: GIN_MODE

helm/prometheus-kafka-adapter/values.yaml

@@ -63,6 +63,8 @@ environment:
   # Kafka SSL broker CA certificate file, defaults to "/ca_cert/ssl_ca_cert.pem" if
   # KAFKA_SSL_CA_CERT is provided, "" otherwise
   KAFKA_SSL_CA_CERT_FILE:
+  # defines the match rules, simple metric name match and label match
+  MATCH:
 
 serviceAccount:
   # Specifies whether a service account should be created

serializers.go

@@ -47,12 +47,16 @@ func Serialize(s Serializer, req *prompb.WriteRequest) (map[string][][]byte, err
 		t := topic(labels)
 
 		for _, sample := range ts.Samples {
-			epoch := time.Unix(sample.Timestamp/1000, 0).UTC()
+			name := string(labels["__name__"])
+			if !filter(name, labels) {
+				continue
+			}
 
+			epoch := time.Unix(sample.Timestamp/1000, 0).UTC()
 			m := map[string]interface{}{
 				"timestamp": epoch.Format(time.RFC3339),
 				"value":     strconv.FormatFloat(sample.Value, 'f', -1, 64),
-				"name":      string(labels["__name__"]),
+				"name":      name,
 				"labels":    labels,
 			}
 
@@ -115,3 +119,33 @@ func topic(labels map[string]string) string {
 	}
 	return buf.String()
 }
+
+func filter(name string, labels map[string]string) bool {
+	if len(match) == 0 {
+		return true
+	}
+	mf, ok := match[name]
+	if !ok {
+		return false
+	}
+
+	for _, m := range mf.Metric {
+		if len(m.Label) == 0 {
+			return true
+		}
+
+		labelMatch := true
+		for _, label := range m.Label {
+			val, ok := labels[label.GetName()]
+			if !ok || val != label.GetValue() {
+				labelMatch = false
+				break
+			}
+		}
+
+		if labelMatch {
+			return true
+		}
+	}
+	return false
+}

serializers_test.go

@@ -100,6 +100,38 @@ func TestTemplatedTopic(t *testing.T) {
 	}
 }
 
+func TestFilter(t *testing.T) {
+	rulesText := `['foo{y="2"}','foo', 'bar{x="1"}',
+'up{x="1",y="2"}', 'baz{key="valu
+e1;value2"}','bar{y="2"}']`
+
+	rules, _ := parseMatchList(rulesText)
+	for _, mf := range rules {
+		match[mf.GetName()] = mf
+	}
+	type TestCase struct {
+		Name   string
+		Labels map[string]string
+		Expect bool
+	}
+
+	testList := []TestCase{
+		{Name: "foo", Labels: map[string]string{"z": "3"}, Expect: true},
+		{Name: "bar", Labels: map[string]string{"x": "1"}, Expect: true},
+		{Name: "bar", Labels: map[string]string{"x": "2"}, Expect: false},
+		{Name: "bar", Labels: map[string]string{"y": "2"}, Expect: true},
+		{Name: "bar", Labels: map[string]string{"y": "1"}, Expect: false},
+		{Name: "up", Labels: map[string]string{"x": "1", "y": "2"}, Expect: true},
+		{Name: "up", Labels: map[string]string{"x": "1", "y": "2", "z": "3"}, Expect: true},
+		{Name: "up", Labels: map[string]string{"x": "2", "y": "1"}, Expect: false},
+		{Name: "go", Labels: map[string]string{"x": "1", "y": "2"}, Expect: false},
+	}
+
+	for _, tcase := range testList {
+		assert.Equal(t, tcase.Expect, filter(tcase.Name, tcase.Labels))
+	}
+}
+
 func BenchmarkSerializeToAvroJSON(b *testing.B) {
 	serializer, _ := NewAvroJSONSerializer("schemas/metric.avsc")
 	writeRequest := NewWriteRequest()