로키의 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