refactor(compute): add tracing related args

Signed-off-by: MrCroxx <[email protected]>

Cargo.toml

@@ -76,7 +76,7 @@ license = "Apache-2.0"
 repository = "https://github.com/risingwavelabs/risingwave"
 
 [workspace.dependencies]
-foyer = { version = "0.9.4", features = ["nightly"] }
+foyer = { version = "0.9.4", features = ["nightly", "mtrace"] }
 apache-avro = { git = "https://github.com/risingwavelabs/avro", rev = "25113ba88234a9ae23296e981d8302c290fdaa4b", features = [
     "snappy",
     "zstandard",
@@ -340,6 +340,8 @@ deno_web = { git = "https://github.com/bakjos/deno", rev = "787a232" }
 deno_websocket = { git = "https://github.com/bakjos/deno", rev = "787a232" }
 # patch to remove preserve_order from serde_json
 bson = { git = "https://github.com/risingwavelabs/bson-rust", rev = "e5175ec" }
+# temporarily patch until foyer release v0.10
+foyer = { git = "https://github.com/MrCroxx/foyer", rev = "fe904bc" }
 
 [workspace.metadata.dylint]
 libraries = [{ path = "./lints" }]

src/common/src/config.rs

@@ -487,6 +487,9 @@ pub struct ServerConfig {
     #[serde(default = "default::server::grpc_max_reset_stream_size")]
     pub grpc_max_reset_stream: u32,
 
+    #[serde(default)]
+    pub tracing: TracingConfig,
+
     #[serde(default, flatten)]
     #[config_doc(omitted)]
     pub unrecognized: Unrecognized<Self>,
@@ -1246,6 +1249,20 @@ impl SystemConfig {
     }
 }
 
+#[derive(Clone, Debug, Serialize, Deserialize, DefaultFromSerde)]
+pub struct TracingConfig {
+    /// Url of jaeger for tracing data collection.
+    ///
+    /// Tracing will be disabled if no valid jaeger url is set.
+    ///
+    /// Example: "127.0.0.1:6831".
+    #[serde(default = "default::tracing_config::jaeger")]
+    pub jaeger: String,
+    /// Interval for minitrace reporter reports in ms.
+    #[serde(default = "default::tracing_config::report_interval_ms")]
+    pub report_interval_ms: u64,
+}
+
 pub mod default {
     pub mod meta {
         use crate::config::{DefaultParallelism, MetaBackend};
@@ -2075,6 +2092,16 @@ pub mod default {
             }
         }
     }
+
+    pub mod tracing_config {
+        pub fn jaeger() -> String {
+            "".to_string()
+        }
+
+        pub fn report_interval_ms() -> u64 {
+            10
+        }
+    }
 }
 
 #[derive(Debug, Clone)]

src/compute/Cargo.toml

@@ -26,6 +26,8 @@ futures-async-stream = { workspace = true }
 hyper = "0.14" # required by tonic
 itertools = { workspace = true }
 maplit = "1.0.2"
+minitrace = "0.6.7"
+minitrace-jaeger = "0.6.7"
 pprof = { version = "0.13", features = ["flamegraph"] }
 prometheus = { version = "0.13" }
 prost = { workspace = true }

src/compute/src/server.rs

@@ -395,6 +395,20 @@ pub async fn compute_node_serve(
         tracing::info!("Telemetry didn't start due to config");
     }
 
+    let service = format!("compute-{worker_id}");
+    if let Ok(agent) = config.server.tracing.jaeger.parse() {
+        let reporter = minitrace_jaeger::JaegerReporter::new(agent, &service).unwrap();
+        minitrace::set_reporter(
+            reporter,
+            minitrace::collector::Config::default().report_interval(Duration::from_millis(
+                config.server.tracing.report_interval_ms,
+            )),
+        );
+        tracing::info!("Jaeger exporter for {service} is set at {agent:?}.");
+    } else {
+        tracing::info!("Jaeger exporter for {service} is disabled.")
+    }
+
     // Clean up the spill directory.
     #[cfg(not(madsim))]
     SpillOp::clean_spill_directory().await.unwrap();