Automatic detection of metric interval

If a metric is received with an Interval field set to 0, then we
need to automatically detect the interval.  This is achieved by buffering
the last 3 points in memory.  We then sort the points by timestamp
and get the time difference between the first 2, and last 2.  The lower
difference is used as the interval. Metrics are delayed from being processed
until the Interval is known, so when a new series is sent it will take 3x interval
before any data is queriable.

Future effort needs to be done to adjust
the interval to known valid intervals. eg 1s, 5s, 10s, 30s, 1m, 5m, etc....

Author: woodsaj

input/carbon/carbon.go

@@ -33,7 +33,6 @@ type Carbon struct {
 	handlerWaitGroup sync.WaitGroup
 	quit             chan struct{}
 	connTrack        *ConnTrack
-	intervalGetter   IntervalGetter
 }
 
 type ConnTrack struct {
@@ -102,10 +101,6 @@ func New() *Carbon {
 	}
 }
 
-func (c *Carbon) IntervalGetter(i IntervalGetter) {
-	c.intervalGetter = i
-}
-
 func (c *Carbon) Start(handler input.Handler, fatal chan struct{}) error {
 	c.Handler = handler
 	l, err := net.ListenTCP("tcp", c.addr)
@@ -186,17 +181,18 @@ func (c *Carbon) handle(conn net.Conn) {
 		}
 		nameSplits := strings.Split(string(key), ";")
 		md := &schema.MetricData{
+			Id:       nameSplits[0], //this is a temporary ID. It will be updated once we determine the Interval
 			Name:     nameSplits[0],
 			Metric:   nameSplits[0],
-			Interval: c.intervalGetter.GetInterval(nameSplits[0]),
+			Interval: 0,
 			Value:    val,
 			Unit:     "unknown",
 			Time:     int64(ts),
 			Mtype:    "gauge",
 			Tags:     nameSplits[1:],
 			OrgId:    1, // admin org
 		}
-		md.SetId()
+
 		metricsPerMessage.ValueUint32(1)
 		c.Handler.Process(md, int32(partitionId))
 	}

input/carbon/intervalgetter.go

