From 71a3172abba28bafbc5d5ae5f00eb91533dcd981 Mon Sep 17 00:00:00 2001
From: Julius Volz <julius@soundcloud.com>
Date: Mon, 13 May 2013 18:14:14 +0200
Subject: [PATCH] Fix and optimize getValuesAtIntervalOp data extraction.

- only the data extracted in the last loop iteration of ExtractSamples() was
  emitted as output
- if e.g. op interval < sample interval, there were situations where the same
  sample was added multiple times to the output
---
 storage/metric/operation.go      | 11 +++++---
 storage/metric/operation_test.go | 44 ++++++++++++++++++++++++++++++++
 2 files changed, 52 insertions(+), 3 deletions(-)

diff --git a/storage/metric/operation.go b/storage/metric/operation.go
index 8566f59e3..9c3408c0d 100644
--- a/storage/metric/operation.go
+++ b/storage/metric/operation.go
@@ -156,12 +156,17 @@ func (g *getValuesAtIntervalOp) ExtractSamples(in model.Values) (out model.Value
 		return
 	}
 	lastChunkTime := in[len(in)-1].Timestamp
-	for {
-		out = extractValuesAroundTime(g.from, in)
+	for len(in) > 0 {
+		out = append(out, extractValuesAroundTime(g.from, in)...)
+		lastExtractedTime := out[len(out)-1].Timestamp
+		in = in.TruncateBefore(lastExtractedTime.Add(1))
 		g.from = g.from.Add(g.interval)
-		if g.from.After(lastChunkTime) {
+		if lastExtractedTime.Equal(lastChunkTime) {
 			break
 		}
+		for !g.from.After(lastExtractedTime) {
+			g.from = g.from.Add(g.interval)
+		}
 		if g.from.After(g.through) {
 			break
 		}
diff --git a/storage/metric/operation_test.go b/storage/metric/operation_test.go
index 6ec5eae2d..772f439da 100644
--- a/storage/metric/operation_test.go
+++ b/storage/metric/operation_test.go
@@ -1559,6 +1559,50 @@ func TestGetValuesAtIntervalOp(t *testing.T) {
 				},
 			},
 		},
+		// Operator interval skips over several values and ends past the last
+		// available value. This is to verify that we still include the last value
+		// of a series even if we target a time past it and haven't extracted that
+		// value yet as part of a previous interval step (thus the necessity to
+		// skip over values for the test).
+		{
+			op: getValuesAtIntervalOp{
+				from:     testInstant.Add(30 * time.Second),
+				through:  testInstant.Add(4 * time.Minute),
+				interval: 3 * time.Minute,
+			},
+			in: model.Values{
+				{
+					Timestamp: testInstant,
+					Value:     1,
+				},
+				{
+					Timestamp: testInstant.Add(1 * time.Minute),
+					Value:     1,
+				},
+				{
+					Timestamp: testInstant.Add(2 * time.Minute),
+					Value:     1,
+				},
+				{
+					Timestamp: testInstant.Add(3 * time.Minute),
+					Value:     1,
+				},
+			},
+			out: model.Values{
+				{
+					Timestamp: testInstant,
+					Value:     1,
+				},
+				{
+					Timestamp: testInstant.Add(1 * time.Minute),
+					Value:     1,
+				},
+				{
+					Timestamp: testInstant.Add(3 * time.Minute),
+					Value:     1,
+				},
+			},
+		},
 	}
 	for i, scenario := range scenarios {
 		actual := scenario.op.ExtractSamples(scenario.in)