input/idx: update for new MetricPoint type and messages

Author: Dieterbe

idx/cassandra/cassandra.go

@@ -264,17 +264,13 @@ func (c *CasIdx) Stop() {
 	c.session.Close()
 }
 
-func (c *CasIdx) UpdateMaybe(point schema.MetricPointId2, partition int32) (idx.Archive, bool) {
+func (c *CasIdx) UpdateMaybe(point schema.MetricPoint, partition int32) (idx.Archive, bool) {
 	pre := time.Now()
-	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)
+	existing, inMemory := c.MemoryIdx.Get(point.MKey)
 	archive, inMemory2 := c.MemoryIdx.UpdateMaybe(point, partition)
 
 	if !updateCassIdx {

idx/idx.go

@@ -58,7 +58,7 @@ type MetricIndex interface {
 
 	// 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)
+	UpdateMaybe(point schema.MetricPoint, partition int32) (Archive, bool)
 
 	// AddOrUpdate makes sure a metric is known in the index,
 	// and should be called for every received metric.

idx/memory/memory.go

@@ -208,21 +208,16 @@ func (m *MemoryIdx) Stop() {
 
 // 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) {
+func (m *MemoryIdx) UpdateMaybe(point schema.MetricPoint, 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]
+	existing, ok := m.defById[point.MKey]
 	if ok {
-		log.Debug("metricDef with id %v already in index", mkey)
-		existing.LastUpdate = int64(point.MetricPointId1.Time)
+		log.Debug("metricDef with id %v already in index", point.MKey)
+		existing.LastUpdate = int64(point.Time)
 		existing.Partition = partition
 		statUpdate.Inc()
 		statUpdateDuration.Value(time.Since(pre))

input/input.go

@@ -16,7 +16,7 @@ import (
 
 type Handler interface {
 	ProcessMetricData(md *schema.MetricData, partition int32)
-	ProcessMetricPoint(point schema.MetricPointId2, partition int32)
+	ProcessMetricPoint(point schema.MetricPoint, partition int32)
 }
 
 // TODO: clever way to document all metrics for all different inputs
@@ -48,17 +48,13 @@ func NewDefaultHandler(metrics mdata.Metrics, metricIndex idx.MetricIndex, input
 
 // ProcessMetricPoint updates the index if possible, and stores the data if we have an index entry
 // concurrency-safe.
-func (in DefaultHandler) ProcessMetricPoint(point schema.MetricPointId2, partition int32) {
+func (in DefaultHandler) ProcessMetricPoint(point schema.MetricPoint, partition int32) {
 	in.metricsReceived.Inc()
 	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)
@@ -69,8 +65,8 @@ func (in DefaultHandler) ProcessMetricPoint(point schema.MetricPointId2, partiti
 	}
 
 	pre = time.Now()
-	m := in.metrics.GetOrCreate(mkey, archive.SchemaId, archive.AggId)
-	m.Add(point.MetricPointId1.Time, point.MetricPointId1.Value)
+	m := in.metrics.GetOrCreate(point.MKey, archive.SchemaId, archive.AggId)
+	m.Add(point.Time, point.Value)
 	in.pressureTank.Add(int(time.Since(pre).Nanoseconds()))
 
 }

input/kafkamdm/kafkamdm.go

@@ -9,6 +9,7 @@ import (
 	"time"
 
 	schema "gopkg.in/raintank/schema.v1"
+	"gopkg.in/raintank/schema.v1/msg"
 
 	"github.com/Shopify/sarama"
 	"github.com/raintank/worldping-api/pkg/log"
@@ -71,7 +72,7 @@ var partitionLag map[int32]*stats.Gauge64
 func ConfigSetup() {
 	inKafkaMdm := flag.NewFlagSet("kafka-mdm-in", flag.ExitOnError)
 	inKafkaMdm.BoolVar(&Enabled, "enabled", false, "")
-	inKafkaMdm.UintVar(&orgId, "org-id", 0, "For incoming MetricPointId1 messages, assume this org id")
+	inKafkaMdm.UintVar(&orgId, "org-id", 0, "For incoming MetricPoint messages, assume this org id")
 	inKafkaMdm.StringVar(&brokerStr, "brokers", "kafka:9092", "tcp address for kafka (may be be given multiple times as a comma-separated list)")
 	inKafkaMdm.StringVar(&topicStr, "topics", "mdm", "kafka topic (may be given multiple times as a comma-separated list)")
 	inKafkaMdm.StringVar(&offsetStr, "offset", "last", "Set the offset to start consuming from. Can be one of newest, oldest,last or a time duration")
@@ -352,17 +353,13 @@ func (k *KafkaMdm) consumePartition(topic string, partition int32, currentOffset
 }
 
 func (k *KafkaMdm) handleMsg(data []byte, partition int32) {
-	if len(data) == 29 && data[0] == 0 {
-		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 {
-		var point schema.MetricPointId2
-		point.UnmarshalDirect(data[1:])
+	if msg.IsPointMsg(data) {
+		_, point, err := msg.ReadPointMsg(data, uint32(orgId))
+		if err != nil {
+			metricsDecodeErr.Inc()
+			log.Error(3, "kafka-mdm decode error, skipping message. %s", err)
+			return
+		}
 		k.Handler.ProcessMetricPoint(point, partition)
 		return
 	}