Tokio-console subscriber configuration

crates/local-cluster-runner/src/node/mod.rs

@@ -46,8 +46,14 @@ pub struct Node {
     #[builder(mutators(
             #[mutator(requires = [base_dir])]
             pub fn with_node_socket(self) {
+                let ts = std::time::SystemTime::now()
+                    .duration_since(std::time::UNIX_EPOCH)
+                    .expect("to get timestamp")
+                    .as_secs();
                 let node_socket: PathBuf = PathBuf::from(self.base_config.node_name()).join("node.sock");
+                let tokio_console_socket: PathBuf = PathBuf::from(self.base_config.node_name()).join(format!("tokio_console-{ts}.sock"));
                 self.base_config.common.bind_address = Some(BindAddress::Uds(node_socket.clone()));
+                self.base_config.common.tokio_console_bind_address = Some(BindAddress::Uds(tokio_console_socket));
                 self.base_config.common.advertised_address = AdvertisedAddress::Uds(node_socket);
             }
 
@@ -235,6 +241,11 @@ impl Node {
         if let AdvertisedAddress::Uds(file) = &mut self.base_config.common.advertised_address {
             *file = base_dir.join(&*file)
         }
+        if let Some(BindAddress::Uds(file)) =
+            &mut self.base_config.common.tokio_console_bind_address
+        {
+            *file = base_dir.join(&*file)
+        }
 
         self.base_config.common.set_base_dir(base_dir);
         self.base_config.common.set_cluster_name(cluster_name);
@@ -298,12 +309,6 @@ impl Node {
             &mut cmd
         }
         .env("RESTATE_CONFIG", node_config_file)
-        .env(
-            "TOKIO_CONSOLE_BIND",
-            random_socket_address()
-                .expect("to find a random port for tokio console")
-                .to_string(),
-        )
         .envs(env)
         .stdin(Stdio::null())
         .stdout(Stdio::piped())

crates/tracing-instrumentation/Cargo.toml

@@ -22,16 +22,17 @@ rt-tokio = ["dep:tokio"]
 restate-types = { workspace = true }
 
 arc-swap = { workspace = true }
-console-subscriber = { version = "0.4.1", optional = true }
+console-subscriber = { version = "0.4.1", features = ["parking_lot"], optional = true }
 derive_builder = { workspace = true }
 futures = { workspace = true }
+metrics-tracing-context = { version = "0.16.0" }
 nu-ansi-term = "0.46.0"
 once_cell = { workspace = true }
 opentelemetry = { workspace = true }
 opentelemetry-contrib = { version = "0.16.0", features = ["jaeger_json_exporter", "rt-tokio"] }
 opentelemetry-otlp = { version = "0.17.0" }
-opentelemetry_sdk = { workspace = true, features = ["rt-tokio"] }
 opentelemetry-semantic-conventions = "0.16.0"
+opentelemetry_sdk = { workspace = true, features = ["rt-tokio"] }
 schemars = { workspace = true, optional = true }
 serde = { workspace = true }
 thiserror = { workspace = true }
@@ -40,7 +41,6 @@ tonic = { workspace = true }
 tracing = { workspace = true }
 tracing-opentelemetry = { workspace = true }
 tracing-subscriber = { workspace = true, features = ["json"] }
-metrics-tracing-context = { version = "0.16.0" }
 
 [dev-dependencies]
 tokio = { workspace = true }

crates/tracing-instrumentation/src/lib.rs

@@ -37,6 +37,8 @@ use tracing_subscriber::util::SubscriberInitExt;
 use tracing_subscriber::{EnvFilter, Layer, Registry};
 
 use restate_types::config::{CommonOptions, LogFormat};
+#[cfg(feature = "console-subscriber")]
+use restate_types::net::BindAddress;
 
 use crate::exporter::UserServiceModifierSpanExporter;
 use crate::pretty::PrettyFields;
@@ -271,7 +273,22 @@ pub fn init_tracing_and_logging(
 
     // Console subscriber layer
     #[cfg(feature = "console-subscriber")]
-    let layers = layers.with(console_subscriber::spawn());
+    let layers = {
+        let tokio_console_bind_address = match common_opts
+            .tokio_console_bind_address
+            .clone()
+            .expect("falls back to default value")
+        {
+            BindAddress::Uds(p) => console_subscriber::ServerAddr::Unix(p),
+            BindAddress::Socket(s) => console_subscriber::ServerAddr::Tcp(s),
+        };
+
+        layers.with(
+            console_subscriber::ConsoleLayer::builder()
+                .server_addr(tokio_console_bind_address)
+                .spawn(),
+        )
+    };
 
     build_services_tracing(common_opts)?;
 

crates/types/src/config/common.rs

@@ -15,10 +15,11 @@ use std::time::Duration;
 
 use enumset::EnumSet;
 use once_cell::sync::Lazy;
-use restate_serde_util::{NonZeroByteCount, SerdeableHeaderHashMap};
 use serde::{Deserialize, Serialize};
 use serde_with::serde_as;
 
+use restate_serde_util::{NonZeroByteCount, SerdeableHeaderHashMap};
+
 use super::{AwsOptions, HttpOptions, PerfStatsLevel, RocksDbOptions};
 use crate::net::{AdvertisedAddress, BindAddress};
 use crate::nodes_config::Role;
@@ -121,6 +122,12 @@ pub struct CommonOptions {
     /// Disable ANSI terminal codes for logs. This is useful when the log collector doesn't support processing ANSI terminal codes.
     pub log_disable_ansi_codes: bool,
 
+    /// Address to bind for the tokio-console tracing subscriber. If unset and restate-server is
+    /// built with tokio-console support, it'll listen on `0.0.0.0:6669`.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    #[cfg_attr(feature = "schemars", schemars(with = "String"))]
+    pub tokio_console_bind_address: Option<BindAddress>,
+
     /// Timeout for idle histograms.
     ///
     /// The duration after which a histogram is considered idle and will be removed from
@@ -359,6 +366,7 @@ impl Default for CommonOptions {
             log_filter: "warn,restate=info".to_string(),
             log_format: Default::default(),
             log_disable_ansi_codes: false,
+            tokio_console_bind_address: Some(BindAddress::Socket("0.0.0.0:6669".parse().unwrap())),
             default_thread_pool_size: None,
             storage_high_priority_bg_threads: None,
             storage_low_priority_bg_threads: None,