diff --git a/opentelemetry-app/.gitignore b/opentelemetry-app/.gitignore index ab94fdfe..96b378b7 100644 --- a/opentelemetry-app/.gitignore +++ b/opentelemetry-app/.gitignore @@ -1 +1,2 @@ -fakemetricload +/fakemetricload +/e2esdkbench diff --git a/opentelemetry-app/README.md b/opentelemetry-app/README.md index 10e86104..ba96b7ef 100644 --- a/opentelemetry-app/README.md +++ b/opentelemetry-app/README.md @@ -37,3 +37,101 @@ data point. The collector processor merges incoming sketches and emits query res The benchmark script (`opentelemetry-collector-contrib-patch/cmd/bench.sh`) runs end-to-end benchmarks for all sketch types using `fakemetricload` as the SDK load generator. See the bench script for the full list of targets. + +--- + +## SDK Pipeline Configuration Files + +YAML configuration files define the full SDK→collector pipeline in one place, +replacing scattered CLI flags. A per-sketch-type file is provided plus a raw-gauge +baseline. + +| File | Sketch type | `transmit_sketch` | +|------|------------|-------------------| +| `config.yaml` | DDSketch (base template) | `true` | +| `config-ddsketch.yaml` | DDSketch | `true` | +| `config-kll.yaml` | KLL | `true` | +| `config-hll.yaml` | HLL | `true` | +| `config-countminsketch.yaml` | CountMinSketch | `true` | +| `config-countsketch.yaml` | CountSketch | `true` | +| `config-baseline.yaml` | none (raw gauge) | `false` | + +### `transmit_sketch` + +Controls which side of the pipeline performs sketch aggregation: + +- **`transmit_sketch: true`** — SDK attaches an `Aggregation` view to the instrument + and pre-aggregates all recorded values into a sketch before each export tick. The + collector receives a single compact sketch data point per series per window. +- **`transmit_sketch: false`** — SDK emits raw `Float64Gauge` (LastValue) observations. + The collector-side processor (e.g. `kllprocessor`, `hllprocessor`) performs the + aggregation. Used for the `baseline` type and for collector-driven sketch modes. + +### Config schema + +```yaml +exporter: + endpoint: localhost:4317 # OTLP gRPC endpoint + insecure: true + max_send_msg_size_mb: 64 # raise for large HLL/CountSketch payloads + +reader: + interval: 1s # SDK export interval + +sketch: + type: ddsketch # ddsketch | kll | hll | countsketch | countminsketch | baseline + transmit_sketch: true # true = SDK aggregates; false = collector aggregates + ddsketch: + relative_accuracy: 0.01 + kll: + k: 256 + hll: {} + countsketch: + rows: 5 + cols: 2000 + epsilon: 0.01 + delta: 0.99 + countminsketch: + rows: 5 + cols: 2000 + +instruments: + - name: benchmark.latency + unit: ms + instrument_kind: histogram # histogram | gauge + group_by: # attribute keys that form the sketch dimension + - host.name + - service.name + series_per_sketch: 1 # 1=per-series, 0=all-in-one, N=group N series + +# Mirrors ControllerConfig from asap-planner-rs — can be fed directly to asap-planner-rs +asap_query: + metrics: + - metric: benchmark.latency + labels: [host.name, service.name] + query_groups: + - id: 1 + queries: ["quantile_over_time(0.99, benchmark_latency[5m])"] + repetition_delay: 5000 + controller_options: + accuracy_sla: 0.01 + latency_sla: 100.0 + sketch_parameters: # optional overrides consumed by asap-planner-rs + DatasketchesKLL: + K: 256 + CountMinSketch: + depth: 5 + width: 2000 + +load: + workers: 10 + series: 1000 + samples_per_sec_per_series: 50.0 + duration: 60s + distribution: + type: zipf + s: 1.1 + v: 1.0 + max: 500 + mean: 250.0 +``` diff --git a/opentelemetry-app/cmd/e2esdkbench/main.go b/opentelemetry-app/cmd/e2esdkbench/main.go index 5f1fb013..448a8cbe 100644 --- a/opentelemetry-app/cmd/e2esdkbench/main.go +++ b/opentelemetry-app/cmd/e2esdkbench/main.go @@ -49,8 +49,9 @@ import ( // --------------------------------------------------------------------------- var ( + configPath = flag.String("config", "", "Path to YAML pipeline config file; when set, overrides sketch/series/interval flags") endpoint = flag.String("endpoint", "localhost:4317", "OTLP gRPC endpoint of the collector") - sketchArg = flag.String("sketch-type", "ddsketch", "Sketch type: ddsketch|kll|hll") + sketchArg = flag.String("sketch-type", "ddsketch", "Sketch type: ddsketch|kll|hll|countsketch|countminsketch|baseline") seriesCount = flag.Int("series", 1000, "Total number of distinct time series") samplesPerSecPerSeries = flag.Float64("samples-per-sec-per-series", 50.0, "Samples per second per series (controls worker record rate)") workers = flag.Int("workers", 10, "Number of worker goroutines (internal parallelism)") @@ -246,11 +247,53 @@ func resourceSampler(ctx context.Context, samples *[]sample, mu *sync.Mutex) { func main() { flag.Parse() + // --- Config file: when --config is set, load YAML and propagate into flags --- + var pipelineCfg *sdkmetric.PipelineConfig + if *configPath != "" { + var err error + pipelineCfg, err = sdkmetric.LoadPipelineConfig(*configPath) + if err != nil { + log.Fatalf("failed to load config: %v", err) + } + *sketchArg = pipelineCfg.Sketch.Type + *endpoint = pipelineCfg.Exporter.Endpoint + if pipelineCfg.Load.Series > 0 { + *seriesCount = pipelineCfg.Load.Series + } + if pipelineCfg.Load.SamplesPerSecPerSeries > 0 { + *samplesPerSecPerSeries = pipelineCfg.Load.SamplesPerSecPerSeries + } + if pipelineCfg.Load.Workers > 0 { + *workers = pipelineCfg.Load.Workers + } + if pipelineCfg.Load.Duration > 0 { + *duration = pipelineCfg.Load.Duration + } + if pipelineCfg.Load.Distribution.S > 0 { + *zipfS = pipelineCfg.Load.Distribution.S + } + if pipelineCfg.Load.Distribution.V > 0 { + *zipfV = pipelineCfg.Load.Distribution.V + } + if pipelineCfg.Load.Distribution.Max > 0 { + *zipfMax = pipelineCfg.Load.Distribution.Max + } + if pipelineCfg.Load.Distribution.Mean > 0 { + *zipfMean = pipelineCfg.Load.Distribution.Mean + } + if pipelineCfg.Sketch.DDSketch.RelativeAccuracy > 0 { + *ddsketchAccuracy = pipelineCfg.Sketch.DDSketch.RelativeAccuracy + } + if pipelineCfg.Sketch.KLL.K > 0 { + *kllK = pipelineCfg.Sketch.KLL.K + } + } + mode := strings.ToLower(*sketchArg) switch mode { - case "ddsketch", "kll", "hll", "baseline": + case "ddsketch", "kll", "hll", "countsketch", "countminsketch", "baseline": default: - log.Fatalf("invalid --sketch-type %q; valid: ddsketch|kll|hll", mode) + log.Fatalf("invalid --sketch-type %q; valid: ddsketch|kll|hll|countsketch|countminsketch|baseline", mode) } if err := os.MkdirAll(*outputDir, 0o755); err != nil { @@ -304,6 +347,11 @@ func main() { agg = sdkmetric.AggregationKLLSketch{K: *kllK} case "hll": agg = sdkmetric.AggregationHLLSketch{} + case "countsketch": + agg = sdkmetric.AggregationCountSketch{Epsilon: *csEpsilon, Delta: *csDelta} + case "countminsketch": + agg = sdkmetric.AggregationCountMinSketch{Rows: *cmsRows, Cols: *cmsCols} + // "baseline": agg stays nil — Float64Gauge uses default LastValue aggregation. } ctx, cancel := context.WithTimeout(context.Background(), *duration+10*time.Second) @@ -320,15 +368,20 @@ func main() { log.Fatalf("failed to create OTLP exporter: %v", err) } - // View overrides aggregation for sketch types. Baseline skips the view so - // the Float64Gauge uses its natural LastValue aggregation — no nil-aggregation - // side-effect that would suppress reporting. + // Build provider options. When a config file was loaded, use its views so + // that transmit_sketch semantics (SDK-side vs collector-side aggregation) + // are driven by the config rather than hardcoded flag logic. Otherwise fall + // back to the existing flag-driven view construction. providerOpts := []sdkmetric.Option{ sdkmetric.WithReader( sdkmetric.NewPeriodicReader(exp, sdkmetric.WithInterval(readerInterval)), ), } - if agg != nil { + if pipelineCfg != nil { + for _, v := range pipelineCfg.ToViews() { + providerOpts = append(providerOpts, sdkmetric.WithView(v)) + } + } else if agg != nil { providerOpts = append(providerOpts, sdkmetric.WithView( sdkmetric.NewView( sdkmetric.Instrument{Name: "benchmark.latency"}, diff --git a/opentelemetry-app/cmd/fakemetricload/main.go b/opentelemetry-app/cmd/fakemetricload/main.go new file mode 100644 index 00000000..e7c7baeb --- /dev/null +++ b/opentelemetry-app/cmd/fakemetricload/main.go @@ -0,0 +1,343 @@ +package main + +import ( + "context" + "flag" + "fmt" + "log" + "math/rand" + "strings" + "sync" + "time" + + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc" + "go.opentelemetry.io/otel/metric" + sdkmetric "go.opentelemetry.io/otel/sdk/metric" +) + +var ( + configPath = flag.String("config", "", "Path to YAML pipeline config file; when set, overrides sketch/interval flags") + endpoint = flag.String("endpoint", "localhost:4317", "OTLP gRPC endpoint") + workers = flag.Int("workers", 4, "Number of worker goroutines") + hosts = flag.Int("hosts", 10, "Hosts per worker") + metrics = flag.Int("metrics", 10, "Metrics per host") + interval = flag.Duration("interval", 10*time.Second, "SDK export interval — controls data points/sec to collector") + duration = flag.Duration("duration", 60*time.Second, "Run duration (0 = forever)") + sketchArg = flag.String("sketch-type", "ddsketch", "Sketch aggregation: ddsketch|kll|countsketch|countminsketch|hll|baseline") + + // How many Zipf values to record per series per export window. + // For ddsketch this controls sketch richness; for others only the last value is kept (gauge). + samplesPerInterval = flag.Int("samples-per-interval", 1, + "Values recorded per series per export window (ddsketch: sketch richness; gauge types: last value wins)") + + // DDSketch + ddsketchAccuracy = flag.Float64("ddsketch-accuracy", 0.01, "DDSketch relative accuracy (0,1)") + + // KLL + kllK = flag.Int("kll-k", 256, "KLL sketch parameter k (>= 2)") + + // CountSketch + csEpsilon = flag.Float64("countsketch-epsilon", 0.01, "CountSketch epsilon (0,1)") + csDelta = flag.Float64("countsketch-delta", 0.99, "CountSketch delta (0,1)") + + // CountMinSketch + cmsRows = flag.Int("countmin-rows", 5, "CountMinSketch rows") + cmsCols = flag.Int("countmin-cols", 2000, "CountMinSketch columns") + + // Zipf distribution + zipfS = flag.Float64("zipf-s", 1.1, "Zipf s parameter (must be > 1)") + zipfV = flag.Float64("zipf-v", 1.0, "Zipf v parameter (must be >= 1)") + zipfMax = flag.Uint64("zipf-max", 500, "Zipf imax") + zipfMean = flag.Float64("zipf-mean", 250.0, "Target mean for scaling Zipf values") +) + +// sketchMode describes how measurements are delivered to the collector: +// +// - sdkSketch: Float64Histogram with a sketch aggregation view. The SDK +// pre-aggregates all recorded values into a sketch before export. The +// collector receives a single compressed data point per series per window. +// Only ddsketch is supported here because the ddsketchprocessor explicitly +// handles incoming DDSketch pdata; the other processors accept only Gauge. +// +// - sdkGauge: Float64Gauge with default (LastValue) aggregation. The SDK +// emits the last recorded value per series on each export tick. The +// collector's sketch processor (KLL, CountSketch, CountMinSketch) then +// performs the aggregation from these raw gauge data points. +type sketchMode int + +const ( + sdkSketch sketchMode = iota + sdkGauge +) + +func main() { + flag.Parse() + + // --- Config file path: when --config is set, load YAML and override flags --- + var pipelineCfg *sdkmetric.PipelineConfig + if *configPath != "" { + var err error + pipelineCfg, err = sdkmetric.LoadPipelineConfig(*configPath) + if err != nil { + log.Fatalf("failed to load config: %v", err) + } + // Propagate config values into flag variables so the rest of main is + // driven by a single set of variables regardless of the source. + *sketchArg = pipelineCfg.Sketch.Type + *endpoint = pipelineCfg.Exporter.Endpoint + *interval = pipelineCfg.ReaderInterval() + if pipelineCfg.Load.Workers > 0 { + *workers = pipelineCfg.Load.Workers + } + if pipelineCfg.Load.Duration > 0 { + *duration = pipelineCfg.Load.Duration + } + if pipelineCfg.Load.Distribution.S > 0 { + *zipfS = pipelineCfg.Load.Distribution.S + } + if pipelineCfg.Load.Distribution.V > 0 { + *zipfV = pipelineCfg.Load.Distribution.V + } + if pipelineCfg.Load.Distribution.Max > 0 { + *zipfMax = pipelineCfg.Load.Distribution.Max + } + if pipelineCfg.Load.Distribution.Mean > 0 { + *zipfMean = pipelineCfg.Load.Distribution.Mean + } + // Sketch-specific params + if pipelineCfg.Sketch.DDSketch.RelativeAccuracy > 0 { + *ddsketchAccuracy = pipelineCfg.Sketch.DDSketch.RelativeAccuracy + } + if pipelineCfg.Sketch.KLL.K > 0 { + *kllK = pipelineCfg.Sketch.KLL.K + } + if pipelineCfg.Sketch.CountSketch.Epsilon > 0 { + *csEpsilon = pipelineCfg.Sketch.CountSketch.Epsilon + } + if pipelineCfg.Sketch.CountSketch.Delta > 0 { + *csDelta = pipelineCfg.Sketch.CountSketch.Delta + } + if pipelineCfg.Sketch.CountMinSketch.Rows > 0 { + *cmsRows = pipelineCfg.Sketch.CountMinSketch.Rows + } + if pipelineCfg.Sketch.CountMinSketch.Cols > 0 { + *cmsCols = pipelineCfg.Sketch.CountMinSketch.Cols + } + } + + mode := strings.ToLower(*sketchArg) + switch mode { + case "ddsketch", "kll", "countsketch", "countminsketch", "hll", "baseline": + default: + log.Fatalf("invalid sketch-type %q — valid: ddsketch|kll|countsketch|countminsketch|hll|baseline", mode) + } + if *zipfS <= 1.0 { + log.Fatalf("zipf-s must be > 1.0, got %.2f", *zipfS) + } + if *zipfV < 1.0 { + log.Fatalf("zipf-v must be >= 1.0, got %.2f", *zipfV) + } + + totalSeries := *workers * *hosts * *metrics + throughput := float64(totalSeries) / interval.Seconds() + + fmt.Printf("--- SDK Sketch Load Generator ---\n") + fmt.Printf("Sketch type: %s\n", strings.ToUpper(mode)) + fmt.Printf("Target: %s\n", *endpoint) + fmt.Printf("Series: %d workers × %d hosts × %d metrics = %d total\n", + *workers, *hosts, *metrics, totalSeries) + fmt.Printf("Interval: %s → %.0f data-points/s to collector\n", *interval, throughput) + fmt.Printf("Samples/win: %d (values recorded per series per export window)\n", *samplesPerInterval) + if *duration > 0 { + fmt.Printf("Duration: %v\n", *duration) + } else { + fmt.Printf("Duration: Until Ctrl+C\n") + } + + var ctx context.Context + var cancel context.CancelFunc + if *duration > 0 { + ctx, cancel = context.WithTimeout(context.Background(), *duration) + } else { + ctx, cancel = context.WithCancel(context.Background()) + } + defer cancel() + + // Choose aggregation. For ddsketch the SDK pre-aggregates; for the rest we + // fall back to gauge so the collector-side processor can do the aggregation. + var delivery sketchMode + var agg sdkmetric.Aggregation + switch mode { + case "ddsketch": + delivery = sdkSketch + agg = sdkmetric.AggregationDDSketch{RelativeAccuracy: *ddsketchAccuracy} + case "kll": + // SDK pre-aggregates values into a KLL sketch before export. + delivery = sdkSketch + agg = sdkmetric.AggregationKLLSketch{K: *kllK} + case "countsketch": + // SDK pre-aggregates values into a CountSketch before export. + delivery = sdkSketch + agg = sdkmetric.AggregationCountSketch{Epsilon: *csEpsilon, Delta: *csDelta} + case "countminsketch": + // SDK pre-aggregates values into a CountMinSketch before export. + delivery = sdkSketch + agg = sdkmetric.AggregationCountMinSketch{Rows: *cmsRows, Cols: *cmsCols} + case "hll": + // HLL: SDK pre-aggregates values into an HLL sketch before export, analogous + // to DDSketch. The collector receives HLLSketch-typed data points. + delivery = sdkSketch + agg = sdkmetric.AggregationHLLSketch{} + case "baseline": + // Baseline: no sketch aggregation. SDK emits raw Float64Gauge (LastValue) + // observations; the collector-side processor performs any aggregation. + delivery = sdkGauge + } + + exp, err := otlpmetricgrpc.New(ctx, + otlpmetricgrpc.WithEndpoint(*endpoint), + otlpmetricgrpc.WithInsecure(), + ) + if err != nil { + log.Fatalf("failed to create OTLP exporter: %v", err) + } + + // Build provider options. When a config file was loaded use its views + // (which carry transmit_sketch semantics); otherwise build a single view + // from the resolved flag values. + providerOpts := []sdkmetric.Option{ + sdkmetric.WithReader( + sdkmetric.NewPeriodicReader(exp, sdkmetric.WithInterval(*interval)), + ), + } + if pipelineCfg != nil { + for _, v := range pipelineCfg.ToViews() { + providerOpts = append(providerOpts, sdkmetric.WithView(v)) + } + } else { + providerOpts = append(providerOpts, sdkmetric.WithView( + sdkmetric.NewView( + sdkmetric.Instrument{Name: "benchmark.latency"}, + sdkmetric.Stream{Aggregation: agg}, + ), + )) + } + + provider := sdkmetric.NewMeterProvider(providerOpts...) + defer func() { + shutCtx, shutCancel := context.WithTimeout(context.Background(), 5*time.Second) + defer shutCancel() + if err := provider.Shutdown(shutCtx); err != nil { + log.Printf("MeterProvider shutdown: %v", err) + } + }() + + meter := provider.Meter("fakemetricload") + + var wg sync.WaitGroup + switch delivery { + case sdkSketch: + // Pre-aggregated path: histogram with sketch view. + hist, err := meter.Float64Histogram("benchmark.latency", + metric.WithDescription("Benchmark latency measurement"), + metric.WithUnit("ms"), + ) + if err != nil { + log.Fatalf("failed to create histogram: %v", err) + } + for w := 0; w < *workers; w++ { + wg.Add(1) + go runHistogramWorker(ctx, w, hist, mode, &wg) + } + + case sdkGauge: + // Raw-value path: gauge so the collector processor aggregates. + gauge, err := meter.Float64Gauge("benchmark.latency", + metric.WithDescription("Benchmark latency measurement"), + metric.WithUnit("ms"), + ) + if err != nil { + log.Fatalf("failed to create gauge: %v", err) + } + for w := 0; w < *workers; w++ { + wg.Add(1) + go runGaugeWorker(ctx, w, gauge, mode, &wg) + } + } + + <-ctx.Done() + fmt.Println("\nLoad generation complete. Flushing final export...") + wg.Wait() +} + +// runHistogramWorker records samplesPerInterval Zipf values for each series on +// every export tick. For the ddsketch path the SDK accumulates them all into a +// single sketch per series before the periodic export fires. +func runHistogramWorker(ctx context.Context, id int, hist metric.Float64Histogram, mode string, wg *sync.WaitGroup) { + defer wg.Done() + + src := rand.NewSource(time.Now().UnixNano() + int64(id)) + rng := rand.New(src) + zipf := rand.NewZipf(rng, *zipfS, *zipfV, *zipfMax) + + ticker := time.NewTicker(*interval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + for h := 0; h < *hosts; h++ { + hostAttr := attribute.String("host.name", fmt.Sprintf("worker-%d-host-%02d", id, h)) + for m := 0; m < *metrics; m++ { + metricAttr := attribute.String("metric.name", fmt.Sprintf("metric.%03d", m)) + attrs := metric.WithAttributes(hostAttr, metricAttr) + for s := 0; s < *samplesPerInterval; s++ { + hist.Record(ctx, generateZipfValue(zipf), attrs) + } + } + } + fmt.Printf("Worker %d: recorded batch (%s, %d samples/series)\n", + id, mode, *samplesPerInterval) + } + } +} + +// runGaugeWorker records one Zipf value per series per export tick. The SDK +// keeps the last observation and exports it as a Gauge data point; the +// collector-side KLL/CountSketch/CountMinSketch processor aggregates the stream. +func runGaugeWorker(ctx context.Context, id int, gauge metric.Float64Gauge, mode string, wg *sync.WaitGroup) { + defer wg.Done() + + src := rand.NewSource(time.Now().UnixNano() + int64(id)) + rng := rand.New(src) + zipf := rand.NewZipf(rng, *zipfS, *zipfV, *zipfMax) + + ticker := time.NewTicker(*interval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + for h := 0; h < *hosts; h++ { + hostAttr := attribute.String("host.name", fmt.Sprintf("worker-%d-host-%02d", id, h)) + for m := 0; m < *metrics; m++ { + metricAttr := attribute.String("metric.name", fmt.Sprintf("metric.%03d", m)) + gauge.Record(ctx, generateZipfValue(zipf), + metric.WithAttributes(hostAttr, metricAttr), + ) + } + } + fmt.Printf("Worker %d: recorded batch (%s)\n", id, mode) + } + } +} + +func generateZipfValue(zipf *rand.Zipf) float64 { + scaleFactor := *zipfMean / (float64(*zipfMax) / 2.0) + return float64(zipf.Uint64()+1) * scaleFactor +} diff --git a/opentelemetry-app/config-baseline.yaml b/opentelemetry-app/config-baseline.yaml new file mode 100644 index 00000000..5a91d89b --- /dev/null +++ b/opentelemetry-app/config-baseline.yaml @@ -0,0 +1,60 @@ +# SDK pipeline configuration — Baseline (raw gauge, no sketch aggregation) +# +# transmit_sketch: false — SDK emits raw Float64Gauge (LastValue) observations. +# The collector-side sketch processor (kllprocessor, hllprocessor, etc.) is +# responsible for aggregation. Use this when you want the collector to own +# the sketch type and parameters rather than the SDK. +# +# This is also the reference "uncompressed" baseline for bandwidth and CPU +# benchmarking against sketch modes. + +exporter: + endpoint: localhost:4317 + insecure: true + max_send_msg_size_mb: 64 + +reader: + # For baseline, match the worker record interval so every sample is exported + # individually (no SDK-side accumulation between export ticks). + interval: 20ms # ~50 samples/s/series at default load settings + +sketch: + type: baseline + transmit_sketch: false + +instruments: + - name: benchmark.latency + description: Benchmark latency measurement + unit: ms + instrument_kind: gauge # raw LastValue — no aggregation view applied + group_by: + - host.name + - service.name + series_per_sketch: 1 # not meaningful for baseline; kept for schema consistency + +asap_query: + metrics: + - metric: benchmark.latency + labels: + - host.name + - service.name + query_groups: + - id: 1 + queries: + - "quantile_over_time(0.99, benchmark_latency[5m])" + repetition_delay: 5000 + controller_options: + accuracy_sla: 0.01 + latency_sla: 100.0 + +load: + workers: 10 + series: 1000 + samples_per_sec_per_series: 50.0 + duration: 60s + distribution: + type: zipf + s: 1.1 + v: 1.0 + max: 500 + mean: 250.0 diff --git a/opentelemetry-app/config-countminsketch.yaml b/opentelemetry-app/config-countminsketch.yaml new file mode 100644 index 00000000..4ecfa357 --- /dev/null +++ b/opentelemetry-app/config-countminsketch.yaml @@ -0,0 +1,62 @@ +# SDK pipeline configuration — CountMinSketch, transmit_sketch: true +# +# SDK pre-aggregates into a CountMinSketch (frequency estimation). +# The collector's countminsketchprocessor handles the sketch payload. +# Note: large payloads at high series counts may require max_send_msg_size_mb +# above the default 4 MiB gRPC limit. + +exporter: + endpoint: localhost:4317 + insecure: true + max_send_msg_size_mb: 64 + +reader: + interval: 1s + +sketch: + type: countminsketch + transmit_sketch: true + countminsketch: + rows: 5 + cols: 2000 + +instruments: + - name: benchmark.latency + description: Benchmark latency measurement + unit: ms + instrument_kind: histogram + group_by: + - host.name + - service.name + series_per_sketch: 1 + +asap_query: + metrics: + - metric: benchmark.latency + labels: + - host.name + - service.name + query_groups: + - id: 1 + queries: + - "frequency_over_time(benchmark_latency[5m])" + repetition_delay: 5000 + controller_options: + accuracy_sla: 0.01 + latency_sla: 100.0 + sketch_parameters: + CountMinSketch: + depth: 5 + width: 2000 + +load: + workers: 10 + series: 1000 + samples_per_sec_per_series: 50.0 + duration: 60s + distribution: + type: zipf + s: 1.1 + v: 1.0 + max: 500 + mean: 250.0 diff --git a/opentelemetry-app/config-countsketch.yaml b/opentelemetry-app/config-countsketch.yaml new file mode 100644 index 00000000..4c81bc5b --- /dev/null +++ b/opentelemetry-app/config-countsketch.yaml @@ -0,0 +1,59 @@ +# SDK pipeline configuration — CountSketch, transmit_sketch: true +# +# SDK pre-aggregates into a CountSketch (signed frequency estimation with +# better accuracy than CountMinSketch for certain workloads). + +exporter: + endpoint: localhost:4317 + insecure: true + max_send_msg_size_mb: 64 + +reader: + interval: 1s + +sketch: + type: countsketch + transmit_sketch: true + countsketch: + rows: 5 + cols: 2000 + epsilon: 0.01 + delta: 0.99 + dimension: "" # optional: label describing the sketched dimension + +instruments: + - name: benchmark.latency + description: Benchmark latency measurement + unit: ms + instrument_kind: histogram + group_by: + - host.name + - service.name + series_per_sketch: 1 + +asap_query: + metrics: + - metric: benchmark.latency + labels: + - host.name + - service.name + query_groups: + - id: 1 + queries: + - "frequency_over_time(benchmark_latency[5m])" + repetition_delay: 5000 + controller_options: + accuracy_sla: 0.01 + latency_sla: 100.0 + +load: + workers: 10 + series: 1000 + samples_per_sec_per_series: 50.0 + duration: 60s + distribution: + type: zipf + s: 1.1 + v: 1.0 + max: 500 + mean: 250.0 diff --git a/opentelemetry-app/config-ddsketch.yaml b/opentelemetry-app/config-ddsketch.yaml new file mode 100644 index 00000000..b9a9c595 --- /dev/null +++ b/opentelemetry-app/config-ddsketch.yaml @@ -0,0 +1,58 @@ +# SDK pipeline configuration — DDSketch, transmit_sketch: true +# +# SDK pre-aggregates values into a DDSketch before each export tick. +# The collector receives DDSketch-typed data points and can either +# re-emit quantile gauges (transmit_sketch: false on collector side) or +# forward the raw sketch payload downstream (transmit_sketch: true). + +exporter: + endpoint: localhost:4317 + insecure: true + max_send_msg_size_mb: 64 + +reader: + interval: 1s + +sketch: + type: ddsketch + transmit_sketch: true + ddsketch: + relative_accuracy: 0.01 + no_min_max: false + +instruments: + - name: benchmark.latency + description: Benchmark latency measurement + unit: ms + instrument_kind: histogram + group_by: + - host.name + - service.name + series_per_sketch: 1 + +asap_query: + metrics: + - metric: benchmark.latency + labels: + - host.name + - service.name + query_groups: + - id: 1 + queries: + - "quantile_over_time(0.99, benchmark_latency[5m])" + repetition_delay: 5000 + controller_options: + accuracy_sla: 0.01 + latency_sla: 100.0 + +load: + workers: 10 + series: 1000 + samples_per_sec_per_series: 50.0 + duration: 60s + distribution: + type: zipf + s: 1.1 + v: 1.0 + max: 500 + mean: 250.0 diff --git a/opentelemetry-app/config-hll.yaml b/opentelemetry-app/config-hll.yaml new file mode 100644 index 00000000..4ad27818 --- /dev/null +++ b/opentelemetry-app/config-hll.yaml @@ -0,0 +1,54 @@ +# SDK pipeline configuration — HyperLogLog (HLL) sketch, transmit_sketch: true +# +# HLL estimates cardinality. SDK pre-aggregates into HLL sketch registers; +# the collector's hllprocessor decodes them into cardinality gauges. + +exporter: + endpoint: localhost:4317 + insecure: true + max_send_msg_size_mb: 64 + +reader: + interval: 1s + +sketch: + type: hll + transmit_sketch: true + hll: {} # HLL has no tunable parameters + +instruments: + - name: benchmark.latency + description: Benchmark unique-value cardinality + unit: "1" + instrument_kind: histogram + group_by: + - host.name + - service.name + series_per_sketch: 1 + +asap_query: + metrics: + - metric: benchmark.latency + labels: + - host.name + - service.name + query_groups: + - id: 1 + queries: + - "cardinality_over_time(benchmark_latency[5m])" + repetition_delay: 5000 + controller_options: + accuracy_sla: 0.02 + latency_sla: 100.0 + +load: + workers: 10 + series: 1000 + samples_per_sec_per_series: 50.0 + duration: 60s + distribution: + type: zipf + s: 1.1 + v: 1.0 + max: 500 + mean: 250.0 diff --git a/opentelemetry-app/config-kll.yaml b/opentelemetry-app/config-kll.yaml new file mode 100644 index 00000000..fe50aeb5 --- /dev/null +++ b/opentelemetry-app/config-kll.yaml @@ -0,0 +1,58 @@ +# SDK pipeline configuration — KLL sketch, transmit_sketch: true +# +# SDK pre-aggregates values into a KLL sketch. The collector's kllprocessor +# can receive these and either re-emit quantile gauges or forward the sketch. + +exporter: + endpoint: localhost:4317 + insecure: true + max_send_msg_size_mb: 64 + +reader: + interval: 1s + +sketch: + type: kll + transmit_sketch: true + kll: + k: 256 + +instruments: + - name: benchmark.latency + description: Benchmark latency measurement + unit: ms + instrument_kind: histogram + group_by: + - host.name + - service.name + series_per_sketch: 1 + +asap_query: + metrics: + - metric: benchmark.latency + labels: + - host.name + - service.name + query_groups: + - id: 1 + queries: + - "quantile_over_time(0.99, benchmark_latency[5m])" + repetition_delay: 5000 + controller_options: + accuracy_sla: 0.01 + latency_sla: 100.0 + sketch_parameters: + DatasketchesKLL: + K: 256 + +load: + workers: 10 + series: 1000 + samples_per_sec_per_series: 50.0 + duration: 60s + distribution: + type: zipf + s: 1.1 + v: 1.0 + max: 500 + mean: 250.0 diff --git a/opentelemetry-app/config.yaml b/opentelemetry-app/config.yaml new file mode 100644 index 00000000..3d1e2ec0 --- /dev/null +++ b/opentelemetry-app/config.yaml @@ -0,0 +1,137 @@ +# SDK pipeline configuration — base template (DDSketch, transmit_sketch: true) +# +# transmit_sketch: true → SDK pre-aggregates into a sketch payload; the +# collector receives a single compressed data point +# per series per export window (sdkSketch mode). +# transmit_sketch: false → SDK emits raw Float64Gauge values; the collector- +# side processor (kllprocessor, hllprocessor, …) +# performs aggregation (sdkGauge mode). + +# --------------------------------------------------------------------------- +# Exporter +# --------------------------------------------------------------------------- +exporter: + endpoint: localhost:4317 + insecure: true + # Raise gRPC send limit for large sketch payloads (CountSketch, HLL at high + # series counts can exceed the default 4 MiB limit). + max_send_msg_size_mb: 64 + +# --------------------------------------------------------------------------- +# Reader (SDK export interval) +# --------------------------------------------------------------------------- +reader: + interval: 1s + +# --------------------------------------------------------------------------- +# Sketch +# --------------------------------------------------------------------------- +sketch: + # Sketch type: ddsketch | kll | hll | countsketch | countminsketch | baseline + # "baseline" disables sketch aggregation — raw LastValue gauges are sent. + type: ddsketch + + # Whether the SDK performs the aggregation before export. + # true → SDK attaches an Aggregation View to the instrument; the collector + # receives sketch-typed data points (DDSketchDataPoint, etc.). + # false → SDK records raw gauge observations; the collector processor + # (e.g. kllprocessor, hllprocessor) aggregates the stream. + transmit_sketch: true + + # --- DDSketch parameters (used when type: ddsketch) --- + ddsketch: + relative_accuracy: 0.01 # target relative error; 0 uses SDK default + no_min_max: false + + # --- KLL parameters (used when type: kll) --- + kll: + k: 256 # compaction parameter; 0 uses SDK default (>= 2) + + # --- HLL parameters (used when type: hll) --- + hll: {} # HyperLogLog has no tunable parameters + + # --- CountSketch parameters (used when type: countsketch) --- + countsketch: + rows: 5 # number of hash functions + cols: 2000 # buckets per row + epsilon: 0.01 # accuracy parameter (0,1) + delta: 0.99 # failure probability (0,1) + dimension: "" # free-form label describing the sketched dimension + + # --- CountMinSketch parameters (used when type: countminsketch) --- + countminsketch: + rows: 5 + cols: 2000 + +# --------------------------------------------------------------------------- +# Instruments — what to aggregate and along which label dimensions +# --------------------------------------------------------------------------- +instruments: + - name: benchmark.latency + description: Benchmark latency measurement + unit: ms + + # OTel instrument kind used when transmit_sketch: true. + # "histogram" → Float64Histogram with a sketch Aggregation View. + # "gauge" → Float64Gauge (also used when transmit_sketch: false). + instrument_kind: histogram + + # Attribute keys that form the grouping dimension. + # One sketch (or gauge) is produced per unique combination of these values. + group_by: + - host.name + - service.name + + # How many distinct label-value combinations are collapsed into one sketch. + # 1 → one sketch per series (finest granularity, existing per-series behavior) + # 0 → all series collapsed into a single sketch (coarsest) + # N → every N consecutive series share one sketch (e.g. for matrix aggregation) + series_per_sketch: 1 + +# --------------------------------------------------------------------------- +# ASAPQuery controller integration +# Mirrors the ControllerConfig schema in asap-planner-rs/src/config/input.rs +# so the same file (or a shared section) can be fed to asap-planner-rs. +# --------------------------------------------------------------------------- +asap_query: + metrics: + - metric: benchmark.latency + labels: + - host.name + - service.name + + query_groups: + - id: 1 + queries: + - "quantile_over_time(0.99, benchmark_latency[5m])" + repetition_delay: 5000 # milliseconds + controller_options: + accuracy_sla: 0.01 + latency_sla: 100.0 + + # Optional per-sketch-type parameter overrides consumed by asap-planner-rs. + # These shadow the top-level sketch.* params when asap-planner-rs is driving + # sketch selection. + sketch_parameters: + DatasketchesKLL: + K: 256 + CountMinSketch: + depth: 5 + width: 2000 + +# --------------------------------------------------------------------------- +# Load generation (fakemetricload / e2esdkbench) +# --------------------------------------------------------------------------- +load: + workers: 10 + series: 1000 + samples_per_sec_per_series: 50.0 + duration: 60s + + # Synthetic data distribution + distribution: + type: zipf + s: 1.1 # shape parameter (must be > 1) + v: 1.0 # shift parameter (must be >= 1) + max: 500 # upper bound of integer range + mean: 250.0 # target mean used for value scaling diff --git a/opentelemetry-go-patch/sdk/metric/go.mod b/opentelemetry-go-patch/sdk/metric/go.mod index 2b0f8ed0..dfcf4385 100644 --- a/opentelemetry-go-patch/sdk/metric/go.mod +++ b/opentelemetry-go-patch/sdk/metric/go.mod @@ -13,6 +13,7 @@ require ( go.opentelemetry.io/otel/metric v1.41.0 go.opentelemetry.io/otel/sdk v1.41.0 go.opentelemetry.io/otel/trace v1.41.0 + go.yaml.in/yaml/v2 v2.4.3 google.golang.org/protobuf v1.36.10 ) @@ -28,7 +29,6 @@ require ( github.com/prometheus/prometheus v0.307.1 // indirect github.com/zeebo/xxh3 v1.0.2 // indirect go.opentelemetry.io/auto/sdk v1.2.1 // indirect - go.yaml.in/yaml/v2 v2.4.3 // indirect golang.org/x/sys v0.41.0 // indirect golang.org/x/text v0.29.0 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect diff --git a/opentelemetry-go-patch/sdk/metric/pipelineconfig.go b/opentelemetry-go-patch/sdk/metric/pipelineconfig.go new file mode 100644 index 00000000..5788674b --- /dev/null +++ b/opentelemetry-go-patch/sdk/metric/pipelineconfig.go @@ -0,0 +1,230 @@ +// Copyright The OpenTelemetry Authors +// SPDX-License-Identifier: Apache-2.0 + +package metric // import "go.opentelemetry.io/otel/sdk/metric" + +import ( + "fmt" + "os" + "time" + + "go.yaml.in/yaml/v2" +) + +// PipelineConfig is the top-level SDK pipeline configuration loaded from a +// YAML file. It covers the full SDK→collector path: exporter destination, +// reader interval, sketch type and transmit mode, instrument definitions, an +// ASAPQuery controller block, and load-generation parameters. +type PipelineConfig struct { + Exporter PipelineExporterConfig `yaml:"exporter"` + Reader PipelineReaderConfig `yaml:"reader"` + Sketch PipelineSketchConfig `yaml:"sketch"` + Instruments []PipelineInstrument `yaml:"instruments"` + ASAPQuery *PipelineASAPQuery `yaml:"asap_query,omitempty"` + Load PipelineLoadConfig `yaml:"load"` +} + +// PipelineExporterConfig holds the OTLP gRPC exporter settings. +type PipelineExporterConfig struct { + Endpoint string `yaml:"endpoint"` + Insecure bool `yaml:"insecure"` + MaxSendMsgSizeMB int `yaml:"max_send_msg_size_mb"` +} + +// PipelineReaderConfig controls the SDK periodic export interval. +type PipelineReaderConfig struct { + Interval time.Duration `yaml:"interval"` +} + +// PipelineSketchConfig selects the sketch type and whether the SDK or the +// collector performs aggregation (transmit_sketch). +type PipelineSketchConfig struct { + // Type is one of: ddsketch | kll | hll | countsketch | countminsketch | baseline. + // "baseline" disables sketch aggregation — raw LastValue gauges are sent. + Type string `yaml:"type"` + + // TransmitSketch controls which side of the pipeline aggregates. + // true → SDK attaches a sketch Aggregation View; the collector receives + // a single compact sketch data point per series per window. + // false → SDK emits raw Float64Gauge observations; the collector-side + // processor (kllprocessor, hllprocessor, …) aggregates. + TransmitSketch bool `yaml:"transmit_sketch"` + + DDSketch PipelineDDSketchParams `yaml:"ddsketch"` + KLL PipelineKLLParams `yaml:"kll"` + CountSketch PipelineCountSketchParams `yaml:"countsketch"` + CountMinSketch PipelineCountMinSketchParams `yaml:"countminsketch"` + // HLL has no tunable parameters; its presence in the YAML is informational. +} + +// PipelineDDSketchParams are the tuning knobs for DDSketch aggregation. +type PipelineDDSketchParams struct { + RelativeAccuracy float64 `yaml:"relative_accuracy"` + NoMinMax bool `yaml:"no_min_max"` +} + +// PipelineKLLParams are the tuning knobs for KLL sketch aggregation. +type PipelineKLLParams struct { + K int `yaml:"k"` +} + +// PipelineCountSketchParams are the tuning knobs for CountSketch aggregation. +type PipelineCountSketchParams struct { + Rows int `yaml:"rows"` + Cols int `yaml:"cols"` + Epsilon float64 `yaml:"epsilon"` + Delta float64 `yaml:"delta"` + Dimension string `yaml:"dimension"` +} + +// PipelineCountMinSketchParams are the tuning knobs for CountMinSketch aggregation. +type PipelineCountMinSketchParams struct { + Rows int `yaml:"rows"` + Cols int `yaml:"cols"` +} + +// PipelineInstrument describes a single OTel instrument to be aggregated. +type PipelineInstrument struct { + Name string `yaml:"name"` + Description string `yaml:"description"` + Unit string `yaml:"unit"` + + // InstrumentKind is "histogram" or "gauge". + // "histogram" is used for transmit_sketch: true (sketch Aggregation View). + // "gauge" is used for transmit_sketch: false (raw LastValue observations). + InstrumentKind string `yaml:"instrument_kind"` + + // GroupBy lists the attribute keys that form the sketch grouping dimension. + // One sketch (or gauge) is produced per unique combination of these values. + GroupBy []string `yaml:"group_by"` + + // SeriesPerSketch controls how many distinct label-value combinations are + // collapsed into a single sketch: + // 1 → one sketch per series (fine-grained, default) + // 0 → all series collapsed into one sketch (coarsest) + // N → every N series share one sketch (matrix grouping) + SeriesPerSketch int `yaml:"series_per_sketch"` +} + +// PipelineASAPQuery mirrors the ControllerConfig schema from asap-planner-rs +// so that the same YAML block can be consumed by asap-planner-rs directly. +type PipelineASAPQuery struct { + Metrics []PipelineASAPMetric `yaml:"metrics"` + QueryGroups []PipelineASAPQueryGroup `yaml:"query_groups"` + SketchParameters map[string]any `yaml:"sketch_parameters,omitempty"` +} + +// PipelineASAPMetric names a metric and its label dimensions for the planner. +type PipelineASAPMetric struct { + Metric string `yaml:"metric"` + Labels []string `yaml:"labels"` +} + +// PipelineASAPQueryGroup is a group of related queries sharing SLA targets. +type PipelineASAPQueryGroup struct { + ID *uint32 `yaml:"id,omitempty"` + Queries []string `yaml:"queries"` + RepetitionDelay uint64 `yaml:"repetition_delay"` + ControllerOptions PipelineASAPControllerOpts `yaml:"controller_options"` +} + +// PipelineASAPControllerOpts carries the accuracy and latency SLA targets. +type PipelineASAPControllerOpts struct { + AccuracySLA float64 `yaml:"accuracy_sla"` + LatencySLA float64 `yaml:"latency_sla"` +} + +// PipelineLoadConfig holds load-generation parameters for fakemetricload and +// e2esdkbench. +type PipelineLoadConfig struct { + Workers int `yaml:"workers"` + Series int `yaml:"series"` + SamplesPerSecPerSeries float64 `yaml:"samples_per_sec_per_series"` + Duration time.Duration `yaml:"duration"` + Distribution PipelineDistributionConfig `yaml:"distribution"` +} + +// PipelineDistributionConfig parameterises the synthetic data distribution. +type PipelineDistributionConfig struct { + Type string `yaml:"type"` + S float64 `yaml:"s"` + V float64 `yaml:"v"` + Max uint64 `yaml:"max"` + Mean float64 `yaml:"mean"` +} + +// LoadPipelineConfig reads and parses a YAML pipeline config file. +func LoadPipelineConfig(path string) (*PipelineConfig, error) { + data, err := os.ReadFile(path) + if err != nil { + return nil, fmt.Errorf("reading pipeline config %s: %w", path, err) + } + var cfg PipelineConfig + if err := yaml.Unmarshal(data, &cfg); err != nil { + return nil, fmt.Errorf("parsing pipeline config %s: %w", path, err) + } + return &cfg, nil +} + +// ToAggregation returns the SDK Aggregation corresponding to sketch.type and +// the sketch-specific parameters. Returns nil for "baseline" (no aggregation +// override — the instrument's default LastValue aggregation is used). +func (c *PipelineConfig) ToAggregation() Aggregation { + s := c.Sketch + switch s.Type { + case "ddsketch": + return AggregationDDSketch{ + RelativeAccuracy: s.DDSketch.RelativeAccuracy, + NoMinMax: s.DDSketch.NoMinMax, + } + case "kll": + return AggregationKLLSketch{K: s.KLL.K} + case "hll": + return AggregationHLLSketch{} + case "countsketch": + return AggregationCountSketch{ + Rows: s.CountSketch.Rows, + Cols: s.CountSketch.Cols, + Epsilon: s.CountSketch.Epsilon, + Delta: s.CountSketch.Delta, + Dimension: s.CountSketch.Dimension, + } + case "countminsketch": + return AggregationCountMinSketch{ + Rows: s.CountMinSketch.Rows, + Cols: s.CountMinSketch.Cols, + } + default: // "baseline" or unrecognised + return nil + } +} + +// ToViews returns one View per configured instrument that attaches the sketch +// aggregation. Returns nil when transmit_sketch is false or the type is +// "baseline" (no aggregation override needed). +func (c *PipelineConfig) ToViews() []View { + if !c.Sketch.TransmitSketch { + return nil + } + agg := c.ToAggregation() + if agg == nil { + return nil + } + views := make([]View, 0, len(c.Instruments)) + for _, inst := range c.Instruments { + views = append(views, NewView( + Instrument{Name: inst.Name}, + Stream{Aggregation: agg}, + )) + } + return views +} + +// ReaderInterval returns the configured reader export interval, defaulting to +// 1 second when unset. +func (c *PipelineConfig) ReaderInterval() time.Duration { + if c.Reader.Interval <= 0 { + return time.Second + } + return c.Reader.Interval +} diff --git a/opentelemetry-go-patch/sdk/metric/pipelineconfig_test.go b/opentelemetry-go-patch/sdk/metric/pipelineconfig_test.go new file mode 100644 index 00000000..0f4b08a4 --- /dev/null +++ b/opentelemetry-go-patch/sdk/metric/pipelineconfig_test.go @@ -0,0 +1,318 @@ +// Copyright The OpenTelemetry Authors +// SPDX-License-Identifier: Apache-2.0 + +package metric + +import ( + "os" + "path/filepath" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// writeTemp writes content to a temporary YAML file and returns the path. +func writeTemp(t *testing.T, content string) string { + t.Helper() + f, err := os.CreateTemp(t.TempDir(), "pipelinecfg-*.yaml") + require.NoError(t, err) + _, err = f.WriteString(content) + require.NoError(t, err) + require.NoError(t, f.Close()) + return f.Name() +} + +// --------------------------------------------------------------------------- +// LoadPipelineConfig +// --------------------------------------------------------------------------- + +func TestLoadPipelineConfig_DDSketch(t *testing.T) { + yaml := ` +sketch: + type: ddsketch + transmit_sketch: true + ddsketch: + relative_accuracy: 0.02 + no_min_max: true +reader: + interval: 2s +instruments: + - name: my.metric + unit: ms + instrument_kind: histogram + group_by: [host.name] + series_per_sketch: 1 +` + cfg, err := LoadPipelineConfig(writeTemp(t, yaml)) + require.NoError(t, err) + assert.Equal(t, "ddsketch", cfg.Sketch.Type) + assert.True(t, cfg.Sketch.TransmitSketch) + assert.InDelta(t, 0.02, cfg.Sketch.DDSketch.RelativeAccuracy, 1e-9) + assert.True(t, cfg.Sketch.DDSketch.NoMinMax) + assert.Equal(t, 2*time.Second, cfg.Reader.Interval) + require.Len(t, cfg.Instruments, 1) + assert.Equal(t, "my.metric", cfg.Instruments[0].Name) + assert.Equal(t, []string{"host.name"}, cfg.Instruments[0].GroupBy) +} + +func TestLoadPipelineConfig_KLL(t *testing.T) { + yaml := ` +sketch: + type: kll + transmit_sketch: true + kll: + k: 512 +` + cfg, err := LoadPipelineConfig(writeTemp(t, yaml)) + require.NoError(t, err) + assert.Equal(t, "kll", cfg.Sketch.Type) + assert.Equal(t, 512, cfg.Sketch.KLL.K) +} + +func TestLoadPipelineConfig_HLL(t *testing.T) { + yaml := ` +sketch: + type: hll + transmit_sketch: true +` + cfg, err := LoadPipelineConfig(writeTemp(t, yaml)) + require.NoError(t, err) + assert.Equal(t, "hll", cfg.Sketch.Type) + assert.True(t, cfg.Sketch.TransmitSketch) +} + +func TestLoadPipelineConfig_CountSketch(t *testing.T) { + yaml := ` +sketch: + type: countsketch + transmit_sketch: true + countsketch: + rows: 7 + cols: 1024 + epsilon: 0.05 + delta: 0.95 + dimension: latency +` + cfg, err := LoadPipelineConfig(writeTemp(t, yaml)) + require.NoError(t, err) + assert.Equal(t, "countsketch", cfg.Sketch.Type) + assert.Equal(t, 7, cfg.Sketch.CountSketch.Rows) + assert.Equal(t, 1024, cfg.Sketch.CountSketch.Cols) + assert.InDelta(t, 0.05, cfg.Sketch.CountSketch.Epsilon, 1e-9) + assert.InDelta(t, 0.95, cfg.Sketch.CountSketch.Delta, 1e-9) + assert.Equal(t, "latency", cfg.Sketch.CountSketch.Dimension) +} + +func TestLoadPipelineConfig_CountMinSketch(t *testing.T) { + yaml := ` +sketch: + type: countminsketch + transmit_sketch: true + countminsketch: + rows: 5 + cols: 2000 +` + cfg, err := LoadPipelineConfig(writeTemp(t, yaml)) + require.NoError(t, err) + assert.Equal(t, "countminsketch", cfg.Sketch.Type) + assert.Equal(t, 5, cfg.Sketch.CountMinSketch.Rows) + assert.Equal(t, 2000, cfg.Sketch.CountMinSketch.Cols) +} + +func TestLoadPipelineConfig_Baseline(t *testing.T) { + yaml := ` +sketch: + type: baseline + transmit_sketch: false +` + cfg, err := LoadPipelineConfig(writeTemp(t, yaml)) + require.NoError(t, err) + assert.Equal(t, "baseline", cfg.Sketch.Type) + assert.False(t, cfg.Sketch.TransmitSketch) +} + +func TestLoadPipelineConfig_ASAPQuery(t *testing.T) { + yaml := ` +sketch: + type: kll + transmit_sketch: true +asap_query: + metrics: + - metric: req.latency + labels: [host, region] + query_groups: + - id: 1 + queries: ["quantile_over_time(0.99, req_latency[5m])"] + repetition_delay: 3000 + controller_options: + accuracy_sla: 0.01 + latency_sla: 50.0 +` + cfg, err := LoadPipelineConfig(writeTemp(t, yaml)) + require.NoError(t, err) + require.NotNil(t, cfg.ASAPQuery) + require.Len(t, cfg.ASAPQuery.Metrics, 1) + assert.Equal(t, "req.latency", cfg.ASAPQuery.Metrics[0].Metric) + assert.Equal(t, []string{"host", "region"}, cfg.ASAPQuery.Metrics[0].Labels) + require.Len(t, cfg.ASAPQuery.QueryGroups, 1) + assert.InDelta(t, 0.01, cfg.ASAPQuery.QueryGroups[0].ControllerOptions.AccuracySLA, 1e-9) + assert.InDelta(t, 50.0, cfg.ASAPQuery.QueryGroups[0].ControllerOptions.LatencySLA, 1e-9) +} + +func TestLoadPipelineConfig_FileNotFound(t *testing.T) { + _, err := LoadPipelineConfig(filepath.Join(t.TempDir(), "nonexistent.yaml")) + assert.Error(t, err) +} + +func TestLoadPipelineConfig_InvalidYAML(t *testing.T) { + _, err := LoadPipelineConfig(writeTemp(t, ":: invalid: yaml: [")) + assert.Error(t, err) +} + +// --------------------------------------------------------------------------- +// ToAggregation +// --------------------------------------------------------------------------- + +func TestToAggregation(t *testing.T) { + tests := []struct { + name string + yaml string + wantType Aggregation + }{ + { + name: "ddsketch", + yaml: `sketch: {type: ddsketch, ddsketch: {relative_accuracy: 0.02}}`, + wantType: AggregationDDSketch{RelativeAccuracy: 0.02}, + }, + { + name: "kll", + yaml: `sketch: {type: kll, kll: {k: 128}}`, + wantType: AggregationKLLSketch{K: 128}, + }, + { + name: "hll", + yaml: `sketch: {type: hll}`, + wantType: AggregationHLLSketch{}, + }, + { + name: "countsketch", + yaml: `sketch: {type: countsketch, countsketch: {rows: 3, cols: 500, epsilon: 0.1, delta: 0.9}}`, + wantType: AggregationCountSketch{Rows: 3, Cols: 500, Epsilon: 0.1, Delta: 0.9}, + }, + { + name: "countminsketch", + yaml: `sketch: {type: countminsketch, countminsketch: {rows: 4, cols: 800}}`, + wantType: AggregationCountMinSketch{Rows: 4, Cols: 800}, + }, + { + name: "baseline returns nil", + yaml: `sketch: {type: baseline}`, + wantType: nil, + }, + { + name: "unknown type returns nil", + yaml: `sketch: {type: bogus}`, + wantType: nil, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + cfg, err := LoadPipelineConfig(writeTemp(t, tc.yaml)) + require.NoError(t, err) + assert.Equal(t, tc.wantType, cfg.ToAggregation()) + }) + } +} + +// --------------------------------------------------------------------------- +// ToViews +// --------------------------------------------------------------------------- + +func TestToViews_TransmitSketchTrue(t *testing.T) { + yaml := ` +sketch: + type: kll + transmit_sketch: true + kll: + k: 256 +instruments: + - name: metric.a + - name: metric.b +` + cfg, err := LoadPipelineConfig(writeTemp(t, yaml)) + require.NoError(t, err) + + views := cfg.ToViews() + // One View per instrument. + assert.Len(t, views, 2) +} + +func TestToViews_TransmitSketchFalse(t *testing.T) { + yaml := ` +sketch: + type: kll + transmit_sketch: false +instruments: + - name: metric.a +` + cfg, err := LoadPipelineConfig(writeTemp(t, yaml)) + require.NoError(t, err) + + assert.Nil(t, cfg.ToViews()) +} + +func TestToViews_Baseline(t *testing.T) { + yaml := ` +sketch: + type: baseline + transmit_sketch: true # transmit_sketch: true but type: baseline → nil +instruments: + - name: metric.a +` + cfg, err := LoadPipelineConfig(writeTemp(t, yaml)) + require.NoError(t, err) + + // baseline has no aggregation regardless of transmit_sketch. + assert.Nil(t, cfg.ToViews()) +} + +func TestToViews_NoInstruments(t *testing.T) { + yaml := ` +sketch: + type: ddsketch + transmit_sketch: true +` + cfg, err := LoadPipelineConfig(writeTemp(t, yaml)) + require.NoError(t, err) + + assert.Empty(t, cfg.ToViews()) +} + +// --------------------------------------------------------------------------- +// ReaderInterval +// --------------------------------------------------------------------------- + +func TestReaderInterval(t *testing.T) { + t.Run("configured", func(t *testing.T) { + yaml := `reader: {interval: 5s}` + cfg, err := LoadPipelineConfig(writeTemp(t, yaml)) + require.NoError(t, err) + assert.Equal(t, 5*time.Second, cfg.ReaderInterval()) + }) + + t.Run("default when unset", func(t *testing.T) { + cfg, err := LoadPipelineConfig(writeTemp(t, `sketch: {type: kll}`)) + require.NoError(t, err) + assert.Equal(t, time.Second, cfg.ReaderInterval()) + }) + + t.Run("default when zero", func(t *testing.T) { + yaml := `reader: {interval: 0}` + cfg, err := LoadPipelineConfig(writeTemp(t, yaml)) + require.NoError(t, err) + assert.Equal(t, time.Second, cfg.ReaderInterval()) + }) +}