tests, docs, utilization value

Author: Dav1dde

relay-server/src/service.rs

@@ -290,7 +290,7 @@ impl ServiceState {
 
         services.start(RelayStats::new(
             config.clone(),
-            handle.metrics(),
+            handle.clone(),
             upstream_relay.clone(),
             #[cfg(feature = "processing")]
             redis_pools.clone(),

relay-server/src/services/stats.rs

@@ -8,15 +8,16 @@ use relay_redis::AsyncRedisClient;
 #[cfg(feature = "processing")]
 use relay_redis::{RedisPool, RedisPools, Stats};
 use relay_statsd::metric;
-use relay_system::{Addr, RuntimeMetrics, Service};
+use relay_system::{Addr, Handle, RuntimeMetrics, Service};
 use tokio::time::interval;
 
 /// Relay Stats Service.
 ///
 /// Service which collects stats periodically and emits them via statsd.
 pub struct RelayStats {
     config: Arc<Config>,
-    runtime: RuntimeMetrics,
+    runtime: Handle,
+    rt_metrics: RuntimeMetrics,
     upstream_relay: Addr<UpstreamRelay>,
     #[cfg(feature = "processing")]
     redis_pools: Option<RedisPools>,
@@ -25,88 +26,106 @@ pub struct RelayStats {
 impl RelayStats {
     pub fn new(
         config: Arc<Config>,
-        runtime: RuntimeMetrics,
+        runtime: Handle,
         upstream_relay: Addr<UpstreamRelay>,
         #[cfg(feature = "processing")] redis_pools: Option<RedisPools>,
     ) -> Self {
         Self {
             config,
             upstream_relay,
+            rt_metrics: runtime.metrics(),
             runtime,
             #[cfg(feature = "processing")]
             redis_pools,
         }
     }
 
+    async fn service_metrics(&self) {
+        for (service, metrics) in self.runtime.current_services_metrics().iter() {
+            metric!(
+                gauge(RelayGauges::ServiceUtilization) = metrics.utilization as u64,
+                service = service.name(),
+                instance_id = &service.instance_id().to_string(),
+            );
+        }
+    }
+
     async fn tokio_metrics(&self) {
-        metric!(gauge(RuntimeGauges::NumIdleThreads) = self.runtime.num_idle_threads() as u64);
-        metric!(gauge(RuntimeGauges::NumAliveTasks) = self.runtime.num_alive_tasks() as u64);
+        metric!(gauge(RuntimeGauges::NumIdleThreads) = self.rt_metrics.num_idle_threads() as u64);
+        metric!(gauge(RuntimeGauges::NumAliveTasks) = self.rt_metrics.num_alive_tasks() as u64);
         metric!(
-            gauge(RuntimeGauges::BlockingQueueDepth) = self.runtime.blocking_queue_depth() as u64
+            gauge(RuntimeGauges::BlockingQueueDepth) =
+                self.rt_metrics.blocking_queue_depth() as u64
         );
         metric!(
-            gauge(RuntimeGauges::NumBlockingThreads) = self.runtime.num_blocking_threads() as u64
+            gauge(RuntimeGauges::NumBlockingThreads) =
+                self.rt_metrics.num_blocking_threads() as u64
         );
         metric!(
             gauge(RuntimeGauges::NumIdleBlockingThreads) =
-                self.runtime.num_idle_blocking_threads() as u64
+                self.rt_metrics.num_idle_blocking_threads() as u64
         );
 
         metric!(
             counter(RuntimeCounters::BudgetForcedYieldCount) +=
-                self.runtime.budget_forced_yield_count()
+                self.rt_metrics.budget_forced_yield_count()
         );
 
-        metric!(gauge(RuntimeGauges::NumWorkers) = self.runtime.num_workers() as u64);
-        for worker in 0..self.runtime.num_workers() {
+        metric!(gauge(RuntimeGauges::NumWorkers) = self.rt_metrics.num_workers() as u64);
+        for worker in 0..self.rt_metrics.num_workers() {
             let worker_name = worker.to_string();
 
             metric!(
                 gauge(RuntimeGauges::WorkerLocalQueueDepth) =
-                    self.runtime.worker_local_queue_depth(worker) as u64,
+                    self.rt_metrics.worker_local_queue_depth(worker) as u64,
                 worker = &worker_name,
             );
             metric!(
                 gauge(RuntimeGauges::WorkerMeanPollTime) =
-                    self.runtime.worker_mean_poll_time(worker).as_secs_f64(),
+                    self.rt_metrics.worker_mean_poll_time(worker).as_secs_f64(),
                 worker = &worker_name,
             );
 
             metric!(
                 counter(RuntimeCounters::WorkerLocalScheduleCount) +=
-                    self.runtime.worker_local_schedule_count(worker),
+                    self.rt_metrics.worker_local_schedule_count(worker),
                 worker = &worker_name,
             );
             metric!(
-                counter(RuntimeCounters::WorkerNoopCount) += self.runtime.worker_noop_count(worker),
+                counter(RuntimeCounters::WorkerNoopCount) +=
+                    self.rt_metrics.worker_noop_count(worker),
                 worker = &worker_name,
             );
             metric!(
                 counter(RuntimeCounters::WorkerOverflowCount) +=
-                    self.runtime.worker_overflow_count(worker),
+                    self.rt_metrics.worker_overflow_count(worker),
                 worker = &worker_name,
             );
             metric!(
-                counter(RuntimeCounters::WorkerParkCount) += self.runtime.worker_park_count(worker),
+                counter(RuntimeCounters::WorkerParkCount) +=
+                    self.rt_metrics.worker_park_count(worker),
                 worker = &worker_name,
             );
             metric!(
-                counter(RuntimeCounters::WorkerPollCount) += self.runtime.worker_poll_count(worker),
+                counter(RuntimeCounters::WorkerPollCount) +=
+                    self.rt_metrics.worker_poll_count(worker),
                 worker = &worker_name,
             );
             metric!(
                 counter(RuntimeCounters::WorkerStealCount) +=
-                    self.runtime.worker_steal_count(worker),
+                    self.rt_metrics.worker_steal_count(worker),
                 worker = &worker_name,
             );
             metric!(
                 counter(RuntimeCounters::WorkerStealOperations) +=
-                    self.runtime.worker_steal_operations(worker),
+                    self.rt_metrics.worker_steal_operations(worker),
                 worker = &worker_name,
             );
             metric!(
                 counter(RuntimeCounters::WorkerTotalBusyDuration) +=
-                    self.runtime.worker_total_busy_duration(worker).as_millis() as u64,
+                    self.rt_metrics
+                        .worker_total_busy_duration(worker)
+                        .as_millis() as u64,
                 worker = &worker_name,
             );
         }
@@ -171,6 +190,7 @@ impl Service for RelayStats {
         loop {
             let _ = tokio::join!(
                 self.upstream_status(),
+                self.service_metrics(),
                 self.tokio_metrics(),
                 self.redis_pools(),
             );

relay-server/src/statsd.rs

@@ -42,6 +42,12 @@ pub enum RelayGauges {
     /// - `namespace`: the metric namespace.
     #[cfg(feature = "processing")]
     MetricDelayMax,
+    /// Estimated percentage [0-100] of how busy Relay's internal services are.
+    ///
+    /// This metric is tagged with:
+    /// - `service`: the service name.
+    /// - `instance_id`: a for the service name unique identifier for the running service
+    ServiceUtilization,
 }
 
 impl GaugeMetric for RelayGauges {
@@ -63,6 +69,7 @@ impl GaugeMetric for RelayGauges {
             RelayGauges::ServerActiveConnections => "server.http.connections",
             #[cfg(feature = "processing")]
             RelayGauges::MetricDelayMax => "metrics.delay.max",
+            RelayGauges::ServiceUtilization => "service.utilization",
         }
     }
 }

relay-system/Cargo.toml

@@ -28,3 +28,4 @@ pin-project-lite = { workspace = true }
 insta = { workspace = true }
 relay-statsd = { workspace = true, features = ["test"] }
 tokio = { workspace = true, features = ["test-util"] }
+futures =  { workspace = true, features = ["async-await"] }

relay-system/src/runtime/handle.rs

@@ -0,0 +1,101 @@
+use std::sync::Arc;
+
+use futures::stream::FuturesUnordered;
+use futures::StreamExt as _;
+
+use crate::runtime::metrics::TokioCallbackMetrics;
+use crate::{RuntimeMetrics, ServiceJoinHandle, ServiceRegistry, ServiceSpawn, ServicesMetrics};
+
+#[derive(Debug)]
+struct HandleInner {
+    name: &'static str,
+    services: ServiceRegistry,
+    tokio: tokio::runtime::Handle,
+    tokio_cb_metrics: Arc<TokioCallbackMetrics>,
+}
+
+/// Handle to the [`Runtime`].
+///
+/// The handle is internally reference-counted and can be freely cloned.
+/// A handle can be obtained using the [`Runtime::handle`] method.
+#[derive(Debug, Clone)]
+pub struct Handle {
+    inner: Arc<HandleInner>,
+}
+
+impl Handle {
+    pub(crate) fn new(
+        name: &'static str,
+        tokio: tokio::runtime::Handle,
+        tokio_cb_metrics: Arc<TokioCallbackMetrics>,
+    ) -> Self {
+        Self {
+            inner: Arc::new(HandleInner {
+                name,
+                services: ServiceRegistry::new(),
+                tokio,
+                tokio_cb_metrics,
+            }),
+        }
+    }
+
+    /// Returns a new [`RuntimeMetrics`] handle for this runtime.
+    pub fn metrics(&self) -> RuntimeMetrics {
+        Arc::clone(&self.inner.tokio_cb_metrics)
+            .into_metrics(self.inner.name, self.inner.tokio.metrics())
+    }
+
+    /// Returns all service metrics of all currently running services.
+    ///
+    /// Unlike [`Self::metrics`], this is not a handle to the metrics.
+    pub fn current_services_metrics(&self) -> ServicesMetrics {
+        self.inner.services.metrics()
+    }
+
+    /// Returns a new unique [`ServiceSet`] to spawn services and await their termination.
+    pub fn service_set(&self) -> ServiceSet {
+        ServiceSet {
+            handle: Arc::clone(&self.inner),
+            services: Default::default(),
+        }
+    }
+}
+
+impl ServiceSpawn for Handle {
+    fn start_obj(&self, service: crate::ServiceObj) {
+        self.inner.services.start_in(&self.inner.tokio, service);
+    }
+}
+
+/// Spawns and keeps track of running services.
+///
+/// A [`ServiceSet`] can be awaited for the completion of all started services
+/// on this [`ServiceSet`].
+///
+/// Every service started on this [`ServiceSet`] is attached to the [`Handle`]
+/// this set was created from, using [`Handle::service_set`].
+pub struct ServiceSet {
+    handle: Arc<HandleInner>,
+    services: FuturesUnordered<ServiceJoinHandle>,
+}
+
+impl ServiceSet {
+    /// Awaits until all services have finished.
+    ///
+    /// Panics if one of the spawned services has panicked.
+    pub async fn join(&mut self) {
+        while let Some(res) = self.services.next().await {
+            if let Some(panic) = res.err().and_then(|e| e.into_panic()) {
+                // Re-trigger panic to terminate the process:
+                std::panic::resume_unwind(panic);
+            }
+        }
+    }
+}
+
+impl ServiceSpawn for ServiceSet {
+    fn start_obj(&self, service: crate::ServiceObj) {
+        let handle = self.handle.services.start_in(&self.handle.tokio, service);
+        self.services.push(handle);
+    }
+}

relay-system/src/runtime/metrics.rs

@@ -242,7 +242,7 @@ mod tests {
     fn test_metric_diff() {
         let rt = crate::Runtime::builder("test").worker_threads(1).build();
 
-        let metrics = rt.metrics();
+        let metrics = rt.handle().metrics();
 
         rt.block_on(async move {
             let tokio_metrics = tokio::runtime::Handle::current().metrics();

relay-system/src/runtime/mod.rs

@@ -1,3 +1,4 @@
+mod handle;
 mod metrics;
 #[expect(
     clippy::module_inception,
@@ -6,6 +7,7 @@ mod metrics;
 mod runtime;
 mod spawn;
 
+pub use self::handle::{Handle, ServiceSet};
 pub use self::metrics::RuntimeMetrics;
-pub use self::runtime::{Builder, Handle, Runtime, ServiceSet};
+pub use self::runtime::{Builder, Runtime};
 pub use self::spawn::{spawn, spawn_in, TaskId};