add writeQueue buffer to memoryIdx

Add support for a writeQueue  that holds new MetricDefinitions in a
buffer, then writes them into the index as a single batch.

Author: woodsaj

docker/docker-chaos/metrictank.ini

@@ -454,6 +454,12 @@ find-cache-invalidate-max-size = 100
 find-cache-invalidate-max-wait = 5s
 # amount of time to disable the findCache when the invalidate queue fills up.
 find-cache-backoff-time = 60s
+# enable buffering new metricDefinitions and writing them to the index in batches
+write-queue-enabled = false
+# maximum delay between flushing buffered metric writes to the index
+write-queue-delay = 30s
+# maximum number of metricDefinitions that can be added to the index in a single batch
+write-max-batch-size = 5000
 
 ### Bigtable index
 [bigtable-idx]

docker/docker-cluster/metrictank.ini

@@ -454,6 +454,12 @@ find-cache-invalidate-max-size = 100
 find-cache-invalidate-max-wait = 5s
 # amount of time to disable the findCache when the invalidate queue fills up.
 find-cache-backoff-time = 60s
+# enable buffering new metricDefinitions and writing them to the index in batches
+write-queue-enabled = false
+# maximum delay between flushing buffered metric writes to the index
+write-queue-delay = 30s
+# maximum number of metricDefinitions that can be added to the index in a single batch
+write-max-batch-size = 5000
 
 ### Bigtable index
 [bigtable-idx]

docker/docker-dev-custom-cfg-kafka/metrictank.ini

@@ -454,6 +454,12 @@ find-cache-invalidate-max-size = 100
 find-cache-invalidate-max-wait = 5s
 # amount of time to disable the findCache when the invalidate queue fills up.
 find-cache-backoff-time = 60s
+# enable buffering new metricDefinitions and writing them to the index in batches
+write-queue-enabled = false
+# maximum delay between flushing buffered metric writes to the index
+write-queue-delay = 30s
+# maximum number of metricDefinitions that can be added to the index in a single batch
+write-max-batch-size = 5000
 
 ### Bigtable index
 [bigtable-idx]

docs/config.md

@@ -528,6 +528,12 @@ find-cache-invalidate-max-size = 100
 find-cache-invalidate-max-wait = 5s
 # amount of time to disable the findCache when the invalidate queue fills up.
 find-cache-backoff-time = 60s
