Merge pull request #1348 from grafana/reject_invalid_tags

Reject invalid tags

Author: Dieterbe

CHANGELOG.md

@@ -1,6 +1,11 @@
 # master
 
 ## breaking changes
+* we now reject metrics with invalid tags on ingest by default, this can be disabled via the `input.reject-invalid-tags` flag.
+  if you're unsure whether you're currently sending invalid tags, it's a good idea to first disable the invalid tag rejection and watch the
+  new counter called `input.<input name>.metricdata.discarded.invalid_tag`, if invalid tags get ingested this counter will increase without
+  rejecting them. once you're sure that you don't ingest invalid tags you can enable rejection to enforce the validation.
+  more information on #1348
 * refactor jaeger configuration + many more options #1341
   the following config options have been renamed:
   - `tracing-enabled -> jaeger.enabled`

Gopkg.lock

@@ -111,7 +111,7 @@ unused-packages = true
 
 [[constraint]]
   name = "github.com/raintank/schema"
-  revision = "49c3f30550ed40c22931bc7f46e5c3a7676236b4"
+  revision = "5611429bd204bf26a3de166316cf3cad0de377c9"
 
 [[constraint]]
   name = "github.com/rs/cors"

Gopkg.toml

@@ -101,6 +101,10 @@ func main() {
 		fmt.Fprintf(os.Stderr, "FATAL: configuration file error: %s", err)
 		os.Exit(1)
 	}
+
+	// input handlers
+	input.ConfigSetup()
+
 	// load config for metric ingestors
 	inCarbon.ConfigSetup()
 	inKafkaMdm.ConfigSetup()

cmd/metrictank/metrictank.go

@@ -155,7 +155,7 @@
         },
         {
           "refId": "C",
-          "target": "groupByNodes(perSecond(metrictank.stats.$environment.$instance.input.*.*.discarded.invalid.counter32), 'sum', 5, 6, 7, 8)"
+          "target": "groupByNodes(perSecond(metrictank.stats.$environment.$instance.input.*.*.discarded.invalid*.counter32), 'sum', 5, 6, 7, 8)"
         },
         {
           "refId": "D",
@@ -4763,4 +4763,4 @@
   "title": "Metrictank",
   "uid": "tQW3QShiz",
   "version": 5
-}
+}

dashboards/main/metrictank.json

@@ -237,6 +237,10 @@ speculation-threshold = 1
 
 ## metric data inputs ##
 
+[input]
+# reject received metrics that have invalid tags
+reject-invalid-tags = true
+
 ### carbon input (optional)
 [carbon-in]
 enabled = true

docker/docker-chaos/metrictank.ini

@@ -237,6 +237,10 @@ speculation-threshold = 1
 
 ## metric data inputs ##
 
+[input]
+# reject received metrics that have invalid tags
+reject-invalid-tags = true
+
 ### carbon input (optional)
 [carbon-in]
 enabled = true

docker/docker-cluster-query/metrictank.ini

@@ -237,6 +237,10 @@ speculation-threshold = 1
 
 ## metric data inputs ##
 
+[input]
+# reject received metrics that have invalid tags
+reject-invalid-tags = true
+
 ### carbon input (optional)
 [carbon-in]
 enabled = true

docker/docker-cluster/metrictank.ini

@@ -237,6 +237,10 @@ speculation-threshold = 1
 
 ## metric data inputs ##
 
+[input]
+# reject received metrics that have invalid tags
+reject-invalid-tags = true
+
 ### carbon input (optional)
 [carbon-in]
 enabled = true

docker/docker-dev-custom-cfg-kafka/metrictank.ini

@@ -286,6 +286,13 @@ speculation-threshold = 1
 ```
 
 ## metric data inputs ##
+
+```
+[input]
+# reject received metrics that have invalid tags
+reject-invalid-tags = true
+```
+
 ### carbon input (optional)
 
 ```

docs/config.md

@@ -214,6 +214,9 @@ the duration of (successful) update of a metric to the memory idx
 the number of currently known metrics in the index
 * `input.%s.metricdata.discarded.invalid`:  
 a count of times a metricdata was invalid by input plugin
+* `input.%s.metricdata.discarded.invalid_tags`:  
+a count of times a metricdata was considered invalid due to
+invalid tags in the metric definition. all rejected metrics counted here are also counted in the above "invalid" counter
 * `input.%s.metricdata.received`:  
 the count of metricdata datapoints received by input plugin
 * `input.%s.metricpoint.discarded.invalid`:  

docs/metrics.md

@@ -276,8 +276,10 @@ func (m *UnpartitionedMemoryIdx) Init() error {
 }
 
 func (m *UnpartitionedMemoryIdx) Stop() {
-	m.findCache.Shutdown()
-	m.findCache = nil
+	if m.findCache != nil {
+		m.findCache.Shutdown()
+		m.findCache = nil
+	}
 	return
 }
 

idx/memory/memory.go

