Demote per-5s datagram-loss WARN to debug, export dropped-total as a metric

Dockerfile

@@ -43,7 +43,7 @@ RUN mkdir -p \
 
 RUN --mount=type=cache,target=/usr/local/cargo/registry,sharing=locked \
     --mount=type=cache,target=/build/target,sharing=locked \
-    cargo build --release
+    cargo build --release --features moq-pub-mmtp/metrics-prometheus
 
 COPY . ./
 
@@ -52,10 +52,17 @@ COPY . ./
 # There's also issues with the cache mount since it builds into /usr/local/cargo/bin
 # We can't mount that without clobbering cargo itself.
 # We instead we build the binaries and copy them to the cargo bin directory.
+#
+# moq-pub-mmtp/metrics-prometheus (BLO-22882): compiled in by default so the
+# MOQ_PUB_METRICS_ADDR env var (set via Helm) is live without a separate
+# opt-in image variant — unlike profiling/heap-profiling below, the exporter
+# itself stays inert (no listener bound) until that env var is set, so this
+# does not change default runtime behavior.
 RUN --mount=type=cache,target=/usr/local/cargo/registry,sharing=locked \
     --mount=type=cache,target=/build/target,sharing=locked \
     find . -path '*/src/*.rs' -o -path '*/src/**/*.rs' | xargs touch && \
-    cargo build --release && cp /build/target/release/moq-* /usr/local/cargo/bin
+    cargo build --release --features moq-pub-mmtp/metrics-prometheus && \
+    cp /build/target/release/moq-* /usr/local/cargo/bin
 
 # Optional: overwrite moq-pub-mmtp with a profiling-enabled build. CPU profiling
 # uses PROFILING=1; retained-allocation profiling uses HEAP_PROFILING=1 and the
@@ -68,11 +75,11 @@ RUN --mount=type=cache,target=/usr/local/cargo/registry,sharing=locked \
     if [ -n "$HEAP_PROFILING" ]; then \
       JEMALLOC_SYS_WITH_MALLOC_CONF="prof:true,prof_active:false,lg_prof_sample:19" \
         RUSTFLAGS="-C force-frame-pointers=yes" \
-        cargo rustc --release -p moq-pub-mmtp --features heap-profiling -- \
+        cargo rustc --release -p moq-pub-mmtp --features heap-profiling,metrics-prometheus -- \
           -C link-arg=-no-pie && \
       cp /build/target/release/moq-pub-mmtp /usr/local/cargo/bin/moq-pub-mmtp; \
     elif [ -n "$PROFILING" ]; then \
-      cargo build --release -p moq-pub-mmtp --features profiling && \
+      cargo build --release -p moq-pub-mmtp --features profiling,metrics-prometheus && \
       cp /build/target/release/moq-pub-mmtp /usr/local/cargo/bin/moq-pub-mmtp; \
     fi
 

moq-pub-mmtp/Cargo.toml

@@ -30,10 +30,12 @@ heap-profiling = [
 	"dep:tikv-jemallocator",
 	"dep:tikv-jemalloc-ctl",
 ]
-# Off by default. Compiles the Prometheus exporter behind --metrics-addr
-# (installed in main.rs). Without this feature, --metrics-addr is parsed but
-# produces a startup warning instead of serving anything — the shipped
-# binary otherwise never links metrics-exporter-prometheus.
+# Off by default. Adds the Prometheus HTTP exporter for the `metrics` crate
+# facade (src/metrics_endpoint.rs), mirroring moq-relay-ietf's wiring. The
+# facade itself is always compiled in (see the `metrics` dependency below);
+# this feature only adds the exporter, which is ADDITIONALLY runtime-gated by
+# the MOQ_PUB_METRICS_ADDR env var (same activation pattern as
+# MOQ_PUB_PROFILE_ADDR in profiling.rs).
 metrics-prometheus = ["dep:metrics-exporter-prometheus"]
 
 [dependencies]
@@ -60,11 +62,10 @@ tracing-subscriber = { workspace = true }
 anyhow = { version = "1", features = ["backtrace"] }
 serde_json = "1"
 
