fix(coprocessor): restore service otel wiring and align span op names

Author: Eikix

coprocessor/fhevm-engine/gw-listener/src/bin/gw_listener.rs

@@ -103,19 +103,18 @@ async fn main() -> anyhow::Result<()> {
 
     let conf = Conf::parse();
 
-    tracing_subscriber::fmt()
-        .json()
-        .with_level(true)
-        .with_max_level(conf.log_level)
-        .init();
-
-    let _otel_guard = match telemetry::init_otel(&conf.service_name) {
-        Ok(otel_guard) => otel_guard,
-        Err(err) => {
-            error!(error = %err, "Failed to setup OTLP");
-            None
-        }
-    };
+    let mut otlp_setup_error: Option<String> = None;
+    let _otel_guard =
+        match telemetry::init_json_subscriber(conf.log_level, &conf.service_name, "otlp-layer") {
+            Ok(guard) => guard,
+            Err(err) => {
+                otlp_setup_error = Some(err.to_string());
+                None
+            }
+        };
+    if let Some(err) = otlp_setup_error {
+        error!(error = %err, "Failed to setup OTLP");
+    }
 
     info!(gateway_url = %conf.gw_url, max_retries = %conf.provider_max_retries,
          retry_interval = ?conf.provider_retry_interval, "Connecting to Gateway");

coprocessor/fhevm-engine/host-listener/src/bin/main.rs

@@ -1,14 +1,26 @@
 use clap::Parser;
+use fhevm_engine_common::telemetry;
+use tracing::error;
 
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
     let args = host_listener::cmd::Args::parse();
 
-    tracing_subscriber::fmt()
-        .json()
-        .with_level(true)
-        .with_max_level(args.log_level)
-        .init();
+    let mut otlp_setup_error: Option<String> = None;
+    let _otel_guard = match telemetry::init_json_subscriber(
+        args.log_level,
+        &args.service_name,
+        "otlp-layer",
+    ) {
+        Ok(guard) => guard,
+        Err(err) => {
+            otlp_setup_error = Some(err.to_string());
+            None
+        }
+    };
+    if let Some(err) = otlp_setup_error {
+        error!(error = %err, "Failed to setup OTLP");
+    }
 
     host_listener::cmd::main(args).await
 }

coprocessor/fhevm-engine/host-listener/src/bin/poller.rs

@@ -3,10 +3,10 @@ use std::time::Duration;
 use alloy::primitives::Address;
 use clap::Parser;
 use tokio_util::sync::CancellationToken;
-use tracing::Level;
+use tracing::{error, Level};
 
-use fhevm_engine_common::metrics_server;
 use fhevm_engine_common::utils::DatabaseURL;
