feat: Add step param to Patterns Query API

pkg/loghttp/patterns.go

@@ -2,6 +2,7 @@ package loghttp
 
 import (
 	"net/http"
+	"time"
 
 	"github.com/grafana/loki/v3/pkg/logproto"
 )
@@ -15,6 +16,7 @@ func ParsePatternsQuery(r *http.Request) (*logproto.QueryPatternsRequest, error)
 	}
 	req.Start = start
 	req.End = end
+	req.Step = (time.Duration(defaultQueryRangeStep(start, end)) * time.Second).Milliseconds()
 
 	req.Query = query(r)
 	return req, nil

pkg/logproto/compat.go

@@ -335,6 +335,7 @@ func (m *VolumeRequest) LogToSpan(sp opentracing.Span) {
 		otlog.String("query", m.GetQuery()),
 		otlog.String("start", timestamp.Time(int64(m.From)).String()),
 		otlog.String("end", timestamp.Time(int64(m.Through)).String()),
+		otlog.String("step", time.Duration(m.Step).String()),
 	)
 }
 
@@ -448,8 +449,6 @@ func (m *ShardsRequest) LogToSpan(sp opentracing.Span) {
 
 func (m *QueryPatternsRequest) GetCachingOptions() (res definitions.CachingOptions) { return }
 
-func (m *QueryPatternsRequest) GetStep() int64 { return 0 }
-
 func (m *QueryPatternsRequest) WithStartEnd(start, end time.Time) definitions.Request {
 	clone := *m
 	clone.Start = start
@@ -469,9 +468,10 @@ func (m *QueryPatternsRequest) WithStartEndForCache(start, end time.Time) result
 
 func (m *QueryPatternsRequest) LogToSpan(sp opentracing.Span) {
 	fields := []otlog.Field{
+		otlog.String("query", m.GetQuery()),
 		otlog.String("start", m.Start.String()),
 		otlog.String("end", m.End.String()),
-		otlog.String("query", m.GetQuery()),
+		otlog.String("step", time.Duration(m.Step).String()),
 	}
 	sp.LogFields(fields...)
 }

pkg/logproto/pattern.proto

@@ -24,6 +24,7 @@ message QueryPatternsRequest {
     (gogoproto.stdtime) = true,
     (gogoproto.nullable) = false
   ];
+  int64 step = 4;
 }
 
 message QueryPatternsResponse {

pkg/pattern/drain/chunk.go

@@ -11,7 +11,7 @@ import (
 )
 
 const (
-	timeResolution = model.Time(int64(time.Second*10) / 1e6)
+	TimeResolution = model.Time(int64(time.Second*10) / 1e6)
 
 	defaultVolumeSize = 500
 
@@ -25,7 +25,7 @@ type Chunk struct {
 }
 
 func newChunk(ts model.Time) Chunk {
-	maxSize := int(maxChunkTime.Nanoseconds()/timeResolution.UnixNano()) + 1
+	maxSize := int(maxChunkTime.Nanoseconds()/TimeResolution.UnixNano()) + 1
 	v := Chunk{Samples: make([]logproto.PatternSample, 1, maxSize)}
 	v.Samples[0] = logproto.PatternSample{
 		Timestamp: ts,
@@ -43,9 +43,9 @@ func (c Chunk) spaceFor(ts model.Time) bool {
 }
 
 // ForRange returns samples with only the values
-// in the given range [start:end).
-// start and end are in milliseconds since epoch.
-func (c Chunk) ForRange(start, end model.Time) []logproto.PatternSample {
+// in the given range [start:end) and aggregates them by step duration.
+// start and end are in milliseconds since epoch. step is a duration in milliseconds.
+func (c Chunk) ForRange(start, end, step model.Time) []logproto.PatternSample {
 	if len(c.Samples) == 0 {
 		return nil
 	}
@@ -66,11 +66,36 @@ func (c Chunk) ForRange(start, end model.Time) []logproto.PatternSample {
 			return c.Samples[i].Timestamp >= end
 		})
 	}
-	return c.Samples[lo:hi]
+	if step == TimeResolution {
+		return c.Samples[lo:hi]
+	}
+
+	// Re-scale samples into step-sized buckets
+	currentStep := truncateTimestamp(c.Samples[lo].Timestamp, step)
+	aggregatedSamples := make([]logproto.PatternSample, 0, ((c.Samples[hi-1].Timestamp-currentStep)/step)+1)
+	aggregatedSamples = append(aggregatedSamples, logproto.PatternSample{
+		Timestamp: currentStep,
+		Value:     0,
+	})
+	for _, sample := range c.Samples[lo:hi] {
+		if sample.Timestamp >= currentStep+step {
+			stepForSample := truncateTimestamp(sample.Timestamp, step)
+			for i := currentStep + step; i <= stepForSample; i += step {
+				aggregatedSamples = append(aggregatedSamples, logproto.PatternSample{
+					Timestamp: i,
+					Value:     0,
+				})
+			}
+			currentStep = stepForSample
+		}
+		aggregatedSamples[len(aggregatedSamples)-1].Value += sample.Value
+	}
+
+	return aggregatedSamples
 }
 
 func (c *Chunks) Add(ts model.Time) {
-	t := truncateTimestamp(ts)
+	t := truncateTimestamp(ts, TimeResolution)
 
 	if len(*c) == 0 {
 		*c = append(*c, newChunk(t))
@@ -91,10 +116,10 @@ func (c *Chunks) Add(ts model.Time) {
 	})
 }
 
-func (c Chunks) Iterator(pattern string, from, through model.Time) iter.Iterator {
+func (c Chunks) Iterator(pattern string, from, through, step model.Time) iter.Iterator {
 	iters := make([]iter.Iterator, 0, len(c))
 	for _, chunk := range c {
-		samples := chunk.ForRange(from, through)
+		samples := chunk.ForRange(from, through, step)
 		if len(samples) == 0 {
 			continue
 		}
@@ -173,4 +198,4 @@ func (c *Chunks) size() int {
 	return size
 }
 
-func truncateTimestamp(ts model.Time) model.Time { return ts - ts%timeResolution }
+func truncateTimestamp(ts, step model.Time) model.Time { return ts - ts%step }