engine

Documentation

Overview

Package engine translates Loki API queries into journald journal operations. Methods correspond 1-to-1 with Loki query endpoints and accept only domain types — no HTTP request or response types leak into this package.

Index

• Constants
• Variables
• type Engine
  • func New(p *util.Pool[*journal.Journal], schema *model.Schema) *Engine
  • func (e *Engine) IndexStats(matchers string, start, end time.Time) (*logproto.IndexStatsResponse, error)
  • func (e *Engine) LabelValues(name string) ([]string, error)
  • func (e *Engine) Labels() ([]string, error)
  • func (e *Engine) LogQueryRange(query string, start, end time.Time, limit int, direction logproto.Direction) (loghttp.Streams, error)
  • func (e *Engine) MetricQuery(query string, ts time.Time, direction logproto.Direction) (loghttp.Vector, error)
  • func (e *Engine) MetricQueryRange(query string, start, end time.Time, step time.Duration, ...) (loghttp.Matrix, error)
  • func (e *Engine) Series(filters []*labels.Matcher, start, end time.Time) ([]loghttp.LabelSet, error)
  • func (e *Engine) Tail(ctx context.Context, query string, start time.Time, _ int, ...) error

Constants

const DefaultLabelValuesLimit = 10000

DefaultLabelValuesLimit is the maximum number of distinct values returned by LabelValues. This prevents unbounded memory growth for high-cardinality fields like MESSAGE or timestamps.

Variables

var ErrLabelExcluded = fmt.Errorf("label excluded by configuration")

ErrLabelExcluded is returned when a label is not in the schema's label set.

Types

type Engine

type Engine struct {
	// contains filtered or unexported fields
}

Engine executes Loki-compatible queries against journald journals. For each request it acquires a Journal from the pool, using it for the duration of that request before releasing it back.

func New

func New(p *util.Pool[*journal.Journal], schema *model.Schema) *Engine

New creates an Engine that acquires Journals from p for each request. Non-excluded fields become stream labels; excluded fields become structured metadata.

func (*Engine) IndexStats

func (e *Engine) IndexStats(matchers string, start, end time.Time) (*logproto.IndexStatsResponse, error)

IndexStats returns approximate counts of streams, chunks, entries, and bytes for entries matching the given time range. This corresponds to the /loki/api/v1/index/stats endpoint.

matchers is a LogQL stream selector string (e.g. `{job="sshd"}`). Currently treated as a match-all for any non-empty string; per-stream filtering will be implemented later.

The returned IndexStatsResponse can be marshaled directly to produce Loki-compatible JSON with uint64 fields for streams, chunks, entries, and bytes.

func (*Engine) LabelValues

func (e *Engine) LabelValues(name string) ([]string, error)

LabelValues returns all distinct values for the given label across the journal files, up to DefaultLabelValuesLimit (10000) values. This corresponds to the /loki/api/v1/label/{name}/values endpoint.

name is the label to inspect (case-insensitive). Returns ErrLabelExcluded if the label is in the schema's exclude list.

func (*Engine) Labels

func (e *Engine) Labels() ([]string, error)

Labels returns all label names present in the journal files that are NOT in the schema's exclude list. Names are lowercased to match Loki conventions. This corresponds to the /loki/api/v1/labels endpoint.

func (*Engine) LogQueryRange

func (e *Engine) LogQueryRange(query string, start, end time.Time, limit int, direction logproto.Direction) (loghttp.Streams, error)

LogQueryRange executes a log query over a time range, returning log stream entries that match the LogQL selector. This corresponds to the /loki/api/v1/query_range endpoint for log queries.

query is a LogQL log selector expression (e.g. `{job="sshd"}`). start and end define the time window to search. limit caps the total number of log entries returned. direction controls whether entries are returned newest-first (BACKWARD) or oldest-first (FORWARD).

Returns loghttp.Streams which can be marshaled directly with json-iterator to produce Loki-compatible JSON (entries serialized as ["ts","line"] arrays).

func (*Engine) MetricQuery

func (e *Engine) MetricQuery(query string, ts time.Time, direction logproto.Direction) (loghttp.Vector, error)

MetricQuery executes an instant metric query at a single point in time. This corresponds to the /loki/api/v1/query endpoint for metric expressions.

query is a LogQL sample expression (e.g. `count_over_time({job="sshd"}[5m])`). ts is the evaluation timestamp; the lookback window is determined by the range selector in the query expression (e.g., [5m]). direction is accepted for API compatibility and does not affect evaluation.

Returns a loghttp.Vector — one Sample per distinct stream.

func (*Engine) MetricQueryRange

func (e *Engine) MetricQueryRange(query string, start, end time.Time, step time.Duration, direction logproto.Direction) (loghttp.Matrix, error)

MetricQueryRange executes a metric query over a time range, returning sampled results at the given step interval. This corresponds to the /loki/api/v1/query_range endpoint for metric queries (e.g. rate, count_over_time, bytes_over_time).

query is a LogQL sample expression (e.g. `count_over_time({job="sshd"}[5m])`). start and end define the time window to search. step is the sampling interval; a data point is produced for each step within [start, end]. direction is accepted for API compatibility but does not affect the output order of sampled data points.

Returns a loghttp.Matrix — one SampleStream per distinct stream, each containing a SamplePair for every step that had matching entries.

func (*Engine) Series

func (e *Engine) Series(filters []*labels.Matcher, start, end time.Time) ([]loghttp.LabelSet, error)

Series returns the distinct stream label sets that exist within the given time range. This corresponds to the /loki/api/v1/series endpoint.

filters are parsed stream selectors (e.g. from LogQL) that identify which label sets to include. If filters is empty, nil is returned — callers must pass at least one matcher.

The returned LabelSets use Loki's JSON key "stream" when marshaled, matching the expected format for /loki/api/v1/series responses.

func (*Engine) Tail

func (e *Engine) Tail(ctx context.Context, query string, start time.Time, _ int, cb func(string, labels.Labels, loghttp.Entry)) error

Tail streams log entries matching query starting from start. Positions the journal at start, processes the first matching entry there, then delegates to Journal.Follow for all subsequent entries (existing and new). Blocks until ctx is cancelled. Each matching entry is passed to cb with its stream key, stream labels, and loghttp.Entry.