From 5fcd07e2e15fcb7441bf1ca86f5dd20c8e67cbef Mon Sep 17 00:00:00 2001 From: Shawn Date: Mon, 14 Sep 2026 12:39:52 -0700 Subject: [PATCH] feat(tempo-cli): add benchmark profile command Profile a local block for read-path benchmarking: record the block metadata, the row-group count, and present and absent trace IDs. A profile holds only what has to be read from the block, so a benchmark run does not inspect it and every variant of an experiment works from the same measurements. The row-group count comes from the parquet footer rather than meta.TotalRecords, which can over-count: streamingBlock.Complete increments it and then flushes, so a block whose rows divide evenly into row groups records one more than it has. Present IDs are taken at an even stride over every row, so a trace-by-ID benchmark is not confined to the opening pages of each row group. Each absent ID is the midpoint between a present ID and the ID that follows it in the block: row groups hold contiguous ranges of the sort key and the scan reads every row, so the pair is adjacent and the midpoint is absent by construction. That needs no lookup, so a block whose bloom filters are absent can still be profiled. --- .chloggen/benchmark-profile.yaml | 5 + cmd/tempo-cli/cmd-benchmark-profile.go | 62 ++++ cmd/tempo-cli/cmd-benchmark-profile_test.go | 36 ++ cmd/tempo-cli/main.go | 4 + docs/sources/tempo/operations/tempo_cli.md | 34 ++ pkg/benchmark/helper_test.go | 114 +++++++ pkg/benchmark/profile.go | 109 ++++++ pkg/benchmark/profile_builder.go | 321 ++++++++++++++++++ pkg/benchmark/profile_builder_test.go | 350 ++++++++++++++++++++ pkg/benchmark/profile_test.go | 88 +++++ 10 files changed, 1123 insertions(+) create mode 100644 .chloggen/benchmark-profile.yaml create mode 100644 cmd/tempo-cli/cmd-benchmark-profile.go create mode 100644 cmd/tempo-cli/cmd-benchmark-profile_test.go create mode 100644 pkg/benchmark/helper_test.go create mode 100644 pkg/benchmark/profile.go create mode 100644 pkg/benchmark/profile_builder.go create mode 100644 pkg/benchmark/profile_builder_test.go create mode 100644 pkg/benchmark/profile_test.go diff --git a/.chloggen/benchmark-profile.yaml b/.chloggen/benchmark-profile.yaml new file mode 100644 index 00000000000..e24db4ff4aa --- /dev/null +++ b/.chloggen/benchmark-profile.yaml @@ -0,0 +1,5 @@ +change_type: feature +component: tempo-cli +note: Add `tempo-cli benchmark profile` to profile a block for read-path benchmarking. +issues: [] +user: zhxiaogg diff --git a/cmd/tempo-cli/cmd-benchmark-profile.go b/cmd/tempo-cli/cmd-benchmark-profile.go new file mode 100644 index 00000000000..b609fd976fd --- /dev/null +++ b/cmd/tempo-cli/cmd-benchmark-profile.go @@ -0,0 +1,62 @@ +package main + +import ( + "context" + "errors" + "fmt" + "os" + "strconv" + "strings" + + "github.com/grafana/tempo/pkg/benchmark" +) + +type benchmarkProfileCmd struct { + Block string `arg:"" help:"path to the block directory, laid out as //"` + + TraceIDs string `name:"trace-ids" help:"number of present trace IDs to sample, or 'all' to enumerate at run time; one absent ID is derived per present ID. 0 skips the scan they need" default:"10000"` + Out string `short:"o" help:"file to write the profile to, instead of stdout" default:""` +} + +func (cmd *benchmarkProfileCmd) Run(_ *globalOptions) error { + numTraceIDs, err := parseTraceIDCount(cmd.TraceIDs) + if err != nil { + return err + } + + ctx := context.Background() + meta, r, err := benchmark.LoadLocalBlock(ctx, cmd.Block) + if err != nil { + return err + } + + profile, err := benchmark.ProfileBlock(ctx, meta, r, benchmark.ProfileOptions{NumTraceIDs: numTraceIDs}) + if err != nil { + return err + } + + out := os.Stdout + if cmd.Out != "" { + f, err := os.Create(cmd.Out) + if err != nil { + return err + } + defer f.Close() + out = f + } + return profile.Write(out) +} + +func parseTraceIDCount(s string) (int, error) { + if strings.EqualFold(strings.TrimSpace(s), "all") { + return benchmark.TraceIDsAll, nil + } + n, err := strconv.Atoi(s) + if err != nil { + return 0, fmt.Errorf("--trace-ids must be a number or 'all': %w", err) + } + if n < 0 { + return 0, errors.New("--trace-ids must not be negative") + } + return n, nil +} diff --git a/cmd/tempo-cli/cmd-benchmark-profile_test.go b/cmd/tempo-cli/cmd-benchmark-profile_test.go new file mode 100644 index 00000000000..7e78d4d1fbc --- /dev/null +++ b/cmd/tempo-cli/cmd-benchmark-profile_test.go @@ -0,0 +1,36 @@ +package main + +import ( + "testing" + + "github.com/stretchr/testify/require" + + "github.com/grafana/tempo/pkg/benchmark" +) + +func TestParseTraceIDCount(t *testing.T) { + for _, tc := range []struct { + in string + want int + wantErr bool + }{ + {in: "0", want: 0}, + {in: "10000", want: 10000}, + {in: "all", want: benchmark.TraceIDsAll}, + {in: "ALL", want: benchmark.TraceIDsAll}, + {in: " all ", want: benchmark.TraceIDsAll}, + {in: "-1", wantErr: true}, + {in: "lots", wantErr: true}, + {in: "", wantErr: true}, + } { + t.Run(tc.in, func(t *testing.T) { + got, err := parseTraceIDCount(tc.in) + if tc.wantErr { + require.Error(t, err) + return + } + require.NoError(t, err) + require.Equal(t, tc.want, got) + }) + } +} diff --git a/cmd/tempo-cli/main.go b/cmd/tempo-cli/main.go index 6d48fa987a5..f145c826fac 100644 --- a/cmd/tempo-cli/main.go +++ b/cmd/tempo-cli/main.go @@ -56,6 +56,10 @@ var cli struct { Schema viewSchemaCmd `cmd:"" help:"View parquet schema"` } `cmd:""` + Benchmark struct { + Profile benchmarkProfileCmd `cmd:"" help:"Profile a block for read-path benchmarking"` + } `cmd:""` + Gen struct { AttrIndex attrIndexCmd `cmd:"" help:"Generate an attribute index for a parquet block (EXPERIMENTAL)"` } `cmd:""` diff --git a/docs/sources/tempo/operations/tempo_cli.md b/docs/sources/tempo/operations/tempo_cli.md index fce9c91dc4c..f7fa76abf33 100644 --- a/docs/sources/tempo/operations/tempo_cli.md +++ b/docs/sources/tempo/operations/tempo_cli.md @@ -487,6 +487,40 @@ Example: tempo-cli view schema -c ./tempo.yaml single-tenant ca314fba-efec-4852-ba3f-8d2b0bbf69f1 ``` +## Benchmark profile + +Profile a local block for read-path benchmarking. Writes a JSON file recording +what had to be measured from the block — its metadata, its row-group count, and +present and absent trace IDs to look up — so that a benchmark run does not have +to inspect the block, and every variant of an experiment works from the same +measurements. + +```bash +tempo-cli benchmark profile +``` + +Arguments: + +- `block-path` Path to the block directory on local disk, laid out as + `//`. + +Options: + +- `--trace-ids` Number of present trace IDs to sample, or `all` to enumerate + every ID at run time rather than embedding them. Defaults to `10000`. One + absent ID is derived per present ID. Pass `0` to skip trace IDs, and with them + the full scan they require. +- `-o`, `--out` File to write the profile to. Defaults to stdout. + +Profiles built from a customer block embed real trace IDs. Treat them as local +artifacts. + +Example: + +```bash +tempo-cli benchmark profile /data/traces/single-tenant/ca314fba-efec-4852-ba3f-8d2b0bbf69f1 --trace-ids=10000 -o profile.json +``` + ## Query search command Search blocks in a given time range for a specific key/value pair. diff --git a/pkg/benchmark/helper_test.go b/pkg/benchmark/helper_test.go new file mode 100644 index 00000000000..b0432bddfe7 --- /dev/null +++ b/pkg/benchmark/helper_test.go @@ -0,0 +1,114 @@ +package benchmark + +import ( + "bytes" + "context" + "io" + "math/rand" + "slices" + "sort" + "testing" + "time" + + "github.com/google/uuid" + "github.com/stretchr/testify/require" + + "github.com/grafana/tempo/pkg/tempopb" + "github.com/grafana/tempo/pkg/util/test" + "github.com/grafana/tempo/tempodb/backend" + "github.com/grafana/tempo/tempodb/backend/local" + "github.com/grafana/tempo/tempodb/encoding" + "github.com/grafana/tempo/tempodb/encoding/common" +) + +// testBlock writes a block of numTraces traces, with enough row groups to +// exercise per-row-group work. +func testBlock(t *testing.T, numTraces int) (*backend.BlockMeta, backend.Reader, string) { + t.Helper() + + // Real trace IDs are random across all 16 bytes, and code under test hashes + // them, so a fixed seed keeps the fixture both realistic and reproducible. + rng := rand.New(rand.NewSource(0xbeef)) + + ids := make([][]byte, 0, numTraces) + for range numTraces { + id := make([]byte, 16) + _, err := rng.Read(id) + require.NoError(t, err) + ids = append(ids, test.ValidTraceID(id)) + } + + meta, r, bucket := testBlockWithTraceIDs(t, ids) + + rowGroups, err := rowGroupCount(context.Background(), meta, r) + require.NoError(t, err) + require.Greater(t, rowGroups, 1, "need more than one row group to exercise per-row-group work") + + return meta, r, bucket +} + +// testBlockWithTraceIDs writes a block holding exactly the given trace IDs, +// using a small row group size. It returns the bucket root alongside the block, +// for tests that address it by path. +func testBlockWithTraceIDs(t *testing.T, ids [][]byte) (*backend.BlockMeta, backend.Reader, string) { + t.Helper() + + bucket := t.TempDir() + rawR, rawW, _, err := local.New(&local.Config{Path: bucket}) + require.NoError(t, err) + r, w := backend.NewReader(rawR), backend.NewWriter(rawW) + + ids = slices.Clone(ids) + // vParquet blocks are sorted by trace ID. + sort.Slice(ids, func(i, j int) bool { return bytes.Compare(ids[i], ids[j]) < 0 }) + + iter := &sliceIterator{} + for _, id := range ids { + iter.add(id, test.MakeTraceWithSpanCount(1, 4, id)) + } + + // test.MakeSpan timestamps spans at time.Now(), so the block's window has + // to bracket that or every query filters them all out. + now := time.Now() + + enc := encoding.LatestEncoding() + meta := backend.NewBlockMeta("test-tenant", uuid.New(), enc.Version()) + meta.TotalObjects = int64(len(ids)) + meta.StartTime = now.Add(-time.Hour) + meta.EndTime = now.Add(time.Hour) + + cfg := &common.BlockConfig{ + BloomFP: 0.01, + BloomShardSizeBytes: 100 * 1024, + RowGroupSizeBytes: 8 * 1024, + Version: enc.Version(), + } + + out, err := enc.CreateBlock(context.Background(), cfg, meta, iter, r, w) + require.NoError(t, err) + + return out, r, bucket +} + +type sliceIterator struct { + ids []common.ID + traces []*tempopb.Trace +} + +var _ common.Iterator = (*sliceIterator)(nil) + +func (i *sliceIterator) add(id common.ID, tr *tempopb.Trace) { + i.ids = append(i.ids, id) + i.traces = append(i.traces, tr) +} + +func (i *sliceIterator) Next(context.Context) (common.ID, *tempopb.Trace, error) { + if len(i.ids) == 0 { + return nil, nil, io.EOF + } + id, tr := i.ids[0], i.traces[0] + i.ids, i.traces = i.ids[1:], i.traces[1:] + return id, tr, nil +} + +func (i *sliceIterator) Close() {} diff --git a/pkg/benchmark/profile.go b/pkg/benchmark/profile.go new file mode 100644 index 00000000000..fda6313ba59 --- /dev/null +++ b/pkg/benchmark/profile.go @@ -0,0 +1,109 @@ +// Package benchmark profiles a Tempo block for read-path benchmarking. +// +// A profile records what had to be measured from the block: its metadata, the +// authoritative row-group count, and a sample of trace IDs. Everything a run +// derives from those — shards, query windows, steps — is left to the runner, so +// one profile serves every variant of an experiment. +package benchmark + +import ( + "encoding/json" + "errors" + "fmt" + "io" + "slices" + "time" + + "github.com/grafana/tempo/pkg/util" + "github.com/grafana/tempo/tempodb/backend" +) + +const ProfileSchemaVersion = 1 + +// traceIDHexLen is the length of a padded 16-byte trace ID in hex. +const traceIDHexLen = 32 + +const ( + TraceIDModeSample = "sample" + TraceIDModeAll = "all" +) + +// TraceIDProfile describes which trace IDs a trace-by-ID benchmark can use. +type TraceIDProfile struct { + Mode string `json:"mode"` + Present []string `json:"present,omitempty"` + // Absent IDs are verified missing, so the miss path is measured against a + // real negative rather than a guess. + Absent []string `json:"absent,omitempty"` +} + +type BuildInfo struct { + TempoVersion string `json:"tempoVersion,omitempty"` + GitSHA string `json:"gitSHA,omitempty"` +} + +// BlockProfile is what a benchmark needs to know about a block. Block is +// recorded in full because two runs are only comparable if they agree on every +// property of the block except the one under test. +type BlockProfile struct { + SchemaVersion int `json:"schemaVersion"` + GeneratedAt time.Time `json:"generatedAt"` + GeneratedBy BuildInfo `json:"generatedBy"` + Block *backend.BlockMeta `json:"block"` + // RowGroups comes from the parquet footer, not Block.TotalRecords, which + // Tempo's own sharding calls an estimate. + RowGroups int `json:"rowGroups"` + TraceIDs TraceIDProfile `json:"traceIDs"` +} + +func (p *BlockProfile) Write(w io.Writer) error { + enc := json.NewEncoder(w) + enc.SetIndent("", " ") + return enc.Encode(p) +} + +func LoadProfile(r io.Reader) (*BlockProfile, error) { + var p BlockProfile + if err := json.NewDecoder(r).Decode(&p); err != nil { + return nil, err + } + if p.SchemaVersion != ProfileSchemaVersion { + return nil, fmt.Errorf("profile schema version %d is not supported, expected %d", p.SchemaVersion, ProfileSchemaVersion) + } + if err := p.Validate(); err != nil { + return nil, err + } + return &p, nil +} + +// Validate runs on load and after profiling, so an unusable profile is +// rejected where it is produced rather than midway through a run. +func (p *BlockProfile) Validate() error { + if p.Block == nil { + return errors.New("profile has no block metadata") + } + if p.RowGroups <= 0 { + return fmt.Errorf("profile reports %d row groups", p.RowGroups) + } + + switch p.TraceIDs.Mode { + case TraceIDModeSample, TraceIDModeAll: + default: + return fmt.Errorf("unknown trace ID mode %q", p.TraceIDs.Mode) + } + if p.TraceIDs.Mode == TraceIDModeAll && len(p.TraceIDs.Present) > 0 { + return errors.New(`trace ID mode is "all" but present IDs are embedded`) + } + + for _, id := range slices.Concat(p.TraceIDs.Present, p.TraceIDs.Absent) { + // Require the padded form: HexStringToTraceID accepts a short string, + // so a truncated ID would otherwise pass and decode to the wrong trace. + if len(id) != traceIDHexLen { + return fmt.Errorf("trace ID %q is %d characters, expected %d", id, len(id), traceIDHexLen) + } + if _, err := util.HexStringToTraceID(id); err != nil { + return fmt.Errorf("invalid trace ID %q: %w", id, err) + } + } + return nil +} diff --git a/pkg/benchmark/profile_builder.go b/pkg/benchmark/profile_builder.go new file mode 100644 index 00000000000..17f54541943 --- /dev/null +++ b/pkg/benchmark/profile_builder.go @@ -0,0 +1,321 @@ +package benchmark + +import ( + "bytes" + "context" + "errors" + "fmt" + "math" + "math/big" + "path/filepath" + "runtime/debug" + "slices" + "time" + + "github.com/google/uuid" + "github.com/parquet-go/parquet-go" + "github.com/prometheus/common/version" + + "github.com/grafana/tempo/pkg/tempopb" + "github.com/grafana/tempo/pkg/traceql" + "github.com/grafana/tempo/pkg/util" + "github.com/grafana/tempo/tempodb/backend" + "github.com/grafana/tempo/tempodb/backend/local" + "github.com/grafana/tempo/tempodb/encoding" + "github.com/grafana/tempo/tempodb/encoding/common" +) + +// dataFileName is the parquet file in every vParquet block. +const dataFileName = "data.parquet" + +// TraceIDsAll asks for every trace ID in the block rather than a sample. +const TraceIDsAll = -1 + +// ProfileOptions control what is measured. +type ProfileOptions struct { + // NumTraceIDs is how many present IDs to sample: 0 for none, or TraceIDsAll + // to defer enumeration to run time. One absent ID is derived per present + // ID, so the hit and miss paths are measured over the same number of + // samples. + NumTraceIDs int +} + +// LoadLocalBlock opens the block at path, which must be a directory laid out as +// // because that is how backend.KeyPathForBlock +// addresses a block. +func LoadLocalBlock(ctx context.Context, path string) (*backend.BlockMeta, backend.Reader, error) { + path = filepath.Clean(path) + + blockDir, tenantID := filepath.Base(path), filepath.Base(filepath.Dir(path)) + bucket := filepath.Dir(filepath.Dir(path)) + + blockID, err := uuid.Parse(blockDir) + if err != nil { + return nil, nil, fmt.Errorf("%q is not a block directory: its name must be a block ID: %w", path, err) + } + if tenantID == "." || tenantID == string(filepath.Separator) { + return nil, nil, fmt.Errorf("%q has no tenant directory above the block", path) + } + + rawR, _, _, err := local.New(&local.Config{Path: bucket}) + if err != nil { + return nil, nil, err + } + + r := backend.NewReader(rawR) + meta, err := r.BlockMeta(ctx, blockID, tenantID) + if err != nil { + return nil, nil, fmt.Errorf("reading block meta for %s in tenant %s: %w", blockID, tenantID, err) + } + return meta, r, nil +} + +// ProfileBlock measures the block. The read cost is paid once here, and every +// variant of an experiment then works from the same profile. +func ProfileBlock(ctx context.Context, meta *backend.BlockMeta, r backend.Reader, o ProfileOptions) (*BlockProfile, error) { + if meta == nil { + return nil, errors.New("block metadata is required") + } + + pf, err := openParquetFile(ctx, meta, r) + if err != nil { + return nil, err + } + + blk, err := encoding.OpenBlock(meta, r) + if err != nil { + return nil, fmt.Errorf("opening block: %w", err) + } + + traceIDs, err := profileTraceIDs(ctx, blk, pf, o.NumTraceIDs) + if err != nil { + return nil, err + } + + p := &BlockProfile{ + SchemaVersion: ProfileSchemaVersion, + GeneratedAt: time.Now().UTC(), + GeneratedBy: buildInfo(), + Block: meta, + RowGroups: len(pf.RowGroups()), + TraceIDs: traceIDs, + } + if err := p.Validate(); err != nil { + return nil, fmt.Errorf("produced an invalid profile: %w", err) + } + return p, nil +} + +// openParquetFile reads the footer of the block's data file. Its row-group +// count is authoritative: meta.TotalRecords can over-count, because +// streamingBlock.Complete increments it and then flushes, so a block whose rows +// divide evenly into row groups records one more than it has. +func openParquetFile(ctx context.Context, meta *backend.BlockMeta, r backend.Reader) (*parquet.File, error) { + pf, err := parquet.OpenFile(&blockReaderAt{ctx: ctx, r: r, meta: meta}, int64(meta.Size_)) + if err != nil { + return nil, fmt.Errorf("opening %s for block %s: %w", dataFileName, meta.BlockID, err) + } + if len(pf.RowGroups()) == 0 { + return nil, fmt.Errorf("block %s has no row groups", meta.BlockID) + } + return pf, nil +} + +func rowGroupCount(ctx context.Context, meta *backend.BlockMeta, r backend.Reader) (int, error) { + pf, err := openParquetFile(ctx, meta, r) + if err != nil { + return 0, err + } + return len(pf.RowGroups()), nil +} + +// blockReaderAt reads data.parquet through the backend, so the footer can be +// opened without importing a specific block encoding. +type blockReaderAt struct { + ctx context.Context + r backend.Reader + meta *backend.BlockMeta +} + +func (b *blockReaderAt) ReadAt(p []byte, off int64) (int, error) { + err := b.r.ReadRange(b.ctx, dataFileName, uuid.UUID(b.meta.BlockID), b.meta.TenantID, uint64(off), p, nil) + if err != nil { + return 0, err + } + return len(p), nil +} + +// profileTraceIDs samples present trace IDs and derives an absent one next to +// each of them. +func profileTraceIDs(ctx context.Context, blk common.BackendBlock, pf *parquet.File, num int) (TraceIDProfile, error) { + if num == 0 { + return TraceIDProfile{Mode: TraceIDModeSample}, nil + } + if num == TraceIDsAll { + // Embedding every ID would make the profile as large as the block's ID + // column, so record the intent and let the runner enumerate both the + // present IDs and their absent counterparts. + return TraceIDProfile{Mode: TraceIDModeAll}, nil + } + + present, absent, err := sampleTraceIDs(ctx, blk, pf, num) + if err != nil { + return TraceIDProfile{}, err + } + return TraceIDProfile{Mode: TraceIDModeSample, Present: present, Absent: absent}, nil +} + +// sampleTraceIDs takes num present IDs at an even stride over every row of the +// block, and pairs each with an absent ID. +// +// Striding over all rows costs a full scan, but taking the head of each row +// group instead would leave a trace-by-ID benchmark reading only the opening +// pages of each row group — well under 1% of a large block's pages, on a path +// whose cost is dominated by page reads. The scan is paid once per block and +// every variant then reuses the sample, which needs no seed because the block's +// row order is fixed. +// +// Each absent ID is the midpoint between a sampled ID and the one that follows +// it in the block. Because a row group holds a contiguous range of the trace ID +// sort key, and the scan reads every row of it, those two IDs are adjacent in +// the whole block: no trace can lie between them, so the midpoint is absent by +// construction rather than by a lookup. That keeps profiling off the bloom +// filters, and spreads the absent IDs over the block's ID range, which matters +// because the bloom shard a lookup reads is a hash of the whole ID. +func sampleTraceIDs(ctx context.Context, blk common.BackendBlock, pf *parquet.File, num int) (present, absent []string, err error) { + var total int64 + for _, rg := range pf.RowGroups() { + total += rg.NumRows() + } + if total == 0 { + return nil, nil, errors.New("block has no rows") + } + + stride := max(total/int64(num), 1) + + present = make([]string, 0, num) + absent = make([]string, 0, num) + + // An ID can only be paired once its successor is known, and a row group's + // last ID is followed by the next group's first: row groups hold + // contiguous ranges of the sort key, so that pair is adjacent too. + var ( + prev []byte + prevOnStep bool + row int64 + prevGroupID int + ) + for rg, group := range pf.RowGroups() { + ids, err := listTraceIDs(ctx, blk, group, rg) + if err != nil { + return nil, nil, fmt.Errorf("listing trace IDs in row group %d: %w", rg, err) + } + slices.SortFunc(ids, bytes.Compare) + + // The midpoints below are only absent if this row group's IDs all sort + // above the previous one's, so check rather than assume it. + if len(ids) > 0 && prev != nil && bytes.Compare(prev, ids[0]) >= 0 { + return nil, nil, fmt.Errorf("row group %d overlaps row group %d, cannot derive absent IDs", rg, prevGroupID) + } + prevGroupID = rg + + for _, id := range ids { + if prevOnStep { + if mid := midpointTraceID(prev, id); !bytes.Equal(mid, prev) { + present = append(present, util.PadTraceIDString(util.TraceIDToHexString(prev))) + absent = append(absent, util.PadTraceIDString(util.TraceIDToHexString(mid))) + if len(present) == num { + return present, absent, nil + } + } + } + prev, prevOnStep = id, row%stride == 0 + row++ + } + } + + if len(present) == 0 { + return nil, nil, errors.New("found no trace IDs in the block") + } + // Fewer IDs than asked for is a property of the block, not an error. + return present, absent, nil +} + +// listTraceIDs returns every trace ID in one row group. +// +// The time range spans everything representable rather than the block's own +// window: a block's declared start and end can clip traces whose spans fall in +// its ingestion slack, and a partial list would break the adjacency the absent +// IDs rely on. +func listTraceIDs(ctx context.Context, blk common.BackendBlock, group parquet.RowGroup, rowGroup int) ([][]byte, error) { + opts := common.DefaultSearchOptions() + opts.StartPage, opts.TotalPages = rowGroup, 1 + + fetcher := traceql.NewSpansetFetcherWrapperBoth( + func(ctx context.Context, req traceql.FetchSpansRequest) (traceql.FetchSpansResponse, error) { + return blk.Fetch(ctx, req, opts) + }, + func(ctx context.Context, req traceql.FetchSpansRequest) (traceql.FetchSpansOnlyResponse, error) { + return blk.FetchSpans(ctx, req, opts) + }, + ) + + req := &tempopb.SearchRequest{ + Query: "{}", + Limit: uint32(group.NumRows()), + Start: 1, + End: math.MaxUint32, + } + + resp, err := traceql.NewEngine().ExecuteSearch(ctx, req, fetcher) + if err != nil { + return nil, err + } + if resp == nil { + return nil, nil + } + + ids := make([][]byte, 0, len(resp.Traces)) + for _, tr := range resp.Traces { + if tr.TraceID == "" { + continue + } + id, err := util.HexStringToTraceID(tr.TraceID) + if err != nil { + return nil, fmt.Errorf("decoding trace ID %q: %w", tr.TraceID, err) + } + ids = append(ids, id) + } + return ids, nil +} + +// midpointTraceID returns the ID halfway between a and b, read as big-endian +// integers. +func midpointTraceID(a, b []byte) []byte { + mid := new(big.Int).Add(new(big.Int).SetBytes(a), new(big.Int).SetBytes(b)) + mid.Rsh(mid, 1) + + out := make([]byte, len(a)) + mid.FillBytes(out) + return out +} + +// buildInfo falls back to the VCS stamp because the ldflags that populate +// prometheus/common/version are only set by the Makefile. +func buildInfo() BuildInfo { + info := BuildInfo{TempoVersion: version.Version, GitSHA: version.Revision} + if info.TempoVersion != "" && info.GitSHA != "" { + return info + } + + bi, ok := debug.ReadBuildInfo() + if !ok { + return info + } + for _, setting := range bi.Settings { + if setting.Key == "vcs.revision" && info.GitSHA == "" { + info.GitSHA = setting.Value + } + } + return info +} diff --git a/pkg/benchmark/profile_builder_test.go b/pkg/benchmark/profile_builder_test.go new file mode 100644 index 00000000000..62bc2fbf594 --- /dev/null +++ b/pkg/benchmark/profile_builder_test.go @@ -0,0 +1,350 @@ +package benchmark + +import ( + "bytes" + "context" + "os" + "path/filepath" + "slices" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/grafana/tempo/pkg/util" + "github.com/grafana/tempo/tempodb/encoding" + "github.com/grafana/tempo/tempodb/encoding/common" +) + +func TestProfileBlock(t *testing.T) { + ctx := context.Background() + meta, r, _ := testBlock(t, 300) + + p, err := ProfileBlock(ctx, meta, r, ProfileOptions{NumTraceIDs: 50}) + require.NoError(t, err) + require.NoError(t, p.Validate()) + + footer, err := rowGroupCount(ctx, meta, r) + require.NoError(t, err) + + require.Equal(t, ProfileSchemaVersion, p.SchemaVersion) + require.Equal(t, meta, p.Block) + require.Equal(t, footer, p.RowGroups) + + require.Equal(t, TraceIDModeSample, p.TraceIDs.Mode) + require.Len(t, p.TraceIDs.Present, 50) + require.Len(t, p.TraceIDs.Absent, 50) +} + +func TestProfileBlockPresentIDsAreFound(t *testing.T) { + ctx := context.Background() + meta, r, _ := testBlock(t, 300) + + p, err := ProfileBlock(ctx, meta, r, ProfileOptions{NumTraceIDs: 20}) + require.NoError(t, err) + + blk, err := encoding.OpenBlock(meta, r) + require.NoError(t, err) + + for _, hexID := range p.TraceIDs.Present { + id, err := util.HexStringToTraceID(hexID) + require.NoError(t, err) + + resp, err := blk.FindTraceByID(ctx, id, common.DefaultSearchOptions()) + require.NoError(t, err) + require.NotNil(t, resp, "profiled present ID %s was not found", hexID) + require.NotNil(t, resp.Trace) + } +} + +func TestProfileBlockAbsentIDsAreNotFound(t *testing.T) { + ctx := context.Background() + meta, r, _ := testBlock(t, 300) + + p, err := ProfileBlock(ctx, meta, r, ProfileOptions{NumTraceIDs: 5}) + require.NoError(t, err) + + blk, err := encoding.OpenBlock(meta, r) + require.NoError(t, err) + + for _, hexID := range p.TraceIDs.Absent { + id, err := util.HexStringToTraceID(hexID) + require.NoError(t, err) + + resp, err := blk.FindTraceByID(ctx, id, common.DefaultSearchOptions()) + require.NoError(t, err) + if resp != nil { + require.Nil(t, resp.Trace, "profiled absent ID %s was found", hexID) + } + } +} + +// The sample must be stable across calls, otherwise two variants of an +// experiment would look up different IDs and their latencies would not compare. +func TestProfileBlockIsDeterministic(t *testing.T) { + ctx := context.Background() + meta, r, _ := testBlock(t, 300) + + first, err := ProfileBlock(ctx, meta, r, ProfileOptions{NumTraceIDs: 40}) + require.NoError(t, err) + second, err := ProfileBlock(ctx, meta, r, ProfileOptions{NumTraceIDs: 40}) + require.NoError(t, err) + + require.Equal(t, first.TraceIDs, second.TraceIDs) +} + +// The sample must spread over each row group's rows, not sit at its head, or a +// The sample must spread over each row group's rows, not sit at its head, or a +// trace-by-ID benchmark reads only the opening pages of each group. +func TestSampleTraceIDsSpreadsOverAllRows(t *testing.T) { + ctx := context.Background() + meta, r, _ := testBlock(t, 300) + + blk, err := encoding.OpenBlock(meta, r) + require.NoError(t, err) + + pf, err := openParquetFile(ctx, meta, r) + require.NoError(t, err) + require.Greater(t, len(pf.RowGroups()), 1) + + // Where each ID sits within its row group, in sorted order. + type place struct{ group, offset, groupLen int } + placeOf := make(map[string]place) + for rg, group := range pf.RowGroups() { + ids, err := listTraceIDs(ctx, blk, group, rg) + require.NoError(t, err) + require.Equal(t, int(group.NumRows()), len(ids), "the ID list must be complete") + + slices.SortFunc(ids, bytes.Compare) + for i, id := range ids { + placeOf[string(id)] = place{group: rg, offset: i, groupLen: len(ids)} + } + } + + const num = 20 + present, _, err := sampleTraceIDs(ctx, blk, pf, num) + require.NoError(t, err) + require.Len(t, present, num) + + groups := make(map[int]struct{}) + deepest := 0.0 + for _, hexID := range present { + id, err := util.HexStringToTraceID(hexID) + require.NoError(t, err) + + pl, ok := placeOf[string(id)] + require.True(t, ok, "sampled ID is not in the block") + groups[pl.group] = struct{}{} + deepest = max(deepest, float64(pl.offset)/float64(pl.groupLen)) + } + + require.Greater(t, len(groups), 1, "sample should span row groups") + require.Greater(t, deepest, 0.5, "at least one sampled ID should come from the back half of its row group") +} + +// A complete per-row-group ID list is what makes the absent IDs provably +// absent, so the scan must return every row. +func TestListTraceIDsReturnsEveryRow(t *testing.T) { + ctx := context.Background() + meta, r, _ := testBlock(t, 300) + + blk, err := encoding.OpenBlock(meta, r) + require.NoError(t, err) + pf, err := openParquetFile(ctx, meta, r) + require.NoError(t, err) + + total := 0 + for rg, group := range pf.RowGroups() { + ids, err := listTraceIDs(ctx, blk, group, rg) + require.NoError(t, err) + require.Equal(t, int(group.NumRows()), len(ids), "row group %d", rg) + total += len(ids) + } + require.Equal(t, int(meta.TotalObjects), total) +} + +func TestProfileBlockTraceIDsAll(t *testing.T) { + ctx := context.Background() + meta, r, _ := testBlock(t, 100) + + p, err := ProfileBlock(ctx, meta, r, ProfileOptions{NumTraceIDs: TraceIDsAll}) + require.NoError(t, err) + + require.Equal(t, TraceIDModeAll, p.TraceIDs.Mode) + require.Empty(t, p.TraceIDs.Present) + require.Empty(t, p.TraceIDs.Absent) + require.NoError(t, p.Validate()) +} + +// A block whose bloom filters are not present locally can still be profiled, +// so asking for no trace IDs must not read them. +func TestProfileBlockNoTraceIDs(t *testing.T) { + ctx := context.Background() + meta, r, _ := testBlock(t, 50) + + p, err := ProfileBlock(ctx, meta, r, ProfileOptions{NumTraceIDs: 0}) + require.NoError(t, err) + + require.Equal(t, TraceIDModeSample, p.TraceIDs.Mode) + require.Empty(t, p.TraceIDs.Present) + require.Empty(t, p.TraceIDs.Absent) + require.NoError(t, p.Validate()) +} + +// The footer is authoritative: meta.TotalRecords must not be reported, since it +// can over-count and a shard past the real end of the file reads nothing. +func TestProfileBlockIgnoresStaleRowGroupCount(t *testing.T) { + ctx := context.Background() + meta, r, _ := testBlock(t, 300) + + truth, err := rowGroupCount(ctx, meta, r) + require.NoError(t, err) + + meta.TotalRecords = uint32(truth) + 7 + + p, err := ProfileBlock(ctx, meta, r, ProfileOptions{NumTraceIDs: 0}) + require.NoError(t, err) + require.Equal(t, truth, p.RowGroups) + require.NoError(t, p.Validate()) +} + +func TestLoadLocalBlock(t *testing.T) { + want, _, bucket := testBlock(t, 20) + + path := filepath.Join(bucket, want.TenantID, want.BlockID.String()) + got, r, err := LoadLocalBlock(context.Background(), path) + require.NoError(t, err) + require.NotNil(t, r) + + // meta.json is JSON, so times lose their monotonic reading; compare fields. + require.Equal(t, want.BlockID, got.BlockID) + require.Equal(t, want.TenantID, got.TenantID) + require.Equal(t, want.Version, got.Version) + require.Equal(t, want.TotalObjects, got.TotalObjects) + require.Equal(t, want.TotalRecords, got.TotalRecords) + require.Equal(t, want.Size_, got.Size_) + require.True(t, want.StartTime.Equal(got.StartTime)) + require.True(t, want.EndTime.Equal(got.EndTime)) + + // A trailing separator must address the same block. + got, _, err = LoadLocalBlock(context.Background(), path+string(filepath.Separator)) + require.NoError(t, err) + require.Equal(t, want.BlockID, got.BlockID) +} + +func TestLoadLocalBlockRejectsBadPaths(t *testing.T) { + ctx := context.Background() + + for _, tc := range []struct { + name string + path string + wantErr string + }{ + {"block dir is not a UUID", filepath.Join(t.TempDir(), "tenant", "not-a-uuid"), "not a block directory"}, + {"no tenant above the block", "/00000000-0000-0000-0000-000000000000", "no tenant directory"}, + } { + t.Run(tc.name, func(t *testing.T) { + _, _, err := LoadLocalBlock(ctx, tc.path) + require.ErrorContains(t, err, tc.wantErr) + }) + } +} + +// Each absent ID must sit strictly between two present IDs: that is what +// Each absent ID must be the midpoint between a present ID and the ID that +// follows it in the block, which is what makes it absent by construction. +func TestAbsentTraceIDsLieNextToPresentIDs(t *testing.T) { + ctx := context.Background() + meta, r, _ := testBlock(t, 300) + + blk, err := encoding.OpenBlock(meta, r) + require.NoError(t, err) + pf, err := openParquetFile(ctx, meta, r) + require.NoError(t, err) + + var all [][]byte + for rg, group := range pf.RowGroups() { + ids, err := listTraceIDs(ctx, blk, group, rg) + require.NoError(t, err) + all = append(all, ids...) + } + slices.SortFunc(all, bytes.Compare) + + present, absent, err := sampleTraceIDs(ctx, blk, pf, 30) + require.NoError(t, err) + require.Len(t, absent, len(present)) + + for i, hexPresent := range present { + p, err := util.HexStringToTraceID(hexPresent) + require.NoError(t, err) + a, err := util.HexStringToTraceID(absent[i]) + require.NoError(t, err) + + at, found := slices.BinarySearchFunc(all, p, bytes.Compare) + require.True(t, found, "present ID is not in the block") + require.Less(t, at+1, len(all), "sample should not include the last ID") + + require.Equal(t, midpointTraceID(all[at], all[at+1]), a) + require.Negative(t, bytes.Compare(p, a), "absent ID should sort after its present ID") + require.Negative(t, bytes.Compare(a, all[at+1]), "absent ID should sort before the next present ID") + } +} + +// IDs must land across the block's shards rather than on a few of them. +func TestAbsentTraceIDsCoverBloomShards(t *testing.T) { + ctx := context.Background() + meta, r, _ := testBlock(t, 300) + + p, err := ProfileBlock(ctx, meta, r, ProfileOptions{NumTraceIDs: 200}) + require.NoError(t, err) + + const shardCount = 28 + seen := make(map[int]struct{}, shardCount) + for _, hexID := range p.TraceIDs.Absent { + id, err := util.HexStringToTraceID(hexID) + require.NoError(t, err) + seen[common.ShardKeyForTraceID(id, shardCount)] = struct{}{} + } + require.Equal(t, shardCount, len(seen)) +} + +func TestMidpointTraceID(t *testing.T) { + for _, tc := range []struct{ name, a, b, want string }{ + {"halfway", "00000000000000000000000000000000", "00000000000000000000000000000010", "00000000000000000000000000000008"}, + {"odd gap rounds down", "00000000000000000000000000000000", "00000000000000000000000000000003", "00000000000000000000000000000001"}, + {"adjacent yields the lower", "00000000000000000000000000000004", "00000000000000000000000000000005", "00000000000000000000000000000004"}, + {"equal yields itself", "1111111111111111111111111111111f", "1111111111111111111111111111111f", "1111111111111111111111111111111f"}, + {"carry across the whole width", "00000000000000000000000000000000", "ffffffffffffffffffffffffffffffff", "7fffffffffffffffffffffffffffffff"}, + } { + t.Run(tc.name, func(t *testing.T) { + a, err := util.HexStringToTraceID(tc.a) + require.NoError(t, err) + b, err := util.HexStringToTraceID(tc.b) + require.NoError(t, err) + + got := midpointTraceID(a, b) + require.Len(t, got, 16) + require.Equal(t, tc.want, util.PadTraceIDString(util.TraceIDToHexString(got))) + }) + } +} + +// Absent IDs are derived from adjacency rather than looked up, so a block whose +// bloom filters are missing can still be profiled in full. +func TestProfileBlockWithoutBloomFilters(t *testing.T) { + ctx := context.Background() + meta, r, bucket := testBlock(t, 300) + + blockDir := filepath.Join(bucket, meta.TenantID, meta.BlockID.String()) + blooms, err := filepath.Glob(filepath.Join(blockDir, "bloom-*")) + require.NoError(t, err) + require.NotEmpty(t, blooms, "fixture should have bloom filters to remove") + for _, f := range blooms { + require.NoError(t, os.Remove(f)) + } + + p, err := ProfileBlock(ctx, meta, r, ProfileOptions{NumTraceIDs: 20}) + require.NoError(t, err) + require.Len(t, p.TraceIDs.Present, 20) + require.Len(t, p.TraceIDs.Absent, 20) + require.NoError(t, p.Validate()) +} diff --git a/pkg/benchmark/profile_test.go b/pkg/benchmark/profile_test.go new file mode 100644 index 00000000000..158f75f62d6 --- /dev/null +++ b/pkg/benchmark/profile_test.go @@ -0,0 +1,88 @@ +package benchmark + +import ( + "bytes" + "strings" + "testing" + "time" + + "github.com/google/uuid" + "github.com/stretchr/testify/require" + + "github.com/grafana/tempo/tempodb/backend" +) + +func validProfile() *BlockProfile { + meta := backend.NewBlockMeta("test-tenant", uuid.MustParse("00000000-0000-0000-0000-00000000beef"), "vParquet5") + meta.StartTime = time.Unix(1000, 0).UTC() + meta.EndTime = time.Unix(4600, 0).UTC() + meta.TotalObjects = 10 + meta.TotalRecords = 3 + + return &BlockProfile{ + SchemaVersion: ProfileSchemaVersion, + GeneratedAt: time.Unix(5000, 0).UTC(), + GeneratedBy: BuildInfo{TempoVersion: "0.0.0-test", GitSHA: "deadbeef"}, + Block: meta, + RowGroups: 3, + TraceIDs: TraceIDProfile{ + Mode: TraceIDModeSample, + Present: []string{"0102030405060708090a0b0c0d0e0f10"}, + Absent: []string{strings.Repeat("00", 16)}, + }, + } +} + +func TestProfileRoundTrip(t *testing.T) { + want := validProfile() + + var buf bytes.Buffer + require.NoError(t, want.Write(&buf)) + + got, err := LoadProfile(&buf) + require.NoError(t, err) + require.Equal(t, want, got) +} + +func TestLoadProfileRejectsOtherSchemaVersion(t *testing.T) { + p := validProfile() + p.SchemaVersion = ProfileSchemaVersion + 1 + + var buf bytes.Buffer + require.NoError(t, p.Write(&buf)) + + _, err := LoadProfile(&buf) + require.ErrorContains(t, err, "schema version") +} + +func TestProfileValidate(t *testing.T) { + for _, tc := range []struct { + name string + mutate func(*BlockProfile) + wantErr string + }{ + {"no block", func(p *BlockProfile) { p.Block = nil }, "no block metadata"}, + {"no row groups", func(p *BlockProfile) { p.RowGroups = 0 }, "reports 0 row groups"}, + {"unknown id mode", func(p *BlockProfile) { p.TraceIDs.Mode = "some" }, "unknown trace ID mode"}, + {"short trace id", func(p *BlockProfile) { p.TraceIDs.Present = []string{"0102"} }, "is 4 characters"}, + {"empty trace id", func(p *BlockProfile) { p.TraceIDs.Absent = []string{""} }, "is 0 characters"}, + { + name: "non-hex trace id", + mutate: func(p *BlockProfile) { p.TraceIDs.Present = []string{strings.Repeat("z", 32)} }, + wantErr: "invalid trace ID", + }, + { + name: "mode all with embedded ids", + mutate: func(p *BlockProfile) { p.TraceIDs.Mode = TraceIDModeAll }, + wantErr: "present IDs are embedded", + }, + } { + t.Run(tc.name, func(t *testing.T) { + p := validProfile() + tc.mutate(p) + require.ErrorContains(t, p.Validate(), tc.wantErr) + }) + } + + require.NoError(t, validProfile().Validate()) +}