|
|
|
|
@ -2,17 +2,25 @@ package dataobj |
|
|
|
|
|
|
|
|
|
import ( |
|
|
|
|
"context" |
|
|
|
|
"errors" |
|
|
|
|
"fmt" |
|
|
|
|
"io" |
|
|
|
|
"iter" |
|
|
|
|
"maps" |
|
|
|
|
"slices" |
|
|
|
|
"sort" |
|
|
|
|
"strconv" |
|
|
|
|
"time" |
|
|
|
|
|
|
|
|
|
"github.com/prometheus/prometheus/model/labels" |
|
|
|
|
|
|
|
|
|
"github.com/grafana/loki/pkg/push" |
|
|
|
|
|
|
|
|
|
"github.com/grafana/loki/v3/pkg/dataobj/internal/dataset" |
|
|
|
|
"github.com/grafana/loki/v3/pkg/dataobj/internal/encoding" |
|
|
|
|
"github.com/grafana/loki/v3/pkg/dataobj/internal/metadata/datasetmd" |
|
|
|
|
"github.com/grafana/loki/v3/pkg/dataobj/internal/metadata/filemd" |
|
|
|
|
"github.com/grafana/loki/v3/pkg/dataobj/internal/metadata/logsmd" |
|
|
|
|
"github.com/grafana/loki/v3/pkg/dataobj/internal/result" |
|
|
|
|
"github.com/grafana/loki/v3/pkg/dataobj/internal/sections/logs" |
|
|
|
|
) |
|
|
|
|
@ -27,14 +35,18 @@ type Record struct { |
|
|
|
|
|
|
|
|
|
// LogsReader reads the set of logs from an [Object].
|
|
|
|
|
type LogsReader struct { |
|
|
|
|
obj *Object |
|
|
|
|
idx int |
|
|
|
|
obj *Object |
|
|
|
|
idx int |
|
|
|
|
ready bool |
|
|
|
|
|
|
|
|
|
matchIDs map[int64]struct{} |
|
|
|
|
predicate LogsPredicate |
|
|
|
|
|
|
|
|
|
next func() (result.Result[logs.Record], bool) |
|
|
|
|
stop func() |
|
|
|
|
buf []dataset.Row |
|
|
|
|
|
|
|
|
|
reader *dataset.Reader |
|
|
|
|
columns []dataset.Column |
|
|
|
|
columnDesc []*logsmd.ColumnDesc |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// NewLogsReader creates a new LogsReader that reads from the logs section of
|
|
|
|
|
@ -53,7 +65,7 @@ func NewLogsReader(obj *Object, sectionIndex int) *LogsReader { |
|
|
|
|
// MatchStreams may only be called before reading begins or after a call to
|
|
|
|
|
// [LogsReader.Reset].
|
|
|
|
|
func (r *LogsReader) MatchStreams(ids iter.Seq[int64]) error { |
|
|
|
|
if r.next != nil { |
|
|
|
|
if r.ready { |
|
|
|
|
return fmt.Errorf("cannot change matched streams after reading has started") |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
@ -72,7 +84,7 @@ func (r *LogsReader) MatchStreams(ids iter.Seq[int64]) error { |
|
|
|
|
// A predicate may only be set before reading begins or after a call to
|
|
|
|
|
// [LogsReader.Reset].
|
|
|
|
|
func (r *LogsReader) SetPredicate(p LogsPredicate) error { |
|
|
|
|
if r.next != nil { |
|
|
|
|
if r.ready { |
|
|
|
|
return fmt.Errorf("cannot change predicate after reading has started") |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
@ -84,72 +96,91 @@ func (r *LogsReader) SetPredicate(p LogsPredicate) error { |
|
|
|
|
// into s. It returns the number of records read and any error encountered. At
|
|
|
|
|
// the end of the logs section, Read returns 0, io.EOF.
|
|
|
|
|
func (r *LogsReader) Read(ctx context.Context, s []Record) (int, error) { |
|
|
|
|
// TODO(rfratto): The implementation below is the initial, naive approach. It
|
|
|
|
|
// lacks a few features that will be needed at scale:
|
|
|
|
|
//
|
|
|
|
|
// * Read columns/pages in batches of len(s), rather than one row at a time,
|
|
|
|
|
//
|
|
|
|
|
// * Add page-level filtering based on min/max page values to quickly filter
|
|
|
|
|
// out batches of rows without needing to download or decode them.
|
|
|
|
|
//
|
|
|
|
|
// * Download pages in batches, rather than one at a time.
|
|
|
|
|
//
|
|
|
|
|
// * Only download/decode non-predicate columns following finding rows that
|
|
|
|
|
// match all predicate columns.
|
|
|
|
|
//
|
|
|
|
|
// * Reuse as much memory as possible from a combination of s and the state
|
|
|
|
|
// of LogsReader.
|
|
|
|
|
//
|
|
|
|
|
// These details can change internally without changing the API exposed by
|
|
|
|
|
// LogsReader, which is designed to permit efficient use in the future.
|
|
|
|
|
|
|
|
|
|
if r.obj == nil { |
|
|
|
|
return 0, io.EOF |
|
|
|
|
} else if r.idx < 0 { |
|
|
|
|
return 0, fmt.Errorf("invalid section index %d", r.idx) |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
if r.next == nil { |
|
|
|
|
err := r.initIter(ctx) |
|
|
|
|
if !r.ready { |
|
|
|
|
err := r.initReader(ctx) |
|
|
|
|
if err != nil { |
|
|
|
|
return 0, err |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
for i := range s { |
|
|
|
|
res, ok := r.nextMatching() |
|
|
|
|
if !ok { |
|
|
|
|
return i, io.EOF |
|
|
|
|
} |
|
|
|
|
r.buf = slices.Grow(r.buf, len(s)) |
|
|
|
|
r.buf = r.buf[:len(s)] |
|
|
|
|
|
|
|
|
|
n, err := r.reader.Read(ctx, r.buf) |
|
|
|
|
if err != nil && !errors.Is(err, io.EOF) { |
|
|
|
|
return 0, fmt.Errorf("reading rows: %w", err) |
|
|
|
|
} else if n == 0 && errors.Is(err, io.EOF) { |
|
|
|
|
return 0, io.EOF |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
record, err := res.Value() |
|
|
|
|
for i := range r.buf[:n] { |
|
|
|
|
readRecord, err := logs.Decode(r.columnDesc, r.buf[i]) |
|
|
|
|
if err != nil { |
|
|
|
|
return i, fmt.Errorf("reading record: %w", err) |
|
|
|
|
return i, fmt.Errorf("decoding record: %w", err) |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
s[i] = Record{ |
|
|
|
|
StreamID: record.StreamID, |
|
|
|
|
Timestamp: record.Timestamp, |
|
|
|
|
Metadata: convertMetadata(record.Metadata), |
|
|
|
|
Line: record.Line, |
|
|
|
|
StreamID: readRecord.StreamID, |
|
|
|
|
Timestamp: readRecord.Timestamp, |
|
|
|
|
Metadata: convertMetadata(readRecord.Metadata), |
|
|
|
|
Line: readRecord.Line, |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
return len(s), nil |
|
|
|
|
return n, nil |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
func (r *LogsReader) initIter(ctx context.Context) error { |
|
|
|
|
func (r *LogsReader) initReader(ctx context.Context) error { |
|
|
|
|
dec := r.obj.dec.LogsDecoder() |
|
|
|
|
sec, err := r.findSection(ctx) |
|
|
|
|
if err != nil { |
|
|
|
|
return fmt.Errorf("finding section: %w", err) |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
if r.stop != nil { |
|
|
|
|
r.stop() |
|
|
|
|
columnDescs, err := dec.Columns(ctx, sec) |
|
|
|
|
if err != nil { |
|
|
|
|
return fmt.Errorf("reading columns: %w", err) |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
dset := encoding.LogsDataset(dec, sec) |
|
|
|
|
columns, err := result.Collect(dset.ListColumns(ctx)) |
|
|
|
|
if err != nil { |
|
|
|
|
return fmt.Errorf("reading columns: %w", err) |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
seq := logs.IterSection(ctx, r.obj.dec.LogsDecoder(), sec) |
|
|
|
|
r.next, r.stop = result.Pull(seq) |
|
|
|
|
// r.predicate doesn't contain mappings of stream IDs; we need to build
|
|
|
|
|
// that as a separate predicate and AND them together.
|
|
|
|
|
predicate := streamIDPredicate(maps.Keys(r.matchIDs), columns, columnDescs) |
|
|
|
|
if r.predicate != nil { |
|
|
|
|
predicate = dataset.AndPredicate{ |
|
|
|
|
Left: predicate, |
|
|
|
|
Right: translateLogsPredicate(r.predicate, columns, columnDescs), |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
readerOpts := dataset.ReaderOptions{ |
|
|
|
|
Dataset: dset, |
|
|
|
|
Columns: columns, |
|
|
|
|
Predicate: predicate, |
|
|
|
|
|
|
|
|
|
TargetCacheSize: 16_000_000, // Permit up to 16MB of cache pages.
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
if r.reader == nil { |
|
|
|
|
r.reader = dataset.NewReader(readerOpts) |
|
|
|
|
} else { |
|
|
|
|
r.reader.Reset(readerOpts) |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
r.columnDesc = columnDescs |
|
|
|
|
r.columns = columns |
|
|
|
|
r.ready = true |
|
|
|
|
return nil |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
@ -173,96 +204,184 @@ func (r *LogsReader) findSection(ctx context.Context) (*filemd.SectionInfo, erro |
|
|
|
|
return nil, fmt.Errorf("section index %d not found", r.idx) |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
func (r *LogsReader) nextMatching() (result.Result[logs.Record], bool) { |
|
|
|
|
if r.next == nil { |
|
|
|
|
return result.Result[logs.Record]{}, false |
|
|
|
|
func convertMetadata(md push.LabelsAdapter) labels.Labels { |
|
|
|
|
l := make(labels.Labels, 0, len(md)) |
|
|
|
|
for _, label := range md { |
|
|
|
|
l = append(l, labels.Label{Name: label.Name, Value: label.Value}) |
|
|
|
|
} |
|
|
|
|
sort.Sort(l) |
|
|
|
|
return l |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
NextRow: |
|
|
|
|
res, ok := r.next() |
|
|
|
|
if !ok { |
|
|
|
|
return res, ok |
|
|
|
|
} |
|
|
|
|
// Reset resets the LogsReader with a new object and section index to read
|
|
|
|
|
// from. Reset allows reusing a LogsReader without allocating a new one.
|
|
|
|
|
//
|
|
|
|
|
// Any set predicate is cleared when Reset is called.
|
|
|
|
|
//
|
|
|
|
|
// Reset may be called with a nil object and a negative section index to clear
|
|
|
|
|
// the LogsReader without needing a new object.
|
|
|
|
|
func (r *LogsReader) Reset(obj *Object, sectionIndex int) { |
|
|
|
|
r.obj = obj |
|
|
|
|
r.idx = sectionIndex |
|
|
|
|
r.ready = false |
|
|
|
|
|
|
|
|
|
record, err := res.Value() |
|
|
|
|
if err != nil { |
|
|
|
|
return res, true |
|
|
|
|
clear(r.matchIDs) |
|
|
|
|
r.predicate = nil |
|
|
|
|
|
|
|
|
|
r.columns = nil |
|
|
|
|
r.columnDesc = nil |
|
|
|
|
|
|
|
|
|
// We leave r.reader as-is to avoid reallocating; it'll be reset on the first
|
|
|
|
|
// call to Read.
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
func streamIDPredicate(ids iter.Seq[int64], columns []dataset.Column, columnDesc []*logsmd.ColumnDesc) dataset.Predicate { |
|
|
|
|
var res dataset.Predicate |
|
|
|
|
|
|
|
|
|
streamIDColumn := findColumnFromDesc(columns, columnDesc, func(desc *logsmd.ColumnDesc) bool { |
|
|
|
|
return desc.Type == logsmd.COLUMN_TYPE_STREAM_ID |
|
|
|
|
}) |
|
|
|
|
if streamIDColumn == nil { |
|
|
|
|
return dataset.FalsePredicate{} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
if r.matchIDs != nil { |
|
|
|
|
if _, ok := r.matchIDs[record.StreamID]; !ok { |
|
|
|
|
goto NextRow |
|
|
|
|
for id := range ids { |
|
|
|
|
p := dataset.EqualPredicate{ |
|
|
|
|
Column: streamIDColumn, |
|
|
|
|
Value: dataset.Int64Value(id), |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
if !matchLogsPredicate(r.predicate, record) { |
|
|
|
|
goto NextRow |
|
|
|
|
if res == nil { |
|
|
|
|
res = p |
|
|
|
|
} else { |
|
|
|
|
res = dataset.OrPredicate{ |
|
|
|
|
Left: res, |
|
|
|
|
Right: p, |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
return res, true |
|
|
|
|
return res |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
func matchLogsPredicate(p Predicate, record logs.Record) bool { |
|
|
|
|
func translateLogsPredicate(p LogsPredicate, columns []dataset.Column, columnDesc []*logsmd.ColumnDesc) dataset.Predicate { |
|
|
|
|
if p == nil { |
|
|
|
|
return true |
|
|
|
|
return nil |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
switch p := p.(type) { |
|
|
|
|
case AndPredicate[LogsPredicate]: |
|
|
|
|
return matchLogsPredicate(p.Left, record) && matchLogsPredicate(p.Right, record) |
|
|
|
|
return dataset.AndPredicate{ |
|
|
|
|
Left: translateLogsPredicate(p.Left, columns, columnDesc), |
|
|
|
|
Right: translateLogsPredicate(p.Right, columns, columnDesc), |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
case OrPredicate[LogsPredicate]: |
|
|
|
|
return matchLogsPredicate(p.Left, record) || matchLogsPredicate(p.Right, record) |
|
|
|
|
return dataset.OrPredicate{ |
|
|
|
|
Left: translateLogsPredicate(p.Left, columns, columnDesc), |
|
|
|
|
Right: translateLogsPredicate(p.Right, columns, columnDesc), |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
case NotPredicate[LogsPredicate]: |
|
|
|
|
return !matchLogsPredicate(p.Inner, record) |
|
|
|
|
return dataset.NotPredicate{ |
|
|
|
|
Inner: translateLogsPredicate(p.Inner, columns, columnDesc), |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
case TimeRangePredicate[LogsPredicate]: |
|
|
|
|
return matchTimestamp(p, record.Timestamp) |
|
|
|
|
timeColumn := findColumnFromDesc(columns, columnDesc, func(desc *logsmd.ColumnDesc) bool { |
|
|
|
|
return desc.Type == logsmd.COLUMN_TYPE_TIMESTAMP |
|
|
|
|
}) |
|
|
|
|
if timeColumn == nil { |
|
|
|
|
return dataset.FalsePredicate{} |
|
|
|
|
} |
|
|
|
|
return convertLogsTimePredicate(p, timeColumn) |
|
|
|
|
|
|
|
|
|
case MetadataMatcherPredicate: |
|
|
|
|
return getMetadata(record.Metadata, p.Key) == p.Value |
|
|
|
|
metadataColumn := findColumnFromDesc(columns, columnDesc, func(desc *logsmd.ColumnDesc) bool { |
|
|
|
|
return desc.Type == logsmd.COLUMN_TYPE_METADATA && desc.Info.Name == p.Key |
|
|
|
|
}) |
|
|
|
|
if metadataColumn == nil { |
|
|
|
|
return dataset.FalsePredicate{} |
|
|
|
|
} |
|
|
|
|
return dataset.EqualPredicate{ |
|
|
|
|
Column: metadataColumn, |
|
|
|
|
Value: dataset.StringValue(p.Value), |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
case MetadataFilterPredicate: |
|
|
|
|
return p.Keep(p.Key, getMetadata(record.Metadata, p.Key)) |
|
|
|
|
metadataColumn := findColumnFromDesc(columns, columnDesc, func(desc *logsmd.ColumnDesc) bool { |
|
|
|
|
return desc.Type == logsmd.COLUMN_TYPE_METADATA && desc.Info.Name == p.Key |
|
|
|
|
}) |
|
|
|
|
if metadataColumn == nil { |
|
|
|
|
return dataset.FalsePredicate{} |
|
|
|
|
} |
|
|
|
|
return dataset.FuncPredicate{ |
|
|
|
|
Column: metadataColumn, |
|
|
|
|
Keep: func(_ dataset.Column, value dataset.Value) bool { |
|
|
|
|
return p.Keep(p.Key, valueToString(value)) |
|
|
|
|
}, |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
default: |
|
|
|
|
// Unsupported predicates should already be caught by
|
|
|
|
|
// [LogsReader.SetPredicate].
|
|
|
|
|
panic(fmt.Sprintf("unsupported predicate type %T", p)) |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
func getMetadata(md push.LabelsAdapter, key string) string { |
|
|
|
|
for _, l := range md { |
|
|
|
|
if l.Name == key { |
|
|
|
|
return l.Value |
|
|
|
|
func convertLogsTimePredicate(p TimeRangePredicate[LogsPredicate], column dataset.Column) dataset.Predicate { |
|
|
|
|
var start dataset.Predicate = dataset.GreaterThanPredicate{ |
|
|
|
|
Column: column, |
|
|
|
|
Value: dataset.Int64Value(p.StartTime.UnixNano()), |
|
|
|
|
} |
|
|
|
|
if p.IncludeStart { |
|
|
|
|
start = dataset.OrPredicate{ |
|
|
|
|
Left: start, |
|
|
|
|
Right: dataset.EqualPredicate{ |
|
|
|
|
Column: column, |
|
|
|
|
Value: dataset.Int64Value(p.StartTime.UnixNano()), |
|
|
|
|
}, |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
return "" |
|
|
|
|
} |
|
|
|
|
var end dataset.Predicate = dataset.LessThanPredicate{ |
|
|
|
|
Column: column, |
|
|
|
|
Value: dataset.Int64Value(p.EndTime.UnixNano()), |
|
|
|
|
} |
|
|
|
|
if p.IncludeEnd { |
|
|
|
|
end = dataset.OrPredicate{ |
|
|
|
|
Left: end, |
|
|
|
|
Right: dataset.EqualPredicate{ |
|
|
|
|
Column: column, |
|
|
|
|
Value: dataset.Int64Value(p.EndTime.UnixNano()), |
|
|
|
|
}, |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
func convertMetadata(md push.LabelsAdapter) labels.Labels { |
|
|
|
|
l := make(labels.Labels, 0, len(md)) |
|
|
|
|
for _, label := range md { |
|
|
|
|
l = append(l, labels.Label{Name: label.Name, Value: label.Value}) |
|
|
|
|
return dataset.AndPredicate{ |
|
|
|
|
Left: start, |
|
|
|
|
Right: end, |
|
|
|
|
} |
|
|
|
|
sort.Sort(l) |
|
|
|
|
return l |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// Reset resets the LogsReader with a new object and section index to read
|
|
|
|
|
// from. Reset allows reusing a LogsReader without allocating a new one.
|
|
|
|
|
//
|
|
|
|
|
// Any set predicate is cleared when Reset is called.
|
|
|
|
|
//
|
|
|
|
|
// Reset may be called with a nil object and a negative section index to clear
|
|
|
|
|
// the LogsReader without needing a new object.
|
|
|
|
|
func (r *LogsReader) Reset(obj *Object, sectionIndex int) { |
|
|
|
|
if r.stop != nil { |
|
|
|
|
r.stop() |
|
|
|
|
func findColumnFromDesc[Desc any](columns []dataset.Column, descs []Desc, check func(Desc) bool) dataset.Column { |
|
|
|
|
for i, desc := range descs { |
|
|
|
|
if check(desc) { |
|
|
|
|
return columns[i] |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
return nil |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
r.obj = obj |
|
|
|
|
r.idx = sectionIndex |
|
|
|
|
r.next = nil |
|
|
|
|
r.stop = nil |
|
|
|
|
|
|
|
|
|
clear(r.matchIDs) |
|
|
|
|
r.predicate = nil |
|
|
|
|
func valueToString(value dataset.Value) string { |
|
|
|
|
switch value.Type() { |
|
|
|
|
case datasetmd.VALUE_TYPE_UNSPECIFIED: |
|
|
|
|
return "" |
|
|
|
|
case datasetmd.VALUE_TYPE_INT64: |
|
|
|
|
return strconv.FormatInt(value.Int64(), 10) |
|
|
|
|
case datasetmd.VALUE_TYPE_UINT64: |
|
|
|
|
return strconv.FormatUint(value.Uint64(), 10) |
|
|
|
|
case datasetmd.VALUE_TYPE_STRING: |
|
|
|
|
return value.String() |
|
|
|
|
default: |
|
|
|
|
panic(fmt.Sprintf("unsupported value type %s", value.Type())) |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
Robert Fratto
1 year ago
committed by