Add the telemetry feature to the op-rbuilder

Author: therustmonk

Cargo.lock

@@ -88,6 +88,12 @@ tokio-tungstenite = "0.26.2"
 rand = "0.9.0"
 tracing-subscriber = { version = "0.3.18", features = ["env-filter", "json"] }
 
+# OpenTelemetry dependencies
+opentelemetry = { version = "0.29.1", optional = true }
+opentelemetry_sdk = { version = "0.29", optional = true }
+opentelemetry-otlp = { version = "0.29.0", optional = true, features = ["grpc-tonic"] }
+tracing-opentelemetry = { version = "0.30.0", optional = true }
+
 # `flashblocks` branch
 rollup-boost = { git = "http://github.com/flashbots/rollup-boost", rev = "60885346d4cf7f241de82790478195747433d472" }
 
@@ -118,6 +124,13 @@ min-trace-logs = ["tracing/release_max_level_trace"]
 integration = []
 flashblocks = []
 
+telemetry = [
+	"opentelemetry",
+	"opentelemetry_sdk",
+	"opentelemetry-otlp",
+	"tracing-opentelemetry",
+]
+
 [[bin]]
 name = "op-rbuilder"
 path = "src/main.rs"

crates/op-rbuilder/Cargo.toml

@@ -41,4 +41,12 @@ pub struct OpRbuilderArgs {
     /// Signals whether to log pool transaction events
     #[arg(long = "builder.log-pool-transactions", default_value = "false")]
     pub log_pool_transactions: bool,
+    /// OpenTelemetry endpoint for metrics
+    #[cfg(feature = "telemetry")]
+    #[arg(long, value_name = "TRACING")]
+    pub tracing: bool,
+    /// OpenTelemetry endpoint for metrics
+    #[cfg(feature = "telemetry")]
+    #[arg(long, value_name = "TRACING_ENDPOINT")]
+    pub tracing_endpoint: String,
 }

crates/op-rbuilder/src/args.rs

@@ -1,4 +1,5 @@
 use clap::Parser;
+use eyre::Result;
 use monitoring::Monitoring;
 use reth::providers::CanonStateSubscriptions;
 use reth_optimism_cli::{chainspec::OpChainSpecParser, Cli};
@@ -24,59 +25,72 @@ pub mod payload_builder;
 #[cfg(not(feature = "flashblocks"))]
 mod payload_builder_vanilla;
 mod primitives;
+#[cfg(feature = "telemetry")]
+mod telemetry;
 #[cfg(test)]
 mod tester;
 mod tx_signer;
 use monitor_tx_pool::monitor_tx_pool;
 
-fn main() {
-    Cli::<OpChainSpecParser, args::OpRbuilderArgs>::parse()
-        .run(|builder, builder_args| async move {
-            let rollup_args = builder_args.rollup_args;
+fn main() -> Result<()> {
+    let cli = Cli::<OpChainSpecParser, args::OpRbuilderArgs>::parse();
+    cli.run(|builder, builder_args| async move {
+        #[cfg(feature = "telemetry")]
+        let mut provider = telemetry::TelemetryControl::init(&builder_args)?;
 
-            let op_node = OpNode::new(rollup_args.clone());
-            let handle = builder
-                .with_types::<OpNode>()
-                .with_components(op_node.components().payload(CustomOpPayloadBuilder::new(
-                    builder_args.builder_signer,
-                    builder_args.flashblocks_ws_url,
-                    builder_args.chain_block_time,
-                    builder_args.flashblock_block_time,
-                )))
-                .with_add_ons(
-                    OpAddOnsBuilder::default()
-                        .with_sequencer(rollup_args.sequencer.clone())
-                        .with_enable_tx_conditional(rollup_args.enable_tx_conditional)
-                        .build(),
-                )
-                .on_node_started(move |ctx| {
-                    let new_canonical_blocks = ctx.provider().canonical_state_stream();
-                    let builder_signer = builder_args.builder_signer;
+        // Wrap the main execution in a span
+        let span = tracing::info_span!("op-rbuilder");
+        let _enter = span.enter();
 
-                    if builder_args.log_pool_transactions {
-                        tracing::info!("Logging pool transactions");
-                        ctx.task_executor.spawn_critical(
-                            "txlogging",
-                            Box::pin(async move {
-                                monitor_tx_pool(ctx.pool.all_transactions_event_listener()).await;
-                            }),
-                        );
-                    }
+        let rollup_args = builder_args.rollup_args;
 
+        let op_node = OpNode::new(rollup_args.clone());
+        let handle = builder
+            .with_types::<OpNode>()
+            .with_components(op_node.components().payload(CustomOpPayloadBuilder::new(
+                builder_args.builder_signer,
+                builder_args.flashblocks_ws_url,
+                builder_args.chain_block_time,
+                builder_args.flashblock_block_time,
+            )))
+            .with_add_ons(
+                OpAddOnsBuilder::default()
+                    .with_sequencer(rollup_args.sequencer.clone())
+                    .with_enable_tx_conditional(rollup_args.enable_tx_conditional)
+                    .build(),
+            )
+            .on_node_started(move |ctx| {
+                let new_canonical_blocks = ctx.provider().canonical_state_stream();
+                let builder_signer = builder_args.builder_signer;
+
+                if builder_args.log_pool_transactions {
+                    tracing::info!("Logging pool transactions");
                     ctx.task_executor.spawn_critical(
-                        "monitoring",
+                        "txlogging",
                         Box::pin(async move {
-                            let monitoring = Monitoring::new(builder_signer);
-                            let _ = monitoring.run_with_stream(new_canonical_blocks).await;
+                            monitor_tx_pool(ctx.pool.all_transactions_event_listener()).await;
                         }),
                     );
+                }
+
+                ctx.task_executor.spawn_critical(
+                    "monitoring",
+                    Box::pin(async move {
+                        let monitoring = Monitoring::new(builder_signer);
+                        let _ = monitoring.run_with_stream(new_canonical_blocks).await;
+                    }),
+                );
+
+                Ok(())
+            })
+            .launch()
+            .await?;
 
-                    Ok(())
-                })
-                .launch()
-                .await?;
+        handle.node_exit_future.await?;
+        #[cfg(feature = "telemetry")]
+        provider.shutdown()?;
+        Ok(())
+    })?;
 
-            handle.node_exit_future.await
-        })
-        .unwrap();
+    Ok(())
 }