Merge pull request #21 from bloomberg/multiDef

Fix regressing timestamps, missing defs

Author: yogeswaran

api/dataprocessor.go

@@ -170,7 +170,7 @@ func (s *Server) getTargets(ctx context.Context, reqs []models.Req) ([]models.Se
 	if len(remoteReqs) > 0 {
 		wg.Add(1)
 		go func() {
-			// all errors returned returned are *response.Error.
+			// all errors returned are *response.Error.
 			series, err := s.getTargetsRemote(getCtx, remoteReqs)
 			if err != nil {
 				cancel()

api/graphite.go

@@ -924,8 +924,6 @@ func (s *Server) graphiteTagFindSeries(ctx *middleware.Context, request models.G
 }
 
 func (s *Server) clusterFindByTag(ctx context.Context, orgId uint32, expressions []string, from int64) ([]Series, error) {
-	seriesSet := make(map[string]Series)
-
 	result, err := s.MetricIndex.FindByTag(orgId, expressions, from)
 	if err != nil {
 		return nil, err
@@ -938,12 +936,14 @@ func (s *Server) clusterFindByTag(ctx context.Context, orgId uint32, expressions
 	default:
 	}
 
+	var allSeries []Series
+
 	for _, series := range result {
-		seriesSet[series.Path] = Series{
+		allSeries = append(allSeries, Series{
 			Pattern: series.Path,
 			Node:    cluster.Manager.ThisNode(),
 			Series:  []idx.Node{series},
-		}
+		})
 	}
 
 	data := models.IndexFindByTag{OrgId: orgId, Expr: expressions, From: from}
@@ -966,20 +966,15 @@ func (s *Server) clusterFindByTag(ctx context.Context, orgId uint32, expressions
 			return nil, err
 		}
 		for _, series := range resp.Metrics {
-			seriesSet[series.Path] = Series{
+			allSeries = append(allSeries, Series{
 				Pattern: series.Path,
 				Node:    r.peer,
 				Series:  []idx.Node{series},
-			}
+			})
 		}
 	}
 
-	series := make([]Series, 0, len(seriesSet))
-	for _, s := range seriesSet {
-		series = append(series, s)
-	}
-
-	return series, nil
+	return allSeries, nil
 }
 
 func (s *Server) graphiteTags(ctx *middleware.Context, request models.GraphiteTags) {

idx/memory/memory.go

@@ -222,7 +222,10 @@ func (m *MemoryIdx) Update(point schema.MetricPoint, partition int32) (idx.Archi
 		if LogLevel < 2 {
 			log.Debug("metricDef with id %v already in index", point.MKey)
 		}
-		existing.LastUpdate = int64(point.Time)
+
+		if existing.LastUpdate < int64(point.Time) {
+			existing.LastUpdate = int64(point.Time)
+		}
 		existing.Partition = partition
 		statUpdate.Inc()
 		statUpdateDuration.Value(time.Since(pre))
@@ -244,7 +247,9 @@ func (m *MemoryIdx) AddOrUpdate(mkey schema.MKey, data *schema.MetricData, parti
 	if ok {
 		oldPart := existing.Partition
 		log.Debug("metricDef with id %s already in index.", mkey)
-		existing.LastUpdate = data.Time
+		if existing.LastUpdate < int64(data.Time) {
+			existing.LastUpdate = int64(data.Time)
+		}
 		existing.Partition = partition
 		statUpdate.Inc()
 		statUpdateDuration.Value(time.Since(pre))
@@ -833,6 +838,7 @@ func (m *MemoryIdx) FindByTag(orgId uint32, expressions []string, from int64) ([
 
 	ids := m.idsByTagQuery(orgId, query)
 	res := make([]idx.Node, 0, len(ids))
+	seen := make(map[string]struct{})
 	for id := range ids {
 		def, ok := m.defById[id]
 		if !ok {
@@ -841,12 +847,27 @@ func (m *MemoryIdx) FindByTag(orgId uint32, expressions []string, from int64) ([
 			continue
 		}
 
-		res = append(res, idx.Node{
-			Path:        def.NameWithTags(),
-			Leaf:        true,
-			HasChildren: false,
-			Defs:        []idx.Archive{*def},
-		})
+		if _, ok := seen[def.NameWithTags()]; !ok {
+
+			defMap := m.defByTagSet.defs(orgId, def.NameWithTags())
+
+			defs := make([]idx.Archive, 0, len(defMap))
+
+			for d := range defMap {
+				if from != 0 && def.LastUpdate < from {
+					continue
+				}
+				defs = append(defs, *m.defById[d.Id])
+			}
+
+			res = append(res, idx.Node{
+				Path:        def.NameWithTags(),
+				Leaf:        true,
+				HasChildren: false,
+				Defs:        defs,
+			})
+			seen[def.NameWithTags()] = struct{}{}
+		}
 	}
 	return res, nil
 }

idx/memory/memory_test.go

@@ -126,6 +126,19 @@ func testGetAddKey(t *testing.T) {
 			So(defs, ShouldHaveLength, 15)
 		})
 	})
+
+	if TagSupport {
+		Convey("When adding metricDefs with the same series name as existing metricDefs (tagged)", t, func() {
+			Convey("then findByTag", func() {
+				nodes, _ := ix.FindByTag(1, []string{"name!="}, 0)
+				defs := make([]idx.Archive, 0, len(nodes))
+				for i := range nodes {
+					defs = append(defs, nodes[i].Defs...)
+				}
+				So(defs, ShouldHaveLength, 10)
+			})
+		})
+	}
 }
 
 func TestFind(t *testing.T) {
@@ -673,7 +686,12 @@ func TestPruneTaggedSeriesWithCollidingTagSets(t *testing.T) {
 	Convey("After purge", t, func() {
 		nodes, err := ix.FindByTag(1, findExpressions, 0)
 		So(err, ShouldBeNil)
-		So(nodes, ShouldHaveLength, 2)
+		So(nodes, ShouldHaveLength, 1)
+		defs := make([]idx.Archive, 0, len(nodes))
+		for i := range nodes {
+			defs = append(defs, nodes[i].Defs...)
+		}
+		So(defs, ShouldHaveLength, 2)
 	})
 
 	Convey("When purging newer series", t, func() {