From 7a577b86b7dcbeffab92144f79f9715505dcbbb7 Mon Sep 17 00:00:00 2001
From: Julius Volz <julius@soundcloud.com>
Date: Tue, 25 Mar 2014 19:07:27 +0100
Subject: [PATCH] Fix interval op special case.

In the case that a getValuesAtIntervalOp's ExtractSamples() is called
with a current time after the last chunk time, we return without
extracting any further values beyond the last one in the chunk
(correct), but also without advancing the op's time (incorrect). This
leads to an infinite loop in renderView(), since the op is called
repeatedly without ever being advanced and consumed.

This adds handling for this special case. When detecting this case, we
immediately set the op to be consumed, since we would always get a value
after the current time passed in if there was one.

Change-Id: Id99149e07b5188d655331382b8b6a461b677005c
---
 storage/metric/operation.go   |  7 ++++
 storage/metric/tiered_test.go | 66 +++++++++++++++++++++++++++++++++++
 2 files changed, 73 insertions(+)

diff --git a/storage/metric/operation.go b/storage/metric/operation.go
index dbf484c847..d6cb93638a 100644
--- a/storage/metric/operation.go
+++ b/storage/metric/operation.go
@@ -203,7 +203,14 @@ func (g *getValuesAtIntervalOp) ExtractSamples(in Values) (out Values) {
 	if len(in) == 0 {
 		return
 	}
+
 	lastChunkTime := in[len(in)-1].Timestamp
+
+	if g.current.After(lastChunkTime) {
+		g.current = g.through.Add(clientmodel.MinimumTick)
+		return Values{in[len(in)-1]}
+	}
+
 	for len(in) > 0 {
 		out = append(out, extractValuesAroundTime(g.current, in)...)
 		if g.current.After(lastChunkTime) {
diff --git a/storage/metric/tiered_test.go b/storage/metric/tiered_test.go
index 25685dc204..75ea8a3e08 100644
--- a/storage/metric/tiered_test.go
+++ b/storage/metric/tiered_test.go
@@ -524,6 +524,72 @@ func testMakeView(t test.Tester, flushToDisk bool) {
 				// extracted value at the end of the second chunk.
 				diskOnly: true,
 			},
+			// Single sample, getValuesAtIntervalOp starting after the sample.
+			{
+				data: clientmodel.Samples{
+					{
+						Metric:    metric,
+						Value:     0,
+						Timestamp: instant,
+					},
+				},
+				in: in{
+					atInterval: []getValuesAtIntervalOp{
+						{
+							getValuesAlongRangeOp: getValuesAlongRangeOp{
+								baseOp:  baseOp{current: instant.Add(time.Second)},
+								through: instant.Add(time.Second * 2),
+							},
+							interval: time.Second,
+						},
+					},
+				},
+				out: out{
+					atInterval: []Values{
+						{
+							{
+								Timestamp: instant,
+								Value:     0,
+							},
+						},
+					},
+				},
+			},
+			// Single sample, getValuesAtIntervalOp starting before the sample.
+			{
+				data: clientmodel.Samples{
+					{
+						Metric:    metric,
+						Value:     0,
+						Timestamp: instant.Add(time.Second),
+					},
+				},
+				in: in{
+					atInterval: []getValuesAtIntervalOp{
+						{
+							getValuesAlongRangeOp: getValuesAlongRangeOp{
+								baseOp:  baseOp{current: instant},
+								through: instant.Add(time.Second * 2),
+							},
+							interval: time.Second,
+						},
+					},
+				},
+				out: out{
+					atInterval: []Values{
+						{
+							{
+								Timestamp: instant.Add(time.Second),
+								Value:     0,
+							},
+							{
+								Timestamp: instant.Add(time.Second),
+								Value:     0,
+							},
+						},
+					},
+				},
+			},
 		}
 	)