[core][metrics] Native raw-datapoint ingestion (cumulative counters, pre-aggregated histograms)

[xyuzh]

Description

Follow-up from review discussion on #64671 (PrometheusCollector, which mirrors any Prometheus exposition into Ray custom metrics).

Today, piping externally-produced Prometheus metrics into Ray requires translation, because ray.util.metrics only accepts point samples: cumulative counters must be converted to deltas (stateful prev map), and pre-aggregated histogram _bucket/_sum/_count series must be mirrored as gauges (losing Prometheus TYPE metadata). The cleaner long-term solution is for Ray Core to accept raw metric data points — cumulative counter values and pre-aggregated histogram buckets — so collectors can pass them through untranslated.

Why this is tractable

• The OTLP wire already carries everything needed: cumulative Sum with start_time_unix_nano; HistogramDataPoint with exact bucket_counts + explicit_bounds + sum. No proto changes.
• Pre-aggregated histogram ingestion already exists — OpenTelemetryMetricRecorder.record_histogram_aggregated_batch (python/ray/_private/telemetry/open_telemetry_metric_recorder.py) — but degrades fidelity by replaying synthetic samples at bucket midpoints; the exact incoming sum is discarded.
• The legacy OpenCensus path did exact cumulative pass-through (OpencensusProxyMetric.record in python/ray/_private/metrics_agent.py stores absolute values; exact-bucket Prometheus emission in to_prometheus_metrics). The default OTel path downgraded to delta + agent-side re-accumulation (kDelta hard-coded in src/ray/observability/open_telemetry_metric_recorder.cc, Start()).
• The metrics agent's OTLP/gRPC MetricsService.Export endpoint is a standard OTLP receiver (python/ray/dashboard/modules/reporter/reporter_agent.py) — a collector can push raw OTLP data points directly, so the mirroring use case needs no C++ changes.

Proposed phases

Phase 1 — agent-side fidelity (Python-only)

• _export_number_data (reporter_agent.py): honor aggregation_temporality + start_time_unix_nano; cumulative monotonic sums get a store-absolute path (today set_metric_value always re-accumulates += value); reset detection via start_time bumps.
• _export_histogram_data / record_histogram_aggregated_batch: keep exact bucket_counts and the incoming sum; emit an exact Prometheus HistogramMetricFamily (as the legacy OC collector does) instead of midpoint replay. OTel's synchronous Histogram has no set-buckets API, so this bypasses the instrument.
• Keep routing through metric_cardinality.get_high_cardinality_labels_to_drop so cardinality governance is preserved.

Phase 2 — direct OTLP pipe for collectors
ray.util.prometheus_collector.PrometheusCollector (from #64671) gains a raw sink: exposition → OTLP data points → export to the local metrics agent's OTLP endpoint. Deletes the cumulative→delta bookkeeping and gauge-mirroring; Prometheus TYPE metadata becomes exact. Needs a supported way to discover the local agent's OTLP port (currently internal).

Phase 3 (optional) — worker-origin raw APIs in ray.util.metrics
C++ ObservableCounter (supported by vendored opentelemetry-cpp v1.19.0, currently unused) + per-instrument cumulative temporality + a set-cumulative entry point through stats::Metric → Cython → Python. Only needed if first-party code wants to record absolute totals. Risk: mixed delta/cumulative temporality for the same metric name double-counts.

Sibling primitive gaps (surfaced by vLLM's / SGLang's ray_wrappers.py, which inject Ray-backed prometheus_client wrappers directly into the engine — the zero-translation path for engines running as Ray actors):

• ray.util.metrics has no Summary primitive (wrappers approximate with a default-bucket Histogram).
• Histogram rejects boundaries <= 0 (python/ray/util/metrics.py), silently dropping engines' le="0.0" buckets.

Known blockers / notes

• Bounds-mismatch assert in record_histogram_aggregated_batch (registered bounds must equal incoming explicit_bounds).
• Histogram cross-worker aggregation is intentionally skipped in cardinality reduction (metrics_agent.py).
• The agent OTLP endpoint is internal/undocumented; exposing it needs an API + security review (per Ray's auth-token model for new RPC surfaces).

Use case

Surfacing inference-engine metrics (SGLang, vLLM, TGI) and node exporters in Ray/Anyscale Prometheus + Grafana without hard-coding metric names. Short-term this works via #64670/#64671 (scrape + translate); this issue tracks removing the translation.