always align volume range timestamp to end of step

pkg/querier/queryrange/roundtrip_test.go

@@ -509,7 +509,7 @@ func TestVolumeTripperware(t *testing.T) {
 			From:        model.TimeFromUnixNano(testTime.Add(-25 * time.Hour).UnixNano()), // bigger than split by interval limit
 			Through:     model.TimeFromUnixNano(testTime.UnixNano()),
 			Limit:       10,
-			Step:        0, // Travis/Trevor: this should be ignored and set to 0. Karsten: Why?
+			Step:        0, // Setting this value to 0 is what makes this an instant query
 			AggregateBy: seriesvolume.DefaultAggregateBy,
 		}
 
@@ -567,7 +567,7 @@ func TestVolumeTripperware(t *testing.T) {
 			Matchers:    `{job="varlogs"}`,
 			From:        model.TimeFromUnixNano(start.UnixNano()), // bigger than split by interval limit
 			Through:     model.TimeFromUnixNano(end.UnixNano()),
-			Step:        time.Hour.Milliseconds(),
+			Step:        time.Hour.Milliseconds(), // a non-zero value makes this a range query
 			Limit:       10,
 			AggregateBy: seriesvolume.DefaultAggregateBy,
 		}
@@ -587,21 +587,21 @@ func TestVolumeTripperware(t *testing.T) {
 		require.Equal(t, 6, *count) // 6 queries from splitting into step buckets
 
 		barBazExpectedSamples := []logproto.LegacySample{}
-		util.ForInterval(time.Hour, start, end, true, func(s, _ time.Time) {
+		util.ForInterval(time.Hour, start, end, true, func(_, e time.Time) {
 			barBazExpectedSamples = append(barBazExpectedSamples, logproto.LegacySample{
 				Value:       3350,
-				TimestampMs: s.Unix() * 1e3,
+				TimestampMs: e.UnixMilli(),
 			})
 		})
 		sort.Slice(barBazExpectedSamples, func(i, j int) bool {
 			return barBazExpectedSamples[i].TimestampMs < barBazExpectedSamples[j].TimestampMs
 		})
 
 		fooBarExpectedSamples := []logproto.LegacySample{}
-		util.ForInterval(time.Hour, start, end, true, func(s, _ time.Time) {
+		util.ForInterval(time.Hour, start, end, true, func(_, e time.Time) {
 			fooBarExpectedSamples = append(fooBarExpectedSamples, logproto.LegacySample{
 				Value:       1024,
-				TimestampMs: s.Unix() * 1e3,
+				TimestampMs: e.UnixMilli(),
 			})
 		})
 		sort.Slice(fooBarExpectedSamples, func(i, j int) bool {

pkg/querier/queryrange/volume.go

@@ -34,14 +34,10 @@ func NewVolumeMiddleware() queryrangebase.Middleware {
 			interval := time.Duration(volReq.Step * 1e6)
 
 			util.ForInterval(interval, startTS, endTS, true, func(start, end time.Time) {
-				// Range query buckets are aligned to the starting timestamp
-				// Instant queries are for "this instant", which aligns to the end of the requested range
-				bucket := start
-				if interval == 0 {
-					bucket = end
-				}
-
-				reqs[bucket] = &logproto.VolumeRequest{
+				// Always align to the end of the requested range
+				// For range queries, this aligns to the end of the period we're returning a bytes aggregation for
+				// For instant queries, which are for "this instant", this aligns to the end of the requested range
+				reqs[end] = &logproto.VolumeRequest{
 					From:         model.TimeFromUnix(start.Unix()),
 					Through:      model.TimeFromUnix(end.Unix()),
 					Matchers:     volReq.Matchers,

pkg/querier/queryrange/volume_test.go

@@ -328,4 +328,24 @@ func Test_VolumeMiddleware(t *testing.T) {
 		require.Equal(t, 1, len(promResp.Data.Result))
 		require.Equal(t, 2, len(promResp.Data.Result[0].Samples))
 	})
+
+	t.Run("timestamps are aligned with the end of steps", func(t *testing.T) {
+		volumeReq := &logproto.VolumeRequest{
+			From:        1000000000000,
+			Through:     1000000005000, // 5s range
+			Matchers:    `{foo="bar"}`,
+			Limit:       seriesvolume.DefaultLimit,
+			Step:        1000, // 1s
+			AggregateBy: seriesvolume.Series,
+		}
+		promResp := makeVolumeRequest(volumeReq)
+
+		require.Equal(t, int64(1000000000999),
+			promResp.Data.Result[0].Samples[0].TimestampMs,
+			"first timestamp should be one millisecond before the end of the first step")
+		require.Equal(t,
+			int64(1000000005000),
+			promResp.Data.Result[0].Samples[4].TimestampMs,
+			"last timestamp should be equal to the end of the requested query range")
+	})
 }