diff --git a/CMakeLists.txt b/CMakeLists.txt index 2256348c..fb49d655 100644 --- a/CMakeLists.txt +++ b/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 diff --git a/docs/config.md b/docs/config.md index 4fe4ba05..04ca02b7 100644 --- a/docs/config.md +++ b/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. | --- diff --git a/docs/metrics.md b/docs/metrics.md index 9fa10522..abbc412e 100644 --- a/docs/metrics.md +++ b/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=&namespace=&track=&limit= +``` + +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`. diff --git a/src/MoqxRelay.cpp b/src/MoqxRelay.cpp index 0f76a05b..4cb1e770 100644 --- a/src/MoqxRelay.cpp +++ b/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 +#include #include #include @@ -550,9 +552,9 @@ folly::coro::Task> 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::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_ptrsetActivityThreshold(activityThreshold_); return SubscriptionRegistry::FilterChainResult{ .consumer = std::static_pointer_cast(forwarder), - .topNFilter = topNFilter + .topNFilter = topNFilter, + .chainHead = wrapWithTrackStats( + trackStats_, + ftn, + std::static_pointer_cast(topNFilter), + stats::TrackDirection::Ingest + ) }; } @@ -1361,9 +1374,16 @@ MoqxRelay::buildFilterChain(const FullTrackName& ftn, std::shared_ptr(ftn, std::static_pointer_cast(terminationFilter)); topNFilter->setActivityThreshold(activityThreshold_); + auto chainHead = wrapWithTrackStats( + trackStats_, + ftn, + std::static_pointer_cast(topNFilter), + stats::TrackDirection::Ingest + ); return SubscriptionRegistry::FilterChainResult{ - .consumer = std::static_pointer_cast(topNFilter), - .topNFilter = topNFilter + .consumer = chainHead, + .topNFilter = topNFilter, + .chainHead = chainHead }; } @@ -1847,12 +1867,11 @@ folly::coro::Task 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 MoqxRelay::subscribeFromSubscriber auto [localFwd, isNew, localReg] = acquireLocalForwarder(ftn, [&] { return std::make_shared(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 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 #include @@ -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 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 tlForwarders_; + + stats::TrackStatsRegistry trackStats_; + std::unique_ptr cache_; uint64_t maxDeselected_{kDefaultMaxDeselected}; diff --git a/src/MoqxRelayContext.cpp b/src/MoqxRelayContext.cpp index 02e0704f..cb1dbc7a 100644 --- a/src/MoqxRelayContext.cpp +++ b/src/MoqxRelayContext.cpp @@ -13,6 +13,7 @@ #include #include +#include #include #include #include @@ -173,18 +174,32 @@ folly::coro::Task MoqxRelayContext::purgeCache( co_return total; } -void MoqxRelayContext::initThreadStatsCollectors(folly::IOThreadPoolExecutor& ioExecutor) { - if (!statsRegistry_) { - return; - } +void MoqxRelayContext::initThreadStatsCollectors( + folly::IOThreadPoolExecutor& ioExecutor, + bool initTrackStats +) { + XCHECK(statsRegistry_) << "initThreadStatsCollectors: setStatsRegistry must run first"; + std::vector ioExecs; for (auto& ka : ioExecutor.getAllEventBases()) { auto* evb = ka.get(); + ioExecs.push_back(evb); auto collector = stats::MoQStatsCollector::create_moq_stats_collector(statsRegistry_); collector->setExecutor(evb); statsCollectors_.push_back(collector); // Bind on the owning thread; blocks so every thread is bound before serving. evb->runInEventBaseThreadAndWait([this, collector] { *tlStatsCollector_ = collector; }); } + + if (!initTrackStats) { + return; + } + for (auto& [name, entry] : services_) { + auto execs = ioExecs; + if (auto* relayExec = entry.relay->getRelayExec()) { + execs.push_back(relayExec); + } + entry.relay->trackStatsRegistry().bindAll(execs); + } } void MoqxRelayContext::onNewSession(std::shared_ptr clientSession) { @@ -263,6 +278,78 @@ std::vector MoqxRelayContext::getExactServicePaths() const { return serviceMatcher_.allExactPaths(); } +folly::coro::Task MoqxRelayContext::aggregateTrackMetrics( + std::string serviceName, + TrackNamespace nsPrefix, + std::optional trackName, + size_t limit +) const { + struct Service { + std::string name; + std::shared_ptr relay; + MoqxRelay::TrackMatch match; + }; + + std::vector services; + for (const auto& [name, entry] : services_) { + if (serviceName.empty() || name == serviceName) { + services.push_back({name, entry.relay, {}}); + } + } + + auto matchOn = [](std::shared_ptr relay, + const TrackNamespace& ns, + const std::string* track, + size_t limit) -> folly::coro::Task { + co_return relay->matchTracks(ns, track, limit); + }; + + std::vector> matchTasks; + matchTasks.reserve(services.size()); + for (const auto& service : services) { + // registry_ lives on the relay exec when there is one. Without one, config + // forces threads==1, so the worker EVB is that single io thread. + auto* exec = service.relay->getRelayExec(); + matchTasks.push_back(folly::coro::co_withExecutor( + exec ? exec : static_cast(workerEvb_), + matchOn(service.relay, nsPrefix, trackName ? &*trackName : nullptr, limit) + )); + } + + TrackMetricsResult result; + auto matches = co_await folly::coro::collectAllRange(std::move(matchTasks)); + for (size_t i = 0; i < services.size(); ++i) { + result.matched += matches[i].matched; + services[i].match = std::move(matches[i]); + } + + if (result.matched > limit) { + co_return result; + } + + // A service with no matches would cost one executor hop per thread for an + // empty result, so it contributes no task and no name. + std::vector counted; + std::vector> countTasks; + counted.reserve(services.size()); + countTasks.reserve(services.size()); + for (const auto& service : services) { + if (service.match.keys.empty()) { + continue; + } + counted.push_back(service.name); + countTasks.push_back(service.relay->trackStatsRegistry().aggregateAsync(service.match.keys)); + } + + auto counters = co_await folly::coro::collectAllRange(std::move(countTasks)); + for (size_t i = 0; i < counted.size(); ++i) { + for (auto& [ftn, trackCounters] : counters[i]) { + result.tracks.push_back({counted[i], ftn, trackCounters}); + } + } + co_return result; +} + void MoqxRelayContext::dumpState(RelayContextVisitor& visitor) const { // TODO: source active session count for /state (deferred to the /state rework). int64_t activeSessions = 0; diff --git a/src/MoqxRelayContext.h b/src/MoqxRelayContext.h index a2c73e39..0f9c957e 100644 --- a/src/MoqxRelayContext.h +++ b/src/MoqxRelayContext.h @@ -104,13 +104,38 @@ class MoqxRelayContext { // Used by pico listeners to populate the h3zero WebTransport path table. std::vector getExactServicePaths() const; + struct TrackMetricsEntry { + std::string service; + moxygen::FullTrackName ftn; + stats::TrackCounters counters; + }; + + struct TrackMetricsResult { + std::vector tracks; + // Total matches across services, before the limit; tracks is empty when + // this exceeds the limit so the caller can reject rather than truncate. + size_t matched{0}; + }; + + // serviceName empty matches every service; trackName null matches the whole + // namespace prefix. + folly::coro::Task aggregateTrackMetrics( + std::string serviceName, + moxygen::TrackNamespace nsPrefix, + std::optional trackName, + size_t limit + ) const; + // --- Delegation targets for MoqxRelayServer virtual overrides --- void onNewSession(std::shared_ptr session); void onSessionEnd(std::shared_ptr session); // Must run after setStatsRegistry and before any listener accepts sessions. - void initThreadStatsCollectors(folly::IOThreadPoolExecutor& ioExecutor); + // initTrackStats=false skips per-track counting entirely, leaving the + // data-path filters uninstalled rather than installed-and-ignored. + void + initThreadStatsCollectors(folly::IOThreadPoolExecutor& ioExecutor, bool initTrackStats = true); folly::Expected validateAuthority( const moxygen::ClientSetup& clientSetup, diff --git a/src/SubscriptionRegistry.cpp b/src/SubscriptionRegistry.cpp index 8933b655..ce7bbd7d 100644 --- a/src/SubscriptionRegistry.cpp +++ b/src/SubscriptionRegistry.cpp @@ -48,13 +48,14 @@ SubscriptionRegistry::getOrCreateFromSubscribe( if (it == subscriptions_.end()) { auto forwarder = std::make_shared(ftn, largest); forwarder->setCallback(std::move(callback)); - auto [consumer, topNFilter] = chainBuilder(forwarder); + auto [consumer, topNFilter, chainHead] = chainBuilder(forwarder); auto [emplaceIt, inserted] = subscriptions_.emplace( std::piecewise_construct, std::forward_as_tuple(ftn), std::forward_as_tuple(forwarder, nullptr) ); emplaceIt->second.topNFilter = topNFilter; + emplaceIt->second.chainHead = chainHead; return FirstSubscriber{ forwarder, std::move(consumer), @@ -147,8 +148,9 @@ SubscriptionRegistry::PublishEntry SubscriptionRegistry::createFromPublish( rsub.publisher = std::move(publisher); rsub.isPublish = true; - auto [consumer, topNFilter] = chainBuilder(forwarder); + auto [consumer, topNFilter, chainHead] = chainBuilder(forwarder); rsub.topNFilter = topNFilter; + rsub.chainHead = chainHead; topNFilter->setActivityTarget(&rsub.lastObjectTime); return PublishEntry{std::move(consumer), std::move(evicted)}; @@ -170,7 +172,12 @@ SubscriptionRegistry::getTopNView(const moxygen::FullTrackName& ftn) const { if (it == subscriptions_.end()) { return std::nullopt; } - return TopNView{it->second.forwarder, it->second.topNFilter, it->second.lastObjectTime}; + return TopNView{ + it->second.forwarder, + it->second.topNFilter, + it->second.chainHead, + it->second.lastObjectTime + }; } std::optional @@ -241,6 +248,13 @@ void SubscriptionRegistry::removeIf( } } +void SubscriptionRegistry::forEachName(folly::FunctionRef fn +) const { + for (const auto& [ftn, rsub] : subscriptions_) { + fn(ftn); + } +} + void SubscriptionRegistry::forEach(folly::FunctionRef fn) const { for (const auto& [ftn, rsub] : subscriptions_) { fn(EntryView{ftn, rsub.forwarder, rsub.upstream, rsub.isPublish, rsub.lastObjectTime}); diff --git a/src/SubscriptionRegistry.h b/src/SubscriptionRegistry.h index 3a3c3fdb..e6cc014b 100644 --- a/src/SubscriptionRegistry.h +++ b/src/SubscriptionRegistry.h @@ -24,6 +24,9 @@ class SubscriptionRegistry { struct FilterChainResult { std::shared_ptr consumer; std::shared_ptr topNFilter; + // Differs from consumer in LocalForwarder mode, where the relay chain + // hangs off a channel subscriber rather than the publisher's writes. + std::shared_ptr chainHead; }; // === Subscribe path === @@ -125,6 +128,7 @@ class SubscriptionRegistry { struct TopNView { std::shared_ptr forwarder; std::shared_ptr topNFilter; // may be null for subscribe-path tracks + std::shared_ptr chainHead; std::chrono::steady_clock::time_point lastObjectTime; }; std::optional getTopNView(const moxygen::FullTrackName& ftn) const; @@ -175,6 +179,8 @@ class SubscriptionRegistry { void forEach(folly::FunctionRef fn) const; + void forEachName(folly::FunctionRef fn) const; + private: struct RelaySubscription { RelaySubscription( @@ -192,6 +198,7 @@ class SubscriptionRegistry { folly::coro::SharedPromise promise; bool isPublish{false}; std::shared_ptr topNFilter; + std::shared_ptr chainHead; std::chrono::steady_clock::time_point lastObjectTime; }; diff --git a/src/admin/AdminResponse.h b/src/admin/AdminResponse.h new file mode 100644 index 00000000..e48f1cc0 --- /dev/null +++ b/src/admin/AdminResponse.h @@ -0,0 +1,27 @@ +/* + * Copyright (c) OpenMOQ contributors. + * This source code is licensed under the Apache 2.0 license found in the + * LICENSE file in the root directory of this source tree. + */ + +#pragma once + +#include + +#include +#include +#include +#include + +namespace openmoq::moqx::admin { + +inline void +sendError(proxygen::ResponseHandler* downstream, int status, const std::string& message) { + proxygen::ResponseBuilder(downstream) + .status(status, proxygen::HTTPMessage::getDefaultReason(status)) + .header("Content-Type", "text/plain; charset=utf-8") + .body(folly::IOBuf::copyBuffer(message)) + .sendWithEOM(); +} + +} // namespace openmoq::moqx::admin diff --git a/src/admin/TrackMetricsHandler.cpp b/src/admin/TrackMetricsHandler.cpp new file mode 100644 index 00000000..25107f75 --- /dev/null +++ b/src/admin/TrackMetricsHandler.cpp @@ -0,0 +1,306 @@ +/* + * Copyright (c) OpenMOQ contributors. + * This source code is licensed under the Apache 2.0 license found in the + * LICENSE file in the root directory of this source tree. + */ + +#include "admin/TrackMetricsHandler.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "MoqxRelayContext.h" +#include "SafeTrackName.h" +#include "admin/AdminResponse.h" +#include "admin/AdminServer.h" +#include "stats/PrometheusFormat.h" + +namespace openmoq::moqx::admin { + +namespace { + +// TrackClock is steady, so wall-clock output needs an anchor. One anchor per +// response keeps an unchanging timestamp from jittering between scrapes. +struct ClockAnchor { + stats::TrackClock::time_point steady{stats::TrackClock::now()}; + std::chrono::system_clock::time_point wall{std::chrono::system_clock::now()}; + + int64_t toUnixMillis(stats::TrackClock::time_point tp) const { + auto at = wall - std::chrono::duration_cast(steady - tp); + return std::chrono::duration_cast(at.time_since_epoch()).count(); + } +}; + +} // namespace + +std::unique_ptr formatTrackMetrics(const MoqxRelayContext::TrackMetricsResult& result +) { + stats::PrometheusWriter out; + + struct Counter { + std::string_view name; + std::string_view help; + uint64_t (*value)(const stats::TrackCounters&); + }; + + static constexpr std::array kCounters = {{ + {"moqx_track_groups_received_total", + "Groups ingested, counted on group-ID transition", + [](const stats::TrackCounters& c) { return c.received.groups; }}, + {"moqx_track_subgroups_received_total", + "Subgroups ingested", + [](const stats::TrackCounters& c) { return c.received.subgroups; }}, + {"moqx_track_objects_received_total", + "Objects ingested", + [](const stats::TrackCounters& c) { return c.received.objects; }}, + {"moqx_track_datagrams_received_total", + "Objects ingested as datagrams", + [](const stats::TrackCounters& c) { return c.received.datagrams; }}, + {"moqx_track_bytes_received_total", + "Object payload bytes ingested", + [](const stats::TrackCounters& c) { return c.received.bytes; }}, + {"moqx_track_groups_sent_total", + "Groups delivered, summed over subscribers", + [](const stats::TrackCounters& c) { return c.sent.groups; }}, + {"moqx_track_subgroups_sent_total", + "Subgroups delivered, summed over subscribers", + [](const stats::TrackCounters& c) { return c.sent.subgroups; }}, + {"moqx_track_objects_sent_total", + "Objects delivered, summed over subscribers", + [](const stats::TrackCounters& c) { return c.sent.objects; }}, + {"moqx_track_datagrams_sent_total", + "Objects delivered as datagrams, summed over subscribers", + [](const stats::TrackCounters& c) { return c.sent.datagrams; }}, + {"moqx_track_bytes_sent_total", + "Object payload bytes delivered, summed over subscribers", + [](const stats::TrackCounters& c) { return c.sent.bytes; }}, + }}; + + struct Timestamp { + std::string_view name; + std::string_view help; + stats::TrackClock::time_point (*value)(const stats::TrackCounters&); + }; + + static constexpr std::array kTimestamps = {{ + {"moqx_track_publish_start_timestamp_seconds", + "Unix time the relay first saw the track", + [](const stats::TrackCounters& c) { return c.publishStart; }}, + {"moqx_track_last_object_timestamp_seconds", + "Unix time of the most recent ingested object", + [](const stats::TrackCounters& c) { return c.lastObject; }}, + }}; + + // Service names come from config, so they only need Prometheus quoting; track + // and namespace labels go through safeName() instead. + std::vector labels; + labels.reserve(result.tracks.size()); + for (const auto& entry : result.tracks) { + labels.push_back(folly::to( + "{service=\"", + stats::escapeLabelValue(entry.service), + "\",namespace=\"", + safeName(entry.ftn.trackNamespace), + "\",track=\"", + safeName(entry.ftn.trackName), + "\"} " + )); + } + + for (const auto& series : kCounters) { + out.header(series.name, "counter", series.help); + for (size_t i = 0; i < result.tracks.size(); ++i) { + out.append(series.name); + out.append(labels[i]); + out.num(series.value(result.tracks[i].counters)); + out.append("\n"); + } + out.append("\n"); + } + + const ClockAnchor anchor; + for (const auto& series : kTimestamps) { + out.header(series.name, "gauge", series.help); + for (size_t i = 0; i < result.tracks.size(); ++i) { + auto tp = series.value(result.tracks[i].counters); + // A track with no objects yet has no meaningful timestamp; emitting 0 + // would read as 1970 in any time()-based panel. + if (tp == stats::TrackClock::time_point{}) { + continue; + } + // Prometheus values are float64 and time is always expressed in seconds, + // so sub-second resolution is a fraction rather than a different unit. + auto millis = anchor.toUnixMillis(tp); + out.append(series.name); + out.append(labels[i]); + out.num(millis / 1000); + out.append("."); + out.append(folly::sformat("{:03d}", millis % 1000)); + out.append("\n"); + } + out.append("\n"); + } + + out.header("moqx_track_subscribers", "gauge", "Current downstream subscribers"); + for (size_t i = 0; i < result.tracks.size(); ++i) { + out.append("moqx_track_subscribers"); + out.append(labels[i]); + out.num(result.tracks[i].counters.subscribers); + out.append("\n"); + } + out.append("\n"); + + return out.move(); +} + +void registerTrackMetricsRoute( + AdminServer& adminServer, + std::shared_ptr context, + TrackMetricsLimits limits +) { + adminServer.addRoute( + "GET", + "/metrics/track", + [context = std::move(context), limits]( + std::unique_ptr req, + std::unique_ptr /*body*/, + proxygen::ResponseHandler* downstream, + folly::CancellationToken cancelToken + ) { + // Without this the response is an empty scrape, which reads as "no + // live tracks" rather than "counting is off". + if (!limits.enabled) { + sendError( + downstream, + 503, + "per-track metrics are disabled (admin.track_metrics_enabled)\n" + ); + return; + } + + // An empty prefix matches every namespace, as an empty service matches + // every service; limit is what bounds the scrape either way. + moxygen::TrackNamespace nsPrefix; + auto nsParam = req->getDecodedQueryParam("namespace"); + if (!nsParam.empty()) { + // Same form the labels are rendered in, so a scraped value can be + // pasted straight back into a query. + auto parsed = parseSafeNamespace(nsParam); + if (!parsed) { + sendError(downstream, 400, "namespace is not in the MOQT safe name form\n"); + return; + } + nsPrefix = std::move(*parsed); + } + + size_t limit = limits.defaultLimit; // default is clamped to maxLimit in ConfigResolver.cpp + if (req->hasQueryParam("limit")) { + auto parsed = folly::tryTo(req->getDecodedQueryParam("limit")); + if (!parsed.hasValue() || *parsed == 0) { + sendError(downstream, 400, "limit must be a positive integer\n"); + return; + } + if (*parsed > limits.maxLimit) { + sendError( + downstream, + 400, + folly::to( + "limit exceeds admin.track_metrics_endpoint_max_limit (", + limits.maxLimit, + ")\n" + ) + ); + return; + } + } + + std::string service = req->getDecodedQueryParam("service"); + std::optional track; + if (req->hasQueryParam("track")) { + track = parseSafeBytes(req->getDecodedQueryParam("track")); + if (!track) { + sendError(downstream, 400, "track is not in the MOQT safe name form\n"); + return; + } + } + + folly::coro::co_withCancellation( + cancelToken, + folly::coro::co_withExecutor( + folly::EventBaseManager::get()->getEventBase(), + [](auto ctx, + auto* ds, + auto token, + std::string service, + moxygen::TrackNamespace ns, + std::optional track, + size_t limit) -> folly::coro::Task { + if (token.isCancellationRequested()) { + co_return; + } + MoqxRelayContext::TrackMetricsResult result; + try { + result = co_await ctx->aggregateTrackMetrics( + std::move(service), + std::move(ns), + std::move(track), + limit + ); + } catch (const std::exception& e) { + XLOG(ERR) << "TrackMetricsHandler: aggregateTrackMetrics threw: " << e.what(); + if (!token.isCancellationRequested()) { + sendError(ds, 500, "internal error\n"); + } + co_return; + } + if (token.isCancellationRequested()) { + co_return; + } + // Truncating would hand Prometheus an arbitrary, unstable + // subset of series, so reject instead. + if (result.matched > limit) { + sendError( + ds, + 400, + folly::to( + "matched ", + result.matched, + " tracks, exceeds limit=", + limit, + "; narrow the namespace or raise limit\n" + ) + ); + co_return; + } + proxygen::ResponseBuilder(ds) + .status(200, proxygen::HTTPMessage::getDefaultReason(200)) + .header("Content-Type", "text/plain; version=0.0.4; charset=utf-8") + .body(formatTrackMetrics(result)) + .sendWithEOM(); + }(context, + downstream, + cancelToken, + std::move(service), + std::move(nsPrefix), + std::move(track), + limit) + ) + ) + .start(); + } + ); +} + +} // namespace openmoq::moqx::admin diff --git a/src/admin/TrackMetricsHandler.h b/src/admin/TrackMetricsHandler.h new file mode 100644 index 00000000..ae415b06 --- /dev/null +++ b/src/admin/TrackMetricsHandler.h @@ -0,0 +1,40 @@ +/* + * Copyright (c) OpenMOQ contributors. + * This source code is licensed under the Apache 2.0 license found in the + * LICENSE file in the root directory of this source tree. + */ + +#pragma once + +#include +#include + +#include + +#include "MoqxRelayContext.h" + +namespace openmoq::moqx { + +namespace admin { +class AdminServer; + +struct TrackMetricsLimits { + bool enabled{true}; + size_t defaultLimit{10}; + size_t maxLimit{1000}; +}; + +std::unique_ptr formatTrackMetrics(const MoqxRelayContext::TrackMetricsResult& result +); + +// Registers GET /metrics/track, which reports per-track counters in Prometheus +// text format for tracks matching ?service=&namespace=&track=. +void registerTrackMetricsRoute( + AdminServer& adminServer, + std::shared_ptr context, + TrackMetricsLimits limits = {} +); + +} // namespace admin + +} // namespace openmoq::moqx diff --git a/src/config/Config.h b/src/config/Config.h index 4ddb41c3..756bd9f0 100644 --- a/src/config/Config.h +++ b/src/config/Config.h @@ -227,6 +227,9 @@ struct ServiceConfig { struct AdminConfig { folly::SocketAddress address; std::optional tls; + bool trackMetricsEnabled{false}; + uint32_t trackMetricsLimit{10}; + uint32_t trackMetricsMaxLimit{1000}; }; struct MLogConfig { diff --git a/src/config/ConfigResolver.cpp b/src/config/ConfigResolver.cpp index ee274465..98130e67 100644 --- a/src/config/ConfigResolver.cpp +++ b/src/config/ConfigResolver.cpp @@ -1192,11 +1192,36 @@ folly::Expected resolveConfig(const ParsedConfig& c std::move(adminMaterial) ); } - adminConfig = AdminConfig{ + AdminConfig resolved{ .address = folly::SocketAddress(adminOptional->address.value(), adminOptional->port.value()), .tls = std::move(adminTls), }; + if (auto enabled = adminOptional->track_metrics_enabled.value()) { + resolved.trackMetricsEnabled = *enabled; + } + if (auto limit = adminOptional->track_metrics_endpoint_default_limit.value()) { + resolved.trackMetricsLimit = *limit; + } + if (auto maxLimit = adminOptional->track_metrics_endpoint_max_limit.value()) { + if (*maxLimit <= 0) { + return folly::makeUnexpected(folly::to( + "admin.track_metrics_endpoint_max_limit must be greater than 0, got ", + *maxLimit + )); + } + resolved.trackMetricsMaxLimit = *maxLimit; + } + if (resolved.trackMetricsLimit > resolved.trackMetricsMaxLimit) { + return folly::makeUnexpected(folly::to( + "admin.track_metrics_endpoint_default_limit (", + resolved.trackMetricsLimit, + ") exceeds admin.track_metrics_endpoint_max_limit (", + resolved.trackMetricsMaxLimit, + ")" + )); + } + adminConfig = std::move(resolved); } // Resolve relayID: use configured value or generate a random hex string diff --git a/src/config/ConfigSerializer.h b/src/config/ConfigSerializer.h index fea889b3..bf780921 100644 --- a/src/config/ConfigSerializer.h +++ b/src/config/ConfigSerializer.h @@ -262,6 +262,9 @@ inline void serializeConfig(const Config& cfg, ConfigSink& s) { } else { s.nullField("tls"); } + s.boolField("track_metrics_enabled", cfg.admin->trackMetricsEnabled); + s.uintField("track_metrics_endpoint_default_limit", cfg.admin->trackMetricsLimit); + s.uintField("track_metrics_endpoint_max_limit", cfg.admin->trackMetricsMaxLimit); s.endObject(); } else { s.nullField("admin"); diff --git a/src/config/loader/ParsedConfig.h b/src/config/loader/ParsedConfig.h index e66efcae..a83fd939 100644 --- a/src/config/loader/ParsedConfig.h +++ b/src/config/loader/ParsedConfig.h @@ -337,6 +337,18 @@ struct ParsedAdminConfig { rfl::Description<"Bind address", std::string> address; rfl::Description<"Allow plain HTTP (mutually exclusive with tls)", bool> plaintext; rfl::Description<"TLS configuration", std::optional> tls; + rfl::Description< + "Enable per-track counting and the /metrics/track endpoint (default true)", + std::optional> + track_metrics_enabled; + rfl::Description< + "Max tracks /metrics/track reports when the request omits limit (default 10)", + std::optional> + track_metrics_endpoint_default_limit; + rfl::Description< + "Ceiling on the limit parameter accepted by /metrics/track (default 1000)", + std::optional> + track_metrics_endpoint_max_limit; }; struct ParsedUpstreamTlsConfig { diff --git a/src/main.cpp b/src/main.cpp index bde72de3..9dd15e6d 100644 --- a/src/main.cpp +++ b/src/main.cpp @@ -12,6 +12,7 @@ #include "admin/ConfigHandler.h" #include "admin/MetricsHandler.h" #include "admin/StateHandler.h" +#include "admin/TrackMetricsHandler.h" #include "bpf/QuicReuseportSteering.h" #include "config/loader/ConfigInit.h" #include "logging/LogSetup.h" @@ -181,7 +182,10 @@ int main(int argc, char* argv[]) { if (!servers.empty()) { context->setCacheEvb(ioExecutor->getAllEventBases()[0].get()); - context->initThreadStatsCollectors(*ioExecutor); + context->initThreadStatsCollectors( + *ioExecutor, + /*initTrackStats=*/!config.admin || config.admin->trackMetricsEnabled + ); } // === 7. Start health checks / admin endpoints === @@ -190,6 +194,15 @@ int main(int argc, char* argv[]) { admin::registerMetricsRoute(adminServer, statsRegistry); admin::registerCachePurgeRoute(adminServer, context); admin::registerStateRoute(adminServer, context); + admin::TrackMetricsLimits trackLimits; + if (config.admin) { + trackLimits = { + config.admin->trackMetricsEnabled, + config.admin->trackMetricsLimit, + config.admin->trackMetricsMaxLimit + }; + } + admin::registerTrackMetricsRoute(adminServer, context, trackLimits); admin::registerConfigRoute(adminServer, std::make_shared(config)); // === 8. Start serving === diff --git a/src/relay/TrackStatsFilter.cpp b/src/relay/TrackStatsFilter.cpp new file mode 100644 index 00000000..43aae16e --- /dev/null +++ b/src/relay/TrackStatsFilter.cpp @@ -0,0 +1,213 @@ +/* + * Copyright (c) OpenMOQ contributors. + * This source code is licensed under the Apache 2.0 license found in the + * LICENSE file in the root directory of this source tree. + */ + +#include "relay/TrackStatsFilter.h" + +#include + +namespace openmoq::moqx { + +using moxygen::MoQPublishError; +using moxygen::ObjectHeader; +using moxygen::Payload; +using moxygen::SubgroupConsumer; + +namespace { + +uint64_t payloadLength(const Payload& payload) { + return payload ? payload->computeChainDataLength() : 0; +} + +} // namespace + +// Counts per-object work inside one subgroup, delegating the group/subgroup +// bookkeeping to the owning track filter. +class TrackStatsSubgroupFilter : public moxygen::SubgroupConsumerFilter { +public: + TrackStatsSubgroupFilter( + std::shared_ptr owner, + std::shared_ptr downstream + ) + : moxygen::SubgroupConsumerFilter(std::move(downstream)), owner_(std::move(owner)) {} + + folly::Expected object( + uint64_t objectID, + Payload payload, + moxygen::Extensions extensions = moxygen::noExtensions(), + bool finSubgroup = false + ) override { + owner_->countObject(payloadLength(payload)); + return moxygen::SubgroupConsumerFilter::object( + objectID, + std::move(payload), + std::move(extensions), + finSubgroup + ); + } + + folly::Expected beginObject( + uint64_t objectID, + uint64_t length, + Payload initialPayload, + moxygen::Extensions extensions = moxygen::noExtensions() + ) override { + owner_->countObject(payloadLength(initialPayload)); + return moxygen::SubgroupConsumerFilter::beginObject( + objectID, + length, + std::move(initialPayload), + std::move(extensions) + ); + } + + // endOfGroup/endOfTrackAndGroup deliver a real object (a status object with + // no payload), so they count toward the object totals. + folly::Expected endOfGroup(uint64_t endOfGroupObjectID) override { + owner_->countObject(0); + return moxygen::SubgroupConsumerFilter::endOfGroup(endOfGroupObjectID); + } + + folly::Expected endOfTrackAndGroup(uint64_t endOfTrackObjectID + ) override { + owner_->countObject(0); + return moxygen::SubgroupConsumerFilter::endOfTrackAndGroup(endOfTrackObjectID); + } + + folly::Expected + objectPayload(Payload payload, bool finSubgroup = false) override { + // Continuation of an object already counted by beginObject: bytes only. + owner_->countBytes(payloadLength(payload)); + return moxygen::SubgroupConsumerFilter::objectPayload(std::move(payload), finSubgroup); + } + +private: + std::shared_ptr owner_; +}; + +TrackStatsFilter::TrackStatsFilter( + std::shared_ptr stats, + Direction direction, + std::shared_ptr downstream +) + : moxygen::TrackConsumerFilter(std::move(downstream)), stats_(std::move(stats)), + direction_(direction) { + if (direction_ == Direction::Egress) { + ++stats_->counters().subscribers; + } +} + +// The consumer chain can be released from any thread, so the subscriber +// decrement and the final TrackStats ref drop both hop to the owning thread. +TrackStatsFilter::~TrackStatsFilter() { + auto release = [egress = direction_ == + Direction::Egress](const std::shared_ptr& stats) { + if (egress) { + --stats->counters().subscribers; + } + }; + if (stats_->onOwnerThread()) { + release(stats_); + return; + } + // Must read the executor before moving stats_ out. The ref drop hops too: + // ~TrackStats can only evict its collector entry from the owning thread. + auto* exec = stats_->owningExec(); + if (!exec) { + return; + } + exec->add([stats = std::move(stats_), release] { release(stats); }); +} + +folly::Expected, MoQPublishError> TrackStatsFilter::beginSubgroup( + uint64_t groupID, + uint64_t subgroupID, + moxygen::Priority priority, + moxygen::BeginSubgroupOptions options +) { + auto res = moxygen::TrackConsumerFilter::beginSubgroup(groupID, subgroupID, priority, options); + if (!res) { + return res; + } + countGroupIfNew(groupID); + countSubgroup(); + return std::static_pointer_cast(std::make_shared( + std::static_pointer_cast(shared_from_this()), + std::move(res.value()) + )); +} + +folly::Expected +TrackStatsFilter::objectStream(const ObjectHeader& header, Payload payload, bool lastInGroup) { + countGroupIfNew(header.group); + countSubgroup(); + countObject(payloadLength(payload)); + return moxygen::TrackConsumerFilter::objectStream(header, std::move(payload), lastInGroup); +} + +folly::Expected +TrackStatsFilter::datagram(const ObjectHeader& header, Payload payload, bool lastInGroup) { + countGroupIfNew(header.group); + countDatagram(payloadLength(payload)); + return moxygen::TrackConsumerFilter::datagram(header, std::move(payload), lastInGroup); +} + +// LRU rather than insertion order: a group that keeps receiving subgroups stays +// in the window however many other groups open alongside it. +void TrackStatsFilter::countGroupIfNew(uint64_t groupID) { + for (size_t i = 0; i < recentCount_; ++i) { + if (recentGroups_[i] == groupID) { + std::rotate(recentGroups_.begin(), recentGroups_.begin() + i, recentGroups_.begin() + i + 1); + return; + } + } + recentCount_ = std::min(recentCount_ + 1, kRecentGroups); + std::rotate(recentGroups_.begin(), recentGroups_.end() - 1, recentGroups_.end()); + recentGroups_[0] = groupID; + ++stats_->counters().forDirection(direction_).groups; +} + +void TrackStatsFilter::countSubgroup() { + ++stats_->counters().forDirection(direction_).subgroups; +} + +void TrackStatsFilter::countObject(uint64_t bytes) { + auto& counters = stats_->counters(); + ++counters.forDirection(direction_).objects; + // Sampled on the object header only: payload continuations reach countBytes + // directly, and re-stamping per chunk would not be a last-*object* time. + if (direction_ == Direction::Ingest) { + counters.lastObject = stats::TrackClock::now(); + } + countBytes(bytes); +} + +void TrackStatsFilter::countDatagram(uint64_t bytes) { + ++stats_->counters().forDirection(direction_).datagrams; + countObject(bytes); +} + +void TrackStatsFilter::countBytes(uint64_t bytes) { + stats_->counters().forDirection(direction_).bytes += bytes; +} + +std::shared_ptr wrapWithTrackStats( + stats::TrackStatsRegistry& trackStats, + const moxygen::FullTrackName& ftn, + std::shared_ptr downstream, + stats::TrackDirection direction +) { + auto* collector = trackStats.currentCollector(); + if (!collector) { + return downstream; + } + return std::static_pointer_cast(std::make_shared( + collector->getOrCreate(ftn), + direction, + std::move(downstream) + )); +} + +} // namespace openmoq::moqx diff --git a/src/relay/TrackStatsFilter.h b/src/relay/TrackStatsFilter.h new file mode 100644 index 00000000..4a87c8b7 --- /dev/null +++ b/src/relay/TrackStatsFilter.h @@ -0,0 +1,92 @@ +/* + * Copyright (c) OpenMOQ contributors. + * This source code is licensed under the Apache 2.0 license found in the + * LICENSE file in the root directory of this source tree. + */ + +#pragma once + +#include +#include +#include + +#include + +#include "stats/TrackStatsRegistry.h" + +namespace openmoq::moqx { + +// Counts objects flowing through one point of the data plane into the +// per-thread TrackStats for its track. Installed twice: once on the relay +// chain (ingest) and once per downstream subscriber (egress). +// +// Constructed on the thread that will run it — it caches the TrackStats +// pointer and never consults the collector again. +class TrackStatsFilter : public moxygen::TrackConsumerFilter, + public std::enable_shared_from_this { +public: + using Direction = stats::TrackDirection; + + TrackStatsFilter( + std::shared_ptr stats, + Direction direction, + std::shared_ptr downstream + ); + + ~TrackStatsFilter() override; + + folly::Expected, moxygen::MoQPublishError> + beginSubgroup( + uint64_t groupID, + uint64_t subgroupID, + moxygen::Priority priority, + moxygen::BeginSubgroupOptions options = {} + ) override; + + folly::Expected objectStream( + const moxygen::ObjectHeader& header, + moxygen::Payload payload, + bool lastInGroup = false + ) override; + + folly::Expected + datagram(const moxygen::ObjectHeader& header, moxygen::Payload payload, bool lastInGroup = false) + override; + + // Bounds how deep an interleave can be before a group is counted twice: + // subgroups of the N most recent groups can arrive in any order. + static constexpr size_t kRecentGroups = 3; + + // Lookup and LRU update are both linear scans on the data path, so this stays + // an array rather than a map only while the window is tiny. + static_assert( + kRecentGroups <= 8, + "widening past this wants folly::findFixed (or a map), not a longer scan" + ); + +private: + friend class TrackStatsSubgroupFilter; + + void countGroupIfNew(uint64_t groupID); + void countSubgroup(); + void countObject(uint64_t bytes); + void countDatagram(uint64_t bytes); + void countBytes(uint64_t bytes); + + std::shared_ptr stats_; + Direction direction_; + // Most-recently-seen first. + std::array recentGroups_{}; + size_t recentCount_{0}; +}; + +// Returns downstream unwrapped when no collector is bound on this thread, so +// tests and benchmarks that skip binding still publish. +std::shared_ptr wrapWithTrackStats( + stats::TrackStatsRegistry& trackStats, + const moxygen::FullTrackName& ftn, + std::shared_ptr downstream, + stats::TrackDirection direction +); + +} // namespace openmoq::moqx diff --git a/src/stats/PrometheusFormat.h b/src/stats/PrometheusFormat.h new file mode 100644 index 00000000..52a6fa19 --- /dev/null +++ b/src/stats/PrometheusFormat.h @@ -0,0 +1,72 @@ +/* + * Copyright (c) OpenMOQ contributors. + * This source code is licensed under the Apache 2.0 license found in the + * LICENSE file in the root directory of this source tree. + */ + +#pragma once + +#include +#include +#include + +#include +#include +#include +#include + +namespace openmoq::moqx::stats { + +// Escapes a label value for the Prometheus text exposition format, which only +// defines \\, \" and \n. +inline std::string escapeLabelValue(std::string_view value) { + std::string out; + out.reserve(value.size()); + for (char c : value) { + switch (c) { + case '\\': + out += "\\\\"; + break; + case '"': + out += "\\\""; + break; + case '\n': + out += "\\n"; + break; + default: + out += c; + } + } + return out; +} + +// Accumulates a text exposition format v0.0.4 response: +// https://prometheus.io/docs/instrumenting/exposition_formats/ +class PrometheusWriter { +public: + void append(std::string_view s) { + appender_.push(reinterpret_cast(s.data()), s.size()); + } + + template void num(T value) { append(folly::to(value)); } + + void header(std::string_view name, std::string_view type, std::string_view help) { + append("# HELP "); + append(name); + append(" "); + append(help); + append("\n# TYPE "); + append(name); + append(" "); + append(type); + append("\n"); + } + + std::unique_ptr move() { return queue_.move(); } + +private: + folly::IOBufQueue queue_{folly::IOBufQueue::cacheChainLength()}; + folly::io::QueueAppender appender_{&queue_, 8192}; +}; + +} // namespace openmoq::moqx::stats diff --git a/src/stats/TrackStats.h b/src/stats/TrackStats.h new file mode 100644 index 00000000..8690cb40 --- /dev/null +++ b/src/stats/TrackStats.h @@ -0,0 +1,226 @@ +/* + * Copyright (c) OpenMOQ contributors. + * This source code is licensed under the Apache 2.0 license found in the + * LICENSE file in the root directory of this source tree. + */ + +#pragma once + +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include + +namespace openmoq::moqx::stats { + +// Coarse clock: sampled per object on the data path, so it must stay cheap. +// Converted to wall clock at scrape time against a same-instant anchor pair. +using TrackClock = folly::chrono::coarse_steady_clock; + +// Which side of the relay a filter counts for. +enum class TrackDirection { Ingest, Egress }; + +struct DirectionCounters { + uint64_t groups{0}; + uint64_t subgroups{0}; + uint64_t objects{0}; + uint64_t datagrams{0}; + uint64_t bytes{0}; + + DirectionCounters& operator+=(const DirectionCounters& o) { + groups += o.groups; + subgroups += o.subgroups; + objects += o.objects; + datagrams += o.datagrams; + bytes += o.bytes; + return *this; + } +}; + +// Per-(track, iothread) counters. Aggregation across threads is operator+=, +// so every field must have a meaningful merge. +struct TrackCounters { + DirectionCounters received; + DirectionCounters sent; + + int64_t subscribers{0}; + + TrackClock::time_point publishStart{}; + TrackClock::time_point lastObject{}; + + DirectionCounters& forDirection(TrackDirection d) { + return d == TrackDirection::Ingest ? received : sent; + } + + TrackCounters& operator+=(const TrackCounters& o) { + received += o.received; + sent += o.sent; + subscribers += o.subscribers; + + // Earliest thread to see the track owns the start; latest owns the tail. + if (o.publishStart != TrackClock::time_point{} && + (publishStart == TrackClock::time_point{} || o.publishStart < publishStart)) { + publishStart = o.publishStart; + } + lastObject = std::max(lastObject, o.lastObject); + return *this; + } +}; + +class TrackStatsCollector; + +// Owned by the filters counting into it. The collector back-reference is weak +// because collectors die with the relay while a filter chain held by +// SubscriptionRegistry can outlive them. +class TrackStats : public std::enable_shared_from_this { +public: + TrackStats(std::weak_ptr collector, moxygen::FullTrackName ftn); + ~TrackStats(); + + TrackStats(const TrackStats&) = delete; + TrackStats& operator=(const TrackStats&) = delete; + + const moxygen::FullTrackName& fullTrackName() const { return ftn_; } + + bool onOwnerThread() const { return std::this_thread::get_id() == owner_; } + + // Consumer chains can be released from any thread, so teardown hops here + // before touching counters or dropping the last reference. Null once the + // owning collector is gone. + folly::Executor* owningExec() const; + + // Filters cache this pointer and never re-consult the collector, so the + // thread check has to live here rather than only on collector lookups. + TrackCounters& counters() { + XDCHECK(onOwnerThread()) << "TrackStats mutated off its owning thread"; + return counters_; + } + + const TrackCounters& counters() const { return counters_; } + +private: + std::weak_ptr collector_; + moxygen::FullTrackName ftn_; + TrackCounters counters_; + std::thread::id owner_{std::this_thread::get_id()}; +}; + +// One instance per (service relay, iothread). All methods must be called on the +// owning thread. +class TrackStatsCollector : public std::enable_shared_from_this { +public: + TrackStatsCollector() = default; + + TrackStatsCollector(const TrackStatsCollector&) = delete; + TrackStatsCollector& operator=(const TrackStatsCollector&) = delete; + + // Collectors are constructed at init and handed to their thread; this claims + // ownership from the thread that will actually use them. + void bindToCurrentThread(folly::Executor* exec) { + owner_ = std::this_thread::get_id(); + exec_ = exec; + } + + folly::Executor* exec() const { return exec_; } + + bool onOwnerThread() const { return std::this_thread::get_id() == owner_; } + + // Shares one TrackStats per track per thread, so ingest and every egress + // filter on this thread count into the same entry. + std::shared_ptr getOrCreate(const moxygen::FullTrackName& ftn) { + checkThread(); + if (auto existing = get(ftn)) { + return existing; + } + auto stats = create(ftn); + stats->counters().publishStart = TrackClock::now(); + return stats; + } + + // Creates an entry even when one already exists, displacing it in the slot. + std::shared_ptr create(const moxygen::FullTrackName& ftn) { + checkThread(); + auto stats = std::make_shared(weak_from_this(), ftn); + stats_[ftn] = TrackRef{stats.get(), stats}; + return stats; + } + + std::shared_ptr get(const moxygen::FullTrackName& ftn) const { + checkThread(); + auto it = stats_.find(ftn); + // An entry can outlive its TrackStats when teardown ran off-thread and + // could not reach us; treat the expired weak ref as absent. + return it != stats_.end() ? it->second.weak.lock() : nullptr; + } + + size_t size() const { + checkThread(); + return stats_.size(); + } + + void forEach(folly::FunctionRef fn) const { + checkThread(); + for (const auto& [ftn, ref] : stats_) { + if (auto stats = ref.weak.lock()) { + fn(*stats); + } + } + } + +private: + friend class TrackStats; + + struct TrackRef { + // Identity for removal; never dereferenced (weak covers liveness). + TrackStats* raw{nullptr}; + std::weak_ptr weak; + }; + + // Identity-checked, so a track that is torn down after a successor claimed + // the same name cannot evict the successor's entry. + void remove(const moxygen::FullTrackName& ftn, const TrackStats* expected) { + checkThread(); + auto it = stats_.find(ftn); + if (it == stats_.end() || it->second.raw != expected) { + return; + } + stats_.erase(it); + } + + void checkThread() const { + XCHECK(onOwnerThread()) << "TrackStatsCollector accessed off its owning thread"; + } + + std::thread::id owner_{std::this_thread::get_id()}; + folly::Executor* exec_{nullptr}; + folly::F14FastMap stats_; +}; + +inline folly::Executor* TrackStats::owningExec() const { + auto collector = collector_.lock(); + return collector ? collector->exec() : nullptr; +} + +inline TrackStats::TrackStats( + std::weak_ptr collector, + moxygen::FullTrackName ftn +) + : collector_(std::move(collector)), ftn_(std::move(ftn)) {} + +// Skips removal when the collector is already gone, or when teardown could not +// reach the owning thread — the weak ref then reports the entry absent. +inline TrackStats::~TrackStats() { + auto collector = collector_.lock(); + if (collector && collector->onOwnerThread()) { + collector->remove(ftn_, this); + } +} + +} // namespace openmoq::moqx::stats diff --git a/src/stats/TrackStatsRegistry.cpp b/src/stats/TrackStatsRegistry.cpp new file mode 100644 index 00000000..d7907761 --- /dev/null +++ b/src/stats/TrackStatsRegistry.cpp @@ -0,0 +1,65 @@ +/* + * Copyright (c) OpenMOQ contributors. + * This source code is licensed under the Apache 2.0 license found in the + * LICENSE file in the root directory of this source tree. + */ + +#include "stats/TrackStatsRegistry.h" + +#include + +#include +#include +#include + +namespace openmoq::moqx::stats { + +void TrackStatsRegistry::bindAll(const std::vector& execs) { + for (auto* exec : execs) { + XCHECK(exec) << "TrackStatsRegistry::bindAll: null executor"; + // Blocks so every thread is bound before serving, and so a queued bind can + // never outlive this registry; mirrors + // MoqxRelayContext::initThreadStatsCollectors. + folly::via(exec, [this, exec] { bind(exec); }).get(); + } +} + +void TrackStatsRegistry::bind(folly::Executor* exec) { + XCHECK(!*tlCollector_) << "track stats already bound on this thread"; + auto collector = std::make_shared(); + collector->bindToCurrentThread(exec); + collectors_.push_back({exec, collector}); + *tlCollector_ = std::move(collector); +} + +folly::coro::Task +TrackStatsRegistry::aggregateAsync(std::vector keys) const { + auto snapshotOne = [](std::shared_ptr collector, + const std::vector& keys + ) -> folly::coro::Task { + TrackCountersMap partial; + for (const auto& ftn : keys) { + if (auto stats = collector->get(ftn)) { + partial[ftn] = stats->counters(); + } + } + co_return partial; + }; + + std::vector> tasks; + tasks.reserve(collectors_.size()); + for (const auto& bound : collectors_) { + tasks.push_back(folly::coro::co_withExecutor(bound.exec, snapshotOne(bound.collector, keys))); + } + + auto results = co_await folly::coro::collectAllRange(std::move(tasks)); + TrackCountersMap combined; + for (const auto& partial : results) { + for (const auto& [ftn, counters] : partial) { + combined[ftn] += counters; + } + } + co_return combined; +} + +} // namespace openmoq::moqx::stats diff --git a/src/stats/TrackStatsRegistry.h b/src/stats/TrackStatsRegistry.h new file mode 100644 index 00000000..9283279c --- /dev/null +++ b/src/stats/TrackStatsRegistry.h @@ -0,0 +1,54 @@ +/* + * Copyright (c) OpenMOQ contributors. + * This source code is licensed under the Apache 2.0 license found in the + * LICENSE file in the root directory of this source tree. + */ + +#pragma once + +#include +#include + +#include +#include +#include +#include + +#include "stats/TrackStats.h" + +namespace openmoq::moqx::stats { + +using TrackCountersMap = + folly::F14FastMap; + +// Owns one TrackStatsCollector per data-plane thread and merges them on demand. +// +// Collectors are created before serving and never added afterwards, so the +// thread list is read lock-free — the contract StatsRegistry::collectors_ uses. +class TrackStatsRegistry { +public: + // Binds a collector on each executor's thread, blocking until every one is + // bound. Must run before any listener accepts sessions, and not on a thread + // that drives one of execs — the wait would deadlock. + void bindAll(const std::vector& execs); + + // Null on threads that were never bound; callers skip counting rather than + // fail, so tests and benchmarks can leave collectors unbound. + TrackStatsCollector* currentCollector() { return tlCollector_->get(); } + + folly::coro::Task aggregateAsync(std::vector keys + ) const; + +private: + struct BoundCollector { + folly::Executor* exec; + std::shared_ptr collector; + }; + + void bind(folly::Executor* exec); + + folly::ThreadLocal> tlCollector_; + std::vector collectors_; +}; + +} // namespace openmoq::moqx::stats diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt index 40b5f174..2494305a 100644 --- a/test/CMakeLists.txt +++ b/test/CMakeLists.txt @@ -40,6 +40,7 @@ add_executable(moqx_relay_test MoqxRelayNGRTests.cpp MoqxRelayPeerTests.cpp MoqxRelayTracksTests.cpp + MoqxRelayTrackStatsTests.cpp MoqxRelayTestModes.cpp ) target_link_libraries(moqx_relay_test PRIVATE @@ -158,6 +159,45 @@ target_link_libraries(moqx_bounded_histogram_test PRIVATE ) gtest_discover_tests(moqx_bounded_histogram_test) +add_executable(moqx_track_stats_test + stats/TrackStatsTest.cpp +) +target_include_directories(moqx_track_stats_test PRIVATE + ${PROJECT_SOURCE_DIR}/src +) +target_link_libraries(moqx_track_stats_test PRIVATE + moqx_core + GTest::gtest_main + GTest::gmock +) +gtest_discover_tests(moqx_track_stats_test) + +add_executable(moqx_track_metrics_format_test + admin/TrackMetricsFormatTest.cpp +) +target_include_directories(moqx_track_metrics_format_test PRIVATE + ${PROJECT_SOURCE_DIR}/src +) +target_link_libraries(moqx_track_metrics_format_test PRIVATE + moqx_core + GTest::gtest_main + GTest::gmock +) +gtest_discover_tests(moqx_track_metrics_format_test) + +add_executable(moqx_track_stats_registry_test + stats/TrackStatsRegistryTest.cpp +) +target_include_directories(moqx_track_stats_registry_test PRIVATE + ${PROJECT_SOURCE_DIR}/src +) +target_link_libraries(moqx_track_stats_registry_test PRIVATE + moqx_core + GTest::gtest_main + GTest::gmock +) +gtest_discover_tests(moqx_track_stats_registry_test) + add_executable(moqx_logging_multi_flag_test LoggingMultiFlagTest.cpp ) @@ -296,6 +336,10 @@ add_test( NAME admin_metrics_endpoint COMMAND bash ${PROJECT_SOURCE_DIR}/test/test_admin_metrics.sh $ ) +add_test( + NAME admin_track_metrics_endpoint + COMMAND bash ${PROJECT_SOURCE_DIR}/test/test_admin_track_metrics.sh $ +) add_test( NAME admin_cache_purge_endpoint COMMAND bash ${PROJECT_SOURCE_DIR}/test/test_admin_cache_purge.sh $ diff --git a/test/MoqxRelayTrackStatsTests.cpp b/test/MoqxRelayTrackStatsTests.cpp new file mode 100644 index 00000000..3f73e138 --- /dev/null +++ b/test/MoqxRelayTrackStatsTests.cpp @@ -0,0 +1,417 @@ +/* + * Copyright (c) OpenMOQ contributors. + * This source code is licensed under the Apache 2.0 license found in the + * LICENSE file in the root directory of this source tree. + */ + +#include "MoqxRelayTestFixture.h" + +#include +#include + +#include "relay/TrackStatsFilter.h" + +namespace openmoq::moqx::test { + +namespace { + +Payload makePayload(size_t bytes) { + return folly::IOBuf::copyBuffer(std::string(bytes, 'x')); +} + +} // namespace + +class MoQRelayTrackStatsTest : public MoQRelayTest { +protected: + // Only this fixture counts, so the registry is bound here rather than for + // every relay test. bindAll blocks until each executor's thread runs the + // bind, and this thread is what drives exec_, so pump it from here while a + // helper waits. + void SetUp() override { + MoQRelayTest::SetUp(); + std::vector execs{static_cast(exec_.get())}; + if (relayEvb_) { + execs.push_back(relay_->getRelayExec()); + } + std::atomic done{false}; + std::thread binder([&] { + relay_->trackStatsRegistry().bindAll(execs); + done = true; + // Wakes the pump if it is already parked in drive() on an empty queue. + exec_->add([] {}); + }); + while (!done) { + exec_->drive(); + } + binder.join(); + } + + // Which thread counts what varies by mode — ingest is on the relay exec in + // MT/LF, egress on the relay exec in MT but the subscriber thread in LF — so + // merge every thread's counters, exactly as aggregateTrackStats does. + stats::TrackCounters trackCounters(const FullTrackName& ftn) { + auto merged = countersOnThisThread(ftn); + if (relayEvb_) { + stats::TrackCounters relayCounters; + verifyOnRelayExec([&] { relayCounters = countersOnThisThread(ftn); }); + merged += relayCounters; + } + return merged; + } + +private: + stats::TrackCounters countersOnThisThread(const FullTrackName& ftn) { + auto* collector = relay_->trackStatsRegistry().currentCollector(); + EXPECT_NE(collector, nullptr); + auto stats = collector->get(ftn); + return stats ? stats->counters() : stats::TrackCounters{}; + } +}; + +TEST_P(MoQRelayTrackStatsTest, CountsIngestObjectsAndBytes) { + auto publisherSession = createMockSession(); + auto subscriber = createMockSession(); + auto mockConsumer = createMockConsumer(); + auto mockSg = createMockSubgroupConsumer(); + + EXPECT_CALL(*mockConsumer, beginSubgroup(0, 0, _, _)) + .WillRepeatedly([&](uint64_t, uint64_t, uint8_t, moxygen::BeginSubgroupOptions) { + return folly::makeExpected>(mockSg); + }); + EXPECT_CALL(*mockSg, object(_, _, _, _)) + .WillRepeatedly(Return(folly::makeExpected(folly::unit))); + + auto publishConsumer = doPublish(publisherSession, kTestTrackName); + subscribeToTrack(subscriber, kTestTrackName, mockConsumer, RequestID(1)); + + auto sg = publishConsumer->beginSubgroup(0, 0, 0); + ASSERT_TRUE(sg.hasValue()); + EXPECT_TRUE(sg.value()->object(0, makePayload(10)).hasValue()); + EXPECT_TRUE(sg.value()->object(1, makePayload(20)).hasValue()); + EXPECT_TRUE(sg.value()->endOfSubgroup().hasValue()); + driveIfMultiThread(); + + auto ingest = trackCounters(kTestTrackName); + EXPECT_EQ(ingest.received.groups, 1); + EXPECT_EQ(ingest.received.subgroups, 1); + EXPECT_EQ(ingest.received.objects, 2); + EXPECT_EQ(ingest.received.bytes, 30); + EXPECT_NE(ingest.publishStart, stats::TrackClock::time_point{}); + EXPECT_NE(ingest.lastObject, stats::TrackClock::time_point{}); + + removeSession(publisherSession); + removeSession(subscriber); + exec_->drive(); +} + +TEST_P(MoQRelayTrackStatsTest, CountsGroupsByTransition) { + auto publisherSession = createMockSession(); + auto subscriber = createMockSession(); + auto mockConsumer = createMockConsumer(); + auto mockSg = createMockSubgroupConsumer(); + + EXPECT_CALL(*mockConsumer, beginSubgroup(_, _, _, _)) + .WillRepeatedly([&](uint64_t, uint64_t, uint8_t, moxygen::BeginSubgroupOptions) { + return folly::makeExpected>(mockSg); + }); + EXPECT_CALL(*mockSg, object(_, _, _, _)) + .WillRepeatedly(Return(folly::makeExpected(folly::unit))); + + auto publishConsumer = doPublish(publisherSession, kTestTrackName); + subscribeToTrack(subscriber, kTestTrackName, mockConsumer, RequestID(1)); + + for (uint64_t group = 0; group < 3; ++group) { + auto sg = publishConsumer->beginSubgroup(group, 0, 0); + ASSERT_TRUE(sg.hasValue()) << "group " << group << ": " << sg.error().describe(); + EXPECT_TRUE(sg.value()->object(0, makePayload(5)).hasValue()); + EXPECT_TRUE(sg.value()->endOfSubgroup().hasValue()); + driveIfMultiThread(); + } + + auto ingest = trackCounters(kTestTrackName); + EXPECT_EQ(ingest.received.groups, 3); + EXPECT_EQ(ingest.received.subgroups, 3); + EXPECT_EQ(ingest.received.objects, 3); + EXPECT_EQ(ingest.received.bytes, 15); + + removeSession(publisherSession); + removeSession(subscriber); + exec_->drive(); +} + +TEST_P(MoQRelayTrackStatsTest, CountsEgressPerSubscriber) { + auto publisherSession = createMockSession(); + auto sub1 = createMockSession(); + auto sub2 = createMockSession(); + auto mockConsumer1 = createMockConsumer(); + auto mockConsumer2 = createMockConsumer(); + auto mockSg1 = createMockSubgroupConsumer(); + auto mockSg2 = createMockSubgroupConsumer(); + + EXPECT_CALL(*mockConsumer1, beginSubgroup(0, 0, _, _)) + .WillRepeatedly([&](uint64_t, uint64_t, uint8_t, moxygen::BeginSubgroupOptions) { + return folly::makeExpected>(mockSg1); + }); + EXPECT_CALL(*mockConsumer2, beginSubgroup(0, 0, _, _)) + .WillRepeatedly([&](uint64_t, uint64_t, uint8_t, moxygen::BeginSubgroupOptions) { + return folly::makeExpected>(mockSg2); + }); + EXPECT_CALL(*mockSg1, object(_, _, _, _)) + .WillRepeatedly(Return(folly::makeExpected(folly::unit))); + EXPECT_CALL(*mockSg2, object(_, _, _, _)) + .WillRepeatedly(Return(folly::makeExpected(folly::unit))); + + auto publishConsumer = doPublish(publisherSession, kTestTrackName); + subscribeToTrack(sub1, kTestTrackName, mockConsumer1, RequestID(1)); + subscribeToTrack(sub2, kTestTrackName, mockConsumer2, RequestID(2)); + driveIfMultiThread(); + + EXPECT_EQ(trackCounters(kTestTrackName).subscribers, 2); + + auto sg = publishConsumer->beginSubgroup(0, 0, 0); + ASSERT_TRUE(sg.hasValue()); + EXPECT_TRUE(sg.value()->object(0, makePayload(10)).hasValue()); + EXPECT_TRUE(sg.value()->endOfSubgroup().hasValue()); + driveIfMultiThread(); + + // One object fans out to both subscribers. + auto egress = trackCounters(kTestTrackName); + EXPECT_EQ(egress.sent.objects, 2); + EXPECT_EQ(egress.sent.bytes, 20); + EXPECT_EQ(egress.sent.subgroups, 2); + EXPECT_EQ(egress.sent.groups, 2); + + removeSession(publisherSession); + removeSession(sub1); + removeSession(sub2); + exec_->drive(); +} + +TEST_P(MoQRelayTrackStatsTest, SubscriberGaugeDropsOnUnsubscribe) { + auto publisherSession = createMockSession(); + auto subscriber = createMockSession(); + auto mockConsumer = createMockConsumer(); + + auto publishConsumer = doPublish(publisherSession, kTestTrackName); + subscribeToTrack(subscriber, kTestTrackName, mockConsumer, RequestID(1)); + driveIfMultiThread(); + EXPECT_EQ(trackCounters(kTestTrackName).subscribers, 1); + + removeSession(subscriber); + exec_->drive(); + EXPECT_EQ(trackCounters(kTestTrackName).subscribers, 0); + + removeSession(publisherSession); + exec_->drive(); +} + +TEST_P(MoQRelayTrackStatsTest, MatchTracksFiltersAndReportsTruncation) { + auto publisherSession = createMockSession(); + const FullTrackName other{kTestNamespace, "track2"}; + auto publishConsumer1 = doPublish(publisherSession, kTestTrackName); + auto publishConsumer2 = doPublish(publisherSession, other); + driveIfMultiThread(); + + verifyOnRelayExec([&] { + auto all = relay_->matchTracks(kTestNamespace, nullptr, 10); + EXPECT_EQ(all.matched, 2); + EXPECT_EQ(all.keys.size(), 2); + + auto exact = relay_->matchTracks(kTestNamespace, &kTestTrackName.trackName, 10); + EXPECT_EQ(exact.matched, 1); + ASSERT_EQ(exact.keys.size(), 1); + EXPECT_EQ(exact.keys[0], kTestTrackName); + + auto limited = relay_->matchTracks(kTestNamespace, nullptr, 1); + EXPECT_EQ(limited.matched, 2); + EXPECT_EQ(limited.keys.size(), 1); + + const TrackNamespace unrelated{{"test", "other"}}; + EXPECT_EQ(relay_->matchTracks(unrelated, nullptr, 10).matched, 0); + }); + + removeSession(publisherSession); + exec_->drive(); +} + +// The relay also fans out via PUBLISH (SUBSCRIBE_NAMESPACE with the publish +// option, TRACK_FILTER selection), which reaches subscribers through +// startPublish rather than the subscribe path. +TEST_P(MoQRelayTrackStatsTest, CountsEgressOnPublishFanout) { + auto publisherSession = createMockSession(); + auto subscriber = createMockSession(); + auto mockConsumer = createMockConsumer(); + auto mockSg = createMockSubgroupConsumer(); + std::atomic published{false}; + + EXPECT_CALL(*subscriber, publish(testing::_, testing::_)) + .WillOnce([&](const PublishRequest&, auto /*subHandle*/) { + published.store(true); + return Subscriber::PublishResult(Subscriber::PublishConsumerAndReplyTask{ + mockConsumer, + []() -> folly::coro::Task> { + co_return PublishOk{ + RequestID(1), + true, + 0, + GroupOrder::OldestFirst, + LocationType::LargestObject, + std::nullopt, + std::nullopt + }; + }() + }); + }); + EXPECT_CALL(*mockConsumer, beginSubgroup(0, 0, _, _)) + .WillRepeatedly([&](uint64_t, uint64_t, uint8_t, moxygen::BeginSubgroupOptions) { + return folly::makeExpected>(mockSg); + }); + EXPECT_CALL(*mockSg, object(_, _, _, _)) + .WillRepeatedly(Return(folly::makeExpected(folly::unit))); + + doSubscribeNamespace(subscriber, kTestNamespace); + auto publishConsumer = doPublish(publisherSession, kTestTrackName); + ASSERT_TRUE(driveUntil([&] { return published.load(); })) + << "publish was not forwarded to the subscriber"; + + EXPECT_EQ(trackCounters(kTestTrackName).subscribers, 1); + + auto sg = publishConsumer->beginSubgroup(0, 0, 0); + ASSERT_TRUE(sg.hasValue()) << sg.error().describe(); + EXPECT_TRUE(sg.value()->object(0, makePayload(7)).hasValue()); + EXPECT_TRUE(sg.value()->endOfSubgroup().hasValue()); + driveIfMultiThread(); + + auto counters = trackCounters(kTestTrackName); + EXPECT_EQ(counters.received.objects, 1); + EXPECT_EQ(counters.sent.objects, 1); + EXPECT_EQ(counters.sent.bytes, 7); + + removeSession(publisherSession); + removeSession(subscriber); + exec_->drive(); +} + +// Subgroups of concurrently-open groups interleave; the window keeps both. +TEST_P(MoQRelayTrackStatsTest, CountsInterleavedGroupsOnce) { + auto publisherSession = createMockSession(); + auto subscriber = createMockSession(); + auto mockConsumer = createMockConsumer(); + auto mockSg = createMockSubgroupConsumer(); + + EXPECT_CALL(*mockConsumer, beginSubgroup(_, _, _, _)) + .WillRepeatedly([&](uint64_t, uint64_t, uint8_t, moxygen::BeginSubgroupOptions) { + return folly::makeExpected>(mockSg); + }); + EXPECT_CALL(*mockSg, object(_, _, _, _)) + .WillRepeatedly(Return(folly::makeExpected(folly::unit))); + + auto publishConsumer = doPublish(publisherSession, kTestTrackName); + subscribeToTrack(subscriber, kTestTrackName, mockConsumer, RequestID(1)); + + // G0/S0, G1/S0, G0/S1, G1/S1. + for (uint64_t subgroup = 0; subgroup < 2; ++subgroup) { + for (uint64_t group = 0; group < 2; ++group) { + auto sg = publishConsumer->beginSubgroup(group, subgroup, 0); + ASSERT_TRUE(sg.hasValue()) << sg.error().describe(); + EXPECT_TRUE(sg.value()->object(0, makePayload(1)).hasValue()); + EXPECT_TRUE(sg.value()->endOfSubgroup().hasValue()); + driveIfMultiThread(); + } + } + + auto counters = trackCounters(kTestTrackName); + EXPECT_EQ(counters.received.groups, 2); + EXPECT_EQ(counters.received.subgroups, 4); + + removeSession(publisherSession); + removeSession(subscriber); + exec_->drive(); +} + +// LRU, not insertion order: a group still receiving subgroups survives more +// than kRecentGroups other groups opening alongside it. +TEST_P(MoQRelayTrackStatsTest, ActiveGroupSurvivesWindowChurn) { + auto publisherSession = createMockSession(); + auto subscriber = createMockSession(); + auto mockConsumer = createMockConsumer(); + auto mockSg = createMockSubgroupConsumer(); + + EXPECT_CALL(*mockConsumer, beginSubgroup(_, _, _, _)) + .WillRepeatedly([&](uint64_t, uint64_t, uint8_t, moxygen::BeginSubgroupOptions) { + return folly::makeExpected>(mockSg); + }); + + auto publishConsumer = doPublish(publisherSession, kTestTrackName); + subscribeToTrack(subscriber, kTestTrackName, mockConsumer, RequestID(1)); + + const uint64_t kLongLived = 0; + const size_t window = TrackStatsFilter::kRecentGroups; + auto openSubgroup = [&](uint64_t group, uint64_t subgroup) { + auto sg = publishConsumer->beginSubgroup(group, subgroup, 0); + ASSERT_TRUE(sg.hasValue()) << sg.error().describe(); + EXPECT_TRUE(sg.value()->endOfSubgroup().hasValue()); + driveIfMultiThread(); + }; + + openSubgroup(kLongLived, 0); + // Each new group is followed by another subgroup of the long-lived group, + // refreshing it; under insertion-order eviction it would have aged out. + for (uint64_t group = 1; group <= window; ++group) { + openSubgroup(group, 0); + openSubgroup(kLongLived, group); + } + + EXPECT_EQ(trackCounters(kTestTrackName).received.groups, window + 1); + + removeSession(publisherSession); + removeSession(subscriber); + exec_->drive(); +} + +// Past the window a revisited group counts again — the bound is deliberate. +TEST_P(MoQRelayTrackStatsTest, RecountsGroupEvictedFromWindow) { + auto publisherSession = createMockSession(); + auto subscriber = createMockSession(); + auto mockConsumer = createMockConsumer(); + auto mockSg = createMockSubgroupConsumer(); + + EXPECT_CALL(*mockConsumer, beginSubgroup(_, _, _, _)) + .WillRepeatedly([&](uint64_t, uint64_t, uint8_t, moxygen::BeginSubgroupOptions) { + return folly::makeExpected>(mockSg); + }); + + auto publishConsumer = doPublish(publisherSession, kTestTrackName); + subscribeToTrack(subscriber, kTestTrackName, mockConsumer, RequestID(1)); + + const size_t window = TrackStatsFilter::kRecentGroups; + for (uint64_t group = 0; group <= window; ++group) { + auto sg = publishConsumer->beginSubgroup(group, 0, 0); + ASSERT_TRUE(sg.hasValue()) << sg.error().describe(); + EXPECT_TRUE(sg.value()->endOfSubgroup().hasValue()); + driveIfMultiThread(); + } + auto sg = publishConsumer->beginSubgroup(0, 1, 0); + ASSERT_TRUE(sg.hasValue()) << sg.error().describe(); + EXPECT_TRUE(sg.value()->endOfSubgroup().hasValue()); + driveIfMultiThread(); + + EXPECT_EQ(trackCounters(kTestTrackName).received.groups, window + 2); + + removeSession(publisherSession); + removeSession(subscriber); + exec_->drive(); +} + +INSTANTIATE_TEST_SUITE_P( + AllModes, + MoQRelayTrackStatsTest, + ::testing::Values(RelayMode::SingleThread, RelayMode::MultiThread, RelayMode::LocalForwarderMT), + [](const ::testing::TestParamInfo& info) { + std::ostringstream os; + PrintTo(info.param, &os); + return os.str(); + } +); + +} // namespace openmoq::moqx::test diff --git a/test/admin/TrackMetricsFormatTest.cpp b/test/admin/TrackMetricsFormatTest.cpp new file mode 100644 index 00000000..4e400eab --- /dev/null +++ b/test/admin/TrackMetricsFormatTest.cpp @@ -0,0 +1,107 @@ +/* + * Copyright (c) OpenMOQ contributors. + * This source code is licensed under the Apache 2.0 license found in the + * LICENSE file in the root directory of this source tree. + */ + +#include +#include + +#include "admin/TrackMetricsHandler.h" + +namespace openmoq::moqx::admin { + +namespace { + +MoqxRelayContext::TrackMetricsResult +makeResult(stats::TrackCounters counters, std::string service) { + MoqxRelayContext::TrackMetricsResult result; + result.matched = 1; + result.tracks.push_back( + {std::move(service), + moxygen::FullTrackName{moxygen::TrackNamespace("ns", "/"), "track"}, + counters} + ); + return result; +} + +std::string format(const MoqxRelayContext::TrackMetricsResult& result) { + auto body = formatTrackMetrics(result); + return body->moveToFbString().toStdString(); +} + +} // namespace + +TEST(TrackMetricsFormatTest, EmitsCountersFromBothDirections) { + stats::TrackCounters counters; + counters.received.objects = 7; + counters.received.bytes = 100; + counters.sent.objects = 21; + counters.subscribers = 3; + + auto out = format(makeResult(counters, "svc")); + + EXPECT_NE( + out.find( + "moqx_track_objects_received_total{service=\"svc\",namespace=\"ns\",track=\"track\"} 7\n" + ), + std::string::npos + ); + EXPECT_NE( + out.find( + "moqx_track_bytes_received_total{service=\"svc\",namespace=\"ns\",track=\"track\"} 100\n" + ), + std::string::npos + ); + EXPECT_NE( + out.find( + "moqx_track_objects_sent_total{service=\"svc\",namespace=\"ns\",track=\"track\"} 21\n" + ), + std::string::npos + ); + EXPECT_NE( + out.find("moqx_track_subscribers{service=\"svc\",namespace=\"ns\",track=\"track\"} 3\n"), + std::string::npos + ); +} + +// Prometheus expresses time in seconds, so millisecond resolution has to be a +// fraction rather than a different unit or a renamed metric. +TEST(TrackMetricsFormatTest, TimestampsCarryMillisecondFraction) { + stats::TrackCounters counters; + counters.publishStart = stats::TrackClock::now(); + counters.lastObject = counters.publishStart; + + auto out = format(makeResult(counters, "svc")); + + std::regex lastObject{R"(moqx_track_last_object_timestamp_seconds\{[^}]*\} (\d{10,})\.(\d{3})\n)" + }; + std::smatch match; + ASSERT_TRUE(std::regex_search(out, match, lastObject)) << out; + EXPECT_EQ(match[2].str().size(), 3u); + + std::regex publishStart{R"(moqx_track_publish_start_timestamp_seconds\{[^}]*\} \d{10,}\.\d{3}\n)" + }; + EXPECT_TRUE(std::regex_search(out, publishStart)) << out; +} + +// A track that has not seen an object has no meaningful timestamp, and 0 would +// render as 1970 in any time()-based panel. +TEST(TrackMetricsFormatTest, OmitsUnsetTimestamps) { + stats::TrackCounters counters; + counters.received.objects = 1; + + auto out = format(makeResult(counters, "svc")); + + EXPECT_NE(out.find("# TYPE moqx_track_last_object_timestamp_seconds gauge"), std::string::npos); + EXPECT_EQ(out.find("moqx_track_last_object_timestamp_seconds{"), std::string::npos); + EXPECT_EQ(out.find("moqx_track_publish_start_timestamp_seconds{"), std::string::npos); +} + +TEST(TrackMetricsFormatTest, EscapesServiceLabel) { + auto out = format(makeResult(stats::TrackCounters{}, "a\"b\\c")); + + EXPECT_NE(out.find(R"(service="a\"b\\c")"), std::string::npos) << out; +} + +} // namespace openmoq::moqx::admin diff --git a/test/config/ConfigSerializerTest.cpp b/test/config/ConfigSerializerTest.cpp index 3c14a244..876b3dfa 100644 --- a/test/config/ConfigSerializerTest.cpp +++ b/test/config/ConfigSerializerTest.cpp @@ -89,7 +89,7 @@ static_assert( "UpstreamTlsConfig changed — update serializeUpstream()" ); static_assert( - rfl::internal::num_fields == 2, + rfl::internal::num_fields == 5, "AdminConfig changed — update serializeConfig()" ); static_assert( diff --git a/test/make_test_config.sh b/test/make_test_config.sh index 4c9f154c..443ec6ad 100755 --- a/test/make_test_config.sh +++ b/test/make_test_config.sh @@ -52,6 +52,7 @@ services: admin: port: ${ADMIN_PORT} address: "::1" + track_metrics_enabled: true EOF if [[ -n "$CERT" ]]; then diff --git a/test/stats/TrackStatsRegistryTest.cpp b/test/stats/TrackStatsRegistryTest.cpp new file mode 100644 index 00000000..066725d9 --- /dev/null +++ b/test/stats/TrackStatsRegistryTest.cpp @@ -0,0 +1,116 @@ +/* + * Copyright (c) OpenMOQ contributors. + * This source code is licensed under the Apache 2.0 license found in the + * LICENSE file in the root directory of this source tree. + */ + +#include +#include +#include + +#include "stats/TrackStatsRegistry.h" + +namespace openmoq::moqx::stats { + +namespace { + +moxygen::FullTrackName makeFtn(std::string ns, std::string name) { + return moxygen::FullTrackName{moxygen::TrackNamespace(std::move(ns), "/"), std::move(name)}; +} + +} // namespace + +class TrackStatsRegistryTest : public ::testing::Test { +protected: + void SetUp() override { registry_.bindAll({threadA_.getEventBase(), threadB_.getEventBase()}); } + + // Counts on the thread that owns the collector, as the data-plane filters do, + // and holds the entry alive the way a filter in a chain would. + void countOn(folly::EventBase* evb, const moxygen::FullTrackName& ftn, uint64_t objects) { + evb->runInEventBaseThreadAndWait([&] { + auto* collector = registry_.currentCollector(); + ASSERT_NE(collector, nullptr); + auto stats = collector->getOrCreate(ftn); + stats->counters().received.objects += objects; + stats->counters().subscribers += 1; + stats->counters().lastObject = TrackClock::now(); + held_.push_back(std::move(stats)); + }); + } + + TrackCounters countersOn(folly::EventBase* evb, const moxygen::FullTrackName& ftn) { + TrackCounters counters; + evb->runInEventBaseThreadAndWait([&] { + if (auto stats = registry_.currentCollector()->get(ftn)) { + counters = stats->counters(); + } + }); + return counters; + } + + folly::ScopedEventBaseThread threadA_{"track-stats-a"}; + folly::ScopedEventBaseThread threadB_{"track-stats-b"}; + TrackStatsRegistry registry_; + std::vector> held_; +}; + +TEST_F(TrackStatsRegistryTest, MergesCountersAcrossThreads) { + auto ftn = makeFtn("ns", "track"); + countOn(threadA_.getEventBase(), ftn, 3); + countOn(threadB_.getEventBase(), ftn, 4); + + auto merged = folly::coro::blockingWait(registry_.aggregateAsync({ftn})); + + ASSERT_EQ(merged.size(), 1); + EXPECT_EQ(merged[ftn].received.objects, 7); + EXPECT_EQ(merged[ftn].subscribers, 2); +} + +TEST_F(TrackStatsRegistryTest, ReportsOnlyRequestedKeys) { + auto wanted = makeFtn("ns", "wanted"); + auto other = makeFtn("ns", "other"); + countOn(threadA_.getEventBase(), wanted, 1); + countOn(threadA_.getEventBase(), other, 5); + + auto merged = folly::coro::blockingWait(registry_.aggregateAsync({wanted})); + + EXPECT_EQ(merged.size(), 1); + EXPECT_EQ(merged[wanted].received.objects, 1); +} + +TEST_F(TrackStatsRegistryTest, UnknownKeyIsAbsentRatherThanZero) { + auto merged = + folly::coro::blockingWait(registry_.aggregateAsync({makeFtn("ns", "never-published")})); + EXPECT_TRUE(merged.empty()); +} + +TEST_F(TrackStatsRegistryTest, MergeTakesEarliestStartAndLatestObject) { + auto ftn = makeFtn("ns", "track"); + countOn(threadA_.getEventBase(), ftn, 1); + countOn(threadB_.getEventBase(), ftn, 1); + + auto a = countersOn(threadA_.getEventBase(), ftn); + auto b = countersOn(threadB_.getEventBase(), ftn); + auto merged = folly::coro::blockingWait(registry_.aggregateAsync({ftn})); + + EXPECT_EQ(merged[ftn].publishStart, std::min(a.publishStart, b.publishStart)); + EXPECT_EQ(merged[ftn].lastObject, std::max(a.lastObject, b.lastObject)); +} + +TEST_F(TrackStatsRegistryTest, EntryDisappearsWhenLastRefDropped) { + auto ftn = makeFtn("ns", "track"); + countOn(threadA_.getEventBase(), ftn, 1); + ASSERT_EQ(folly::coro::blockingWait(registry_.aggregateAsync({ftn})).size(), 1); + + threadA_.getEventBase()->runInEventBaseThreadAndWait([&] { held_.clear(); }); + + EXPECT_TRUE(folly::coro::blockingWait(registry_.aggregateAsync({ftn})).empty()); +} + +TEST_F(TrackStatsRegistryTest, UnboundRegistryCountsNothing) { + TrackStatsRegistry unbound; + EXPECT_EQ(unbound.currentCollector(), nullptr); + EXPECT_TRUE(folly::coro::blockingWait(unbound.aggregateAsync({makeFtn("ns", "track")})).empty()); +} + +} // namespace openmoq::moqx::stats diff --git a/test/stats/TrackStatsTest.cpp b/test/stats/TrackStatsTest.cpp new file mode 100644 index 00000000..ce0c1360 --- /dev/null +++ b/test/stats/TrackStatsTest.cpp @@ -0,0 +1,130 @@ +/* + * Copyright (c) OpenMOQ contributors. + * This source code is licensed under the Apache 2.0 license found in the + * LICENSE file in the root directory of this source tree. + */ + +#include + +#include "stats/TrackStats.h" + +namespace openmoq::moqx::stats { + +namespace { + +moxygen::FullTrackName makeFtn(std::string ns, std::string name) { + return moxygen::FullTrackName{moxygen::TrackNamespace(std::move(ns), "/"), std::move(name)}; +} + +} // namespace + +TEST(TrackCountersTest, MergeSumsCounters) { + TrackCounters a; + a.received.objects = 3; + a.received.bytes = 100; + a.sent.objects = 6; + a.subscribers = 2; + + TrackCounters b; + b.received.objects = 4; + b.received.bytes = 50; + b.sent.objects = 8; + b.subscribers = 1; + + a += b; + + EXPECT_EQ(a.received.objects, 7); + EXPECT_EQ(a.received.bytes, 150); + EXPECT_EQ(a.sent.objects, 14); + EXPECT_EQ(a.subscribers, 3); +} + +TEST(TrackCountersTest, MergeTakesEarliestStartAndLatestObject) { + auto t0 = TrackClock::now(); + auto t1 = t0 + std::chrono::seconds(5); + + TrackCounters a; + a.publishStart = t1; + a.lastObject = t0; + + TrackCounters b; + b.publishStart = t0; + b.lastObject = t1; + + a += b; + + EXPECT_EQ(a.publishStart, t0); + EXPECT_EQ(a.lastObject, t1); +} + +TEST(TrackCountersTest, MergeIgnoresUnsetStart) { + auto t0 = TrackClock::now(); + + TrackCounters a; + a.publishStart = t0; + + a += TrackCounters{}; + + EXPECT_EQ(a.publishStart, t0); +} + +TEST(TrackStatsCollectorTest, GetOrCreateSharesOneEntryPerTrack) { + auto collectorPtr = std::make_shared(); + auto& collector = *collectorPtr; + auto ftn = makeFtn("ns", "track"); + + auto first = collector.getOrCreate(ftn); + auto second = collector.getOrCreate(ftn); + + EXPECT_EQ(first.get(), second.get()); + EXPECT_EQ(collector.size(), 1); + EXPECT_NE(first->counters().publishStart, TrackClock::time_point{}); +} + +TEST(TrackStatsCollectorTest, EntryRemovedWhenLastRefDropped) { + auto collectorPtr = std::make_shared(); + auto& collector = *collectorPtr; + auto ftn = makeFtn("ns", "track"); + + { + auto stats = collector.getOrCreate(ftn); + EXPECT_EQ(collector.size(), 1); + EXPECT_NE(collector.get(ftn), nullptr); + } + + EXPECT_EQ(collector.size(), 0); + EXPECT_EQ(collector.get(ftn), nullptr); +} + +TEST(TrackStatsCollectorTest, TeardownDoesNotEvictSuccessorWithSameName) { + auto collectorPtr = std::make_shared(); + auto& collector = *collectorPtr; + auto ftn = makeFtn("ns", "track"); + + auto original = collector.getOrCreate(ftn); + // Force a successor into the slot, as a reconnecting publisher would. + auto successor = collector.create(ftn); + ASSERT_NE(original.get(), successor.get()); + ASSERT_EQ(collector.get(ftn).get(), successor.get()); + + original.reset(); + + EXPECT_EQ(collector.size(), 1); + EXPECT_EQ(collector.get(ftn).get(), successor.get()); +} + +TEST(TrackStatsCollectorTest, ForEachVisitsLiveTracks) { + auto collectorPtr = std::make_shared(); + auto& collector = *collectorPtr; + auto a = collector.getOrCreate(makeFtn("ns", "a")); + auto b = collector.getOrCreate(makeFtn("ns", "b")); + a->counters().received.objects = 2; + b->counters().received.objects = 3; + + uint64_t total = 0; + collector.forEach([&](const TrackStats& stats) { total += stats.counters().received.objects; }); + + EXPECT_EQ(total, 5); +} + +} // namespace openmoq::moqx::stats diff --git a/test/test_admin_track_metrics.sh b/test/test_admin_track_metrics.sh new file mode 100755 index 00000000..9aefa568 --- /dev/null +++ b/test/test_admin_track_metrics.sh @@ -0,0 +1,224 @@ +#!/usr/bin/env bash +set -euo pipefail + +BINARY="${1:-$(dirname "$0")/../build/moqx}" +# shellcheck source=test_ports.sh +source "$(dirname "$0")/test_ports.sh" +LISTEN_PORT=$TEST_ADMIN_TRACK_METRICS_LISTEN +ADMIN_PORT=$TEST_ADMIN_TRACK_METRICS_ADMIN +TRACK_URL="http://localhost:${ADMIN_PORT}/metrics/track" +INFO_URL="http://localhost:${ADMIN_PORT}/info" + +if [[ ! -x "$BINARY" ]]; then + echo "ERROR: binary not found or not executable: $BINARY" >&2 + exit 1 +fi + +TMPDIR=$(mktemp -d) +MOQX_PID="" +DATESERVER_PID="" +TEXTCLIENT_PID="" +cleanup() { + for pid in "${TEXTCLIENT_PID:-}" "${DATESERVER_PID:-}" "${MOQX_PID:-}"; do + if [[ -n "$pid" ]]; then + kill "$pid" 2>/dev/null || true + wait "$pid" 2>/dev/null || true + fi + done + rm -rf "$TMPDIR" +} +trap cleanup EXIT + +"$(dirname "$0")/make_test_config.sh" "$LISTEN_PORT" "$ADMIN_PORT" > "$TMPDIR/config.yaml" + +"$BINARY" --config="$TMPDIR/config.yaml" & +MOQX_PID=$! + +for i in $(seq 1 100); do + HTTP_CODE=$(curl -sw "%{http_code}" -o /dev/null "$INFO_URL" 2>/dev/null || echo "000") + if [[ "$HTTP_CODE" == "200" ]]; then + break + fi + sleep 0.1 + if [[ $i -eq 100 ]]; then + echo "ERROR: admin /info endpoint did not become ready in time (HTTP $HTTP_CODE)" >&2 + exit 1 + fi +done + +fail() { + echo "FAIL: $1" >&2 + exit 1 +} + +# wait_sessions