-# Metrics — moq-native-ietf (a dependency) already emits
-# `moq_negotiation_total` via the `metrics` facade on every connect attempt,
-# recorded by whatever global recorder is installed process-wide. This crate
-# only needs the exporter that installs one; it never calls `metrics::`
-# macros itself, so the `metrics` crate is not a direct dependency here.
+# Always compiled in; near-zero overhead with no recorder installed (same
+# facade-vs-exporter split as moq-relay-ietf/src/metrics.rs). The optional
+# metrics-prometheus feature above adds the HTTP exporter.
+metrics = "0.24"
 metrics-exporter-prometheus = { version = "0.16", optional = true }
 
 # --- feature "profiling" only ---

moq-pub-mmtp/src/main.rs

@@ -19,6 +19,7 @@ use tokio::io::AsyncWriteExt;
 mod cli;
 mod datagram;
 mod framing;
+mod metrics_endpoint;
 mod mmtp_parse;
 #[cfg(feature = "profiling")]
 mod profiling;
@@ -58,6 +59,11 @@ async fn main() -> Result<()> {
     #[cfg(feature = "profiling")]
     profiling::spawn_if_enabled();
 
+    // Optional Prometheus metrics exporter (feature `metrics-prometheus` +
+    // MOQ_PUB_METRICS_ADDR). No-op unless the env var is set; see
+    // metrics_endpoint.rs for the activation pattern.
+    metrics_endpoint::spawn_if_enabled();
+
     let args = Args::parse();
 
     // Optional Prometheus metrics exporter (feature `metrics-prometheus` +

moq-pub-mmtp/src/metrics_endpoint.rs

@@ -0,0 +1,80 @@
+// SPDX-License-Identifier: MIT OR Apache-2.0
+//
+// Prometheus metrics endpoint. Mirrors moq-relay-ietf's split (see
+// moq-relay-ietf/src/metrics.rs + src/bin/moq-relay-ietf/main.rs): the
+// `metrics` crate facade is always compiled in (near-zero overhead with no
+// recorder installed, same as the `log` crate with no logger configured);
+// the optional `metrics-prometheus` feature adds the HTTP exporter.
+//
+// Activation is env-var-only — MOQ_PUB_METRICS_ADDR, e.g. "0.0.0.0:9091" —
+// the same runtime-gating pattern MOQ_PUB_PROFILE_ADDR uses in profiling.rs,
+// so a binary built with the feature compiled in stays inert until an
+// operator opts in, and there is no separate clap flag to keep in sync with
+// the env var.
+//
+// # Available Metrics
+//
+// | Name | Description |
+// |------|-------------|
+// | `moq_pub_mmtp_dropped_datagrams_total` | Datagrams dropped by the publisher-side ring buffer (ring-superseded by a lagging subscriber, or over-MTU payloads skipped) — see moq-transport/src/session/subscribed.rs |
+
+/// Register metric descriptions (Prometheus `# HELP` text).
+pub fn describe_metrics() {
+    metrics::describe_counter!(
+        "moq_pub_mmtp_dropped_datagrams_total",
+        "Datagrams dropped by the publisher-side ring buffer: ring-superseded \
+         (lagging subscriber) or over-MTU payloads skipped"
+    );
+}
+
+/// Install the Prometheus exporter iff `MOQ_PUB_METRICS_ADDR` is set. No-op
+/// otherwise, and a no-op (with a warning) when the `metrics-prometheus`
+/// feature was not compiled in.
+pub fn spawn_if_enabled() {
+    let Ok(addr) = std::env::var("MOQ_PUB_METRICS_ADDR") else {
+        return;
+    };
+
+    #[cfg(feature = "metrics-prometheus")]
+    {
+        let parsed: Result<std::net::SocketAddr, _> = addr.parse();
+        match parsed {
+            Ok(socket_addr) => {
+                match metrics_exporter_prometheus::PrometheusBuilder::new()
+                    .with_http_listener(socket_addr)
+                    .install()
+                {
+                    Ok(()) => {
+                        describe_metrics();
+                        tracing::info!(
+                            addr = %socket_addr,
+                            "metrics exporter listening on http://{socket_addr}/metrics"
+                        );
+                    }
+                    Err(error) => {
+                        // Observability plumbing must not take down a live
+                        // publisher: log and keep running without the
+                        // exporter rather than failing the process.
+                        tracing::warn!(%error, "failed to install Prometheus metrics exporter; continuing without it");
+                    }
+                }
+            }
+            Err(error) => {
+                tracing::warn!(
+                    %addr,
+                    %error,
+                    "MOQ_PUB_METRICS_ADDR is not a valid socket address; metrics exporter disabled"
+                );
+            }
+        }
+    }
+
+    #[cfg(not(feature = "metrics-prometheus"))]
+    {
+        tracing::warn!(
+            %addr,
+            "MOQ_PUB_METRICS_ADDR was set but the metrics-prometheus feature is not enabled. \
+             Rebuild with --features metrics-prometheus to enable the Prometheus exporter."
+        );
+    }
+}

moq-transport/Cargo.toml

@@ -32,3 +32,8 @@ futures = "0.3"
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 serde_with = "3"
+
+# Always compiled in; effectively zero-cost when no recorder is installed
+# (same pattern as moq-relay-ietf/src/metrics.rs — the `metrics` crate is a
+# facade, not a Prometheus dependency by itself).
+metrics = "0.24"

moq-transport/src/session/subscribed.rs

@@ -45,6 +45,17 @@ fn subscribe_ok_params(
     Ok(params)
 }
 
+fn record_datagram_loss_metric(
+    dropped_total: u64,
+    reported_dropped: u64,
+    skipped_too_large: u64,
+    reported_too_large: u64,
+) -> u64 {
+    let loss_delta = (dropped_total - reported_dropped) + (skipped_too_large - reported_too_large);
+    metrics::counter!("moq_pub_mmtp_dropped_datagrams_total").increment(loss_delta);
+    loss_delta
+}
+
 #[derive(Debug)]
 struct ObjectForwarderState {
     largest_location: Option<Location>,
@@ -662,16 +673,37 @@ impl ObjectForwarder {
             let loss_grew =
                 dropped_total > reported_dropped || skipped_too_large > reported_too_large;
             let warn_due = last_loss_warn.is_none_or(|at| at.elapsed() >= LOSS_WARN_INTERVAL);
-            if loss_grew && warn_due {
-                tracing::warn!(
+            if loss_grew {
+                let dropped_delta = dropped_total - reported_dropped;
+                let skipped_too_large_delta = skipped_too_large - reported_too_large;
+                let loss_delta = record_datagram_loss_metric(
                     dropped_total,
-                    dropped_delta = dropped_total - reported_dropped,
+                    reported_dropped,
                     skipped_too_large,
-                    "datagram subscriber lossy: ring-superseded and/or over-MTU payloads skipped"
+                    reported_too_large,
                 );
+                // BLO-22882: aggregate loss must remain durably queryable even
+                // when the per-interval log line below is suppressed or evicted
+                // from a retained log window — the counter is the source of
+                // truth, the log line is a summary.
+                if warn_due {
+                    // Demoted from warn! (BLO-22882): at sustained loss this
+                    // line fires every 5s and was crowding out other startup
+                    // diagnostics in the retained log window. The counter
+                    // above is now the durable signal; this stays for local
+                    // debugging.
+                    tracing::debug!(
+                        dropped_total,
+                        dropped_delta,
+                        skipped_too_large,
+                        skipped_too_large_delta,
+                        loss_delta,
+                        "datagram subscriber lossy: ring-superseded and/or over-MTU payloads skipped"
+                    );
+                    last_loss_warn = Some(std::time::Instant::now());
+                }
                 reported_dropped = dropped_total;
                 reported_too_large = skipped_too_large;
-                last_loss_warn = Some(std::time::Instant::now());
             }
 
             if !delivery_filter.allows(datagram.group_id, datagram.object_id) {
@@ -772,6 +804,19 @@ impl ObjectForwarder {
             datagram_count += 1;
         }
 
+        // The final input can be over the datagram limit. The loop normally
+        // accounts for that loss at its next iteration, so flush it when EOF
+        // arrives directly after the skip.
+        let dropped_total = datagrams.dropped();
+        if dropped_total > reported_dropped || skipped_too_large > reported_too_large {
+            record_datagram_loss_metric(
+                dropped_total,
+                reported_dropped,
+                skipped_too_large,
+                reported_too_large,
+            );
+        }
+
         tracing::info!(
             "[PUBLISHER] serve_datagrams: completed ({} datagrams sent, {} skipped over-MTU)",
             datagram_count,
@@ -802,8 +847,96 @@ impl ObjectForwarderRecv {
 
 #[cfg(test)]
 mod tests {
+    use std::sync::{
+        atomic::{AtomicU64, Ordering},
+        Arc,
+    };
+
+    use metrics::{
+        Counter, CounterFn, Gauge, Histogram, Key, KeyName, Metadata, Recorder, SharedString, Unit,
+    };
+
     use super::*;
 
+    const DROPPED_DATAGRAMS_METRIC: &str = "moq_pub_mmtp_dropped_datagrams_total";
+
+    #[derive(Default)]
+    struct CounterRecorder {
+        dropped_datagrams: Arc<AtomicU64>,
+    }
+
+    struct AtomicCounter(Arc<AtomicU64>);
+
+    impl CounterFn for AtomicCounter {
+        fn increment(&self, value: u64) {
+            self.0.fetch_add(value, Ordering::Relaxed);
+        }
+
+        fn absolute(&self, value: u64) {
+            self.0.fetch_max(value, Ordering::Relaxed);
+        }
+    }
+
+    impl Recorder for CounterRecorder {
+        fn describe_counter(&self, _key: KeyName, _unit: Option<Unit>, _description: SharedString) {
+        }
+
+        fn describe_gauge(&self, _key: KeyName, _unit: Option<Unit>, _description: SharedString) {}
+
+        fn describe_histogram(
+            &self,
+            _key: KeyName,
+            _unit: Option<Unit>,
+            _description: SharedString,
+        ) {
+        }
+
+        fn register_counter(&self, key: &Key, _metadata: &Metadata<'_>) -> Counter {
+            if key.name() == DROPPED_DATAGRAMS_METRIC {
+                Counter::from_arc(Arc::new(AtomicCounter(Arc::clone(&self.dropped_datagrams))))
+            } else {
+                Counter::noop()
+            }
+        }
+
+        fn register_gauge(&self, _key: &Key, _metadata: &Metadata<'_>) -> Gauge {
+            Gauge::noop()
+        }
+
+        fn register_histogram(&self, _key: &Key, _metadata: &Metadata<'_>) -> Histogram {
+            Histogram::noop()
+        }
+    }
+
+    #[test]
+    fn dropped_datagram_metric_counts_ring_loss() {
+        let recorder = CounterRecorder::default();
+
+        metrics::with_local_recorder(&recorder, || record_datagram_loss_metric(7, 3, 0, 0));
+
+        assert_eq!(recorder.dropped_datagrams.load(Ordering::Relaxed), 4);
+    }
+
+    #[test]
+    fn dropped_datagram_metric_counts_over_mtu_loss() {
+        let recorder = CounterRecorder::default();
+
+        metrics::with_local_recorder(&recorder, || record_datagram_loss_metric(0, 0, 5, 2));
+
+        assert_eq!(recorder.dropped_datagrams.load(Ordering::Relaxed), 3);
+    }
+
+    #[test]
+    fn dropped_datagram_metric_flushes_a_final_over_mtu_loss() {
+        let recorder = CounterRecorder::default();
+
+        // This is the EOF path: the final over-MTU packet has no following
+        // loop iteration to report its accumulated loss.
+        metrics::with_local_recorder(&recorder, || record_datagram_loss_metric(0, 0, 1, 0));
+
+        assert_eq!(recorder.dropped_datagrams.load(Ordering::Relaxed), 1);
+    }
+
     #[test]
     fn blockcast_profile_requires_publisher_history_window() {
         let error = subscribe_ok_params(WireProfile::Blockcast01, None, None)