@@ -3,19 +3,28 @@
 package input
 
 import (
+	"flag"
 	"fmt"
 	"math"
 	"strconv"
 
-	"github.com/raintank/schema"
-	"github.com/raintank/schema/msg"
-
+	"github.com/grafana/globalconf"
 	"github.com/grafana/metrictank/idx"
 	"github.com/grafana/metrictank/mdata"
 	"github.com/grafana/metrictank/stats"
+	"github.com/raintank/schema"
+	"github.com/raintank/schema/msg"
 	log "github.com/sirupsen/logrus"
 )
 
+var rejectInvalidTags bool
+
+func ConfigSetup() {
+	input := flag.NewFlagSet("input", flag.ExitOnError)
+	input.BoolVar(&rejectInvalidTags, "reject-invalid-tags", true, "reject received metrics that have invalid tags")
+	globalconf.Register("input", input, flag.ExitOnError)
+}
+
 type Handler interface {
 	ProcessMetricData(md *schema.MetricData, partition int32)
 	ProcessMetricPoint(point schema.MetricPoint, format msg.Format, partition int32)
@@ -29,6 +38,7 @@ type DefaultHandler struct {
 	receivedMP   *stats.Counter32
 	receivedMPNO *stats.Counter32
 	invalidMD    *stats.CounterRate32
+	invalidTagMD *stats.CounterRate32
 	invalidMP    *stats.CounterRate32
 	unknownMP    *stats.Counter32
 
@@ -57,6 +67,9 @@ func NewDefaultHandler(metrics mdata.Metrics, metricIndex idx.MetricIndex, input
 		receivedMPNO: stats.NewCounter32(fmt.Sprintf("input.%s.metricpoint_no_org.received", input)),
 		// metric input.%s.metricdata.discarded.invalid is a count of times a metricdata was invalid by input plugin
 		invalidMD: stats.NewCounterRate32(fmt.Sprintf("input.%s.metricdata.discarded.invalid", input)),
+		// metric input.%s.metricdata.discarded.invalid_tags is a count of times a metricdata was considered invalid due to
+		// invalid tags in the metric definition. all rejected metrics counted here are also counted in the above "invalid" counter
+		invalidTagMD: stats.NewCounterRate32(fmt.Sprintf("input.%s.metricdata.discarded.invalid_tag", input)),
 		// metric input.%s.metricpoint.discarded.invalid is a count of times a metricpoint was invalid by input plugin
 		invalidMP: stats.NewCounterRate32(fmt.Sprintf("input.%s.metricpoint.discarded.invalid", input)),
 		// metric input.%s.metricpoint.discarded.unknown is the count of times the ID of a received metricpoint was not in the index, by input plugin
@@ -103,27 +116,42 @@ func (in DefaultHandler) ProcessMetricData(md *schema.MetricData, partition int3
 	err := md.Validate()
 	if err != nil {
 		in.invalidMD.Inc()
-		log.Debugf("in: Invalid metric %v: %s", md, err)
-
-		var reason string
-		switch err {
-		case schema.ErrInvalidIntervalzero:
-			reason = invalidInterval
-		case schema.ErrInvalidOrgIdzero:
-			reason = invalidOrgId
-		case schema.ErrInvalidEmptyName:
-			reason = invalidName
-		case schema.ErrInvalidMtype:
-			reason = invalidMtype
-		case schema.ErrInvalidTagFormat:
-			reason = invalidTagFormat
-		default:
-			reason = "unknown"
+
+		ignoreError := false
+		// assuming that the tag format was the only issue found by Validate()
+		// better make sure that the tag format is the last check which Validate() checks, to not accidentally ignore a potential following error
+		if err == schema.ErrInvalidTagFormat {
+			in.invalidTagMD.Inc()
+			if !rejectInvalidTags {
+				log.Debugf("in: Invalid metric %v, not rejecting it because rejection due to invalid tags is disabled: %s", md, err)
+				ignoreError = true
+			}
 		}
-		mdata.PromDiscardedSamples.WithLabelValues(reason, strconv.Itoa(md.OrgId)).Inc()
 
-		return
+		if !ignoreError {
+			log.Debugf("in: Invalid metric %v: %s", md, err)
+
+			var reason string
+			switch err {
+			case schema.ErrInvalidIntervalzero:
+				reason = invalidInterval
+			case schema.ErrInvalidOrgIdzero:
+				reason = invalidOrgId
+			case schema.ErrInvalidEmptyName:
+				reason = invalidName
+			case schema.ErrInvalidMtype:
+				reason = invalidMtype
+			case schema.ErrInvalidTagFormat:
+				reason = invalidTagFormat
+			default:
+				reason = "unknown"
+			}
+			mdata.PromDiscardedSamples.WithLabelValues(reason, strconv.Itoa(md.OrgId)).Inc()
+
+			return
+		}
 	}
+
 	// in cassandra we store timestamps and interval as 32bit signed integers.
 	// math.MaxInt32 = Jan 19 03:14:07 UTC 2038
 	if md.Time <= 0 || md.Time >= math.MaxInt32 {