stats: per-track counters exposed at /metrics/track

CMakeLists.txt

@@ -159,8 +159,10 @@ add_library(moqx_core STATIC
   src/admin/CachePurgeHandler.cpp
   src/admin/ConfigHandler.cpp
   src/admin/MetricsHandler.cpp
+  src/admin/TrackMetricsHandler.cpp
   src/admin/StateHandler.cpp
   src/stats/StatsRegistry.cpp
+  src/stats/TrackStatsRegistry.cpp
   src/stats/MoQStatsCollector.cpp
   src/stats/PicoQuicStatsCollector.cpp
   src/stats/QuicStatsCollector.cpp
@@ -169,6 +171,7 @@ add_library(moqx_core STATIC
   src/SafeTrackName.cpp
   src/relay/AuthFilters.cpp
   src/relay/TopNFilter.cpp
+  src/relay/TrackStatsFilter.cpp
   src/relay/PropertyRanking.cpp
   src/relay/CrossExecFilter.cpp
   src/relay/CrossExecForwarderCallback.cpp

docs/config.md

@@ -432,16 +432,31 @@ admin:
   #   cert_file: /etc/moqx/admin-cert.pem
   #   key_file:  /etc/moqx/admin-key.pem
   #   alpn: [h2, "http/1.1"]
+  # track_metrics_enabled: true                 # per-track counting and /metrics/track
+  # track_metrics_endpoint_default_limit: 10    # default tracks per /metrics/track scrape
+  # track_metrics_endpoint_max_limit: 1000      # ceiling on the ?limit= parameter
 ```
 
 Either `plaintext: true` or a `tls` block must be set, but not both.
 
+`track_metrics_enabled: false` leaves the counting filters out of the data path
+entirely — nothing is installed, so there is no per-object cost — and
+`/metrics/track` answers `503` rather than an empty scrape that would read as
+"no live tracks".
+
+`track_metrics_endpoint_default_limit` and `track_metrics_endpoint_max_limit`
+bound `/metrics/track`; the default must not exceed the max. The limit is a
+guard rail, not a selection rule: a query matching more tracks than the limit is
+rejected rather than truncated, because an arbitrary subset would give
+Prometheus a series set that reshuffles between scrapes.  See [docs/metrics.md] (metrics.md).
+
 ### Endpoints
 
 | Method | Path | Description |
 |---|---|---|
 | `GET` | `/info` | Returns `{"service":"moqx","version":"..."}`. |
 | `GET` | `/metrics` | Prometheus-format metrics. See [docs/metrics.md](metrics.md) (pending PR #137). |
+| `GET` | `/metrics/track` | Per-track Prometheus metrics for live tracks. See [docs/metrics.md](metrics.md). |
 | `GET` | `/state` | Relay state: connected peers, active subscriptions, namespace tree, and cache stats. Pending PR #146. |
 
 ---

docs/metrics.md

@@ -182,3 +182,82 @@ Counters with per-code breakdowns:
 |--------|-------------|
 | `moqx_quicActiveConnections` | Active QUIC connections |
 | `moqx_quicActiveStreams` | Active QUIC streams across all connections |
+
+## Per-Track Metrics
+
+```
+GET /metrics/track?service=<name>&namespace=<a/b>&track=<name>&limit=<N>
+```
+
+Requires `admin.track_metrics_enabled` (default true); when it is false the
+counting filters are never installed and this endpoint returns `503`.
+
+Reports counters for **live tracks** — a track's series disappear when the track
+goes away, and start from zero if it comes back.
+
+| Parameter | Required | Description |
+|---|---|---|
+| `namespace` | no | Namespace prefix in the safe form below. Matches every track under the prefix unless `track` is given. Default: all namespaces. |
+| `service` | no | Restrict to one service. Default: all services, each labeled. |
+| `track` | no | Exact track name within the namespace. |
+| `limit` | no | Max tracks to report. Default `admin.track_metrics_endpoint_default_limit` (10), clamped to `admin.track_metrics_endpoint_max_limit` (1000). |
+
+Every parameter is optional, so `GET /metrics/track?limit=20` reports every live
+track when fewer than 20 match.
+
+`limit` is a sanity guard, not a top-N selector. A query matching more tracks
+than the limit returns **400** with the match count rather than truncating —
+an arbitrary subset would give Prometheus a series set that reshuffles between
+scrapes, producing gap-filled graphs that look like data. Narrow the namespace
+or raise the limit. A query that matches nothing returns 200 with no series.
+
+Every series carries `{service, namespace, track}` labels.
+
+### Name encoding
+
+Namespaces and track names are arbitrary bytes, so the `namespace` and `track`
+values — both in labels and in the query parameters — use the form RECOMMENDED
+by moq-transport, [Representing Namespace and Track
+Names](https://datatracker.ietf.org/doc/html/draft-ietf-moq-transport#name-representing-namespace-and-t).
+The namespace `conf.example.com` / `room 1` renders as
+`conf.2eexample.2ecom-room.201`.
+
+Two tracks can never collapse onto one label
+set and produce duplicate series, and a scraped label value can be pasted
+straight back into a query. Values outside the form — an unencoded `/` or space,
+say — are rejected with 400 rather than silently matching something else.
+
+| Metric | Type | Description |
+|--------|------|-------------|
+| `moqx_track_groups_received_total` | counter | Groups ingested |
+| `moqx_track_subgroups_received_total` | counter | Subgroups ingested |
+| `moqx_track_objects_received_total` | counter | Objects ingested |
+| `moqx_track_datagrams_received_total` | counter | Objects ingested as datagrams (also counted in objects) |
+| `moqx_track_bytes_received_total` | counter | Object payload bytes ingested |
+| `moqx_track_groups_sent_total` | counter | Groups delivered, summed over subscribers |
+| `moqx_track_subgroups_sent_total` | counter | Subgroups delivered, summed over subscribers |
+| `moqx_track_objects_sent_total` | counter | Objects delivered, summed over subscribers |
+| `moqx_track_datagrams_sent_total` | counter | Objects delivered as datagrams, summed over subscribers |
+| `moqx_track_bytes_sent_total` | counter | Object payload bytes delivered, summed over subscribers |
+| `moqx_track_subscribers` | gauge | Current downstream subscribers |
+| `moqx_track_publish_start_timestamp_seconds` | gauge | Unix time the relay first saw the track |
+| `moqx_track_last_object_timestamp_seconds` | gauge | Unix time of the most recent ingested object |
+
+Group counters track the 3 most recently seen group IDs (LRU), so subgroups of
+concurrently-open groups can arrive interleaved without inflating the count. A
+group revisited after 3 *other* groups have been seen is counted again — the
+window bounds per-filter state, and MOQT group IDs usually advance and high
+concurrency is not expected, so this only shows up in pathological interleaving.
+
+Byte counters measure **object payload bytes**, not wire bytes: no MOQT headers,
+no QUIC or transport framing. For wire-level volume use `moqx_quicBytesRead_total`
+and `moqx_quicBytesWritten_total`.
+
+Sent counters are summed across subscribers, so one ingested object fanned out to
+three subscribers increments `objects_sent` by 3. "Sent" means **passed to the
+transport**, not acknowledged by the peer: objects dropped before the wire — by
+`STOP_SENDING`, a delivery timeout, or session teardown — are still counted.
+
+Timestamps are Unix seconds carrying millisecond precision (`1754236801.234`).
+They are sampled from a coarse monotonic clock whose resolution is one kernel
+tick, so the last digits quantize to 1–4ms depending on `CONFIG_HZ`.

src/MoqxRelay.cpp

@@ -13,8 +13,10 @@
 #include "relay/NullConsumers.h"
 #include "relay/PublisherCrossExecFilter.h"
 #include "relay/SubscriberCrossExecFilter.h"
+#include "relay/TrackStatsFilter.h"
 #include "relay/WeakRelayForwarderCallback.h"
 #include <folly/container/F14Set.h>
+#include <folly/coro/Collect.h>
 #include <moxygen/MoQFilters.h>
 #include <moxygen/MoQTrackProperties.h>
 
@@ -550,9 +552,9 @@ folly::coro::Task<folly::Expected<PublishOk, PublishError>> MoqxRelay::registerP
   }
 
   auto topNView = registry_.getTopNView(ftn);
-  XCHECK(topNView && topNView->topNFilter)
-      << "registerPublishOnRelayExec: topNFilter always present in MT mode";
-  relayChainFilter->setDownstream(topNView->topNFilter);
+  XCHECK(topNView && topNView->chainHead)
+      << "registerPublishOnRelayExec: relay chain always present in MT mode";
+  relayChainFilter->setDownstream(topNView->chainHead);
 
   co_return setup.value().publishOk;
 }
@@ -798,7 +800,12 @@ std::optional<MoqxRelay::PreparedPublish> MoqxRelay::startPublish(
     subscriber->unsubscribe();
     return std::nullopt;
   }
-  subscriber->trackConsumer = std::move(pub->consumer);
+  subscriber->trackConsumer = wrapWithTrackStats(
+      trackStats_,
+      forwarder->fullTrackName(),
+      std::move(pub->consumer),
+      stats::TrackDirection::Egress
+  );
   return PreparedPublish{std::move(subscriber), std::move(pub->reply)};
 }
 
@@ -1339,7 +1346,13 @@ MoqxRelay::buildFilterChain(const FullTrackName& ftn, std::shared_ptr<MoQForward
     topNFilter->setActivityThreshold(activityThreshold_);
     return SubscriptionRegistry::FilterChainResult{
         .consumer = std::static_pointer_cast<TrackConsumer>(forwarder),
-        .topNFilter = topNFilter
+        .topNFilter = topNFilter,
+        .chainHead = wrapWithTrackStats(
+            trackStats_,
+            ftn,
+            std::static_pointer_cast<TrackConsumer>(topNFilter),
+            stats::TrackDirection::Ingest
+        )
     };
   }
 
@@ -1361,9 +1374,16 @@ MoqxRelay::buildFilterChain(const FullTrackName& ftn, std::shared_ptr<MoQForward
   auto topNFilter =
       std::make_shared<TopNFilter>(ftn, std::static_pointer_cast<TrackConsumer>(terminationFilter));
   topNFilter->setActivityThreshold(activityThreshold_);
+  auto chainHead = wrapWithTrackStats(
+      trackStats_,
+      ftn,
+      std::static_pointer_cast<TrackConsumer>(topNFilter),
+      stats::TrackDirection::Ingest
+  );
   return SubscriptionRegistry::FilterChainResult{
-      .consumer = std::static_pointer_cast<TrackConsumer>(topNFilter),
-      .topNFilter = topNFilter
+      .consumer = chainHead,
+      .topNFilter = topNFilter,
+      .chainHead = chainHead
   };
 }
 
@@ -1847,12 +1867,11 @@ folly::coro::Task<MoqxRelay::PublisherAttachment> MoqxRelay::attachNewLocalForwa
   }
   auto upstreamOk = std::move(upstreamResult->value());
 
-  // Wire the relay chain to topNFilter before pending.complete() so buffered objects
-  // see the filter.
+  // Wire the relay chain before pending.complete() so buffered objects see the filters.
   if (relayChainFilter) {
     auto topNView = registry_.getTopNView(ftn);
-    if (topNView && topNView->topNFilter) {
-      relayChainFilter->setDownstream(topNView->topNFilter);
+    if (topNView && topNView->chainHead) {
+      relayChainFilter->setDownstream(topNView->chainHead);
     }
   }
 
@@ -1951,6 +1970,9 @@ folly::coro::Task<Publisher::SubscribeResult> MoqxRelay::subscribeFromSubscriber
   auto [localFwd, isNew, localReg] =
       acquireLocalForwarder(ftn, [&] { return std::make_shared<MoQForwarder>(ftn); });
 
+  consumer =
+      wrapWithTrackStats(trackStats_, ftn, std::move(consumer), stats::TrackDirection::Egress);
+
   if (!isNew) {
     if (auto err = checkRangeNotInPast(*localFwd, subReq)) {
       co_return folly::makeUnexpected(std::move(*err));
@@ -2062,6 +2084,9 @@ MoqxRelay::subscribeImpl(SubscribeRequest subReq, std::shared_ptr<TrackConsumer>
     co_await upstream_->waitForConnected(kUpstreamConnectWaitTimeout);
   }
 
+  consumer =
+      wrapWithTrackStats(trackStats_, ftn, std::move(consumer), stats::TrackDirection::Egress);
+
   auto firstOrSubsequent = registry_.getOrCreateFromSubscribe(
       ftn,
       shared_from_this(),
@@ -2584,6 +2609,25 @@ void MoqxRelay::onTrackEvicted(const FullTrackName& ftn, std::shared_ptr<MoQSess
   evict(registry_.getForwarder(ftn));
 }
 
+MoqxRelay::TrackMatch
+MoqxRelay::matchTracks(const TrackNamespace& nsPrefix, const std::string* trackName, size_t limit)
+    const {
+  TrackMatch match;
+  registry_.forEachName([&](const FullTrackName& ftn) {
+    if (!ftn.trackNamespace.startsWith(nsPrefix)) {
+      return;
+    }
+    if (trackName && ftn.trackName != *trackName) {
+      return;
+    }
+    ++match.matched;
+    if (match.keys.size() < limit) {
+      match.keys.push_back(ftn);
+    }
+  });
+  return match;
+}
+
 void MoqxRelay::dumpState(RelayStateVisitor& visitor) const {
   visitor.onPeersBegin();
   for (const auto& [sess, peer] : peerSubNsHandles_) {

src/MoqxRelay.h

@@ -16,6 +16,7 @@
 #include "relay/LocalForwarderRegistry.h"
 #include "relay/PropertyRanking.h"
 #include "relay/RelayExecUtil.h"
+#include "stats/TrackStatsRegistry.h"
 #include <moxygen/MoQSession.h>
 #include <moxygen/relay/MoQForwarder.h>
 
@@ -130,6 +131,22 @@ class MoqxRelay : public moxygen::Publisher,
 
   folly::Executor* getRelayExec() const { return relayExec_; }
 
+  // execs must cover every thread the data plane runs on (io threads plus
+  // relayExec_).
+  stats::TrackStatsRegistry& trackStatsRegistry() { return trackStats_; }
+  const stats::TrackStatsRegistry& trackStatsRegistry() const { return trackStats_; }
+
+  struct TrackMatch {
+    std::vector<moxygen::FullTrackName> keys;
+    // Total matches before the limit was applied.
+    size_t matched{0};
+  };
+
+  // Must run on the relay exec.
+  TrackMatch
+  matchTracks(const moxygen::TrackNamespace& nsPrefix, const std::string* trackName, size_t limit)
+      const;
+
   void setAllowedNamespacePrefix(moxygen::TrackNamespace allowed) {
     allowedNamespacePrefix_ = std::move(allowed);
   }
@@ -582,6 +599,9 @@ class MoqxRelay : public moxygen::Publisher,
 
   bool useLocalForwarders_{false};
   folly::ThreadLocalPtr<LocalForwarderRegistry> tlForwarders_;
+
+  stats::TrackStatsRegistry trackStats_;
+
   std::unique_ptr<MoqxCache> cache_;
   uint64_t maxDeselected_{kDefaultMaxDeselected};