+# enable buffering new metricDefinitions and writing them to the index in batches
+write-queue-enabled = false
+# maximum delay between flushing buffered metric writes to the index
+write-queue-delay = 30s
+# maximum number of metricDefinitions that can be added to the index in a single batch
+write-max-batch-size = 5000
 ```
 
 ### Bigtable index

idx/bigtable/bigtable.go

@@ -286,7 +286,7 @@ func (b *BigtableIdx) updateBigtable(now uint32, inMemory bool, archive idx.Arch
 		log.Debugf("bigtable-idx: updating def %s in index.", archive.MetricDefinition.Id)
 		b.writeQueue <- writeReq{recvTime: time.Now(), def: &archive.MetricDefinition}
 		archive.LastSave = now
-		b.MemoryIndex.UpdateArchive(archive)
+		b.MemoryIndex.UpdateArchiveLastSave(archive.Id, archive.Partition, now)
 	} else {
 		// perform a non-blocking write to the writeQueue. If the queue is full, then
 		// this will fail and we won't update the LastSave timestamp. The next time
@@ -297,7 +297,7 @@ func (b *BigtableIdx) updateBigtable(now uint32, inMemory bool, archive idx.Arch
 		select {
 		case b.writeQueue <- writeReq{recvTime: time.Now(), def: &archive.MetricDefinition}:
 			archive.LastSave = now
-			b.MemoryIndex.UpdateArchive(archive)
+			b.MemoryIndex.UpdateArchiveLastSave(archive.Id, archive.Partition, now)
 		default:
 			statSaveSkipped.Inc()
 			log.Debugf("bigtable-idx: writeQueue is full, update of %s not saved this time", archive.MetricDefinition.Id)

idx/cassandra/cassandra.go

@@ -320,7 +320,7 @@ func (c *CasIdx) updateCassandra(now uint32, inMemory bool, archive idx.Archive,
 		log.Debugf("cassandra-idx: updating def %s in index.", archive.MetricDefinition.Id)
 		c.writeQueue <- writeReq{recvTime: time.Now(), def: &archive.MetricDefinition}
 		archive.LastSave = now
-		c.MemoryIndex.UpdateArchive(archive)
+		c.MemoryIndex.UpdateArchiveLastSave(archive.Id, archive.Partition, now)
 	} else {
 		// perform a non-blocking write to the writeQueue. If the queue is full, then
 		// this will fail and we won't update the LastSave timestamp. The next time
@@ -331,7 +331,7 @@ func (c *CasIdx) updateCassandra(now uint32, inMemory bool, archive idx.Archive,
 		select {
 		case c.writeQueue <- writeReq{recvTime: time.Now(), def: &archive.MetricDefinition}:
 			archive.LastSave = now
-			c.MemoryIndex.UpdateArchive(archive)
+			c.MemoryIndex.UpdateArchiveLastSave(archive.Id, archive.Partition, now)
 		default:
 			statSaveSkipped.Inc()
 			log.Debugf("cassandra-idx: writeQueue is full, update of %s not saved this time.", archive.MetricDefinition.Id)

idx/cassandra/cassandra_test.go

@@ -248,7 +248,7 @@ func TestAddToWriteQueue(t *testing.T) {
 
 				archive, _ := ix.Get(mkey)
 				archive.LastSave = uint32(time.Now().Unix() - 100)
-				ix.UpdateArchive(archive)
+				ix.UpdateArchiveLastSave(archive.Id, archive.Partition, archive.LastSave)
 			}
 			for _, s := range metrics {
 				mkey, err := schema.MKeyFromString(s.Id)

idx/memory/memory.go

@@ -61,6 +61,9 @@ var (
 	findCacheInvalidateMaxSize   = 100
 	findCacheInvalidateMaxWait   = 5 * time.Second
 	findCacheBackoffTime         = time.Minute
+	writeQueueEnabled            = false
+	writeQueueDelay              = 30 * time.Second
+	writeMaxBatchSize            = 5000
 )
 
 func ConfigSetup() {
@@ -73,6 +76,9 @@ func ConfigSetup() {
 	memoryIdx.IntVar(&findCacheSize, "find-cache-size", 1000, "number of find expressions to cache (per org). 0 disables cache")
 	memoryIdx.IntVar(&findCacheInvalidateQueueSize, "find-cache-invalidate-queue-size", 200, "size of queue for invalidating findCache entries")
 	memoryIdx.IntVar(&findCacheInvalidateMaxSize, "find-cache-invalidate-max-size", 100, "max amount of invalidations to queue up in one batch")
+	memoryIdx.BoolVar(&writeQueueEnabled, "write-queue-enabled", false, "enable buffering new metricDefinitions and writing them to the index in batches")
+	memoryIdx.DurationVar(&writeQueueDelay, "write-queue-delay", 30*time.Second, "maximum delay between flushing buffered metric writes to the index")
+	memoryIdx.IntVar(&writeMaxBatchSize, "write-max-batch-size", 5000, "maximum number of metricDefinitions that can be added to the index in a single batch")
 	memoryIdx.DurationVar(&findCacheInvalidateMaxWait, "find-cache-invalidate-max-wait", 5*time.Second, "max duration to wait building up a batch to invalidate")
 	memoryIdx.DurationVar(&findCacheBackoffTime, "find-cache-backoff-time", time.Minute, "amount of time to disable the findCache when the invalidate queue fills up.")
 	memoryIdx.StringVar(&indexRulesFile, "rules-file", "/etc/metrictank/index-rules.conf", "path to index-rules.conf file")
@@ -109,8 +115,8 @@ func ConfigProcess() {
 type MemoryIndex interface {
 	idx.MetricIndex
 	LoadPartition(int32, []schema.MetricDefinition) int
-	UpdateArchive(idx.Archive)
-	add(*schema.MetricDefinition) idx.Archive
+	UpdateArchiveLastSave(schema.MKey, int32, uint32)
+	add(*idx.Archive)
 	idsByTagQuery(uint32, TagQuery) IdSet
 	PurgeFindCache()
 	ForceInvalidationFindCache()
@@ -254,24 +260,29 @@ type UnpartitionedMemoryIdx struct {
 	metaTagRecords map[uint32]metaTagRecords // by orgId
 
 	findCache *FindCache
+
+	writeQueue *WriteQueue
 }
 
 func NewUnpartitionedMemoryIdx() *UnpartitionedMemoryIdx {
-	m := UnpartitionedMemoryIdx{
+	m := &UnpartitionedMemoryIdx{
 		defById:        make(map[schema.MKey]*idx.Archive),
 		defByTagSet:    make(defByTagSet),
 		tree:           make(map[uint32]*Tree),
 		tags:           make(map[uint32]TagIndex),
 		metaTags:       make(map[uint32]metaTagIndex),
 		metaTagRecords: make(map[uint32]metaTagRecords),
 	}
-	return &m
+	return m
 }
 
 func (m *UnpartitionedMemoryIdx) Init() error {
 	if findCacheSize > 0 {
 		m.findCache = NewFindCache(findCacheSize, findCacheInvalidateQueueSize, findCacheInvalidateMaxSize, findCacheInvalidateMaxWait, findCacheBackoffTime)
 	}
+	if writeQueueEnabled {
+		m.writeQueue = NewWriteQueue(m, writeQueueDelay, writeMaxBatchSize)
+	}
 	return nil
 }
 
@@ -280,6 +291,10 @@ func (m *UnpartitionedMemoryIdx) Stop() {
 		m.findCache.Shutdown()
 		m.findCache = nil
 	}
+	if m.writeQueue != nil {
+		m.writeQueue.Stop()
+		m.writeQueue = nil
+	}
 	return
 }
 
@@ -303,7 +318,6 @@ func (m *UnpartitionedMemoryIdx) Update(point schema.MetricPoint, partition int3
 	pre := time.Now()
 
 	m.RLock()
-	defer m.RUnlock()
 
 	existing, ok := m.defById[point.MKey]
 	if ok {
@@ -316,8 +330,45 @@ func (m *UnpartitionedMemoryIdx) Update(point schema.MetricPoint, partition int3
 		oldPart := atomic.SwapInt32(&existing.Partition, partition)
 		statUpdate.Inc()
 		statUpdateDuration.Value(time.Since(pre))
+		m.RUnlock()
 		return *existing, oldPart, true
 	}
+	m.RUnlock()
+
+	if m.writeQueue != nil {
+		// if we are using the writeQueue, then the archive for this MKey might be queued
+		// and not yet flushed to the index yet.
+		existing, ok := m.writeQueue.Get(point.MKey)
+		if ok {
+			if log.IsLevelEnabled(log.DebugLevel) {
+				log.Debugf("memory-idx: metricDef with id %v is in the writeQueue", point.MKey)
+			}
+
+			bumpLastUpdate(&existing.LastUpdate, int64(point.Time))
+
+			oldPart := atomic.SwapInt32(&existing.Partition, partition)
+			statUpdate.Inc()
+			statUpdateDuration.Value(time.Since(pre))
+			return *existing, oldPart, true
+		}
+
+		// we need to do one final check of m.defById, as the writeQueue may have been flushed between
+		// when we released m.RLock() and when the call to m.writeQueue.Get() as able to obtain its own lock.
+		existing, ok = m.defById[point.MKey]
+		if ok {
+			if log.IsLevelEnabled(log.DebugLevel) {
+				log.Debugf("memory-idx: metricDef with id %v already in index", point.MKey)
+			}
+
+			bumpLastUpdate(&existing.LastUpdate, int64(point.Time))
+
+			oldPart := atomic.SwapInt32(&existing.Partition, partition)
+			statUpdate.Inc()
+			statUpdateDuration.Value(time.Since(pre))
+			m.RUnlock()
+			return *existing, oldPart, true
+		}
+	}
 
 	return idx.Archive{}, 0, false
 }
@@ -345,26 +396,34 @@ func (m *UnpartitionedMemoryIdx) AddOrUpdate(mkey schema.MKey, data *schema.Metr
 	}
 
 	m.RUnlock()
-	m.Lock()
-	defer m.Unlock()
-
 	def := schema.MetricDefinitionFromMetricData(data)
 	def.Partition = partition
-	archive := m.add(def)
-	statMetricsActive.Inc()
-	statAddDuration.Value(time.Since(pre))
+	archive := getArchive(def)
+	if m.writeQueue == nil {
+		m.Lock()
+		m.add(archive)
+		m.Unlock()
+		statMetricsActive.Inc()
+		statAddDuration.Value(time.Since(pre))
+	} else {
+		// push the new archive into the writeQueue
+		m.writeQueue.Queue(archive)
+	}
 
-	return archive, 0, false
+	return *archive, 0, false
 }
 
-// UpdateArchive updates the archive information
-func (m *UnpartitionedMemoryIdx) UpdateArchive(archive idx.Archive) {
-	m.Lock()
-	defer m.Unlock()
-	if _, ok := m.defById[archive.Id]; !ok {
+// UpdateArchiveLastSave updates the LastSave timestamp of the archive
+func (m *UnpartitionedMemoryIdx) UpdateArchiveLastSave(id schema.MKey, partition int32, lastSave uint32) {
+	m.RLock()
+	if _, ok := m.defById[id]; !ok {
+		// This will happen if the metricDefinition was saved to a backend store
+		// before the archive was added to the index.
+		m.RUnlock()
 		return
 	}
-	*(m.defById[archive.Id]) = archive
+	atomic.StoreUint32(&m.defById[id].LastSave, lastSave)
+	m.RUnlock()
 }
 
 // MetaTagRecordUpsert inserts or updates a meta record, depending on whether
@@ -530,7 +589,7 @@ func (m *UnpartitionedMemoryIdx) Load(defs []schema.MetricDefinition) int {
 			continue
 		}
 
-		m.add(def)
+		m.add(getArchive(def))
 
 		// as we are loading the metricDefs from a persistent store, set the lastSave
 		// to the lastUpdate timestamp.  This won't exactly match the true lastSave Timstamp,
@@ -545,32 +604,46 @@ func (m *UnpartitionedMemoryIdx) Load(defs []schema.MetricDefinition) int {
 	return num
 }
 
-func (m *UnpartitionedMemoryIdx) add(def *schema.MetricDefinition) idx.Archive {
+func getArchive(def *schema.MetricDefinition) *idx.Archive {
 	path := def.NameWithTags()
 	schemaId, _ := mdata.MatchSchema(path, def.Interval)
 	aggId, _ := mdata.MatchAgg(path)
 	irId, _ := IndexRules.Match(path)
 
-	archive := &idx.Archive{
+	return &idx.Archive{
 		MetricDefinition: *def,
 		SchemaId:         schemaId,
 		AggId:            aggId,
 		IrId:             irId,
 	}
+}
+
+func (m *UnpartitionedMemoryIdx) add(archive *idx.Archive) {
+	// there is a race condition that can lead to an archive being added
+	// to the writeQueue just after a queued copy of the archive was flushed.
+	// If that happens, we update the contents of the passed 'archive' to be
+	// the contents of the 'existing' archive found.
+	if existing, ok := m.defById[archive.Id]; ok {
+		*archive = *existing
+		return
+	}
+
+	def := &archive.MetricDefinition
+	path := def.NameWithTags()
 
 	if TagSupport {
 		// Even if there are no tags, index at least "name". It's important to use the definition
 		// in the archive pointer that we add to defById, because the pointers must reference the
 		// same underlying object in m.defById and m.defByTagSet
-		m.indexTags(&archive.MetricDefinition)
+		m.indexTags(def)
 
 		if len(def.Tags) > 0 {
 			if _, ok := m.defById[def.Id]; !ok {
 				m.defById[def.Id] = archive
 				statAdd.Inc()
 				log.Debugf("memory-idx: adding %s to DefById", path)
 			}
-			return *archive
+			return
 		}
 	}
 
@@ -602,7 +675,7 @@ func (m *UnpartitionedMemoryIdx) add(def *schema.MetricDefinition) idx.Archive {
 			node.Defs = append(node.Defs, def.Id)
 			m.defById[def.Id] = archive
 			statAdd.Inc()
-			return *archive
+			return
 		}
 	}
 
@@ -649,7 +722,7 @@ func (m *UnpartitionedMemoryIdx) add(def *schema.MetricDefinition) idx.Archive {
 	m.defById[def.Id] = archive
 	statAdd.Inc()
 
-	return *archive
+	return
 }
 
 func (m *UnpartitionedMemoryIdx) Get(id schema.MKey) (idx.Archive, bool) {
@@ -1140,7 +1213,10 @@ func (m *UnpartitionedMemoryIdx) Find(orgId uint32, pattern string, from int64)
 						log.Debugf("memory-idx: from is %d, so skipping %s which has LastUpdate %d", from, def.Id, atomic.LoadInt64(&def.LastUpdate))
 						continue
 					}
-					log.Debugf("memory-idx: Find: adding to path %s archive id=%s name=%s int=%d schemaId=%d aggId=%d irId=%d lastSave=%d", n.Path, def.Id, def.Name, def.Interval, def.SchemaId, def.AggId, def.IrId, def.LastSave)
+					if log.IsLevelEnabled(log.DebugLevel) {
+						lastSave := atomic.LoadUint32(&def.LastSave)
+						log.Debugf("memory-idx: Find: adding to path %s archive id=%s name=%s int=%d schemaId=%d aggId=%d irId=%d lastSave=%d", n.Path, def.Id, def.Name, def.Interval, def.SchemaId, def.AggId, def.IrId, lastSave)
+					}
 					idxNode.Defs = append(idxNode.Defs, *def)
 				}
 				if len(idxNode.Defs) == 0 {

idx/memory/memory_test.go

@@ -1228,7 +1228,7 @@ func testMatchSchemaWithTags(t *testing.T) {
 	defer ix.Stop()
 
 	data := make([]*schema.MetricDefinition, 10)
-	archives := make([]idx.Archive, 10)
+	archives := make([]*idx.Archive, 10)
 	for i := 0; i < 10; i++ {
 		name := fmt.Sprintf("some.id.of.a.metric.%d", i)
 		data[i] = &schema.MetricDefinition{
@@ -1239,7 +1239,8 @@ func testMatchSchemaWithTags(t *testing.T) {
 			Partition: getPartitionFromName(name),
 		}
 		data[i].SetId()
-		archives[i] = ix.add(data[i])
+		archives[i] = getArchive(data[i])
+		ix.add(archives[i])
 	}
 
 	// only those MDs with tag1=value3 or tag1=value5 should get the first schema id