feat: SDK pipeline configuration files (sketch type, transmit_sketch, instruments, ASAPQuery integration)

opentelemetry-app/.gitignore

@@ -1 +1,2 @@
-fakemetricload
+/fakemetricload
+/e2esdkbench

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
+```

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"},