fix: always do a range query

pkg/loghttp/params.go

@@ -18,14 +18,14 @@ import (
 )
 
 const (
-	defaultQueryLimit = 100
+	DefaultQueryLimit = 100
 	defaultLimit      = 1000
 	defaultSince      = 1 * time.Hour
 	defaultDirection  = logproto.BACKWARD
 )
 
 func limit(r *http.Request) (uint32, error) {
-	l, err := parseInt(r.Form.Get("limit"), defaultQueryLimit)
+	l, err := parseInt(r.Form.Get("limit"), DefaultQueryLimit)
 	if err != nil {
 		return 0, err
 	}
@@ -36,7 +36,7 @@ func limit(r *http.Request) (uint32, error) {
 }
 
 func lineLimit(r *http.Request) (uint32, error) {
-	l, err := parseInt(r.Form.Get("line_limit"), defaultQueryLimit)
+	l, err := parseInt(r.Form.Get("line_limit"), DefaultQueryLimit)
 	if err != nil {
 		return 0, err
 	}

pkg/loghttp/query.go

@@ -440,7 +440,7 @@ func NewRangeQueryWithDefaults() *RangeQuery {
 	result := &RangeQuery{
 		Start:     start,
 		End:       end,
-		Limit:     defaultQueryLimit,
+		Limit:     DefaultQueryLimit,
 		Direction: defaultDirection,
 		Interval:  0,
 	}

pkg/querier/queryrange/roundtrip.go

@@ -1205,41 +1205,27 @@ func aggMetricsVolumeHandler(
 		start:       r.GetStart(),
 		end:         r.GetEnd(),
 		aggregateBy: strings.Join(r.GetTargetLabels(), ","),
+		step:        time.Duration(r.GetStep() * int64(time.Millisecond)),
 	}
 
-	qryStr := aggMetricQry.BuildQuery()
+	qryStr, step := aggMetricQry.BuildQuery()
 	expr, err := syntax.ParseExpr(qryStr)
 	if err != nil {
 		return nil, httpgrpc.Errorf(http.StatusBadRequest, "%s", err.Error())
 	}
 
-	var lokiReq base.Request
-	if r.GetStep() <= 0 {
-		lokiReq = &LokiInstantRequest{
-			Query:     expr.String(),
-			Limit:     1000,
-			Direction: logproto.BACKWARD,
-			TimeTs:    r.GetEnd().UTC(),
-			Path:      "/loki/api/v1/query",
-			Plan: &plan.QueryPlan{
-				AST: expr,
-			},
-			CachingOptions: r.GetCachingOptions(),
-		}
-	} else {
-		lokiReq = &LokiRequest{
-			Query:     expr.String(),
-			Limit:     1000,
-			Step:      r.GetStep(),
-			StartTs:   r.GetStart().UTC(),
-			EndTs:     r.GetEnd().UTC(),
-			Direction: logproto.BACKWARD,
-			Path:      "/loki/api/v1/query_range",
-			Plan: &plan.QueryPlan{
-				AST: expr,
-			},
-			CachingOptions: r.GetCachingOptions(),
-		}
+	lokiReq := &LokiRequest{
+		Query:     expr.String(),
+		Limit:     loghttp.DefaultQueryLimit,
+		Step:      step.Milliseconds(),
+		StartTs:   r.GetStart().UTC(),
+		EndTs:     r.GetEnd().UTC(),
+		Direction: logproto.BACKWARD,
+		Path:      "/loki/api/v1/query_range",
+		Plan: &plan.QueryPlan{
+			AST: expr,
+		},
+		CachingOptions: r.GetCachingOptions(),
 	}
 
 	resp, err := logHandler.Do(ctx, lokiReq)
@@ -1262,6 +1248,18 @@ func aggMetricsVolumeHandler(
 			resultType = loghttp.ResultTypeMatrix
 		}
 
+		// sum the values and take the latest timestamp if an instant volume was requested
+		if r.GetStep() == 0 {
+			newSample := stream.Samples[len(stream.Samples)-1]
+			for _, sample := range stream.Samples[0 : len(stream.Samples)-1] {
+				newSample.Value += sample.Value
+			}
+
+			stream.Samples = []logproto.LegacySample{
+				newSample,
+			}
+		}
+
 		lbls := logproto.FromLabelAdaptersToLabels(stream.Labels)
 		sortableResult = append(sortableResult, sortableSampleStream{
 			name:    lbls.String(),

pkg/querier/queryrange/volume.go

@@ -3,6 +3,7 @@ package queryrange
 import (
 	"context"
 	"fmt"
+	"math"
 	"sort"
 	"strings"
 	"time"
@@ -221,6 +222,7 @@ type aggregatedMetricQuery struct {
 	aggregateBy string
 	start       time.Time
 	end         time.Time
+	step        time.Duration
 }
 
 func (a *aggregatedMetricQuery) buildBaseQueryString(
@@ -254,7 +256,7 @@ func (a *aggregatedMetricQuery) buildBaseQueryString(
 	)
 }
 
-func (a *aggregatedMetricQuery) BuildQuery() string {
+func (a *aggregatedMetricQuery) BuildQuery() (string, time.Duration) {
 	// by this point query as been validated and we can assume that there is at least one matcher
 	firstMatcher := a.matchers[0]
 
@@ -287,7 +289,16 @@ func (a *aggregatedMetricQuery) BuildQuery() string {
 		query = query + " | " + strings.Join(filters, " | ")
 	}
 
-	lookBack := a.end.Sub(a.start).Truncate(time.Second)
-	query = query + fmt.Sprintf(` | unwrap bytes(bytes) | __error__=""[%s]))`, lookBack)
-	return query
+	step := a.step
+	if step == 0 {
+		step = time.Duration(defaultQueryRangeStep(a.start, a.end)) * time.Second
+	}
+	query = query + fmt.Sprintf(` | unwrap bytes(bytes) | __error__=""[%s]))`, step)
+	return query, step
+}
+
+// defaultQueryRangeStep returns the default step used in the query range API,
+// which is dynamically calculated based on the time range
+func defaultQueryRangeStep(start time.Time, end time.Time) int {
+	return int(math.Max(math.Floor(end.Sub(start).Seconds()/250), 1))
 }