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()) +}