fix meta record up for merge

Author: replay

api/cluster.go

@@ -13,6 +13,7 @@ import (
 	"github.com/grafana/metrictank/api/models"
 	"github.com/grafana/metrictank/api/response"
 	"github.com/grafana/metrictank/cluster"
+	"github.com/grafana/metrictank/idx"
 	"github.com/grafana/metrictank/stats"
 	log "github.com/sirupsen/logrus"
 	"github.com/tinylib/msgp/msgp"
@@ -314,7 +315,7 @@ type PeerResponse struct {
 // data:         request to be submitted
 // name:         name to be used in logging & tracing
 // path:         path to request on
-func (s *Server) peerQuery(ctx context.Context, data cluster.Traceable, name, path string) (map[string]PeerResponse, error) {
+func (s *Server) peerQuery(ctx context.Context, data cluster.Traceable, name, path string) (map[string]PeerResponse, map[string]error) {
 
 	peers := cluster.Manager.MemberList(false, true)
 	log.Debugf("HTTP %s across %d instances", name, len(peers)-1)
@@ -353,14 +354,16 @@ func (s *Server) peerQuery(ctx context.Context, data cluster.Traceable, name, pa
 	}()
 
 	result := make(map[string]PeerResponse)
+	errors := make(map[string]error)
 	for resp := range responses {
 		if resp.err != nil {
-			return nil, resp.err
+			errors[resp.data.peer.GetName()] = resp.err
+		} else {
+			result[resp.data.peer.GetName()] = resp.data
 		}
-		result[resp.data.peer.GetName()] = resp.data
 	}
 
-	return result, nil
+	return result, errors
 }
 
 // peerQuerySpeculative takes a request and the path to request it on, then fans it out
@@ -505,3 +508,28 @@ func (s *Server) peerQuerySpeculativeChan(ctx context.Context, data cluster.Trac
 
 	return resultChan, errorChan
 }
+
+func (s *Server) indexMetaTagRecordUpsert(ctx *middleware.Context, req models.IndexMetaTagRecordUpsert) {
+	if s.MetricIndex == nil {
+		response.Write(ctx, response.NewMsgp(200, &models.IndexFindByTagResp{}))
+		return
+	}
+
+	record := idx.MetaTagRecord{
+		MetaTags: req.MetaTags,
+		Queries:  req.Queries,
+	}
+
+	result, created, err := s.MetricIndex.MetaTagRecordUpsert(req.OrgId, record)
+	if err != nil {
+		response.Write(ctx, response.NewError(http.StatusBadRequest, err.Error()))
+		return
+	}
+
+	response.Write(ctx, response.NewMsgp(200, &models.MetaTagRecordUpsertResult{
+		MetaTags: result.MetaTags,
+		Queries:  result.Queries,
+		ID:       result.ID,
+		Created:  created,
+	}))
+}

api/graphite.go

@@ -1115,16 +1115,18 @@ func (s *Server) graphiteTagDelSeries(ctx *middleware.Context, request models.Gr
 	}
 
 	data := models.IndexTagDelSeries{OrgId: ctx.OrgId, Paths: request.Paths}
-	responses, err := s.peerQuery(ctx.Req.Context(), data, "clusterTagDelSeries,", "/index/tags/delSeries")
-	if err != nil {
+	responses, errors := s.peerQuery(ctx.Req.Context(), data, "clusterTagDelSeries,", "/index/tags/delSeries")
+
+	// if there are any errors, write one of them and return
+	for _, err := range errors {
 		response.Write(ctx, response.WrapErrorForTagDB(err))
 		return
 	}
 
 	res.Peers = make(map[string]int, len(responses))
 	peerResp := models.IndexTagDelSeriesResp{}
 	for peer, resp := range responses {
-		_, err = peerResp.UnmarshalMsg(resp.buf)
+		_, err := peerResp.UnmarshalMsg(resp.buf)
 		if err != nil {
 			response.Write(ctx, response.WrapErrorForTagDB(err))
 			return
@@ -1191,16 +1193,71 @@ func (s *Server) getMetaTagRecord(ctx *middleware.Context) {
 	response.Write(ctx, response.NewJson(200, metaTagRecords, ""))
 }
 
-func (s *Server) metaTagRecordUpsert(ctx *middleware.Context, metaTagRecord models.MetaTagRecord) {
+func (s *Server) metaTagRecordUpsert(ctx *middleware.Context, upsertRequest models.MetaTagRecordUpsert) {
 	record := idx.MetaTagRecord{
-		MetaTags: metaTagRecord.MetaTags,
-		Queries:  metaTagRecord.TagQueries,
+		MetaTags: upsertRequest.MetaTags,
+		Queries:  upsertRequest.Queries,
 	}
 
-	result, err := s.MetricIndex.MetaTagRecordUpsert(ctx.OrgId, record)
-	if err != nil {
-		response.Write(ctx, response.NewError(http.StatusBadRequest, err.Error()))
+	var localResult idx.MetaTagRecord
+	var created bool
+	if s.MetricIndex != nil {
+		var err error
+		localResult, created, err = s.MetricIndex.MetaTagRecordUpsert(ctx.OrgId, record)
+		if err != nil {
+			response.Write(ctx, response.NewError(http.StatusBadRequest, err.Error()))
+			return
+		}
+
+		if !upsertRequest.Propagate {
+			response.Write(ctx, response.NewJson(200, models.MetaTagRecordUpsertResult{
+				MetaTags: localResult.MetaTags,
+				Queries:  localResult.Queries,
+				ID:       localResult.ID,
+				Created:  created,
+			}, ""))
+			return
+		}
+	} else if !upsertRequest.Propagate {
+		return
 	}
 
-	response.Write(ctx, response.NewJson(200, result, ""))
+	indexUpsertRequest := models.IndexMetaTagRecordUpsert{
+		OrgId:    ctx.OrgId,
+		MetaTags: upsertRequest.MetaTags,
+		Queries:  upsertRequest.Queries,
+	}
+
+	results, errors := s.peerQuery(ctx.Req.Context(), indexUpsertRequest, "metaTagRecordUpsert", "/index/metaTags/upsert")
+
+	res := models.MetaTagRecordUpsertResultByNode{
+		Local: models.MetaTagRecordUpsertResult{
+			MetaTags: localResult.MetaTags,
+			Queries:  localResult.Queries,
+			ID:       localResult.ID,
+			Created:  created,
+		},
+	}
+
+	if len(errors) > 0 {
+		res.PeerErrors = make(map[string]string, len(errors))
+		for peer, err := range errors {
+			res.PeerErrors[peer] = err.Error()
+		}
+	}
+
+	if len(results) > 0 {
+		res.PeerResults = make(map[string]models.MetaTagRecordUpsertResult, len(results))
+		for peer, resp := range results {
+			peerResp := models.MetaTagRecordUpsertResult{}
+			_, err := peerResp.UnmarshalMsg(resp.buf)
+			if err != nil {
+				res.PeerErrors[peer] = fmt.Sprintf("Error when unmarshaling response: %s", err.Error())
+				continue
+			}
+			res.PeerResults[peer] = peerResp
+		}
+	}
+
+	response.Write(ctx, response.NewJson(200, res, ""))
 }

api/models/graphite.go

@@ -151,11 +151,6 @@ type GraphiteFind struct {
 	Jsonp  string `json:"jsonp" form:"jsonp"`
 }
 
-type MetaTagRecord struct {
-	MetaTags   []string
-	TagQueries []string
-}
-
 type MetricsDelete struct {
 	Query string `json:"query" form:"query" binding:"Required"`
 }