+use fhevm_engine_common::{metrics_server, telemetry};
 use host_listener::cmd::{
     DEFAULT_DEPENDENCE_BY_CONNEXITY, DEFAULT_DEPENDENCE_CACHE_SIZE,
     DEFAULT_DEPENDENCE_CROSS_BLOCK,
@@ -126,11 +126,21 @@ struct Args {
 async fn main() -> anyhow::Result<()> {
     let args = Args::parse();
 
-    tracing_subscriber::fmt()
-        .json()
-        .with_level(true)
-        .with_max_level(args.log_level)
-        .init();
+    let mut otlp_setup_error: Option<String> = None;
+    let _otel_guard = match telemetry::init_json_subscriber(
+        args.log_level,
+        &args.service_name,
+        "otlp-layer",
+    ) {
+        Ok(guard) => guard,
+        Err(err) => {
+            otlp_setup_error = Some(err.to_string());
+            None
+        }
+    };
+    if let Some(err) = otlp_setup_error {
+        error!(error = %err, "Failed to setup OTLP");
+    }
 
     let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
 

coprocessor/fhevm-engine/host-listener/src/cmd/mod.rs

@@ -18,7 +18,6 @@ use std::sync::Arc;
 use std::time::Duration;
 
 use fhevm_engine_common::healthz_server::HttpServer as HealthHttpServer;
-use fhevm_engine_common::telemetry;
 use fhevm_engine_common::types::BlockchainProvider;
 use fhevm_engine_common::utils::{DatabaseURL, HeartBeat};
 
@@ -1042,14 +1041,6 @@ pub async fn main(args: Args) -> anyhow::Result<()> {
         )
     };
 
-    let _otel_guard = match telemetry::init_otel(&args.service_name) {
-        Ok(otel_guard) => otel_guard,
-        Err(err) => {
-            error!(error = %err, "Failed to setup OTLP");
-            None
-        }
-    };
-
     let mut log_iter = InfiniteLogIter::new(&args);
     let chain_id = log_iter.get_chain_id().await?;
     info!(chain_id = %chain_id, "Chain ID");

coprocessor/fhevm-engine/host-listener/src/poller/mod.rs

@@ -16,7 +16,6 @@ use tracing::{error, info, warn};
 
 use fhevm_engine_common::chain_id::ChainId;
 use fhevm_engine_common::healthz_server::HttpServer as HealthHttpServer;
-use fhevm_engine_common::telemetry;
 use fhevm_engine_common::utils::{DatabaseURL, HeartBeat};
 
 use crate::cmd::block_history::BlockSummary;
@@ -90,14 +89,6 @@ pub struct PollerConfig {
 }
 
 pub async fn run_poller(config: PollerConfig) -> Result<()> {
-    let _otel_guard = match telemetry::init_otel(&config.service_name) {
-        Ok(otel_guard) => otel_guard,
-        Err(err) => {
-            error!(error = %err, "Failed to setup OTLP");
-            None
-        }
-    };
-
     let acl_address = config.acl_address;
     let tfhe_address = config.tfhe_address;
 

coprocessor/fhevm-engine/tfhe-worker/src/lib.rs

@@ -3,7 +3,7 @@ use fhevm_engine_common::keys::{FhevmKeys, SerializedFhevmKeys};
 use fhevm_engine_common::{healthz_server, metrics_server, telemetry};
 use tokio_util::sync::CancellationToken;
 
-use std::sync::Once;
+use std::sync::{Once, OnceLock};
 use tokio::task::JoinSet;
 
 pub mod daemon_cli;
@@ -49,29 +49,31 @@ pub fn start_runtime(
 
 // Used for testing as we would call `async_main()` multiple times.
 static TRACING_INIT: Once = Once::new();
+static OTLP_SETUP_ERROR: OnceLock<String> = OnceLock::new();
+static OTEL_GUARD: OnceLock<Option<telemetry::TracerProviderGuard>> = OnceLock::new();
 
 pub async fn async_main(
     args: daemon_cli::Args,
 ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
     TRACING_INIT.call_once(|| {
-        tracing_subscriber::fmt()
-            .json()
-            .with_level(true)
-            .with_max_level(args.log_level)
-            .init();
+        let otel_guard =
+            match telemetry::init_json_subscriber(args.log_level, &args.service_name, "otlp-layer")
+            {
+                Ok(guard) => guard,
+                Err(err) => {
+                    let _ = OTLP_SETUP_ERROR.set(err.to_string());
+                    None
+                }
+            };
+        let _ = OTEL_GUARD.set(otel_guard);
     });
+    if let Some(err) = OTLP_SETUP_ERROR.get() {
+        error!(error = %err, "Failed to setup OTLP");
+    }
 
     let cancel_token = CancellationToken::new();
     info!(target: "async_main", args = ?args, "Starting runtime with args");
 
-    let _otel_guard = match telemetry::init_otel(&args.service_name) {
-        Ok(otel_guard) => otel_guard,
-        Err(err) => {
-            error!(error = %err, "Failed to setup OTLP");
-            None
-        }
-    };
-
     let database_url = args.database_url.clone().unwrap_or_default();
     let health_check = health_check::HealthCheck::new(database_url);
 

coprocessor/fhevm-engine/transaction-sender/src/bin/transaction_sender.rs

@@ -286,11 +286,18 @@ async fn main() -> anyhow::Result<()> {
 
     let conf = parse_args();
 
-    tracing_subscriber::fmt()
-        .json()
-        .with_level(true)
-        .with_max_level(conf.log_level)
-        .init();
+    let mut otlp_setup_error: Option<String> = None;
+    let _otel_guard =
+        match telemetry::init_json_subscriber(conf.log_level, &conf.service_name, "otlp-layer") {
+            Ok(guard) => guard,
+            Err(err) => {
+                otlp_setup_error = Some(err.to_string());
+                None
+            }
+        };
+    if let Some(err) = otlp_setup_error {
+        error!(error = %err, "Failed to setup OTLP");
+    }
 
     let cancel_token = CancellationToken::new();
     install_signal_handlers(cancel_token.clone())?;
@@ -308,14 +315,6 @@ async fn main() -> anyhow::Result<()> {
         }
     };
 
-    let _otel_guard = match telemetry::init_otel(&conf.service_name) {
-        Ok(otel_guard) => otel_guard,
-        Err(err) => {
-            error!(error = %err, "Failed to setup OTLP");
-            None
-        }
-    };
-
     let abstract_signer: AbstractSigner;
     match conf.signer_type {
         SignerType::PrivateKey => {

coprocessor/fhevm-engine/zkproof-worker/src/bin/zkproof_worker.rs

@@ -82,13 +82,18 @@ pub fn parse_args() -> Args {
 async fn main() {
     let args = parse_args();
 
-    tracing_subscriber::fmt()
-        .json()
-        .with_current_span(true)
-        .with_span_list(false)
-        .with_level(true)
-        .with_max_level(args.log_level)
-        .init();
+    let mut otlp_setup_error: Option<String> = None;
+    let _otel_guard =
+        match telemetry::init_json_subscriber(args.log_level, &args.service_name, "otlp-layer") {
+            Ok(guard) => guard,
+            Err(err) => {
+                otlp_setup_error = Some(err.to_string());
+                None
+            }
+        };
+    if let Some(err) = otlp_setup_error {
+        error!(error = %err, "Failed to setup OTLP");
+    }
 
     let database_url = args.database_url.clone().unwrap_or_default();
 
@@ -103,14 +108,6 @@ async fn main() {
         pg_auto_explain_with_min_duration: args.pg_auto_explain_with_min_duration,
     };
 
-    let _otel_guard = match telemetry::init_otel(&args.service_name) {
-        Ok(otel_guard) => otel_guard,
-        Err(err) => {
-            error!(error = %err, "Failed to setup OTLP");
-            None
-        }
-    };
-
     let cancel_token = CancellationToken::new();
     let Some(service) = ZkProofService::create(conf, cancel_token.child_token()).await else {
         error!("Failed to create zkproof service");

coprocessor/fhevm-engine/zkproof-worker/src/verifier.rs

@@ -509,7 +509,7 @@ fn expand_verified_list(
 
 /// Creates a ciphertext
 #[tracing::instrument(skip_all, fields(
-    operation = "create_handle",
+    operation = "create_ciphertext",
     ct_type = tracing::field::Empty,
     ct_idx = ct_idx,
     chain_id = %aux_data.chain_id,