Chunks Finder

main.go

package main

import (
        "context"
        "flag"
        "fmt"
        "time"

        "github.com/grafana/dskit/middleware"
        "github.com/grafana/dskit/user"
        "github.com/grafana/loki/v3/pkg/indexgateway"
        "github.com/grafana/loki/v3/pkg/logproto"
        "github.com/grafana/loki/v3/pkg/logql/syntax"
        "github.com/grafana/loki/v3/pkg/querier/plan"
        "github.com/prometheus/common/model"
        "google.golang.org/grpc"
        "google.golang.org/grpc/credentials/insecure"
)

var ()

func main() {
        var tenant string
        var query string
        var addr string
        var start string
        var end string

        flag.StringVar(&tenant, "tenant", "fake", "X-Scope-OrgID")
        flag.StringVar(&query, "query", "", "LogQL query")
        flag.StringVar(&addr, "index-gateway-addr", "localhost:9095", "gRPC address of the index gateway")
        flag.StringVar(&start, "from", "", "defaults to now()-1h")
        flag.StringVar(&end, "through", "", "defaults to now()")
        flag.Parse()

        fmt.Println("Tenant         ", tenant)
        fmt.Println("Query          ", query)
        fmt.Println("Index Gateway  ", addr)

        if query == "" {
                panic("query is required")
        }

        if tenant == "" {
                panic("tenant is required")
        }

        now := model.TimeFromUnix(time.Now().UTC().Unix())
        var from model.Time
        if start == "" {
                from = now.Add(-1 * time.Hour)
        } else {
                ts, err := time.Parse(time.DateTime, start)
                if err != nil {
                        panic(err)
                }
                from = model.TimeFromUnixNano(ts.UnixNano())
        }

        var through model.Time
        if end == "" {
                through = now
        } else {
                ts, err := time.Parse(time.DateTime, end)
                if err != nil {
                        panic(err)
                }
                through = model.TimeFromUnixNano(ts.UnixNano())
        }

        opts := []grpc.DialOption{
                grpc.WithLocalDNSResolution(),
                grpc.WithChainUnaryInterceptor(middleware.ClientUserHeaderInterceptor),
                grpc.WithChainStreamInterceptor(middleware.StreamClientUserHeaderInterceptor),
                grpc.WithTransportCredentials(insecure.NewCredentials()),
        }
        pool, err := indexgateway.NewClientPool(addr, opts)
        if err != nil {
                panic(err)
        }

        var matchers string
        expr := syntax.MustParseExpr(query)
        expr.Walk(func(e syntax.Expr) bool {
                switch e := e.(type) {
                case *syntax.MatchersExpr:
                        matchers = e.String()
                        return false
                }
                return true
        })

        ctx := context.Background()
        ctx = user.InjectOrgID(ctx, tenant)
        ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
        defer cancel()

        req := &logproto.GetChunkRefRequest{
                From:     from,
                Through:  through,
                Matchers: matchers,
                Plan: plan.QueryPlan{
                        AST: expr,
                },
        }

        res, err := pool.GetChunkRef(ctx, req)
        if err != nil {
                panic(err)
        }

        for _, ref := range res.Refs {
                fmt.Println(newerExternalKey(*ref))
        }
}

func newerExternalKey(ref logproto.ChunkRef) string {
        return fmt.Sprintf("%s/%x/%x:%x:%x", ref.UserID, ref.Fingerprint, int64(ref.From), int64(ref.Through), ref.Checksum)
}