Add tests for and fix getValuesAlongRangeOp value extraction.

storage/metric/operation.go

@@ -72,7 +72,7 @@ func (g *getValuesAtTimeOp) ExtractSamples(in []model.SamplePair) (out []model.S
 
 func extractValuesAroundTime(t time.Time, in []model.SamplePair) (out []model.SamplePair) {
 	i := sort.Search(len(in), func(i int) bool {
-		return in[i].Timestamp.After(t) || in[i].Timestamp.Equal(t)
+		return !in[i].Timestamp.Before(t)
 	})
 	fmt.Printf("I: %d\n", i)
 	switch i {
@@ -158,9 +158,26 @@ func (g *getValuesAlongRangeOp) ExtractSamples(in []model.SamplePair) (out []mod
 	if len(in) == 0 {
 		return
 	}
-	lastChunkTime := in[len(in)-1].Timestamp
-	g.from = lastChunkTime.Add(time.Duration(1))
-	return in
+	// Find the first sample where time >= g.from.
+	firstIdx := sort.Search(len(in), func(i int) bool {
+		return !in[i].Timestamp.Before(g.from)
+	})
+	if firstIdx == len(in) {
+		// No samples at or after operator start time.
+		return
+	}
+
+	// Find the first sample where time > g.through.
+	lastIdx := sort.Search(len(in), func(i int) bool {
+		return in[i].Timestamp.After(g.through)
+	})
+	if lastIdx == firstIdx {
+		return
+	}
+
+	lastSampleTime := in[lastIdx - 1].Timestamp
+	g.from = lastSampleTime.Add(time.Duration(1))
+	return in[firstIdx:lastIdx]
 }
 
 func (g getValuesAlongRangeOp) CurrentTime() (currentTime *time.Time) {

storage/metric/operation_test.go

@@ -1477,3 +1477,179 @@ func TestGetValuesAtIntervalOp(t *testing.T) {
 		}
 	}
 }
+
+func TestGetValuesAlongRangeOp(t *testing.T) {
+	var scenarios = []struct {
+		op  getValuesAlongRangeOp
+		in  []model.SamplePair
+		out []model.SamplePair
+	}{
+		// No values.
+		{
+			op: getValuesAlongRangeOp{
+				from:     testInstant,
+				through:  testInstant.Add(1 * time.Minute),
+			},
+		},
+		// Entire operator range before first value.
+		{
+			op: getValuesAlongRangeOp{
+				from:     testInstant,
+				through:  testInstant.Add(1 * time.Minute),
+			},
+			in: []model.SamplePair{
+				{
+					Timestamp: testInstant.Add(2 * time.Minute),
+					Value:     1,
+				},
+				{
+					Timestamp: testInstant.Add(3 * time.Minute),
+					Value:     1,
+				},
+			},
+			out: []model.SamplePair{},
+		},
+		// Operator range starts before first value, ends within available values.
+		{
+			op: getValuesAlongRangeOp{
+				from:     testInstant,
+				through:  testInstant.Add(2 * time.Minute),
+			},
+			in: []model.SamplePair{
+				{
+					Timestamp: testInstant.Add(1 * time.Minute),
+					Value:     1,
+				},
+				{
+					Timestamp: testInstant.Add(3 * time.Minute),
+					Value:     1,
+				},
+			},
+			out: []model.SamplePair{
+				{
+					Timestamp: testInstant.Add(1 * time.Minute),
+					Value:     1,
+				},
+			},
+		},
+		// Entire operator range is within available values.
+		{
+			op: getValuesAlongRangeOp{
+				from:     testInstant.Add(1 * time.Minute),
+				through:  testInstant.Add(2 * time.Minute),
+			},
+			in: []model.SamplePair{
+				{
+					Timestamp: testInstant,
+					Value:     1,
+				},
+				{
+					Timestamp: testInstant.Add(1 * time.Minute),
+					Value:     1,
+				},
+				{
+					Timestamp: testInstant.Add(3 * time.Minute),
+					Value:     1,
+				},
+			},
+			out: []model.SamplePair{
+				{
+					Timestamp: testInstant.Add(1 * time.Minute),
+					Value:     1,
+				},
+			},
+		},
+		// Operator range begins before first value, ends after last.
+		{
+			op: getValuesAlongRangeOp{
+				from:     testInstant,
+				through:  testInstant.Add(3 * time.Minute),
+			},
+			in: []model.SamplePair{
+				{
+					Timestamp: testInstant.Add(1 * time.Minute),
+					Value:     1,
+				},
+				{
+					Timestamp: testInstant.Add(2 * time.Minute),
+					Value:     1,
+				},
+			},
+			out: []model.SamplePair{
+				{
+					Timestamp: testInstant.Add(1 * time.Minute),
+					Value:     1,
+				},
+				{
+					Timestamp: testInstant.Add(2 * time.Minute),
+					Value:     1,
+				},
+			},
+		},
+		// Operator range begins within available values, ends after the last value.
+		{
+			op: getValuesAlongRangeOp{
+				from:     testInstant.Add(2 * time.Minute),
+				through:  testInstant.Add(4 * time.Minute),
+			},
+			in: []model.SamplePair{
+				{
+					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.SamplePair{
+				{
+					Timestamp: testInstant.Add(2 * time.Minute),
+					Value:     1,
+				},
+				{
+					Timestamp: testInstant.Add(3 * time.Minute),
+					Value:     1,
+				},
+			},
+		},
+		// Entire operator range after the last available value.
+		{
+			op: getValuesAlongRangeOp{
+				from:     testInstant.Add(2 * time.Minute),
+				through:  testInstant.Add(3 * time.Minute),
+			},
+			in: []model.SamplePair{
+				{
+					Timestamp: testInstant,
+					Value:     1,
+				},
+				{
+					Timestamp: testInstant.Add(1 * time.Minute),
+					Value:     1,
+				},
+			},
+			out: []model.SamplePair{},
+		},
+	}
+	for i, scenario := range scenarios {
+		actual := scenario.op.ExtractSamples(scenario.in)
+		if len(actual) != len(scenario.out) {
+			t.Fatalf("%d. expected length %d, got %d: %v", i, len(scenario.out), len(actual), scenario.op)
+			t.Fatalf("%d. expected length %d, got %d", i, len(scenario.out), len(actual))
+		}
+		for j, out := range scenario.out {
+			if out != actual[j] {
+				t.Fatal("%d. expected output %v, got %v", i, scenario.out, actual)
+			}
+		}
+	}
+}