📈 feat(rt): expose tokio runtime metrics

.devcontainer/devcontainer.json

@@ -43,6 +43,7 @@
 	"remoteUser": "code",
 	"containerEnv": {
 		"CXX": "clang++-14",
+		"RUSTFLAGS": "--cfg tokio_unstable"
 	},
 	"mounts": [
 		{

.github/workflows/beta.yml

@@ -15,7 +15,7 @@ env:
   CARGO_INCREMENTAL: 0
   CARGO_NET_RETRY: 10
   RUSTUP_MAX_RETRIES: 10
-  RUSTFLAGS: "-D warnings"
+  RUSTFLAGS: "-D warnings --cfg tokio_unstable"
 
 permissions:
   contents: read

.github/workflows/coverage.yml

@@ -15,7 +15,7 @@ concurrency:
 env:
   CARGO_INCREMENTAL: 0
   CARGO_NET_RETRY: 10
-  RUSTFLAGS: "-D warnings -A deprecated -C debuginfo=2"
+  RUSTFLAGS: "-D warnings -A deprecated --cfg tokio_unstable -C debuginfo=2"
   RUSTUP_MAX_RETRIES: 10
 
 jobs:

.github/workflows/fuzzers.yml

@@ -17,7 +17,7 @@ env:
   CARGO_INCREMENTAL: 0
   CARGO_NET_RETRY: 10
   RUST_BACKTRACE: short
-  RUSTFLAGS: "-D warnings -A deprecated -C debuginfo=0"
+  RUSTFLAGS: "-D warnings -A deprecated --cfg tokio_unstable -C debuginfo=0"
   RUSTUP_MAX_RETRIES: 10
 
 permissions:

.github/workflows/nightly.yml

@@ -14,7 +14,7 @@ on:
 env:
   CARGO_INCREMENTAL: 0
   CARGO_NET_RETRY: 10
-  RUSTFLAGS: "-D warnings -A opaque_hidden_inferred_bound -C debuginfo=0"
+  RUSTFLAGS: "-D warnings -A opaque_hidden_inferred_bound --cfg tokio_unstable -C debuginfo=0"
   RUSTUP_MAX_RETRIES: 10
 
 permissions:

.github/workflows/pr.yml

@@ -5,7 +5,7 @@ env:
   CARGO_INCREMENTAL: 0
   CARGO_NET_RETRY: 10
   RUSTUP_MAX_RETRIES: 10
-  RUSTFLAGS: "-D warnings -D deprecated -C debuginfo=0"
+  RUSTFLAGS: "-D warnings -D deprecated --cfg tokio_unstable -C debuginfo=0"
 
 concurrency:
   group: ${{ github.workflow }}-${{ github.head_ref }}

.github/workflows/release.yml

@@ -48,7 +48,7 @@ on:
 env:
   CARGO_INCREMENTAL: 0
   CARGO_NET_RETRY: 10
-  RUSTFLAGS: "-D warnings -A deprecated"
+  RUSTFLAGS: "-D warnings -A deprecated --cfg tokio_unstable"
   RUSTUP_MAX_RETRIES: 10
 
 concurrency:

Cargo.lock

@@ -653,6 +653,17 @@ version = "0.3.30"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a44623e20b9681a318efdd71c299b6b222ed6f231972bfe2f224ebad6311f0c1"
 
+[[package]]
+name = "futures-macro"
+version = "0.3.30"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "87750cf4b7a4c0625b1529e4c543c2182106e4dedc60a2a6455e00d212c489ac"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
 [[package]]
 name = "futures-sink"
 version = "0.3.30"
@@ -674,6 +685,7 @@ dependencies = [
  "futures-channel",
  "futures-core",
  "futures-io",
+ "futures-macro",
  "futures-sink",
  "futures-task",
  "memchr",
@@ -1020,6 +1032,18 @@ dependencies = [
  "libc",
 ]
 
+[[package]]
+name = "kubert-prometheus-tokio"
+version = "0.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a101fa3df488c89130664aaf4652986da49e204fb1725d089122f75b22ff6cbb"
+dependencies = [
+ "prometheus-client",
+ "tokio",
+ "tokio-metrics",
+ "tracing",
+]
+
 [[package]]
 name = "lazy_static"
 version = "1.5.0"
@@ -2296,6 +2320,7 @@ version = "0.1.0"
 dependencies = [
  "futures",
  "jemallocator",
+ "kubert-prometheus-tokio",
  "linkerd-app",
  "linkerd-meshtls",
  "linkerd-metrics",
@@ -3378,6 +3403,18 @@ dependencies = [
  "syn",
 ]
 
+[[package]]
+name = "tokio-metrics"
+version = "0.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "eace09241d62c98b7eeb1107d4c5c64ca3bd7da92e8c218c153ab3a78f9be112"
+dependencies = [
+ "futures-util",
+ "pin-project-lite",
+ "tokio",
+ "tokio-stream",
+]
+
 [[package]]
 name = "tokio-rustls"
 version = "0.24.1"

Dockerfile

@@ -32,7 +32,7 @@ RUN --mount=type=cache,id=cargo,target=/usr/local/cargo/registry \
 # Build the proxy.
 FROM fetch as build
 ENV CARGO_INCREMENTAL=0
-ENV RUSTFLAGS="-D warnings -A deprecated"
+ENV RUSTFLAGS="-D warnings -A deprecated --cfg tokio_unstable"
 ARG TARGETARCH="amd64"
 ARG PROFILE="release"
 ARG LINKERD2_PROXY_VERSION=""

linkerd2-proxy/Cargo.toml

@@ -18,13 +18,14 @@ pprof = ["linkerd-app/pprof"]
 
 [dependencies]
 futures = { version = "0.3", default-features = false }
-num_cpus = { version = "1", optional = true }
+kubert-prometheus-tokio = { version = "0.1", features = ["rt"] }
 linkerd-app = { path = "../linkerd/app" }
 linkerd-metrics = { path = "../linkerd/metrics" }
 # We don't actually use code from this crate in `main`; it's here only so we can
 # control its feature flags.
 linkerd-meshtls = { path = "../linkerd/meshtls" }
 linkerd-signal = { path = "../linkerd/signal" }
+num_cpus = { version = "1", optional = true }
 tokio = { version = "1", features = ["rt", "time", "net"] }
 tracing = "0.1"
 

linkerd2-proxy/src/main.rs

@@ -42,7 +42,7 @@ fn main() {
         vendor = BUILD_INFO.vendor,
     );
 
-    let metrics = linkerd_metrics::prom::Registry::default();
+    let mut metrics = linkerd_metrics::prom::Registry::default();
 
     // Load configuration from the environment without binding ports.
     let config = match Config::try_from_env() {
@@ -57,6 +57,9 @@ fn main() {
     // `LINKERD2_PROXY_CORES` env or the number of available CPUs (as provided
     // by cgroups, when possible).
     rt::build().block_on(async move {
+        // Spawn a task to run in the background, exporting runtime metrics at a regular interval.
+        rt::spawn_metrics_exporter(&mut metrics);
+
         let (shutdown_tx, mut shutdown_rx) = mpsc::unbounded_channel();
         let shutdown_grace_period = config.shutdown_grace_period;
 

linkerd2-proxy/src/rt.rs

@@ -64,3 +64,37 @@ pub(crate) fn build() -> Runtime {
         .build()
         .expect("failed to build basic runtime!")
 }
+
+// Spawns a task to scrape metrics for the given runtime at a regular interval.
+///
+/// Note that this module requires unstable tokio functionality that must be
+/// enabled via the `tokio_unstable` feature. When it is not enabled, no metrics
+/// will be registered.
+///
+/// `RUSTFLAGS="--cfg tokio_unstable"` must be set at build-time to use this feature.
+pub fn spawn_metrics_exporter(registry: &mut linkerd_metrics::prom::Registry) {
+    #[cfg(tokio_unstable)]
+    {
+        use {std::time::Duration, tracing::Instrument};
+
+        /// The fixed interval at which tokio runtime metrics are updated.
+        //
+        //  TODO(kate): perhaps this could be configurable eventually. for now, it's hard-coded.
+        const INTERVAL: Duration = Duration::from_secs(1);
+
+        let mut interval = tokio::time::interval(INTERVAL);
+
+        let registry = registry.sub_registry_with_prefix("tokio_rt");
+        let runtime = tokio::runtime::Handle::current();
+        let metrics = kubert_prometheus_tokio::Runtime::register(registry, runtime);
+
+        tokio::spawn(
+            async move { metrics.updated(&mut interval).await }
+                .instrument(tracing::info_span!("kubert-prom-tokio-rt")),
+        );
+    }
+    #[cfg(not(tokio_unstable))]
+    {
+        tracing::debug!("Tokio runtime metrics cannot be monitored without the tokio_unstable cfg");
+    }
+}