@@ -4,6 +4,8 @@ package input
 
 import (
 	"fmt"
+	"sort"
+	"sync"
 	"time"
 
 	"github.com/grafana/metrictank/idx"
@@ -29,10 +31,21 @@ type DefaultHandler struct {
 
 	metrics     mdata.Metrics
 	metricIndex idx.MetricIndex
+
+	buffer sync.Map
+}
+
+type IntervalLookupRecord struct {
+	sync.Mutex
+
+	Interval   int
+	DataPoints []*schema.MetricData
+	Last       time.Time
+	pos        int
 }
 
-func NewDefaultHandler(metrics mdata.Metrics, metricIndex idx.MetricIndex, input string) DefaultHandler {
-	return DefaultHandler{
+func NewDefaultHandler(metrics mdata.Metrics, metricIndex idx.MetricIndex, input string) *DefaultHandler {
+	return &DefaultHandler{
 		metricsReceived: stats.NewCounter32(fmt.Sprintf("input.%s.metrics_received", input)),
 		MetricInvalid:   stats.NewCounter32(fmt.Sprintf("input.%s.metric_invalid", input)),
 		MsgsAge:         stats.NewMeter32(fmt.Sprintf("input.%s.message_age", input), false),
@@ -46,10 +59,17 @@ func NewDefaultHandler(metrics mdata.Metrics, metricIndex idx.MetricIndex, input
 
 // process makes sure the data is stored and the metadata is in the index
 // concurrency-safe.
-func (in DefaultHandler) Process(metric *schema.MetricData, partition int32) {
+func (in *DefaultHandler) Process(metric *schema.MetricData, partition int32) {
 	if metric == nil {
 		return
 	}
+
+	if metric.Interval <= 0 {
+		// we need to buffer enough of this metric to be able to work out its interval.
+		// once that is done, we can update the Id and call in.Process again
+		in.SetInterval(metric, partition)
+		return
+	}
 	in.metricsReceived.Inc()
 	err := metric.Validate()
 	if err != nil {
@@ -72,3 +92,99 @@ func (in DefaultHandler) Process(metric *schema.MetricData, partition int32) {
 	m.Add(uint32(metric.Time), metric.Value)
 	in.pressureTank.Add(int(time.Since(pre).Nanoseconds()))
 }
+
+func (in *DefaultHandler) SetInterval(metric *schema.MetricData, partition int32) {
+	b, ok := in.buffer.Load(metric.Id)
+	if !ok {
+		dp := make([]*schema.MetricData, 1, 3)
+		dp[0] = metric
+		in.buffer.Store(metric.Id, &IntervalLookupRecord{
+			DataPoints: dp,
+		})
+		return
+	}
+	ilr := b.(*IntervalLookupRecord)
+	ilr.Lock()
+	// check for duplicate TS
+	if ilr.DataPoints[ilr.pos].Time == metric.Time {
+		//drop the metric as it is a duplicate.
+		ilr.Unlock()
+		return
+	}
+
+	// add this datapoint to our circular buffer
+	if len(ilr.DataPoints) < 3 {
+		ilr.DataPoints = append(ilr.DataPoints, metric)
+		ilr.pos++
+	} else {
+		if ilr.pos == 2 {
+			ilr.pos = 0
+		} else {
+			ilr.pos++
+		}
+		ilr.DataPoints[ilr.pos] = metric
+	}
+
+	// if the interval is already known and was updated in the last 24hours, use it.
+	if ilr.Interval != 0 && time.Since(ilr.Last) > 84600 {
+		metric.Interval = ilr.Interval
+		metric.SetId()
+		in.Process(metric, partition)
+		ilr.Unlock()
+		return
+	}
+
+	if len(ilr.DataPoints) < 3 {
+		// we dont have 3 points yet.
+		ilr.Unlock()
+		return
+	}
+	log.Debug("input: calculating interval of %s", metric.Id)
+
+	delta1 := ilr.DataPoints[1].Time - ilr.DataPoints[0].Time
+	// make sure the points are not out of order
+	if delta1 < 0 {
+		delta1 = -1 * delta1
+	}
+	// make sure the points are not out of order
+	delta2 := ilr.DataPoints[2].Time - ilr.DataPoints[1].Time
+	if delta2 < 0 {
+		delta2 = -1 * delta2
+	}
+	interval := 0
+
+	// To protect against dropped metrics and out of order metrics, use the smallest delta as the interval.
+	// Because we have 3 datapoints, it doesnt matter what order they are in.  The smallest delta is always
+	// going to be correct. (unless their are out of order and dropped metrics)
+	if delta1 <= delta2 {
+		interval = int(delta1)
+	} else {
+		interval = int(delta2)
+	}
+
+	// TODO: align the interval to the likely value. eg. if the value is 27, make it 30. if it is 68, use 60. etc...
+
+	if ilr.Last.IsZero() {
+		// sort the datapoints then update their interval and process them properly.
+		sort.Sort(SortedMetricData(ilr.DataPoints))
+		ilr.pos = 2 // as the metrics are sorted, the oldest point is at the end of the slice
+		for _, md := range ilr.DataPoints {
+			md.Interval = interval
+			md.SetId()
+			in.Process(md, partition)
+		}
+	} else {
+		metric.Interval = interval
+		metric.SetId()
+		in.Process(metric, partition)
+	}
+	ilr.Interval = interval
+	ilr.Last = time.Now()
+	ilr.Unlock()
+}
+
+type SortedMetricData []*schema.MetricData
+
+func (a SortedMetricData) Len() int           { return len(a) }
+func (a SortedMetricData) Swap(i, j int)      { a[i], a[j] = a[j], a[i] }
+func (a SortedMetricData) Less(i, j int) bool { return a[i].Time < a[j].Time }

input/input.go

@@ -379,9 +379,6 @@ func main() {
 	***********************************/
 	pluginFatal := make(chan struct{})
 	for _, plugin := range inputs {
-		if carbonPlugin, ok := plugin.(*inCarbon.Carbon); ok {
-			carbonPlugin.IntervalGetter(inCarbon.NewIndexIntervalGetter(metricIndex))
-		}
 		err = plugin.Start(input.NewDefaultHandler(metrics, metricIndex, plugin.Name()), pluginFatal)
 		if err != nil {
 			shutdown()