idx & input: changing approach. no need for composite msg.Point type

instead, use different codepaths in the input handler and index
this is a bit simpler

Author: Dieterbe

idx/cassandra/cassandra.go

@@ -264,10 +264,40 @@ func (c *CasIdx) Stop() {
 	c.session.Close()
 }
 
-func (c *CasIdx) AddOrUpdate(point msg.Point, partition int32) idx.Archive {
+func (c *CasIdx) UpdateMaybe(point schema.MetricPointId2, partition int32) (idx.Archive, bool) {
 	pre := time.Now()
-	existing, inMemory := c.MemoryIdx.Get(data.Id)
-	archive := c.MemoryIdx.AddOrUpdate(data, partition)
+	mkey := schema.MKey{
+		Key: point.MetricPointId1.Id,
+		Org: point.Org,
+	}
+
+	// note that both functions return an 'ok' bool.
+	// abeit very unlikely,
+	// the idx entry could be pruned in between the two calls and so they could be different
+	existing, inMemory := c.MemoryIdx.Get(mkey)
+	archive, inMemory2 := c.MemoryIdx.UpdateMaybe(point, partition)
+
+	if !updateCassIdx {
+		statUpdateDuration.Value(time.Since(pre))
+		return archive, inMemory2
+	}
+
+	if inMemory {
+		c.deleteOldIfMoved(existing, partition)
+	}
+
+	if inMemory2 {
+		archive = c.updateCassandraIfStale(inMemory, archive, partition)
+	}
+
+	statUpdateDuration.Value(time.Since(pre))
+	return archive, inMemory2
+}
+
+func (c *CasIdx) AddOrUpdate(mkey schema.MKey, data *schema.MetricData, partition int32) idx.Archive {
+	pre := time.Now()
+	existing, inMemory := c.MemoryIdx.Get(mkey)
+	archive := c.MemoryIdx.AddOrUpdate(mkey, data, partition)
 	stat := statUpdateDuration
 	if !inMemory {
 		stat = statAddDuration
@@ -277,22 +307,35 @@ func (c *CasIdx) AddOrUpdate(point msg.Point, partition int32) idx.Archive {
 		return archive
 	}
 
-	now := uint32(time.Now().Unix())
+	if inMemory {
+		c.deleteOldIfMoved(existing, partition)
+	}
+
+	archive = c.updateCassandraIfStale(inMemory, archive, partition)
+	stat.Value(time.Since(pre))
+	return archive
+}
 
-	// Cassandra uses partition id as the partitioning key, so an "update" that changes the partition for
-	// an existing metricDef will just create a new row in the table and wont remove the old row.
-	// So we need to explicitly delete the old entry.
-	if inMemory && existing.Partition != partition {
+// Cassandra uses partition id as the partitioning key, so an "update" that changes the partition for
+// an existing metricDef will just create a new row in the table and wont remove the old row.
+// So we need to explicitly delete the old entry.
+func (c *CasIdx) deleteOldIfMoved(existing idx.Archive, partition int32) {
+	if existing.Partition != partition {
 		go func() {
 			if err := c.deleteDef(&existing); err != nil {
 				log.Error(3, err.Error())
 			}
 		}()
 	}
+}
+
+// updateCassandraIfStale saves the archive to cassandra if needed and
+// updates the memory index with the updated fields.
+func (c *CasIdx) updateCassandraIfStale(inMemory bool, archive idx.Archive, partition int32) idx.Archive {
+	now := uint32(time.Now().Unix())
 
 	// check if we need to save to cassandra.
 	if archive.LastSave >= (now - updateInterval32) {
-		stat.Value(time.Since(pre))
 		return archive
 	}
 
@@ -320,7 +363,6 @@ func (c *CasIdx) AddOrUpdate(point msg.Point, partition int32) idx.Archive {
 		}
 	}
 
-	stat.Value(time.Since(pre))
 	return archive
 }
 

idx/idx.go

@@ -6,8 +6,6 @@ import (
 	"errors"
 	"time"
 
-	"github.com/grafana/metrictank/msg"
-
 	schema "gopkg.in/raintank/schema.v1"
 )
 
@@ -58,9 +56,13 @@ type MetricIndex interface {
 	// Stop shuts down the index.
 	Stop()
 
+	// UpdateMaybe updates an existing archive, if found.
+	// it returns the existing archive (if any), and whether it was found
+	UpdateMaybe(point schema.MetricPointId2, partition int32) (Archive, bool)
+
 	// AddOrUpdate makes sure a metric is known in the index,
 	// and should be called for every received metric.
-	AddOrUpdate(point msg.Point, partition int32) Archive
+	AddOrUpdate(mkey schema.MKey, data *schema.MetricData, partition int32) Archive
 
 	// Get returns the archive for the requested id.
 	Get(key schema.MKey) (Archive, bool)

idx/memory/memory.go

@@ -206,13 +206,43 @@ func (m *MemoryIdx) Stop() {
 	return
 }
 
-func (m *MemoryIdx) AddOrUpdate(data *schema.MetricData, partition int32) idx.Archive {
+// UpdateMaybe updates an existing archive, if found.
+// it returns the existing archive (if any), and whether it was found
+func (m *MemoryIdx) UpdateMaybe(point schema.MetricPointId2, partition int32) (idx.Archive, bool) {
+	pre := time.Now()
+
+	mkey := schema.MKey{
+		Key: point.MetricPointId1.Id,
+		Org: point.Org,
+	}
+
+	m.Lock()
+	defer m.Unlock()
+
+	existing, ok := m.defById[mkey]
+	if ok {
+		log.Debug("metricDef with id %v already in index", mkey)
+		existing.LastUpdate = int64(point.MetricPointId1.Time)
+		existing.Partition = partition
+		statUpdate.Inc()
+		statUpdateDuration.Value(time.Since(pre))
+		return *existing, true
+	}
+
+	return idx.Archive{}, false
+}
+
+// AddOrUpdate returns the corresponding Archive for the MetricData.
+// if it is existing -> updates lastUpdate based on .Time, and partition
+// if was new        -> adds new MetricDefinition to index
+func (m *MemoryIdx) AddOrUpdate(mkey schema.MKey, data *schema.MetricData, partition int32) idx.Archive {
 	pre := time.Now()
 	m.Lock()
 	defer m.Unlock()
-	existing, ok := m.defById[data.Id]
+
+	existing, ok := m.defById[mkey]
 	if ok {
-		log.Debug("metricDef with id %s already in index.", data.Id)
+		log.Debug("metricDef with id %s already in index.", mkey)
 		existing.LastUpdate = data.Time
 		existing.Partition = partition
 		statUpdate.Inc()

input/carbon/carbon.go

@@ -12,7 +12,6 @@ import (
 
 	"github.com/grafana/metrictank/cluster"
 	"github.com/grafana/metrictank/input"
-	"github.com/grafana/metrictank/msg"
 	"github.com/grafana/metrictank/stats"
 	"github.com/metrics20/go-metrics20/carbon20"
 	"github.com/raintank/worldping-api/pkg/log"
@@ -186,8 +185,7 @@ func (c *Carbon) handle(conn net.Conn) {
 			continue
 		}
 		nameSplits := strings.Split(string(key), ";")
-		pointMsg := msg.Point{}
-		pointMsg.Md = &schema.MetricData{
+		md := &schema.MetricData{
 			Name:     nameSplits[0],
 			Metric:   nameSplits[0],
 			Interval: c.intervalGetter.GetInterval(nameSplits[0]),
@@ -198,9 +196,9 @@ func (c *Carbon) handle(conn net.Conn) {
 			Tags:     nameSplits[1:],
 			OrgId:    1, // admin org
 		}
-		pointMsg.Md.SetId()
+		md.SetId()
 		metricsPerMessage.ValueUint32(1)
-		c.Handler.Process(pointMsg, int32(partitionId))
+		c.Handler.ProcessMetricData(md, int32(partitionId))
 	}
 	c.handlerWaitGroup.Done()
 }

input/input.go

@@ -10,13 +10,13 @@ import (
 
 	"github.com/grafana/metrictank/idx"
 	"github.com/grafana/metrictank/mdata"
-	"github.com/grafana/metrictank/msg"
 	"github.com/grafana/metrictank/stats"
 	"github.com/raintank/worldping-api/pkg/log"
 )
 
 type Handler interface {
-	Process(pointMsg msg.Point, partition int32)
+	ProcessMetricData(md *schema.MetricData, partition int32)
+	ProcessMetricPoint(point schema.MetricPointId2, partition int32)
 }
 
 // TODO: clever way to document all metrics for all different inputs
@@ -46,55 +46,63 @@ func NewDefaultHandler(metrics mdata.Metrics, metricIndex idx.MetricIndex, input
 	}
 }
 
-// process makes sure the data is stored and the metadata is in the index
+// ProcessMetricPoint updates the index if possible, and stores the data if we have an index entry
 // concurrency-safe.
-func (in DefaultHandler) Process(pointMsg msg.Point, partition int32) {
+func (in DefaultHandler) ProcessMetricPoint(point schema.MetricPointId2, partition int32) {
 	in.metricsReceived.Inc()
-	var mkey schema.MKey
-	var timestamp uint32
-	var value float64
-
-	if pointMsg.Val == 0 {
-		err := pointMsg.Md.Validate()
-		if err != nil {
-			in.MetricInvalid.Inc()
-			log.Debug("in: Invalid metric %v: %s", pointMsg.Md, err)
-			return
-		}
-		if pointMsg.Md.Time == 0 {
-			in.MetricInvalid.Inc()
-			log.Warn("in: invalid metric. metric.Time is 0. %s", pointMsg.Md.Id)
-			return
-		}
-
-		mkey, err = schema.MKeyFromString(pointMsg.Md.Id)
-		if err != nil {
-			log.Debug("in: Invalid metric %v: %s", pointMsg.Md, err)
-		}
-
-		timestamp = uint32(pointMsg.Md.Time)
-		value = pointMsg.Md.Value
-	} else {
-		if !pointMsg.Point.Valid() {
-			in.MetricInvalid.Inc()
-			log.Debug("in: Invalid metric %v", pointMsg.Point)
-			return
-		}
-		mkey = schema.MKey{
-			Key: pointMsg.Point.MetricPointId1.Id,
-			Org: pointMsg.Point.Org,
-		}
-		timestamp = pointMsg.Point.MetricPointId1.Time
-		value = pointMsg.Point.MetricPointId1.Value
+	if !point.Valid() {
+		in.MetricInvalid.Inc()
+		log.Debug("in: Invalid metric %v", point)
+		return
+	}
+	mkey := schema.MKey{
+		Key: point.MetricPointId1.Id,
+		Org: point.Org,
+	}
+
+	pre := time.Now()
+	archive, ok := in.metricIndex.UpdateMaybe(point, partition)
+	in.pressureIdx.Add(int(time.Since(pre).Nanoseconds()))
+
+	if !ok {
+		return
+	}
+
+	pre = time.Now()
+	m := in.metrics.GetOrCreate(mkey, archive.SchemaId, archive.AggId)
+	m.Add(point.MetricPointId1.Time, point.MetricPointId1.Value)
+	in.pressureTank.Add(int(time.Since(pre).Nanoseconds()))
+
+}
+
+// ProcessMetricData assures the data is stored and the metadata is in the index
+// concurrency-safe.
+func (in DefaultHandler) ProcessMetricData(md *schema.MetricData, partition int32) {
+	in.metricsReceived.Inc()
+	err := md.Validate()
+	if err != nil {
+		in.MetricInvalid.Inc()
+		log.Debug("in: Invalid metric %v: %s", md, err)
+		return
+	}
+	if md.Time == 0 {
+		in.MetricInvalid.Inc()
+		log.Warn("in: invalid metric. metric.Time is 0. %s", md.Id)
+		return
+	}
 
+	mkey, err := schema.MKeyFromString(md.Id)
+	if err != nil {
+		log.Error(3, "in: Invalid metric %v: could not parse ID: %s", md, err)
+		return
 	}
 
 	pre := time.Now()
-	archive := in.metricIndex.AddOrUpdate(pointMsg, partition)
+	archive := in.metricIndex.AddOrUpdate(mkey, md, partition)
 	in.pressureIdx.Add(int(time.Since(pre).Nanoseconds()))
 
 	pre = time.Now()
 	m := in.metrics.GetOrCreate(mkey, archive.SchemaId, archive.AggId)
-	m.Add(timestamp, value)
+	m.Add(uint32(md.Time), md.Value)
 	in.pressureTank.Add(int(time.Since(pre).Nanoseconds()))
 }

input/kafkamdm/kafkamdm.go

@@ -8,14 +8,15 @@ import (
 	"sync"
 	"time"
 
+	schema "gopkg.in/raintank/schema.v1"
+
 	"github.com/Shopify/sarama"
 	"github.com/raintank/worldping-api/pkg/log"
 	"github.com/rakyll/globalconf"
 
 	"github.com/grafana/metrictank/cluster"
 	"github.com/grafana/metrictank/input"
 	"github.com/grafana/metrictank/kafka"
-	"github.com/grafana/metrictank/msg"
 	"github.com/grafana/metrictank/stats"
 )
 
@@ -351,30 +352,30 @@ func (k *KafkaMdm) consumePartition(topic string, partition int32, currentOffset
 }
 
 func (k *KafkaMdm) handleMsg(data []byte, partition int32) {
-	pointMsg := msg.Point{}
 	if len(data) == 29 && data[0] == 0 {
-		pointMsg.Val = 1
-		pointMsg.Point.MetricPointId1.UnmarshalDirect(data[1:])
-		pointMsg.Point.Org = uint32(orgId)
-		k.Handler.Process(pointMsg, partition)
+		var point schema.MetricPointId2
+		point.MetricPointId1.UnmarshalDirect(data[1:])
+		point.Org = uint32(orgId)
+		k.Handler.ProcessMetricPoint(point, partition)
 		return
 	}
 
 	if len(data) == 33 && data[0] == 1 {
-		pointMsg.Val = 1
-		pointMsg.Point.UnmarshalDirect(data[1:])
-		k.Handler.Process(pointMsg, partition)
+		var point schema.MetricPointId2
+		point.UnmarshalDirect(data[1:])
+		k.Handler.ProcessMetricPoint(point, partition)
 		return
 	}
 
-	_, err := pointMsg.Md.UnmarshalMsg(data)
+	md := schema.MetricData{}
+	_, err := md.UnmarshalMsg(data)
 	if err != nil {
 		metricsDecodeErr.Inc()
 		log.Error(3, "kafka-mdm decode error, skipping message. %s", err)
 		return
 	}
 	metricsPerMessage.ValueUint32(1)
-	k.Handler.Process(pointMsg, partition)
+	k.Handler.ProcessMetricData(&md, partition)
 }
 
 // Stop will initiate a graceful stop of the Consumer (permanent)