add writeQueue buffer to memoryIdx

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)