adds two more unit tests which are testing:

- does aggmetric call the cache push callback every time a metric gets
  evicted from the ring buffer?
- does aggmetric add the chunk into the store if the node is primary?
- does aggmetric not add the chunk into the store if the node is not primary?

mdata/aggmetric_test.go

@@ -73,6 +73,68 @@ func (c *Checker) Verify(primary bool, from, to, first, last uint32) {
 	cluster.ThisNode.SetPrimary(currentClusterStatus)
 }
 
+func TestMetricPersistBeingPrimary(t *testing.T) {
+	testMetricPersistOptionalPrimary(t, true)
+}
+
+func TestMetricPersistNotBeingPrimary(t *testing.T) {
+	testMetricPersistOptionalPrimary(t, false)
+}
+
+func testMetricPersistOptionalPrimary(t *testing.T, primary bool) {
+	// always reset the counter when entering and leaving the test
+	dnstore.Reset()
+	defer dnstore.Reset()
+
+	cluster.Init("default", "test", time.Now())
+	cluster.ThisNode.SetPrimary(primary)
+
+	callCount := uint32(0)
+	calledCb := make(chan bool)
+	var countingCacheCb cache.CacheCb = func(string, uint32, *chunk.IterGen) {
+		calledCb <- true
+	}
+
+	numChunks, chunkAddCount, chunkSpan := uint32(5), uint32(10), uint32(300)
+	agg := NewAggMetric(dnstore, countingCacheCb, "foo", chunkSpan, numChunks, 1, []AggSetting{}...)
+
+	ts := uint32(1000)
+	for i := uint32(0); i < chunkAddCount; i++ {
+		agg.Add(ts, 1)
+		ts = ts + chunkSpan
+	}
+
+	timeout := make(chan bool, 1)
+	oneSecTimeout := func() {
+		time.Sleep(1 * time.Second)
+		timeout <- true
+	}
+
+	for i := uint32(0); i < chunkAddCount-1; i++ {
+		go oneSecTimeout()
+		select {
+		case <-timeout:
+			t.Fatalf("timed out waiting for a callback call")
+		case <-calledCb:
+			callCount = callCount + 1
+		}
+	}
+
+	if callCount < chunkAddCount-1 {
+		t.Fatalf("there should have been %d chunk pushes, but go %d", chunkAddCount-1, callCount)
+	}
+
+	if primary {
+		if dnstore.AddCount != chunkAddCount-1 {
+			t.Fatalf("there should have been %d chunk adds on store, but go %d", chunkAddCount-1, dnstore.AddCount)
+		}
+	} else {
+		if dnstore.AddCount != 0 {
+			t.Fatalf("there should have been %d chunk adds on store, but go %d", 0, dnstore.AddCount)
+		}
+	}
+}
+
 func TestAggMetric(t *testing.T) {
 	cluster.Init("default", "test", time.Now())
 

mdata/store_devnull.go

@@ -3,6 +3,7 @@ package mdata
 import "github.com/raintank/metrictank/mdata/chunk"
 
 type devnullStore struct {
+	AddCount uint32
 }
 
 func NewDevnullStore() *devnullStore {
@@ -11,6 +12,11 @@ func NewDevnullStore() *devnullStore {
 }
 
 func (c *devnullStore) Add(cwr *ChunkWriteRequest) {
+	c.AddCount++
+}
+
+func (c *devnullStore) Reset() {
+	c.AddCount = 0
 }
 
 func (c *devnullStore) Search(key string, start, end uint32) ([]chunk.IterGen, error) {