Skip to content

Commit 2587657

Browse files
trevorwhitneytrevorwhitney
committed
fix: grouping
1 parent b897fc5 commit 2587657

File tree

8 files changed

+163
-84
lines changed

8 files changed

+163
-84
lines changed

‎cmd/loki/loki-local-config.yaml

+2
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,8 @@ schema_config:
3535

3636
pattern_ingester:
3737
enabled: true
38+
metric_aggregation:
39+
enabled: true
3840

3941
ruler:
4042
alertmanager_url: http://localhost:9093

‎pkg/pattern/ingester.go

+2-2
Original file line numberDiff line numberDiff line change
@@ -48,11 +48,11 @@ type Config struct {
4848
func (cfg *Config) RegisterFlags(fs *flag.FlagSet) {
4949
cfg.LifecyclerConfig.RegisterFlagsWithPrefix("pattern-ingester.", fs, util_log.Logger)
5050
cfg.ClientConfig.RegisterFlags(fs)
51+
cfg.MetricAggregation.RegisterFlagsWithPrefix(fs, "pattern-ingester.")
52+
5153
fs.BoolVar(&cfg.Enabled, "pattern-ingester.enabled", false, "Flag to enable or disable the usage of the pattern-ingester component.")
5254
fs.IntVar(&cfg.ConcurrentFlushes, "pattern-ingester.concurrent-flushes", 32, "How many flushes can happen concurrently from each stream.")
5355
fs.DurationVar(&cfg.FlushCheckPeriod, "pattern-ingester.flush-check-period", 30*time.Second, "How often should the ingester see if there are any blocks to flush. The first flush check is delayed by a random time up to 0.8x the flush check period. Additionally, there is +/- 1% jitter added to the interval.")
54-
55-
cfg.MetricAggregation.RegisterFlagsWithPrefix(fs, "pattern-ingester.")
5656
}
5757

5858
func (cfg *Config) Validate() error {

‎pkg/pattern/instance.go

+1-12
Original file line numberDiff line numberDiff line change
@@ -125,24 +125,13 @@ func (i *instance) QuerySample(
125125
return nil, err
126126
}
127127

128-
typ, err := metric.ExtractMetricType(expr)
129-
if err != nil || typ == metric.Unsupported {
130-
return nil, err
131-
}
132-
133128
var iters []iter.Iterator
134129
err = i.forMatchingStreams(
135130
selector.Matchers(),
136131
func(stream *stream) error {
137132
var iter iter.Iterator
138133
var err error
139-
if typ == metric.Bytes {
140-
iter, err = stream.BytesIterator(ctx, expr, from, through, step)
141-
} else if typ == metric.Count {
142-
iter, err = stream.CountIterator(ctx, expr, from, through, step)
143-
} else {
144-
return fmt.Errorf("unsupported query operation")
145-
}
134+
iter, err = stream.SampleIterator(ctx, expr, from, through, step)
146135

147136
if err != nil {
148137
return err

‎pkg/pattern/metric/chunk.go

+16-3
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@ import (
77
"time"
88

99
"github.com/grafana/loki/v3/pkg/logproto"
10+
"github.com/grafana/loki/v3/pkg/logql/syntax"
1011
"github.com/grafana/loki/v3/pkg/pattern/chunk"
1112
"github.com/grafana/loki/v3/pkg/pattern/iter"
1213
"github.com/prometheus/common/model"
@@ -51,12 +52,23 @@ func (c *Chunks) Observe(bytes, count uint64, ts model.Time) {
5152
func (c *Chunks) Iterator(
5253
ctx context.Context,
5354
typ MetricType,
55+
grouping *syntax.Grouping,
5456
from, through, step model.Time,
5557
) (iter.Iterator, error) {
5658
if typ == Unsupported {
5759
return nil, fmt.Errorf("unsupported metric type")
5860
}
5961

62+
lbls := c.labels
63+
if grouping != nil {
64+
sort.Strings(grouping.Groups)
65+
lbls = make(labels.Labels, 0, len(grouping.Groups))
66+
for _, group := range grouping.Groups {
67+
value := c.labels.Get(group)
68+
lbls = append(lbls, labels.Label{Name: group, Value: value})
69+
}
70+
}
71+
6072
iters := make([]iter.Iterator, 0, len(c.chunks))
6173
for _, chunk := range c.chunks {
6274
samples, err := chunk.ForRangeAndType(typ, from, through, step)
@@ -68,9 +80,10 @@ func (c *Chunks) Iterator(
6880
continue
6981
}
7082

71-
iters = append(iters, iter.NewLabelsSlice(c.labels, samples))
83+
iters = append(iters, iter.NewLabelsSlice(lbls, samples))
7284
}
73-
return iter.NewNonOverlappingLabelsIterator(c.labels, iters), nil
85+
86+
return iter.NewNonOverlappingLabelsIterator(lbls, iters), nil
7487
}
7588

7689
// TODO(twhitney): These values should be float64s (to match prometheus samples) or int64s (to match pattern samples)
@@ -127,7 +140,7 @@ func (c *Chunk) spaceFor(ts model.Time) bool {
127140
return ts.Sub(c.Samples[0].Timestamp) < chunk.MaxChunkTime
128141
}
129142

130-
//TODO(twhitney): any way to remove the duplication between this and the drain chunk ForRange method?
143+
// TODO(twhitney): any way to remove the duplication between this and the drain chunk ForRange method?
131144
// ForRangeAndType returns samples with only the values
132145
// in the given range [start:end] and aggregates them by step duration.
133146
// start and end are in milliseconds since epoch. step is a duration in milliseconds.

‎pkg/pattern/metric/chunk_test.go

+110
Original file line numberDiff line numberDiff line change
@@ -1,11 +1,15 @@
11
package metric
22

33
import (
4+
"context"
45
"reflect"
56
"testing"
67

78
"github.com/grafana/loki/v3/pkg/logproto"
9+
"github.com/grafana/loki/v3/pkg/logql/syntax"
10+
"github.com/grafana/loki/v3/pkg/pattern/iter"
811
"github.com/prometheus/common/model"
12+
"github.com/prometheus/prometheus/model/labels"
913
"github.com/stretchr/testify/require"
1014
)
1115

@@ -327,3 +331,109 @@ func TestForRangeAndType(t *testing.T) {
327331
})
328332
}
329333
}
334+
335+
func Test_Chunks_Iterator(t *testing.T) {
336+
ctx := context.Background()
337+
lbls := labels.Labels{
338+
labels.Label{Name: "foo", Value: "bar"},
339+
labels.Label{Name: "container", Value: "jar"},
340+
}
341+
chunks := Chunks{
342+
chunks: []Chunk{
343+
{
344+
Samples: []MetricSample{
345+
{Timestamp: 2, Bytes: 2, Count: 1},
346+
{Timestamp: 4, Bytes: 4, Count: 3},
347+
{Timestamp: 6, Bytes: 6, Count: 5},
348+
},
349+
mint: 2,
350+
maxt: 6,
351+
},
352+
},
353+
labels: lbls,
354+
}
355+
356+
t.Run("without grouping", func(t *testing.T) {
357+
it, err := chunks.Iterator(ctx, Bytes, nil, 0, 10, 2)
358+
require.NoError(t, err)
359+
360+
res, err := iter.ReadAllWithLabels(it)
361+
require.NoError(t, err)
362+
363+
require.Equal(t, 1, len(res.Series))
364+
require.Equal(t, lbls.String(), res.Series[0].GetLabels())
365+
366+
it, err = chunks.Iterator(ctx, Count, nil, 0, 10, 2)
367+
require.NoError(t, err)
368+
369+
res, err = iter.ReadAllWithLabels(it)
370+
require.NoError(t, err)
371+
372+
require.Equal(t, 1, len(res.Series))
373+
require.Equal(t, lbls.String(), res.Series[0].GetLabels())
374+
})
375+
376+
t.Run("grouping", func(t *testing.T) {
377+
grouping := &syntax.Grouping{
378+
Groups: []string{"container"},
379+
Without: false,
380+
}
381+
382+
expectedLabels := labels.Labels{
383+
labels.Label{
384+
Name: "container",
385+
Value: "jar",
386+
},
387+
}
388+
389+
it, err := chunks.Iterator(ctx, Bytes, grouping, 0, 10, 2)
390+
require.NoError(t, err)
391+
392+
res, err := iter.ReadAllWithLabels(it)
393+
require.NoError(t, err)
394+
395+
require.Equal(t, 1, len(res.Series))
396+
require.Equal(t, expectedLabels.String(), res.Series[0].GetLabels())
397+
398+
it, err = chunks.Iterator(ctx, Count, grouping, 0, 10, 2)
399+
require.NoError(t, err)
400+
401+
res, err = iter.ReadAllWithLabels(it)
402+
require.NoError(t, err)
403+
404+
require.Equal(t, 1, len(res.Series))
405+
require.Equal(t, expectedLabels.String(), res.Series[0].GetLabels())
406+
})
407+
408+
t.Run("grouping by a missing label", func(t *testing.T) {
409+
grouping := &syntax.Grouping{
410+
Groups: []string{"missing"},
411+
Without: false,
412+
}
413+
414+
expectedLabels := labels.Labels{
415+
labels.Label{
416+
Name: "missing",
417+
Value: "",
418+
},
419+
}
420+
421+
it, err := chunks.Iterator(ctx, Bytes, grouping, 0, 10, 2)
422+
require.NoError(t, err)
423+
424+
res, err := iter.ReadAllWithLabels(it)
425+
require.NoError(t, err)
426+
427+
require.Equal(t, 1, len(res.Series))
428+
require.Equal(t, expectedLabels.String(), res.Series[0].GetLabels())
429+
430+
it, err = chunks.Iterator(ctx, Count, grouping, 0, 10, 2)
431+
require.NoError(t, err)
432+
433+
res, err = iter.ReadAllWithLabels(it)
434+
require.NoError(t, err)
435+
436+
require.Equal(t, 1, len(res.Series))
437+
require.Equal(t, expectedLabels.String(), res.Series[0].GetLabels())
438+
})
439+
}

‎pkg/pattern/metric/evaluator.go

+12-10
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,7 @@ import (
1919
)
2020

2121
// TODO(twhitney): duplication with code in NewStepEvaluator
22-
func ExtractMetricType(expr syntax.SampleExpr) (MetricType, error) {
22+
func extractMetricType(expr syntax.SampleExpr) (MetricType, error) {
2323
var typ MetricType
2424
switch e := expr.(type) {
2525
case *syntax.VectorAggregationExpr:
@@ -57,7 +57,6 @@ type SampleEvaluatorFactory interface {
5757
ctx context.Context,
5858
nextEvaluatorFactory SampleEvaluatorFactory,
5959
expr syntax.SampleExpr,
60-
typ MetricType,
6160
from, through, step model.Time,
6261
) (logql.StepEvaluator, error)
6362
}
@@ -66,18 +65,16 @@ type SampleEvaluatorFunc func(
6665
ctx context.Context,
6766
nextEvaluatorFactory SampleEvaluatorFactory,
6867
expr syntax.SampleExpr,
69-
typ MetricType,
7068
from, through, step model.Time,
7169
) (logql.StepEvaluator, error)
7270

7371
func (s SampleEvaluatorFunc) NewStepEvaluator(
7472
ctx context.Context,
7573
nextEvaluatorFactory SampleEvaluatorFactory,
7674
expr syntax.SampleExpr,
77-
typ MetricType,
7875
from, through, step model.Time,
7976
) (logql.StepEvaluator, error) {
80-
return s(ctx, nextEvaluatorFactory, expr, typ, from, through, step)
77+
return s(ctx, nextEvaluatorFactory, expr, from, through, step)
8178
}
8279

8380
type DefaultEvaluatorFactory struct {
@@ -94,9 +91,13 @@ func (ev *DefaultEvaluatorFactory) NewStepEvaluator(
9491
ctx context.Context,
9592
evFactory SampleEvaluatorFactory,
9693
expr syntax.SampleExpr,
97-
typ MetricType,
9894
from, through, step model.Time,
9995
) (logql.StepEvaluator, error) {
96+
metricType, err := extractMetricType(expr)
97+
if err != nil || metricType == Unsupported {
98+
return nil, err
99+
}
100+
100101
switch e := expr.(type) {
101102
case *syntax.VectorAggregationExpr:
102103
if rangExpr, ok := e.Left.(*syntax.RangeAggregationExpr); ok && e.Operation == syntax.OpTypeSum {
@@ -106,12 +107,11 @@ func (ev *DefaultEvaluatorFactory) NewStepEvaluator(
106107
func(ctx context.Context,
107108
_ SampleEvaluatorFactory,
108109
_ syntax.SampleExpr,
109-
typ MetricType,
110110
from, through, step model.Time,
111111
) (logql.StepEvaluator, error) {
112112
fromWithRangeAndOffset := from.Add(-rangExpr.Left.Interval).Add(-rangExpr.Left.Offset)
113113
throughWithOffset := through.Add(-rangExpr.Left.Offset)
114-
it, err := ev.chunks.Iterator(ctx, typ, fromWithRangeAndOffset, throughWithOffset, step)
114+
it, err := ev.chunks.Iterator(ctx, metricType, e.Grouping, fromWithRangeAndOffset, throughWithOffset, step)
115115
if err != nil {
116116
return nil, err
117117
}
@@ -129,7 +129,7 @@ func (ev *DefaultEvaluatorFactory) NewStepEvaluator(
129129
if e.Grouping == nil {
130130
return nil, errors.Errorf("aggregation operator '%q' without grouping", e.Operation)
131131
}
132-
nextEvaluator, err := evFactory.NewStepEvaluator(ctx, evFactory, e.Left, typ, from, through, step)
132+
nextEvaluator, err := evFactory.NewStepEvaluator(ctx, evFactory, e.Left, from, through, step)
133133
if err != nil {
134134
return nil, err
135135
}
@@ -145,7 +145,7 @@ func (ev *DefaultEvaluatorFactory) NewStepEvaluator(
145145
case *syntax.RangeAggregationExpr:
146146
fromWithRangeAndOffset := from.Add(-e.Left.Interval).Add(-e.Left.Offset)
147147
throughWithOffset := through.Add(-e.Left.Offset)
148-
it, err := ev.chunks.Iterator(ctx, typ, fromWithRangeAndOffset, throughWithOffset, step)
148+
it, err := ev.chunks.Iterator(ctx, metricType, e.Grouping, fromWithRangeAndOffset, throughWithOffset, step)
149149
if err != nil {
150150
return nil, err
151151
}
@@ -250,6 +250,8 @@ type SeriesToSampleIterator struct {
250250
lbls labels.Labels
251251
}
252252

253+
// TODO: could this me a matrix iterator that returned multiple samples with
254+
// different labels for the same timestamp?
253255
func NewSeriesToSampleIterator(series *promql.Series) *SeriesToSampleIterator {
254256
return &SeriesToSampleIterator{
255257
floats: series.Floats,

‎pkg/pattern/metric/evaluator_test.go

-4
Original file line numberDiff line numberDiff line change
@@ -30,14 +30,10 @@ func Test_SampleEvaluator(t *testing.T) {
3030
expr, err := syntax.ParseSampleExpr(query)
3131
require.NoError(t, err)
3232

33-
typ, err := ExtractMetricType(expr)
34-
require.NoError(t, err)
35-
3633
evaluator, err := factory.NewStepEvaluator(
3734
context.Background(),
3835
factory,
3936
expr.(syntax.SampleExpr),
40-
typ,
4137
model.Time(now-fiveMin), model.Time(now), model.Time(fiveMin),
4238
)
4339

0 commit comments

Comments
 (0)