feat: controller BackendClient — push StreamingConfig to ASAPQuery

[zzylol]

Closes the producer side of the ASAPQuery PR E hot-reload contract. After a replan, the controller now (optionally) POSTs the new plan as a StreamingConfig YAML to ASAPQuery-backend's /api/v1/streaming-config endpoint, paralleling its existing OpAMP push to agent-role and backend-role collectors.

Why

ASAPQuery #10 and #12 landed the endpoint and made it observable at query time, but the backend's active StreamingConfig only changes when something POSTs to it. Manual curl was the only producer until now. With this PR the controller closes the loop:

query hits SimpleEngine miss
  → ASAPQuery PR #11 fires fire-and-forget POST to controller
  → controller runs the planner, generates a new plan
  → THIS PR pushes the plan to the backend's streaming-config endpoint
  → next query re-snapshots (ASAPQuery PR #12 phase 2) and finds a match

The data path (agent → OTLP sketch → backend ingest) was already covered by the existing OpAMP push in replan.rs. This PR covers the control path back to the query-side state the backend holds.

What's new

controller::config::asapquery_backend (new module)

Converts a CollectionPlan + metric name into the YAML shape StreamingConfig::from_yaml_data consumes:

aggregations:
  - aggregationId: <deterministic u64 from metric name>
    aggregationType: DDSketch | DatasketchesKLL | HLL | CountSketch | CountMinSketch
    metric: <metric>
    labels:
      grouping: [<from plan.agent_config.aggregate_by>]
      rollup: []
      aggregated: []
    parameters: { ... sketch-type-specific ... }
    windowSize: <secs from plan.agent_config.window_duration>
    windowType: tumbling
    spatialFilter: <label_matchers joined by comma>

Key details:

• Maps SketchType → backend AggregationType::Display string (notably KLL → \"DatasketchesKLL\", not the factory string \"KLL\")
• Deterministic u64 aggregationId from the metric name so repeat pushes for the same metric update (not duplicate) the backend's agg map
• Rejects zero-window plans (the backend parser does too)
• Joins label_matchers into a comma-separated spatialFilter string

Separate from existing config::backend which emits OTel YAML for a backend OTel collector running merge processors — two different services, two different formats.

controller::backend_client (new module)

Thin HTTP client wrapping reqwest::Client:

• BackendClient::new(endpoint) — 5-second timeout (symmetric with ASAPQuery's HttpControllerClient in PR update README #11)
• BackendClient::push_streaming_config(yaml) -> Result<()> — POSTs with content-type: application/x-yaml, maps non-2xx to Err
• push_or_log(client, metric, yaml) — fire-and-forget helper used by the replanner, never propagates errors (next replan cycle retries)

Replanner::with_backend_client(client) builder

New optional field. When set, replan_metric() calls generate_streaming_config_yaml(metric, &plan) after the OpAMP pushes and fire-and-forgets the result. Without the builder call, replans behave exactly as before — existing deployments unaffected.

main.rs wiring

New env var CONTROLLER_BACKEND_ENDPOINT. When set, constructs a BackendClient and attaches it to the Replanner. Logs whether the feature is enabled at startup.

Tests

config::asapquery_backend (5 new)

1. deterministic_id_is_stable_across_calls — same metric → same id, different metrics → different ids, never returns 0
2. yaml_round_trips_through_serde_yaml — generate → re-parse → assert every field
3. maps_all_sketch_types — pins all 5 SketchType → AggregationType mappings including the KLL → DatasketchesKLL gotcha
4. rejects_plan_without_window_duration — zero-window plan rejected with clear error
5. spatial_filter_joins_label_matchers — multiple matchers joined with commas

backend_client (3 new)

1. success_path_round_trips_yaml — spawns a local axum mock, asserts the client posts the exact YAML and the server receives it
2. non_2xx_status_is_reported_as_error — mock returns 500 → client returns formatted Err containing the status
3. push_or_log_swallows_errors — points at an unreachable port, verifies fire-and-forget does not propagate the failure

Validation

• ✅ cargo check --all-targets: clean
• ✅ cargo test --bin controller: 353 passed (up 8 from main)
• ✅ rustfmt --check on new files only: clean

Stack

• ASAPQuery #9 (merged) — MessagePack encoding parity
• ASAPQuery #10 (merged) — StreamingConfig hot-reload endpoint (the consumer this PR POSTs to)
• ASAPQuery #11 (merged) — capability-miss → ControllerClient notification (the trigger for the replan this PR publishes)
• ASAPQuery #12 — per-query re-snapshot (makes the push visible at query time)
• This PR — controller-side producer that closes the loop

Follow-ups

• Rate-limit / dedupe — rapid replans may push the same plan repeatedly. Backend tolerates this idempotently via deterministic agg_ids, but a "push only on hash change" filter would cut unnecessary HTTP traffic.
• Retry with backoff — fire-and-forget drops one replan cycle on transient outage. Small retry queue would recover it.
• Merge semantics on the backend POST handler — today the backend's POST REPLACES the entire StreamingConfig. When the controller pushes for metric A, any aggregations the backend had for metric B are wiped. A follow-up should either restrict the controller to full snapshots or extend the backend endpoint to support partial updates keyed by metric.
• labels.rollup / aggregated — today left empty; wire through once the cost model tracks rollup dimensions separately.

🤖 Generated with Claude Code