[36] initial support for tagged graphite metrics

main.go

@@ -24,6 +24,7 @@ import (
 	_ "net/http/pprof"
 	"os"
 	"regexp"
+	"sort"
 	"strconv"
 	"strings"
 	"sync"
@@ -63,13 +64,60 @@ var (
 			Help: "How long in seconds a metric sample is valid for.",
 		},
 	)
+	tagParseFailures = prometheus.NewCounter(
+		prometheus.CounterOpts{
+			Name: "graphite_tag_parse_failures",
+			Help: "Total count of samples with invalid tags",
+		})
+	invalidMetrics = prometheus.NewCounter(
+		prometheus.CounterOpts{
+			Name: "graphite_invalid_metrics",
+			Help: "Total count of metrics dropped due to mismatched label keys",
+		})
 	invalidMetricChars = regexp.MustCompile("[^a-zA-Z0-9_:]")
+
+	metricNameKeysIndex = newMetricNameAndKeys()
 )
 
+// metricNameAndKeys is a cache of metric names and the label keys previously used
+type metricNameAndKeys struct {
+	mtx   sync.Mutex
+	cache map[string]string
+}
+
+func newMetricNameAndKeys() *metricNameAndKeys {
+	x := metricNameAndKeys{
+		cache: make(map[string]string),
+	}
+	return &x
+}
+
+func keysFromLabels(labels prometheus.Labels) string {
+	labelKeys := make([]string, len(labels))
+	for k, _ := range labels {
+		labelKeys = append(labelKeys, k)
+	}
+	sort.Strings(labelKeys)
+	return strings.Join(labelKeys, ",")
+}
+
+// checkNameAndKeys returns true if metric has the same label keys or is new, false if not
+func (c *metricNameAndKeys) checkNameAndKeys(name string, labels prometheus.Labels) bool {
+	c.mtx.Lock()
+	defer c.mtx.Unlock()
+	providedKeys := keysFromLabels(labels)
+	if keys, found := c.cache[name]; found {
+		return keys == providedKeys
+	}
+
+	c.cache[name] = providedKeys
+	return true
+}
+
 type graphiteSample struct {
 	OriginalName string
 	Name         string
-	Labels       map[string]string
+	Labels       prometheus.Labels
 	Help         string
 	Value        float64
 	Type         prometheus.ValueType
@@ -126,6 +174,37 @@ func (c *graphiteCollector) processLines() {
 	}
 }
 
