diff --git a/internal/telemetry/parquet.go b/internal/telemetry/parquet.go index e945e0c8..365e8c3c 100644 --- a/internal/telemetry/parquet.go +++ b/internal/telemetry/parquet.go @@ -49,6 +49,10 @@ type BatchMetadata struct { Spans int `json:"spans"` Logs int `json:"logs"` Metrics int `json:"metrics"` + // Bytes is the batch's on-disk size, derived rather than persisted: it is + // measured when a batch is loaded, from stats the load already performs. + // Zero means not measured. + Bytes int64 `json:"-"` } type TraceQuery struct { @@ -1119,6 +1123,10 @@ func loadStoredBatch(dir string) (*storedBatch, error) { if !info.Mode().IsRegular() { return nil, fmt.Errorf("%s Parquet is not a regular file", signal.name) } + // Size the batch from the stat this validation already performs. + // Compaction prices a merge by the bytes it admits, and measuring here + // costs nothing over the estimate it would otherwise fall back to. + metadata.Bytes += info.Size() file, err := os.Open(path) if err != nil { return nil, err diff --git a/internal/telemetry/store/compaction.go b/internal/telemetry/store/compaction.go index f5f0ec6e..0128103e 100644 --- a/internal/telemetry/store/compaction.go +++ b/internal/telemetry/store/compaction.go @@ -16,8 +16,17 @@ import ( ) const ( - maxCompactionRows = 25_000_000 - minCompactionInputs = 8 + maxCompactionRows = 25_000_000 + maxCompactionBytes = 256 << 20 + + // assumedCompactionBytesPerRow prices a batch whose size was never + // measured. Measured batches run well under this -- span rows with JSON + // attributes compress to roughly 75 bytes on disk -- so the estimate is + // deliberately several times pessimistic: over-charging costs smaller + // groups and more passes, while under-charging costs the merge that the + // byte ceiling exists to prevent. + assumedCompactionBytesPerRow = 256 + minCompactionInputs = 8 ) var parquetSignals = [...]string{"spans", "logs", "metrics"} @@ -198,7 +207,7 @@ func selectBoundedCompactionGroup(group []telemetry.BatchMetadata, maxBatches in return ordered[i].ID < ordered[j].ID }) candidate := make([]telemetry.BatchMetadata, 0, min(maxBatches, len(ordered))) - var rows int64 + var rows, bytes int64 saturated := false var smallest int64 for _, batch := range ordered { @@ -206,6 +215,7 @@ func selectBoundedCompactionGroup(group []telemetry.BatchMetadata, maxBatches in if batchRows <= 0 || batchRows > maxCompactionRows { continue } + batchBytes := compactionBatchBytes(batch) if smallest == 0 { smallest = batchRows } @@ -217,18 +227,47 @@ func selectBoundedCompactionGroup(group []telemetry.BatchMetadata, maxBatches in saturated = true break } + // A merge opens a cursor per row group across every input and holds + // each one's dictionaries at once, so the live cost of a pass tracks + // the bytes it admits. Rows cannot stand in for that: wide spans and + // bare log lines differ by an order of magnitude at equal row counts. + // A pair is always admitted, however large: a merge of two inputs is + // bounded by those inputs, and refusing it would strand files that are + // individually over budget instead of ever shrinking them. Past a pair + // the ceiling binds. + if len(candidate) >= 2 && bytes > maxCompactionBytes-batchBytes { + saturated = true + break + } candidate = append(candidate, batch) rows += batchRows + bytes += batchBytes } if rows == maxCompactionRows || smallest > 0 && smallest > maxCompactionRows-rows { saturated = true } + if bytes == maxCompactionBytes { + saturated = true + } if len(candidate) == maxBatches || saturated && len(candidate) >= 2 { return candidate } return nil } +// compactionBatchBytes prices a batch for the byte ceiling, falling back to a +// pessimistic per-row estimate when the batch was never measured. +func compactionBatchBytes(batch telemetry.BatchMetadata) int64 { + if batch.Bytes > 0 { + return batch.Bytes + } + rows := compactionBatchRows(batch) + if rows <= 0 || rows > math.MaxInt64/assumedCompactionBytesPerRow { + return math.MaxInt64 + } + return rows * assumedCompactionBytesPerRow +} + func compactionBatchRows(batch telemetry.BatchMetadata) int64 { if batch.Spans < 0 { return math.MaxInt64 diff --git a/internal/telemetry/store/compaction_bytes_test.go b/internal/telemetry/store/compaction_bytes_test.go new file mode 100644 index 00000000..868b4cfb --- /dev/null +++ b/internal/telemetry/store/compaction_bytes_test.go @@ -0,0 +1,73 @@ +package store + +import ( + "fmt" + "testing" + + "github.com/labstack/fanout/internal/telemetry" +) + +// A merge holds every input's dictionaries and cursors at once, so the live +// cost of a pass tracks the bytes it admits, not the number of files or the +// number of rows. Row counts cannot stand in for that: a batch of wide spans +// carrying large attribute payloads and a batch of bare log lines can hold the +// same row count and differ by an order of magnitude on disk. +func TestSelectBoundedCompactionGroupStopsAtTheByteBudget(t *testing.T) { + const oneMiB = 1 << 20 + group := make([]telemetry.BatchMetadata, 0, 16) + for i := range 16 { + group = append(group, telemetry.BatchMetadata{ + ID: string(rune('a' + i)), + Spans: 1_000, + MinIngestedNanos: int64(i), + Bytes: 64 * oneMiB, + }) + } + + candidate := selectBoundedCompactionGroup(group, 128) + + if len(candidate) == 0 { + t.Fatal("no group selected; the byte budget must still admit a mergeable group") + } + var admitted int64 + for _, batch := range candidate { + admitted += batch.Bytes + } + if admitted > maxCompactionBytes { + t.Errorf("admitted %d bytes, budget %d: selection ignored the byte ceiling", admitted, maxCompactionBytes) + } + if len(candidate) == len(group) { + t.Errorf("admitted all %d inputs at %d bytes each; the budget never bound", len(group), int64(64*oneMiB)) + } +} + +// Batches whose size is unknown must not be treated as free. Nothing persists +// the byte count -- it is measured when a batch is loaded -- so a zero means +// "not measured". Charging those an estimate keeps the ceiling meaningful; +// treating them as weightless would reopen the hole the budget exists to +// close, and refusing them outright would stop compaction dead if any path +// ever failed to measure. +func TestSelectBoundedCompactionGroupChargesUnsizedBatchesAnEstimate(t *testing.T) { + group := make([]telemetry.BatchMetadata, 0, 128) + for i := range 128 { + group = append(group, telemetry.BatchMetadata{ + ID: fmt.Sprintf("batch-%03d", i), + Spans: 50_000, + MinIngestedNanos: int64(i), + // Bytes deliberately left zero: never measured. + }) + } + + candidate := selectBoundedCompactionGroup(group, 128) + + if len(candidate) == 0 { + t.Fatal("no group selected; unsized batches must still be compactable") + } + if len(candidate) == len(group) { + t.Fatalf("admitted all %d unsized inputs; an unmeasured batch must not count as zero bytes", len(group)) + } + estimated := int64(len(candidate)) * 50_000 * assumedCompactionBytesPerRow + if estimated > maxCompactionBytes { + t.Errorf("estimated %d bytes across %d inputs, budget %d", estimated, len(candidate), int64(maxCompactionBytes)) + } +} diff --git a/internal/telemetry/store/repository_test.go b/internal/telemetry/store/repository_test.go index 4a5bf334..43b2a1b1 100644 --- a/internal/telemetry/store/repository_test.go +++ b/internal/telemetry/store/repository_test.go @@ -511,6 +511,9 @@ func TestSelectCompactionBatchesBuildsRowBoundedGroup(t *testing.T) { batches[i] = telemetry.BatchMetadata{ ID: fmt.Sprintf("batch-%d", i), MaxIngestedNanos: 1, Generation: 2, Spans: 2_000_000, + // Sized well under the byte ceiling so the row ceiling stays the + // constraint under test here; the byte ceiling has its own tests. + Bytes: 1 << 20, } } selected := selectCompactionBatches(batches, maxBatches) @@ -533,6 +536,9 @@ func TestSelectCompactionBatchesCombinesSaturatedLargeFiles(t *testing.T) { batches[i] = telemetry.BatchMetadata{ ID: fmt.Sprintf("large-%d", i), MaxIngestedNanos: 1, Generation: 3, Spans: 6_000_000, + // Sized well under the byte ceiling so saturation here is the row + // ceiling, which is what this test is about. + Bytes: 1 << 20, } } selected := selectCompactionBatches(batches, maxBatches)