Get should not return first chunk when not primary node

The first chunk recorded is likely a partial chunk.  When not the
cluster primary dont serve the first chunk out of memory. Instead
the chunk will be fetched from Cassandra.

Author: woodsaj

metric_tank/aggmetric.go

@@ -29,6 +29,7 @@ type AggMetric struct {
 	aggregators     []*Aggregator
 	writeQueue      chan *Chunk
 	activeWrite     bool
+	firstChunkT0    uint32
 }
 
 // re-order the chunks with the oldest at start of the list and newest at the end.
@@ -254,6 +255,21 @@ func (a *AggMetric) Get(from, to uint32) (uint32, []Iter) {
 		return math.MaxInt32, make([]Iter, 0)
 	}
 
+	// The first chunk is likely only a partial chunk. If we are not the primary node
+	// we should not serve data from this chunk, and should instead get the chunk from cassandra.
+	// if we are the primary node, then there is likely no data in Cassandra anyway.
+	if !clusterStatus.IsPrimary() && oldestChunk.T0 == a.firstChunkT0 {
+		oldestPos++
+		if oldestPos >= len(a.Chunks) {
+			oldestPos = 0
+		}
+		oldestChunk = a.getChunk(oldestPos)
+		if oldestChunk == nil {
+			log.Error(3, "unexpected nil chunk.")
+			return math.MaxInt32, make([]Iter, 0)
+		}
+	}
+
 	if to <= oldestChunk.T0 {
 		// the requested time range ends before any data we have.
 		log.Debug("AggMetric %s Get(): no data for requested range", a.Key)
@@ -446,6 +462,10 @@ func (a *AggMetric) Add(ts uint32, val float64) {
 		// no data has been added to this metric at all.
 		a.Chunks = append(a.Chunks, NewChunk(t0))
 
+		// The first chunk is typically going to be a partial chunk
+		// so we keep a record of it.
+		a.firstChunkT0 = t0
+
 		if err := a.Chunks[0].Push(ts, val); err != nil {
 			panic(fmt.Sprintf("FATAL ERROR: this should never happen. Pushing initial value <%d,%f> to new chunk at pos 0 failed: %q", ts, val, err))
 		}

metric_tank/aggmetric_test.go

@@ -34,7 +34,9 @@ func (c *Checker) Add(ts uint32, val float64) {
 // from to is the range that gets requested from AggMetric
 // first/last is what we use as data range to compare to (both inclusive)
 // these may be different because AggMetric returns broader rangers (due to packed format),
-func (c *Checker) Verify(from, to, first, last uint32) {
+func (c *Checker) Verify(primary bool, from, to, first, last uint32) {
+	currentClusterStatus := clusterStatus.IsPrimary()
+	clusterStatus.Set(primary)
 	_, iters := c.agg.Get(from, to)
 	// we don't do checking or fancy logic, it is assumed that the caller made sure first and last are ts of actual points
 	var pi int // index of first point we want
@@ -61,6 +63,7 @@ func (c *Checker) Verify(from, to, first, last uint32) {
 	if index != pj {
 		c.t.Fatalf("not all values returned. missing %v", c.points[index:pj+1])
 	}
+	clusterStatus.Set(currentClusterStatus)
 }
 
 func TestAggMetric(t *testing.T) {
@@ -72,52 +75,55 @@ func TestAggMetric(t *testing.T) {
 
 	// basic case, single range
 	c.Add(101, 101)
-	c.Verify(100, 200, 101, 101)
+	c.Verify(true, 100, 200, 101, 101)
 	c.Add(105, 105)
-	c.Verify(100, 199, 101, 105)
+	c.Verify(true, 100, 199, 101, 105)
 	c.Add(115, 115)
 	c.Add(125, 125)
 	c.Add(135, 135)
-	c.Verify(100, 199, 101, 135)
+	c.Verify(true, 100, 199, 101, 135)
 
 	// add new ranges, aligned and unaligned
 	c.Add(200, 200)
 	c.Add(315, 315)
-	c.Verify(100, 399, 101, 315)
+	c.Verify(true, 100, 399, 101, 315)
+
+	// verify as secondary node. Data from the first chunk should not be returned.
+	c.Verify(false, 100, 399, 200, 315)
 
 	// get subranges
-	c.Verify(120, 299, 101, 200)
-	c.Verify(220, 299, 200, 200)
-	c.Verify(312, 330, 315, 315)
+	c.Verify(true, 120, 299, 101, 200)
+	c.Verify(true, 220, 299, 200, 200)
+	c.Verify(true, 312, 330, 315, 315)
 
 	// border dancing. good for testing inclusivity and exclusivity
-	c.Verify(100, 199, 101, 135)
-	c.Verify(100, 200, 101, 135)
-	c.Verify(100, 201, 101, 200)
-	c.Verify(198, 199, 101, 135)
-	c.Verify(199, 200, 101, 135)
-	c.Verify(200, 201, 200, 200)
-	c.Verify(201, 202, 200, 200)
-	c.Verify(299, 300, 200, 200)
-	c.Verify(300, 301, 315, 315)
+	c.Verify(true, 100, 199, 101, 135)
+	c.Verify(true, 100, 200, 101, 135)
+	c.Verify(true, 100, 201, 101, 200)
+	c.Verify(true, 198, 199, 101, 135)
+	c.Verify(true, 199, 200, 101, 135)
+	c.Verify(true, 200, 201, 200, 200)
+	c.Verify(true, 201, 202, 200, 200)
+	c.Verify(true, 299, 300, 200, 200)
+	c.Verify(true, 300, 301, 315, 315)
 
 	// skipping
 	c.Add(510, 510)
 	c.Add(512, 512)
-	c.Verify(100, 599, 101, 512)
+	c.Verify(true, 100, 599, 101, 512)
 
 	// basic wraparound
 	c.Add(610, 610)
 	c.Add(612, 612)
 	c.Add(710, 710)
 	c.Add(712, 712)
 	// TODO would be nice to test that it panics when requesting old range. something with recover?
-	//c.Verify(100, 799, 101, 512)
+	//c.Verify(true, 100, 799, 101, 512)
 
 	// largest range we have so far
-	c.Verify(300, 799, 315, 712)
+	c.Verify(true, 300, 799, 315, 712)
 	// a smaller range
-	c.Verify(502, 799, 510, 712)
+	c.Verify(true, 502, 799, 510, 712)
 
 	// the circular buffer had these ranges:
 	// 100 200 300 skipped 500
@@ -130,7 +136,7 @@ func TestAggMetric(t *testing.T) {
 	// but we just check we only get this point
 	c.Add(1299, 1299)
 	// TODO: implement skips and enable this
-	//	c.Verify(800, 1299, 1299, 1299)
+	//	c.Verify(true, 800, 1299, 1299, 1299)
 }
 
 // basic expected RAM usage for 1 iteration (= 1 days)

metric_tank/aggregator_test.go

@@ -36,7 +36,9 @@ func TestAggBoundary(t *testing.T) {
 
 // note that values don't get "committed" to the metric until the aggregation interval is complete
 func TestAggregator(t *testing.T) {
+	clusterStatus = NewClusterStatus("default", false)
 	compare := func(key string, metric Metric, expected []Point) {
+		clusterStatus.Set(true)
 		_, iters := metric.Get(0, 1000)
 		got := make([]Point, 0, len(expected))
 		for _, iter := range iters {
@@ -56,6 +58,7 @@ func TestAggregator(t *testing.T) {
 				}
 			}
 		}
+		clusterStatus.Set(false)
 	}
 	agg := NewAggregator("test", 60, 120, 10, 10)
 	agg.Add(100, 123.4)