[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)
  }
  }