mirror of https://github.com/grafana/loki
Bytes aggregations (#2150)
* Add bytes_rate and bytes_over_time. Still need to works on test. Signed-off-by: Cyril Tovena <cyril.tovena@gmail.com> * Add more tests. Signed-off-by: Cyril Tovena <cyril.tovena@gmail.com> * lint. Signed-off-by: Cyril Tovena <cyril.tovena@gmail.com> * More lint fixes. Signed-off-by: Cyril Tovena <cyril.tovena@gmail.com> * More comments. Signed-off-by: Cyril Tovena <cyril.tovena@gmail.com> * Add documentation. Signed-off-by: Cyril Tovena <cyril.tovena@gmail.com> * Review feedbacks. Signed-off-by: Cyril Tovena <cyril.tovena@gmail.com> * Fix bad conflic resolution. Signed-off-by: Cyril Tovena <cyril.tovena@gmail.com>k20
Cyril Tovena
5 years ago
committed by
GitHub
parent
635dd0add5
commit
18f1b0640d
@ -0,0 +1,63 @@ |
||||
package logql |
||||
|
||||
import ( |
||||
"fmt" |
||||
"time" |
||||
|
||||
"github.com/prometheus/prometheus/promql" |
||||
) |
||||
|
||||
const unsupportedErr = "unsupported range vector aggregation operation: %s" |
||||
|
||||
func (r rangeAggregationExpr) extractor() (SampleExtractor, error) { |
||||
switch r.operation { |
||||
case OpRangeTypeRate, OpRangeTypeCount: |
||||
return extractCount, nil |
||||
case OpRangeTypeBytes, OpRangeTypeBytesRate: |
||||
return extractBytes, nil |
||||
default: |
||||
return nil, fmt.Errorf(unsupportedErr, r.operation) |
||||
} |
||||
} |
||||
|
||||
func (r rangeAggregationExpr) aggregator() (RangeVectorAggregator, error) { |
||||
switch r.operation { |
||||
case OpRangeTypeRate: |
||||
return rateLogs(r.left.interval), nil |
||||
case OpRangeTypeCount: |
||||
return countOverTime, nil |
||||
case OpRangeTypeBytesRate: |
||||
return rateLogBytes(r.left.interval), nil |
||||
case OpRangeTypeBytes: |
||||
return sumOverTime, nil |
||||
default: |
||||
return nil, fmt.Errorf(unsupportedErr, r.operation) |
||||
} |
||||
} |
||||
|
||||
// rateLogs calculates the per-second rate of log lines.
|
||||
func rateLogs(selRange time.Duration) func(samples []promql.Point) float64 { |
||||
return func(samples []promql.Point) float64 { |
||||
return float64(len(samples)) / selRange.Seconds() |
||||
} |
||||
} |
||||
|
||||
// rateLogBytes calculates the per-second rate of log bytes.
|
||||
func rateLogBytes(selRange time.Duration) func(samples []promql.Point) float64 { |
||||
return func(samples []promql.Point) float64 { |
||||
return sumOverTime(samples) / selRange.Seconds() |
||||
} |
||||
} |
||||
|
||||
// countOverTime counts the amount of log lines.
|
||||
func countOverTime(samples []promql.Point) float64 { |
||||
return float64(len(samples)) |
||||
} |
||||
|
||||
func sumOverTime(samples []promql.Point) float64 { |
||||
var sum float64 |
||||
for _, v := range samples { |
||||
sum += v.V |
||||
} |
||||
return sum |
||||
} |
||||
@ -0,0 +1,99 @@ |
||||
package logql |
||||
|
||||
import ( |
||||
"github.com/grafana/loki/pkg/iter" |
||||
"github.com/grafana/loki/pkg/logproto" |
||||
) |
||||
|
||||
var ( |
||||
extractBytes = bytesSampleExtractor{} |
||||
extractCount = countSampleExtractor{} |
||||
) |
||||
|
||||
// SeriesIterator is an iterator that iterate over a stream of logs and returns sample.
|
||||
type SeriesIterator interface { |
||||
Close() error |
||||
Next() bool |
||||
Peek() (Sample, bool) |
||||
} |
||||
|
||||
// Sample is a series sample
|
||||
type Sample struct { |
||||
Labels string |
||||
Value float64 |
||||
TimestampNano int64 |
||||
} |
||||
|
||||
type seriesIterator struct { |
||||
iter iter.PeekingEntryIterator |
||||
sampler SampleExtractor |
||||
|
||||
updated bool |
||||
cur Sample |
||||
} |
||||
|
||||
func newSeriesIterator(it iter.EntryIterator, sampler SampleExtractor) SeriesIterator { |
||||
return &seriesIterator{ |
||||
iter: iter.NewPeekingIterator(it), |
||||
sampler: sampler, |
||||
} |
||||
} |
||||
|
||||
func (e *seriesIterator) Close() error { |
||||
return e.iter.Close() |
||||
} |
||||
|
||||
func (e *seriesIterator) Next() bool { |
||||
e.updated = false |
||||
return e.iter.Next() |
||||
} |
||||
|
||||
func (e *seriesIterator) Peek() (Sample, bool) { |
||||
if e.updated { |
||||
return e.cur, true |
||||
} |
||||
|
||||
for { |
||||
lbs, entry, ok := e.iter.Peek() |
||||
if !ok { |
||||
return Sample{}, false |
||||
} |
||||
|
||||
// transform
|
||||
e.cur, ok = e.sampler.From(lbs, entry) |
||||
if ok { |
||||
break |
||||
} |
||||
if !e.iter.Next() { |
||||
return Sample{}, false |
||||
} |
||||
} |
||||
e.updated = true |
||||
return e.cur, true |
||||
} |
||||
|
||||
// SampleExtractor transforms a log entry into a sample.
|
||||
// In case of failure the second return value will be false.
|
||||
type SampleExtractor interface { |
||||
From(labels string, e logproto.Entry) (Sample, bool) |
||||
} |
||||
|
||||
type countSampleExtractor struct{} |
||||
|
||||
func (countSampleExtractor) From(lbs string, entry logproto.Entry) (Sample, bool) { |
||||
return Sample{ |
||||
Labels: lbs, |
||||
TimestampNano: entry.Timestamp.UnixNano(), |
||||
Value: 1., |
||||
}, true |
||||
} |
||||
|
||||
type bytesSampleExtractor struct{} |
||||
|
||||
func (bytesSampleExtractor) From(lbs string, entry logproto.Entry) (Sample, bool) { |
||||
return Sample{ |
||||
Labels: lbs, |
||||
TimestampNano: entry.Timestamp.UnixNano(), |
||||
Value: float64(len(entry.Line)), |
||||
}, true |
||||
} |
||||
@ -0,0 +1,159 @@ |
||||
package logql |
||||
|
||||
import ( |
||||
"context" |
||||
"testing" |
||||
"time" |
||||
|
||||
"github.com/stretchr/testify/require" |
||||
|
||||
"github.com/grafana/loki/pkg/iter" |
||||
"github.com/grafana/loki/pkg/logproto" |
||||
) |
||||
|
||||
func Test_seriesIterator_Peek(t *testing.T) { |
||||
type expectation struct { |
||||
ok bool |
||||
sample Sample |
||||
} |
||||
for _, test := range []struct { |
||||
name string |
||||
it SeriesIterator |
||||
expectations []expectation |
||||
}{ |
||||
{ |
||||
"count", |
||||
newSeriesIterator(iter.NewStreamIterator(newStream(5, identity, `{app="foo"}`)), extractCount), |
||||
[]expectation{ |
||||
{true, Sample{Labels: `{app="foo"}`, TimestampNano: 0, Value: 1}}, |
||||
{true, Sample{Labels: `{app="foo"}`, TimestampNano: time.Unix(1, 0).UnixNano(), Value: 1}}, |
||||
{true, Sample{Labels: `{app="foo"}`, TimestampNano: time.Unix(2, 0).UnixNano(), Value: 1}}, |
||||
{true, Sample{Labels: `{app="foo"}`, TimestampNano: time.Unix(3, 0).UnixNano(), Value: 1}}, |
||||
{true, Sample{Labels: `{app="foo"}`, TimestampNano: time.Unix(4, 0).UnixNano(), Value: 1}}, |
||||
{false, Sample{}}, |
||||
}, |
||||
}, |
||||
{ |
||||
"bytes empty", |
||||
newSeriesIterator( |
||||
iter.NewStreamIterator( |
||||
newStream( |
||||
3, |
||||
func(i int64) logproto.Entry { |
||||
return logproto.Entry{ |
||||
Timestamp: time.Unix(i, 0), |
||||
} |
||||
}, |
||||
`{app="foo"}`, |
||||
), |
||||
), |
||||
extractBytes, |
||||
), |
||||
[]expectation{ |
||||
{true, Sample{Labels: `{app="foo"}`, TimestampNano: 0, Value: 0}}, |
||||
{true, Sample{Labels: `{app="foo"}`, TimestampNano: time.Unix(1, 0).UnixNano(), Value: 0}}, |
||||
{true, Sample{Labels: `{app="foo"}`, TimestampNano: time.Unix(2, 0).UnixNano(), Value: 0}}, |
||||
{false, Sample{}}, |
||||
}, |
||||
}, |
||||
{ |
||||
"bytes", |
||||
newSeriesIterator( |
||||
iter.NewStreamIterator( |
||||
newStream( |
||||
3, |
||||
func(i int64) logproto.Entry { |
||||
return logproto.Entry{ |
||||
Timestamp: time.Unix(i, 0), |
||||
Line: "foo", |
||||
} |
||||
}, |
||||
`{app="foo"}`, |
||||
), |
||||
), |
||||
extractBytes, |
||||
), |
||||
[]expectation{ |
||||
{true, Sample{Labels: `{app="foo"}`, TimestampNano: 0, Value: 3}}, |
||||
{true, Sample{Labels: `{app="foo"}`, TimestampNano: time.Unix(1, 0).UnixNano(), Value: 3}}, |
||||
{true, Sample{Labels: `{app="foo"}`, TimestampNano: time.Unix(2, 0).UnixNano(), Value: 3}}, |
||||
{false, Sample{}}, |
||||
}, |
||||
}, |
||||
{ |
||||
"bytes backward", |
||||
newSeriesIterator( |
||||
iter.NewStreamsIterator(context.Background(), |
||||
[]logproto.Stream{ |
||||
newStream( |
||||
3, |
||||
func(i int64) logproto.Entry { |
||||
return logproto.Entry{ |
||||
Timestamp: time.Unix(i, 0), |
||||
Line: "foo", |
||||
} |
||||
}, |
||||
`{app="foo"}`, |
||||
), |
||||
newStream( |
||||
3, |
||||
func(i int64) logproto.Entry { |
||||
return logproto.Entry{ |
||||
Timestamp: time.Unix(i, 0), |
||||
Line: "barr", |
||||
} |
||||
}, |
||||
`{app="barr"}`, |
||||
), |
||||
}, |
||||
logproto.BACKWARD, |
||||
), |
||||
extractBytes, |
||||
), |
||||
[]expectation{ |
||||
{true, Sample{Labels: `{app="foo"}`, TimestampNano: 0, Value: 3}}, |
||||
{true, Sample{Labels: `{app="foo"}`, TimestampNano: time.Unix(1, 0).UnixNano(), Value: 3}}, |
||||
{true, Sample{Labels: `{app="foo"}`, TimestampNano: time.Unix(2, 0).UnixNano(), Value: 3}}, |
||||
{true, Sample{Labels: `{app="barr"}`, TimestampNano: 0, Value: 4}}, |
||||
{true, Sample{Labels: `{app="barr"}`, TimestampNano: time.Unix(1, 0).UnixNano(), Value: 4}}, |
||||
{true, Sample{Labels: `{app="barr"}`, TimestampNano: time.Unix(2, 0).UnixNano(), Value: 4}}, |
||||
{false, Sample{}}, |
||||
}, |
||||
}, |
||||
{ |
||||
"skip first", |
||||
newSeriesIterator(iter.NewStreamIterator(newStream(2, identity, `{app="foo"}`)), fakeSampler{}), |
||||
[]expectation{ |
||||
{true, Sample{Labels: `{app="foo"}`, TimestampNano: time.Unix(1, 0).UnixNano(), Value: 10}}, |
||||
{false, Sample{}}, |
||||
}, |
||||
}, |
||||
} { |
||||
t.Run(test.name, func(t *testing.T) { |
||||
for _, e := range test.expectations { |
||||
sample, ok := test.it.Peek() |
||||
require.Equal(t, e.ok, ok) |
||||
if !e.ok { |
||||
continue |
||||
} |
||||
require.Equal(t, e.sample, sample) |
||||
test.it.Next() |
||||
} |
||||
require.NoError(t, test.it.Close()) |
||||
}) |
||||
} |
||||
} |
||||
|
||||
// fakeSampler is a Sampler that returns no value for 0 timestamp otherwise always 10
|
||||
type fakeSampler struct{} |
||||
|
||||
func (fakeSampler) From(lbs string, entry logproto.Entry) (Sample, bool) { |
||||
if entry.Timestamp.UnixNano() == 0 { |
||||
return Sample{}, false |
||||
} |
||||
return Sample{ |
||||
Labels: lbs, |
||||
TimestampNano: entry.Timestamp.UnixNano(), |
||||
Value: 10, |
||||
}, true |
||||
} |
||||
Loading…
Reference in new issue