api: update to adopt new Key types

Author: Dieterbe

api/cluster.go

@@ -163,7 +163,7 @@ func (s *Server) indexFindByTag(ctx *middleware.Context, req models.IndexFindByT
 
 // IndexGet returns a msgp encoded schema.MetricDefinition
 func (s *Server) indexGet(ctx *middleware.Context, req models.IndexGet) {
-	def, ok := s.MetricIndex.Get(req.Id)
+	def, ok := s.MetricIndex.Get(req.MKey)
 	if !ok {
 		response.Write(ctx, response.NewError(http.StatusNotFound, "Not Found"))
 		return

api/dataprocessor.go

@@ -380,16 +380,12 @@ func (s *Server) getTarget(ctx context.Context, req models.Req) (points []schema
 	}
 }
 
-func logLoad(typ, key string, from, to uint32) {
+func logLoad(typ string, key schema.AMKey, from, to uint32) {
 	if LogLevel < 2 {
-		log.Debug("DP load from %-6s %-20s %d - %d (%s - %s) span:%ds", typ, key, from, to, util.TS(from), util.TS(to), to-from-1)
+		log.Debug("DP load from %-6s %20s %d - %d (%s - %s) span:%ds", typ, key, from, to, util.TS(from), util.TS(to), to-from-1)
 	}
 }
 
-func AggMetricKey(key, archive string, aggSpan uint32) string {
-	return fmt.Sprintf("%s_%s_%d", key, archive, aggSpan)
-}
-
 func (s *Server) getSeriesFixed(ctx context.Context, req models.Req, consolidator consolidation.Consolidator) ([]schema.Point, error) {
 	select {
 	case <-ctx.Done():
@@ -474,18 +470,17 @@ func (s *Server) itersToPoints(ctx *requestContext, iters []chunk.Iter) []schema
 func (s *Server) getSeriesAggMetrics(ctx *requestContext) (mdata.Result, error) {
 	_, span := tracing.NewSpan(ctx.ctx, s.Tracer, "getSeriesAggMetrics")
 	defer span.Finish()
-	metric, ok := s.MemoryStore.Get(ctx.Key)
+	metric, ok := s.MemoryStore.Get(ctx.AMKey.MKey)
 	if !ok {
 		return mdata.Result{
 			Oldest: ctx.Req.To,
 		}, nil
 	}
 
+	logLoad("memory", ctx.AMKey, ctx.From, ctx.To)
 	if ctx.Cons != consolidation.None {
-		logLoad("memory", ctx.AggKey, ctx.From, ctx.To)
 		return metric.GetAggregated(ctx.Cons, ctx.Req.ArchInterval, ctx.From, ctx.To)
 	} else {
-		logLoad("memory", ctx.Req.Key, ctx.From, ctx.To)
 		return metric.Get(ctx.From, ctx.To), nil
 	}
 }
@@ -495,22 +490,17 @@ func (s *Server) getSeriesCachedStore(ctx *requestContext, until uint32) ([]chun
 	var iters []chunk.Iter
 	var prevts uint32
 
-	key := ctx.Key
-	if ctx.Cons != consolidation.None {
-		key = ctx.AggKey
-	}
-
 	_, span := tracing.NewSpan(ctx.ctx, s.Tracer, "getSeriesCachedStore")
 	defer span.Finish()
-	span.SetTag("key", key)
+	span.SetTag("key", ctx.AMKey)
 	span.SetTag("from", ctx.From)
 	span.SetTag("until", until)
 
 	reqSpanBoth.ValueUint32(ctx.To - ctx.From)
-	logLoad("cassan", ctx.Key, ctx.From, ctx.To)
+	logLoad("cassan", ctx.AMKey, ctx.From, ctx.To)
 
-	log.Debug("cache: searching query key %s, from %d, until %d", key, ctx.From, until)
-	cacheRes := s.Cache.Search(ctx.ctx, key, ctx.From, until)
+	log.Debug("cache: searching query key %s, from %d, until %d", ctx.AMKey, ctx.From, until)
+	cacheRes := s.Cache.Search(ctx.ctx, ctx.AMKey, ctx.From, until)
 	log.Debug("cache: result start %d, end %d", len(cacheRes.Start), len(cacheRes.End))
 
 	// check to see if the request has been canceled, if so abort now.
@@ -544,7 +534,7 @@ func (s *Server) getSeriesCachedStore(ctx *requestContext, until uint32) ([]chun
 	// the request cannot completely be served from cache, it will require cassandra involvement
 	if !cacheRes.Complete {
 		if cacheRes.From != cacheRes.Until {
-			storeIterGens, err := s.BackendStore.Search(ctx.ctx, key, ctx.Req.TTL, cacheRes.From, cacheRes.Until)
+			storeIterGens, err := s.BackendStore.Search(ctx.ctx, ctx.AMKey, ctx.Req.TTL, cacheRes.From, cacheRes.Until)
 			if err != nil {
 				return iters, err
 			}
@@ -566,7 +556,7 @@ func (s *Server) getSeriesCachedStore(ctx *requestContext, until uint32) ([]chun
 				}
 				// it's important that the itgens get added in chronological order,
 				// currently we rely on cassandra returning results in order
-				s.Cache.Add(key, ctx.Key, prevts, itgen)
+				s.Cache.Add(ctx.AMKey, prevts, itgen)
 				prevts = itgen.Ts
 				iters = append(iters, *it)
 			}
@@ -632,18 +622,18 @@ func mergeSeries(in []models.Series) []models.Series {
 	return merged
 }
 
+// requestContext is a more concrete specification to load data based on a models.Req
 type requestContext struct {
 	ctx context.Context
 
 	// request by external user.
 	Req *models.Req
 
 	// internal request needed to satisfy user request.
-	Cons   consolidation.Consolidator // to satisfy avg request from user, this would be sum or cnt
-	From   uint32                     // may be different than user request, see below
-	To     uint32                     // may be different than user request, see below
-	Key    string                     // key to query
-	AggKey string                     // aggkey to query (if needed)
+	Cons  consolidation.Consolidator // to satisfy avg request from user, this would be sum or cnt
+	From  uint32                     // may be different than user request, see below
+	To    uint32                     // may be different than user request, see below
+	AMKey schema.AMKey               // set by combining Req's key, consolidator and archive info
 }
 
 func prevBoundary(ts uint32, span uint32) uint32 {
@@ -656,7 +646,6 @@ func newRequestContext(ctx context.Context, req *models.Req, consolidator consol
 		ctx:  ctx,
 		Req:  req,
 		Cons: consolidator,
-		Key:  req.Key,
 	}
 
 	// while aggregated archives are quantized, raw intervals are not.  quantizing happens after fetching the data,
@@ -673,10 +662,11 @@ func newRequestContext(ctx context.Context, req *models.Req, consolidator consol
 	if consolidator == consolidation.None {
 		rc.From = prevBoundary(req.From, req.ArchInterval) + 1
 		rc.To = prevBoundary(req.To, req.ArchInterval) + 1
+		rc.AMKey = schema.AMKey{MKey: req.MKey}
 	} else {
 		rc.From = req.From
 		rc.To = req.To
-		rc.AggKey = AggMetricKey(req.Key, consolidator.Archive(), req.ArchInterval)
+		rc.AMKey = schema.GetAMKey(req.MKey, consolidator.Archive(), req.ArchInterval)
 	}
 
 	return &rc

api/graphite.go

@@ -11,6 +11,7 @@ import (
 	"time"
 
 	macaron "gopkg.in/macaron.v1"
+	schema "gopkg.in/raintank/schema.v1"
 
 	"github.com/grafana/metrictank/api/middleware"
 	"github.com/grafana/metrictank/api/models"
@@ -445,7 +446,7 @@ func (s *Server) metricsIndex(ctx *middleware.Context) {
 	}()
 
 	series := make([]idx.Archive, 0)
-	seenDefs := make(map[string]struct{})
+	seenDefs := make(map[schema.MKey]struct{})
 	for resp := range responses {
 		if resp.err != nil {
 			response.Write(ctx, response.WrapError(err))

api/models/node.go

@@ -3,6 +3,7 @@ package models
 import (
 	"github.com/grafana/metrictank/cluster"
 	opentracing "github.com/opentracing/opentracing-go"
+	schema "gopkg.in/raintank/schema.v1"
 )
 
 type NodeStatus struct {
@@ -136,7 +137,7 @@ func (i IndexTagDelSeries) TraceDebug(span opentracing.Span) {
 }
 
 type IndexGet struct {
-	Id string `json:"id" form:"id" binding:"Required"`
+	MKey schema.MKey `json:"id" form:"id" binding:"Required"`
 }
 
 type IndexFind struct {

api/models/request.go

@@ -3,16 +3,19 @@ package models
 import (
 	"fmt"
 
+	schema "gopkg.in/raintank/schema.v1"
+
 	"github.com/grafana/metrictank/cluster"
 	"github.com/grafana/metrictank/consolidation"
 	"github.com/grafana/metrictank/util"
 	opentracing "github.com/opentracing/opentracing-go"
 	"github.com/opentracing/opentracing-go/log"
 )
 
+// Req is a request for data by MKey and parameters such as consolidator, max points, etc
 type Req struct {
 	// these fields can be set straight away:
-	Key          string                     `json:"key"`     // metric key aka metric definition id (orgid.<hash>), often same as target for graphite-metrictank requests
+	MKey         schema.MKey                `json:"key"`     // metric key aka metric definition id (orgid.<hash>), often same as target for graphite-metrictank requests
 	Target       string                     `json:"target"`  // the target we should return either to graphite or as if we're graphite.  simply the graphite metric key from the index
 	Pattern      string                     `json:"pattern"` // the original query pattern specified by user (not wrapped by any functions). e.g. `foo.b*`. To be able to tie the result data back to the data need as requested
 	From         uint32                     `json:"from"`
@@ -36,7 +39,7 @@ type Req struct {
 	AggNum       uint32 `json:"aggNum"`       // how many points to consolidate together at runtime, after fetching from the archive
 }
 
-func NewReq(key, target, patt string, from, to, maxPoints, rawInterval uint32, cons, consReq consolidation.Consolidator, node cluster.Node, schemaId, aggId uint16) Req {
+func NewReq(key schema.MKey, target, patt string, from, to, maxPoints, rawInterval uint32, cons, consReq consolidation.Consolidator, node cluster.Node, schemaId, aggId uint16) Req {
 	return Req{
 		key,
 		target,
@@ -59,18 +62,18 @@ func NewReq(key, target, patt string, from, to, maxPoints, rawInterval uint32, c
 }
 
 func (r Req) String() string {
-	return fmt.Sprintf("%s %d - %d (%s - %s) span:%ds. points <= %d. %s.", r.Key, r.From, r.To, util.TS(r.From), util.TS(r.To), r.To-r.From-1, r.MaxPoints, r.Consolidator)
+	return fmt.Sprintf("%s %d - %d (%s - %s) span:%ds. points <= %d. %s.", r.MKey, r.From, r.To, util.TS(r.From), util.TS(r.To), r.To-r.From-1, r.MaxPoints, r.Consolidator)
 }
 
 func (r Req) DebugString() string {
 	return fmt.Sprintf("Req key=%q target=%q pattern=%q %d - %d (%s - %s) (span %d) maxPoints=%d rawInt=%d cons=%s consReq=%d schemaId=%d aggId=%d archive=%d archInt=%d ttl=%d outInt=%d aggNum=%d",
-		r.Key, r.Target, r.Pattern, r.From, r.To, util.TS(r.From), util.TS(r.To), r.To-r.From-1, r.MaxPoints, r.RawInterval, r.Consolidator, r.ConsReq, r.SchemaId, r.AggId, r.Archive, r.ArchInterval, r.TTL, r.OutInterval, r.AggNum)
+		r.MKey, r.Target, r.Pattern, r.From, r.To, util.TS(r.From), util.TS(r.To), r.To-r.From-1, r.MaxPoints, r.RawInterval, r.Consolidator, r.ConsReq, r.SchemaId, r.AggId, r.Archive, r.ArchInterval, r.TTL, r.OutInterval, r.AggNum)
 }
 
 // Trace puts all request properties as tags in a span
 // good for when a span deals with 1 request
 func (r Req) Trace(span opentracing.Span) {
-	span.SetTag("key", r.Key)
+	span.SetTag("key", r.MKey)
 	span.SetTag("target", r.Target)
 	span.SetTag("pattern", r.Pattern)
 	span.SetTag("from", r.From)
@@ -95,7 +98,7 @@ func (r Req) Trace(span opentracing.Span) {
 // 1000~1500 bytes
 func (r Req) TraceLog(span opentracing.Span) {
 	span.LogFields(
-		log.String("key", r.Key),
+		log.Object("key", r.MKey),
 		log.String("target", r.Target),
 		log.String("pattern", r.Pattern),
 		log.Int("from", int(r.From)),