关于golang:聊聊loki的Query

序

本文次要钻研一下 loki 的 Query

Query

loki/pkg/logql/engine.go

// Query is a LogQL query to be executed.
type Query interface {
    // Exec processes the query.
    Exec(ctx context.Context) (Result, error)
}

// Result is the result of a query execution.
type Result struct {
    Data       promql_parser.Value
    Statistics stats.Result
}

Query 接口定义了 Exec 办法，返回 Result；Result 定义了 Data、Statistics 属性

Exec

loki/pkg/logql/engine.go

// Exec Implements `Query`. It handles instrumentation & defers to Eval.
func (q *query) Exec(ctx context.Context) (Result, error) {log, ctx := spanlogger.New(ctx, "query.Exec")
    defer log.Finish()

    rangeType := GetRangeType(q.params)
    timer := prometheus.NewTimer(queryTime.WithLabelValues(string(rangeType)))
    defer timer.ObserveDuration()

    // records query statistics
    var statResult stats.Result
    start := time.Now()
    ctx = stats.NewContext(ctx)

    data, err := q.Eval(ctx)

    statResult = stats.Snapshot(ctx, time.Since(start))
    statResult.Log(level.Debug(log))

    status := "200"
    if err != nil {
        status = "500"
        if errors.Is(err, ErrParse) || errors.Is(err, ErrPipeline) || errors.Is(err, ErrLimit) {status = "400"}
    }

    if q.record {RecordMetrics(ctx, q.params, status, statResult)
    }

    return Result{
        Data:       data,
        Statistics: statResult,
    }, err
}

Exec 办法执行 q.Eval(ctx) 及 stats.Snapshot

Eval

loki/pkg/logql/engine.go

func (q *query) Eval(ctx context.Context) (promql_parser.Value, error) {ctx, cancel := context.WithTimeout(ctx, q.timeout)
    defer cancel()

    expr, err := q.parse(ctx, q.params.Query())
    if err != nil {return nil, err}

    switch e := expr.(type) {
    case SampleExpr:
        value, err := q.evalSample(ctx, e)
        return value, err

    case LogSelectorExpr:
        iter, err := q.evaluator.Iterator(ctx, e, q.params)
        if err != nil {return nil, err}

        defer helpers.LogErrorWithContext(ctx, "closing iterator", iter.Close)
        streams, err := readStreams(iter, q.params.Limit(), q.params.Direction(), q.params.Interval())
        return streams, err
    default:
        return nil, errors.New("Unexpected type (%T): cannot evaluate")
    }
}

Eval 办法执行 q.parse 解析为 Expr，之后依据 Expr 的类型做不同解决，如果是 SampleExpr 类型执行 q.evalSample；如果是 LogSelectorExpr 类型则执行 q.evaluator.Iterator

Snapshot

loki/pkg/logql/stats/context.go

func Snapshot(ctx context.Context, execTime time.Duration) Result {
    // ingester data is decoded from grpc trailers.
    res := decodeTrailers(ctx)
    // collect data from store.
    s, ok := ctx.Value(storeKey).(*StoreData)
    if ok {
        res.Store.TotalChunksRef = s.TotalChunksRef
        res.Store.TotalChunksDownloaded = s.TotalChunksDownloaded
        res.Store.ChunksDownloadTime = s.ChunksDownloadTime.Seconds()}
    // collect data from chunks iteration.
    c, ok := ctx.Value(chunksKey).(*ChunkData)
    if ok {
        res.Store.HeadChunkBytes = c.HeadChunkBytes
        res.Store.HeadChunkLines = c.HeadChunkLines
        res.Store.DecompressedBytes = c.DecompressedBytes
        res.Store.DecompressedLines = c.DecompressedLines
        res.Store.CompressedBytes = c.CompressedBytes
        res.Store.TotalDuplicates = c.TotalDuplicates
    }

    existing, err := GetResult(ctx)
    if err != nil {res.ComputeSummary(execTime)
        return res
    }

    existing.Merge(res)
    existing.ComputeSummary(execTime)
    return *existing

}

Snapshot 办法从 ctx.Value 取出 StoreData 及 ChunkData 计算 res，而后再取出 Result，进行 Merge 及 ComputeSummary

小结

loki 的 Query 接口定义了 Exec 办法，返回 Result；Result 定义了 Data、Statistics 属性；query 实现了 Query 接口，其 Exec 办法执行 q.Eval(ctx) 及 stats.Snapshot。

doc

• cortex