+func parseMetricNameAndTags(name string, labels prometheus.Labels) (string, error) {
+	if strings.ContainsRune(name, ';') {
+		// name contains tags - parse tags and add to labels
+		if strings.Count(name, ";") != strings.Count(name, "=") {
+			tagParseFailures.Inc()
+			return name, fmt.Errorf("error parsing tags on %s", name)
+		}
+
+		parts := strings.Split(name, ";")
+		parsedName := parts[0]
+		tags := parts[1:]
+
+		for _, tag := range tags {
+			kv := strings.SplitN(tag, "=", 2)
+			if len(kv) != 2 {
+				// we may have added bad labels already...
+				tagParseFailures.Inc()
+				return name, fmt.Errorf("error parsing tags on %s", name)
+			}
+
+			k := kv[0]
+			v := kv[1]
+			labels[k] = v
+		}
+
+		return parsedName, nil
+	}
+
+	return name, nil
+}
+
 func (c *graphiteCollector) processLine(line string) {
 	line = strings.TrimSpace(line)
 	level.Debug(c.logger).Log("msg", "Incoming line", "line", line)
@@ -136,16 +215,42 @@ func (c *graphiteCollector) processLine(line string) {
 	}
 	originalName := parts[0]
 	var name string
-	mapping, labels, present := c.mapper.GetMapping(originalName, mapper.MetricTypeGauge)
+	var err error
+	mapping, labels, mappingPresent := c.mapper.GetMapping(originalName, mapper.MetricTypeGauge)
 
-	if (present && mapping.Action == mapper.ActionTypeDrop) || (!present && c.strictMatch) {
+	if (mappingPresent && mapping.Action == mapper.ActionTypeDrop) || (!mappingPresent && c.strictMatch) {
 		return
 	}
 
-	if present {
+	if mappingPresent {
+		parsedLabels := make(prometheus.Labels)
+		_, err = parseMetricNameAndTags(originalName, parsedLabels)
+		if err != nil {
+			level.Info(c.logger).Log("msg", "Invalid tags", "line", line)
+			return
+		}
+
 		name = invalidMetricChars.ReplaceAllString(mapping.Name, "_")
+		// check to ensure the same tags are present
+		if validKeys := metricNameKeysIndex.checkNameAndKeys(name, parsedLabels); !validKeys {
+			level.Info(c.logger).Log("msg", "Dropped because metric keys do not match previously used keys", "line", line)
+			invalidMetrics.Inc()
+			return
+		}
 	} else {
-		name = invalidMetricChars.ReplaceAllString(originalName, "_")
+		labels = make(prometheus.Labels)
+		name, err = parseMetricNameAndTags(originalName, labels)
+		if err != nil {
+			level.Info(c.logger).Log("msg", "Invalid tags", "line", line)
+			return
+		}
+		name = invalidMetricChars.ReplaceAllString(name, "_")
+		// check to ensure the same tags are present
+		if validKeys := metricNameKeysIndex.checkNameAndKeys(name, labels); !validKeys {
+			level.Info(c.logger).Log("msg", "Dropped because metric keys do not match previously used keys", "line", line)
+			invalidMetrics.Inc()
+			return
+		}
 	}
 
 	value, err := strconv.ParseFloat(parts[1], 64)
@@ -158,6 +263,7 @@ func (c *graphiteCollector) processLine(line string) {
 		level.Info(c.logger).Log("msg", "Invalid timestamp", "line", line)
 		return
 	}
+
 	sample := graphiteSample{
 		OriginalName: originalName,
 		Name:         name,
@@ -257,6 +363,8 @@ func main() {
 	logger := promlog.New(promlogConfig)
 
 	prometheus.MustRegister(sampleExpiryMetric)
+	prometheus.MustRegister(tagParseFailures)
+	prometheus.MustRegister(invalidMetrics)
 	sampleExpiryMetric.Set(sampleExpiry.Seconds())
 
 	level.Info(logger).Log("msg", "Starting graphite_exporter", "version_info", version.Info())

main_test.go

@@ -44,91 +44,157 @@ func (m *mockMapper) InitFromFile(string, int, ...mapper.CacheOption) error {
 func (m *mockMapper) InitCache(int, ...mapper.CacheOption) {
 
 }
+
+func TestParseNameAndTags(t *testing.T) {
+	type testCase struct {
+		line       string
+		parsedName string
+		labels     prometheus.Labels
+		willFail   bool
+	}
+
+	testCases := []testCase{
+		{
+			line:       "my_simple_metric_with_tags;tag1=value1;tag2=value2",
+			parsedName: "my_simple_metric_with_tags",
+			labels: prometheus.Labels{
+				"tag1": "value1",
+				"tag2": "value2",
+			},
+		},
+		{
+			line:       "my_simple_metric_with_bad_tags;tag1=value1;tag2",
+			parsedName: "my_simple_metric_with_bad_tags;tag1=value1;tag2",
+			labels:     prometheus.Labels{},
+			willFail:   true,
+		},
+	}
+
+	for _, testCase := range testCases {
+		labels := prometheus.Labels{}
+		n, err := parseMetricNameAndTags(testCase.line, labels)
+		if !testCase.willFail {
+			assert.NoError(t, err, "Got unexpected error parsing %s", testCase.line)
+		}
+		assert.Equal(t, testCase.parsedName, n)
+		assert.Equal(t, testCase.labels, labels)
+	}
+}
+
 func TestProcessLine(t *testing.T) {
 
 	type testCase struct {
-		line     string
-		name     string
-		labels   map[string]string
-		value    float64
-		present  bool
-		willFail bool
-		action   mapper.ActionType
-		strict   bool
+		line           string
+		name           string
+		mappingLabels  prometheus.Labels
+		parsedLabels   prometheus.Labels
+		value          float64
+		mappingPresent bool
+		willFail       bool
+		action         mapper.ActionType
+		strict         bool
 	}
 
 	testCases := []testCase{
 		{
 			line: "my.simple.metric 9001 1534620625",
 			name: "my_simple_metric",
-			labels: map[string]string{
+			mappingLabels: prometheus.Labels{
 				"foo":  "bar",
 				"zip":  "zot",
 				"name": "alabel",
 			},
-			present: true,
-			value:   float64(9001),
+			mappingPresent: true,
+			value:          float64(9001),
 		},
 		{
 			line: "my.simple.metric.baz 9002 1534620625",
 			name: "my_simple_metric",
-			labels: map[string]string{
+			mappingLabels: prometheus.Labels{
 				"baz": "bat",
 			},
-			present: true,
-			value:   float64(9002),
+			mappingPresent: true,
+			value:          float64(9002),
 		},
 		{
-			line:    "my.nomap.metric 9001 1534620625",
-			name:    "my_nomap_metric",
-			value:   float64(9001),
-			present: false,
+			line:           "my.nomap.metric 9001 1534620625",
+			name:           "my_nomap_metric",
+			value:          float64(9001),
+			parsedLabels:   prometheus.Labels{},
+			mappingPresent: false,
 		},
 		{
-			line:     "my.nomap.metric.novalue 9001 ",
-			name:     "my_nomap_metric_novalue",
-			labels:   nil,
-			value:    float64(9001),
-			willFail: true,
+			line:          "my.nomap.metric.novalue 9001 ",
+			name:          "my_nomap_metric_novalue",
+			mappingLabels: nil,
+			value:         float64(9001),
+			willFail:      true,
 		},
 		{
-			line:     "my.mapped.metric.drop 55 1534620625",
-			name:     "my_mapped_metric_drop",
-			present:  true,
-			willFail: true,
-			action:   mapper.ActionTypeDrop,
+			line:           "my.mapped.metric.drop 55 1534620625",
+			name:           "my_mapped_metric_drop",
+			mappingPresent: true,
+			willFail:       true,
+			action:         mapper.ActionTypeDrop,
+		},
+		{
+			line:           "my.mapped.strict.metric 55 1534620625",
+			name:           "my_mapped_strict_metric",
+			value:          float64(55),
+			mappingPresent: true,
+			willFail:       false,
+			strict:         true,
 		},
 		{
-			line:     "my.mapped.strict.metric 55 1534620625",
-			name:     "my_mapped_strict_metric",
-			value:    float64(55),
-			present:  true,
-			willFail: false,
-			strict:   true,
+			line:           "my.mapped.strict.metric.drop 55 1534620625",
+			name:           "my_mapped_strict_metric_drop",
+			mappingPresent: false,
+			willFail:       true,
+			strict:         true,
+		},
+		{
+			line: "my.simple.metric.with.tags;tag1=value1;tag2=value2 9002 1534620625",
+			name: "my_simple_metric_with_tags",
+			parsedLabels: prometheus.Labels{
+				"tag1": "value1",
+				"tag2": "value2",
+			},
+			mappingPresent: false,
+			value:          float64(9002),
+		},
+		{
+			// same tags, different values, should parse
+			line: "my.simple.metric.with.tags;tag1=value3;tag2=value4 9002 1534620625",
+			name: "my_simple_metric_with_tags",
+			parsedLabels: prometheus.Labels{
+				"tag1": "value3",
+				"tag2": "value4",
+			},
+			mappingPresent: false,
+			value:          float64(9002),
 		},
 		{
-			line:     "my.mapped.strict.metric.drop 55 1534620625",
-			name:     "my_mapped_strict_metric_drop",
-			present:  false,
+			// new tags other than previously used, should drop
+			line:     "my.simple.metric.with.tags;tag1=value1;tag3=value2 9002 1534620625",
+			name:     "my_simple_metric_with_tags",
 			willFail: true,
-			strict:   true,
 		},
 	}
 
 	c := newGraphiteCollector(log.NewNopLogger())
 
 	for _, testCase := range testCases {
 
-		if testCase.present {
+		if testCase.mappingPresent {
 			c.mapper = &mockMapper{
 				name:    testCase.name,
-				labels:  testCase.labels,
+				labels:  testCase.mappingLabels,
 				action:  testCase.action,
-				present: testCase.present,
+				present: testCase.mappingPresent,
 			}
 		} else {
 			c.mapper = &mockMapper{
-				present: testCase.present,
+				present: testCase.mappingPresent,
 			}
 		}
 
@@ -146,7 +212,11 @@ func TestProcessLine(t *testing.T) {
 		} else {
 			if assert.NotNil(t, sample, "Missing %s", k.name) {
 				assert.Equal(t, k.name, sample.Name)
-				assert.Equal(t, k.labels, sample.Labels)
+				if k.mappingPresent {
+					assert.Equal(t, k.mappingLabels, sample.Labels)
+				} else {
+					assert.Equal(t, k.parsedLabels, sample.Labels)
+				}
 				assert.Equal(t, k.value, sample.Value)
 			}
 		}