Merge pull request #398 from raintank/issue-397

Issue #397

Author: Dieterbe

idx/cassandra/cassandra.go

@@ -185,15 +185,15 @@ func (c *CasIdx) Stop() {
 	c.session.Close()
 }
 
-func (c *CasIdx) Add(data *schema.MetricData) {
+func (c *CasIdx) Add(data *schema.MetricData) error {
 	existing, err := c.MemoryIdx.Get(data.Id)
 	inMemory := true
 	if err != nil {
 		if err == idx.DefNotFound {
 			inMemory = false
 		} else {
 			log.Error(3, "cassandra-idx Failed to query Memory Index for %s. %s", data.Id, err)
-			return
+			return err
 		}
 	}
 	if inMemory {
@@ -204,11 +204,14 @@ func (c *CasIdx) Add(data *schema.MetricData) {
 			c.MemoryIdx.AddDef(&existing)
 			c.writeQueue <- writeReq{recvTime: time.Now(), def: &existing}
 		}
-		return
+		return nil
 	}
 	def := schema.MetricDefinitionFromMetricData(data)
-	c.MemoryIdx.AddDef(def)
-	c.writeQueue <- writeReq{recvTime: time.Now(), def: def}
+	err = c.MemoryIdx.AddDef(def)
+	if err == nil {
+		c.writeQueue <- writeReq{recvTime: time.Now(), def: def}
+	}
+	return err
 }
 
 func (c *CasIdx) rebuildIndex() {

idx/elasticsearch/elasticsearch.go

@@ -214,29 +214,32 @@ func (e *EsIdx) Init(stats met.Backend) error {
 	return nil
 }
 
-func (e *EsIdx) Add(data *schema.MetricData) {
+func (e *EsIdx) Add(data *schema.MetricData) error {
 	existing, err := e.MemoryIdx.Get(data.Id)
 	inMemory := true
 	if err != nil {
 		if err == idx.DefNotFound {
 			inMemory = false
 		} else {
 			log.Error(3, "Failed to query Memory Index for %s. %s", data.Id, err)
-			return
+			return err
 		}
 	}
 	if inMemory {
 		log.Debug("def already seen before. Just updating memory Index")
 		existing.LastUpdate = data.Time
 		e.MemoryIdx.AddDef(&existing)
-		return
+		return nil
 	}
 	def := schema.MetricDefinitionFromMetricData(data)
-	e.MemoryIdx.AddDef(def)
-	if err := e.BulkIndexer.Index(esIndex, "metric_index", def.Id, "", "", nil, def); err != nil {
-		log.Error(3, "Failed to add metricDef to BulkIndexer queue. %s", err)
-		e.retryBuf.Queue(def.Id)
+	err = e.MemoryIdx.AddDef(def)
+	if err == nil {
+		if err = e.BulkIndexer.Index(esIndex, "metric_index", def.Id, "", "", nil, def); err != nil {
+			log.Error(3, "Failed to add metricDef to BulkIndexer queue. %s", err)
+			e.retryBuf.Queue(def.Id)
+		}
 	}
+	return err
 }
 
 func (e *EsIdx) bulkSend(buf *bytes.Buffer) error {

idx/idx.go

@@ -11,7 +11,9 @@ import (
 )
 
 var (
-	DefNotFound = errors.New("MetricDef not found")
+	DefNotFound       = errors.New("MetricDef not found")
+	BothBranchAndLeaf = errors.New("node can't be both branch and leaf")
+	BranchUnderLeaf   = errors.New("can't add branch under leaf")
 )
 
 type Node struct {
@@ -46,7 +48,7 @@ Interface
 * Stop():
  This will be called when metrictank is shutting down.
 
-* Add(*schema.MetricData):
+* Add(*schema.MetricData) error:
   Every metric received will result in a call to this method to ensure the
   metric has been added to the index.
 
@@ -83,7 +85,7 @@ Interface
 type MetricIndex interface {
 	Init(met.Backend) error
 	Stop()
-	Add(*schema.MetricData)
+	Add(*schema.MetricData) error
 	Get(string) (schema.MetricDefinition, error)
 	Delete(int, string) ([]schema.MetricDefinition, error)
 	Find(int, string, int64) ([]Node, error)

idx/memory/memory.go

@@ -34,7 +34,7 @@ func ConfigSetup() {
 }
 
 type Tree struct {
-	Items map[string]*Node
+	Items map[string]*Node // key is the full path of the node.
 }
 
 type Node struct {
@@ -54,14 +54,16 @@ func (n *Node) String() string {
 // Implements the the "MetricIndex" interface
 type MemoryIdx struct {
 	sync.RWMutex
-	DefById map[string]*schema.MetricDefinition
-	Tree    map[int]*Tree
+	FailedDefs map[string]error
+	DefById    map[string]*schema.MetricDefinition
+	Tree       map[int]*Tree
 }
 
 func New() *MemoryIdx {
 	return &MemoryIdx{
-		DefById: make(map[string]*schema.MetricDefinition),
-		Tree:    make(map[int]*Tree),
+		FailedDefs: make(map[string]error),
+		DefById:    make(map[string]*schema.MetricDefinition),
+		Tree:       make(map[int]*Tree),
 	}
 }
 
@@ -80,22 +82,32 @@ func (m *MemoryIdx) Stop() {
 	return
 }
 
-func (m *MemoryIdx) Add(data *schema.MetricData) {
+func (m *MemoryIdx) Add(data *schema.MetricData) error {
 	pre := time.Now()
 	m.Lock()
 	defer m.Unlock()
+	err, ok := m.FailedDefs[data.Id]
+	if ok {
+		// if it failed before, it would fail again.
+		// there's not much point in doing the work of trying over
+		// and over again, and flooding the logs with the same failure.
+		// so just trigger the stats metric as if we tried again
+		idxFail.Inc(1)
+		return err
+	}
 	existing, ok := m.DefById[data.Id]
 	if ok {
 		log.Debug("metricDef with id %s already in index.", data.Id)
 		existing.LastUpdate = data.Time
 		idxOk.Inc(1)
 		idxAddDuration.Value(time.Since(pre))
-		return
+		return nil
 	}
 
 	def := schema.MetricDefinitionFromMetricData(data)
-	m.add(def)
+	err = m.add(def)
 	idxAddDuration.Value(time.Since(pre))
+	return err
 }
 
 // Used to rebuild the index from an existing set of metricDefinitions.
@@ -114,7 +126,7 @@ func (m *MemoryIdx) Load(defs []schema.MetricDefinition) {
 	m.Unlock()
 }
 
-func (m *MemoryIdx) AddDef(def *schema.MetricDefinition) {
+func (m *MemoryIdx) AddDef(def *schema.MetricDefinition) error {
 	pre := time.Now()
 	m.Lock()
 	defer m.Unlock()
@@ -123,14 +135,14 @@ func (m *MemoryIdx) AddDef(def *schema.MetricDefinition) {
 		existing.LastUpdate = def.LastUpdate
 		idxOk.Inc(1)
 		idxAddDuration.Value(time.Since(pre))
-		return
+		return nil
 	}
-	m.add(def)
+	err := m.add(def)
 	idxAddDuration.Value(time.Since(pre))
+	return err
 }
 
-func (m *MemoryIdx) add(def *schema.MetricDefinition) {
-	m.DefById[def.Id] = def
+func (m *MemoryIdx) add(def *schema.MetricDefinition) error {
 	path := def.Name
 	//first check to see if a tree has been created for this OrgId
 	tree, ok := m.Tree[def.OrgId]
@@ -153,28 +165,36 @@ func (m *MemoryIdx) add(def *schema.MetricDefinition) {
 			if !node.Leaf {
 				//bad data. A path cant be both a leaf and a branch.
 				log.Info("memory-idx: Bad data, a path can not be both a leaf and a branch. %d - %s", def.OrgId, path)
+				m.FailedDefs[def.Id] = idx.BothBranchAndLeaf
 				idxFail.Inc(1)
-				return
+				return idx.BothBranchAndLeaf
 			}
 			log.Debug("memory-idx: existing index entry for %s. Adding %s as child", path, def.Id)
 			node.Children = append(node.Children, def.Id)
+			m.DefById[def.Id] = def
 			idxOk.Inc(1)
-			return
+			return nil
 		}
 	}
 	// now walk backwards through the node path to find the first branch which exists that
 	// this path extends.
 	nodes := strings.Split(path, ".")
-	startPos := 0
+
+	// if we're trying to insert foo.bar.baz.quux then we see if we can insert it under (in this order):
+	// - foo.bar.baz (if found, startPos is 3)
+	// - foo.bar (if found, startPos is 2)
+	// - foo (if found, startPos is 1)
+	startPos := 0 // the index of the first word that is not part of the prefix
 	var startNode *Node
 	if len(nodes) > 1 {
 		for i := len(nodes) - 1; i > 0; i-- {
 			branch := strings.Join(nodes[0:i], ".")
 			if n, ok := tree.Items[branch]; ok {
 				if n.Leaf {
 					log.Info("memory-idx: Branches cant be added to a leaf node. %d - %s", def.OrgId, path)
+					m.FailedDefs[def.Id] = idx.BranchUnderLeaf
 					idxFail.Inc(1)
-					return
+					return idx.BranchUnderLeaf
 				}
 				log.Debug("memory-idx: Found branch %s which metricDef %s is a descendant of", branch, path)
 				startNode = n
@@ -211,8 +231,9 @@ func (m *MemoryIdx) add(def *schema.MetricDefinition) {
 		Path:     path,
 		Children: []string{def.Id},
 	}
+	m.DefById[def.Id] = def
 	idxOk.Inc(1)
-	return
+	return nil
 }
 
 func (m *MemoryIdx) Get(id string) (schema.MetricDefinition, error) {
@@ -285,6 +306,11 @@ func (m *MemoryIdx) find(orgId int, pattern string) ([]*Node, error) {
 	}
 
 	nodes := strings.Split(pattern, ".")
+
+	// pos is the index of the last node we know for sure
+	// for a query like foo.bar.baz, pos is 2
+	// for a query like foo.bar.* or foo.bar, pos is 1
+	// for a query like foo.b*.baz, pos is 0
 	pos := len(nodes) - 1
 	for i := 0; i < len(nodes); i++ {
 		if strings.ContainsAny(nodes[i], "*{}[]?") {
@@ -435,6 +461,12 @@ func (m *MemoryIdx) Delete(orgId int, pattern string) ([]schema.MetricDefinition
 	if err != nil {
 		return nil, err
 	}
+
+	// by deleting one or more nodes in the tree, any defs that previously failed may now
+	// be able to be added. An easy way to support this is just reset this map and give them
+	// all a chance again
+	m.FailedDefs = make(map[string]error)
+
 	deletedDefs := make([]schema.MetricDefinition, 0)
 	for _, f := range found {
 		deleted, err := m.delete(orgId, f)
@@ -469,7 +501,7 @@ func (m *MemoryIdx) delete(orgId int, n *Node) ([]schema.MetricDefinition, error
 	deletedDefs := make([]schema.MetricDefinition, len(n.Children))
 	// delete the metricDefs
 	for i, id := range n.Children {
-		log.Debug("memory-idx: deleteing %s from index", id)
+		log.Debug("memory-idx: deleting %s from index", id)
 		deletedDefs[i] = *m.DefById[id]
 		delete(m.DefById, id)
 	}