diff --git a/crates/app/src/monitoringapi/checker.rs b/crates/app/src/monitoringapi/checker.rs index 72a0622a7..03813755d 100644 --- a/crates/app/src/monitoringapi/checker.rs +++ b/crates/app/src/monitoringapi/checker.rs @@ -8,7 +8,7 @@ use pluto_eth2api::EthBeaconNodeApiClient; use pluto_p2p::p2p_context::P2PContext; use tokio::{sync::mpsc, time::MissedTickBehavior}; use tokio_util::sync::CancellationToken; -use tracing::{error, warn}; +use tracing::{Instrument as _, Span, error, warn}; use super::{ metrics::MONITORING_METRICS, @@ -52,17 +52,24 @@ pub fn start_ready_checker( let readiness = ReadyState::new(); // Both background tasks are detached; their lifecycle is bound to `ct` and // they stop when the token is cancelled. - let _version_task = tokio::spawn(run_beacon_node_version_metric( - beacon_node.clone(), - ct.clone(), - )); - let _task = tokio::spawn(run_ready_checker( - p2p_context, - beacon_node, - validator_api_calls, - ct, - readiness.clone(), - )); + // + // `tokio::spawn` starts a task with an empty span stack, so both futures + // are re-attached to the caller's span; charon's monitoring API sets no + // topic of its own and runs as a lifecycle hook, which puts its logs on + // `app-start`. + let _version_task = tokio::spawn( + run_beacon_node_version_metric(beacon_node.clone(), ct.clone()).instrument(Span::current()), + ); + let _task = tokio::spawn( + run_ready_checker( + p2p_context, + beacon_node, + validator_api_calls, + ct, + readiness.clone(), + ) + .instrument(Span::current()), + ); readiness } diff --git a/crates/app/src/node/mod.rs b/crates/app/src/node/mod.rs index ad309522f..5db6a8ae1 100644 --- a/crates/app/src/node/mod.rs +++ b/crates/app/src/node/mod.rs @@ -23,6 +23,7 @@ pub use config::AppConfig; use std::{ collections::HashMap, + future::Future, sync::Arc, time::{Duration, SystemTime}, }; @@ -37,6 +38,7 @@ use pluto_testutil::{ }; use tokio::task::JoinSet; use tokio_util::sync::CancellationToken; +use tracing::Instrument as _; use behaviour::{CoreBehaviour, CoreHandles}; use wire::{ParSigExSeam, SlotTickFn, ValidatorInfo, WireInputs, WiredComponents}; @@ -220,8 +222,32 @@ impl App { } } +/// Puts a long-lived background task under the `app-start` topic. +/// +/// Charon's lifecycle manager hands every background hook a +/// `log.WithTopic(context.Background(), "app-start")` context +/// (`app/lifecycle/hook.go`), so a task that never sets a topic of its own +/// still reports under `app-start`. `tokio::spawn` has no such inheritance — +/// it starts the task with an empty span stack — so the span is attached here +/// instead, at the one place background tasks are started. +/// +/// Tasks that open their own topic span (`sched`, `tracker`, `health`, …) +/// shadow this one, exactly as a nested `log.WithTopic` does in Go. +fn background(task: F) -> tracing::instrument::Instrumented { + task.instrument(tracing::debug_span!("app-start", topic = "app-start")) +} + /// Loads the cluster lock + key, builds the consensus component and P2P /// behaviours, wires the core workflow, and drives the node. +/// +/// Carries the `app-start` topic as the catch-all for log metrics not +/// attributed to a more specific component (mirrors charon's `app.Run`). +#[tracing::instrument( + name = "app-start", + level = "debug", + skip_all, + fields(topic = "app-start") +)] async fn run(config: AppConfig, ct: CancellationToken) -> Result<(), AppError> { // ---- (1) Load cluster lock + key, derive peers and this node's index ---- // @@ -690,37 +716,37 @@ async fn run_lifecycle( // Supervise the scheduler actor alongside the other long-lived tasks so // its exit triggers node shutdown (it only exits on cancellation). - tasks.extend([async move { + tasks.extend([background(async move { let _ = scheduler_task.await; Ok::<(), AppError>(()) - }]); + })]); // Swarm drive loop (push-based routing inside behaviours). { let ct = ct.clone(); - tasks.spawn(async move { + tasks.spawn(background(async move { drive_network(node, ct).await; Ok(()) - }); + })); } // ParSigDB trim task. { let parsigdb = Arc::clone(&parsigdb); - tasks.spawn(async move { + tasks.spawn(background(async move { parsigdb.trim(parsigdb_deadliner_rx).await; Ok(()) - }); + })); } // Networked inclusion checker: polls the beacon node once per due slot and // resolves each tracked duty's on-chain inclusion step. { let ct = ct.clone(); - tasks.spawn(async move { + tasks.spawn(background(async move { inclusion_checker.run(ct).await; Ok(()) - }); + })); } // Private-key lock maintenance loop. Only spawn `run` when locking is @@ -730,7 +756,9 @@ async fn run_lifecycle( let svc = Arc::clone(svc); // A lock-maintenance failure fails the run (Charon parity); a graceful // `close()` returns `Ok`. - tasks.spawn(async move { svc.run().await.map_err(|e| AppError::PrivKeyLock(e.into())) }); + tasks.spawn(background(async move { + svc.run().await.map_err(|e| AppError::PrivKeyLock(e.into())) + })); } // ---- Monitoring API ---- @@ -770,10 +798,10 @@ async fn run_lifecycle( Box::new(health::ViseGatherer), num_validators, ); - tasks.spawn(async move { + tasks.spawn(background(async move { checker.run(ct).await; Ok(()) - }); + })); } // Validator API axum server. Each request bumps the readiness "vc @@ -787,20 +815,20 @@ async fn run_lifecycle( } }, )); - tasks.spawn(serve_validator_api( + tasks.spawn(background(serve_validator_api( validator_api_addr, validator_api_router, ct.clone(), - )); + ))); // Monitoring HTTP server (metrics + livez + readyz). - tasks.spawn(serve_monitoring_api( + tasks.spawn(background(serve_monitoring_api( monitoring_addr, monitoringapi::router_with_state( monitoringapi::MonitoringState::new(readiness).with_labels(monitoring_labels), ), ct.clone(), - )); + ))); // Supervise: stop on cancellation or first task completion. A failed task // fails the whole run (Charon parity). diff --git a/crates/app/src/node/wire.rs b/crates/app/src/node/wire.rs index 12b2ef104..916f5bc8d 100644 --- a/crates/app/src/node/wire.rs +++ b/crates/app/src/node/wire.rs @@ -48,6 +48,7 @@ use pluto_eth2api::{ }; use pluto_featureset::{Feature, FeatureSet, Status}; use tokio_util::sync::CancellationToken; +use tracing::Instrument as _; use crate::node::AppError; @@ -518,9 +519,13 @@ pub async fn wire_core_workflow( let duty = duty.clone(); // The core's callback is sync but `inclusion_checked` is // async, so hand the event to the runtime. - tokio::spawn(async move { - tracker.inclusion_checked(duty, pubkey, err).await; - }); + let span = tracing::Span::current(); + tokio::spawn( + async move { + tracker.inclusion_checked(duty, pubkey, err).await; + } + .instrument(span), + ); }), tracker_feature_set, ) @@ -635,26 +640,35 @@ pub async fn wire_core_workflow( move |duty: Duty, value: pbcore::UnsignedDataSet| { let dutydb = Arc::clone(&dutydb); let tracker = Arc::clone(&tracker); - tokio::spawn(async move { - let core_set = - match unsigneddata::unsigned_data_set_from_proto(&duty.duty_type, &value) { + // `tokio::spawn` starts the task with an empty span stack, so + // re-attach the caller's span. + let span = tracing::Span::current(); + tokio::spawn( + async move { + let core_set = match unsigneddata::unsigned_data_set_from_proto( + &duty.duty_type, + &value, + ) { Ok(set) => set, Err(err) => { tracing::warn!(?err, "dutydb: decode unsigned data set"); return; } }; - let pubkeys: Vec = core_set.keys().copied().collect(); - // Logged before the error moves into the tracker's `Arc`. - let step_err = match dutydb.store(duty.clone(), core_set).await { - Ok(()) => None, - Err(err) => { - tracing::warn!(?err, "dutydb: store"); - Some(owned_step_err(err)) - } - }; - tracker.duty_db_stored(duty, &pubkeys, step_err).await; - }); + let pubkeys: Vec = core_set.keys().copied().collect(); + // Logged before the error moves into the tracker's + // `Arc`. + let step_err = match dutydb.store(duty.clone(), core_set).await { + Ok(()) => None, + Err(err) => { + tracing::warn!(?err, "dutydb: store"); + Some(owned_step_err(err)) + } + }; + tracker.duty_db_stored(duty, &pubkeys, step_err).await; + } + .instrument(span), + ); Ok(()) }, )); diff --git a/crates/app/src/sse/mod.rs b/crates/app/src/sse/mod.rs index e5686bcec..36a58b59c 100644 --- a/crates/app/src/sse/mod.rs +++ b/crates/app/src/sse/mod.rs @@ -17,6 +17,7 @@ use chrono::{DateTime, Utc}; use futures::StreamExt; use tokio::sync; use tokio_util::{future::FutureExt, sync::CancellationToken}; +use tracing::Instrument as _; use pluto_eth2api::{BeaconNodeEvent, EthBeaconNodeApiClient, EventTopic}; @@ -118,8 +119,9 @@ impl SseListenerBuilder { let (events_tx, events_rx) = sync::mpsc::channel(CHANNEL_BUFFER_SIZE); let (msg_tx, msg_rx) = sync::mpsc::channel(CHANNEL_BUFFER_SIZE); - tokio::spawn(run_pump(client, addr, events_tx, ct.clone())); - tokio::spawn(actor.run(events_rx, msg_rx, ct)); + let span = tracing::Span::current(); + tokio::spawn(run_pump(client, addr, events_tx, ct.clone()).instrument(span.clone())); + tokio::spawn(actor.run(events_rx, msg_rx, ct).instrument(span)); Ok(SseListenerHandle { sender: msg_tx }) } diff --git a/crates/cli/src/commands/relay.rs b/crates/cli/src/commands/relay.rs index a59bc72ad..073fe9306 100644 --- a/crates/cli/src/commands/relay.rs +++ b/crates/cli/src/commands/relay.rs @@ -196,6 +196,7 @@ pub struct RelayP2PArgs { pub disable_reuseport: bool, } +#[tracing::instrument(name = "relay", level = "debug", skip_all, fields(topic = "relay"))] pub async fn run( config: pluto_relay_server::config::Config, ct: CancellationToken, diff --git a/crates/cli/src/commands/test/beacon.rs b/crates/cli/src/commands/test/beacon.rs index a75265052..7eea7ba86 100644 --- a/crates/cli/src/commands/test/beacon.rs +++ b/crates/cli/src/commands/test/beacon.rs @@ -29,6 +29,7 @@ use tokio::{ time::{Instant, interval, interval_at, sleep}, }; use tokio_util::sync::CancellationToken; +use tracing::Instrument as _; const THRESHOLD_BEACON_MEASURE_AVG: StdDuration = StdDuration::from_millis(40); const THRESHOLD_BEACON_MEASURE_POOR: StdDuration = StdDuration::from_millis(100); @@ -316,10 +317,13 @@ pub async fn run( let endpoint = endpoint.clone(); let shutdown = shutdown.clone(); - set.spawn(async move { - let results = test_single_beacon(&args, &queued, &endpoint, shutdown).await; - (endpoint, results) - }); + set.spawn( + async move { + let results = test_single_beacon(&args, &queued, &endpoint, shutdown).await; + (endpoint, results) + } + .instrument(tracing::Span::current()), + ); } let mut test_results: HashMap> = HashMap::new(); @@ -633,9 +637,9 @@ async fn beacon_ping_load_test( _ = interval.tick() => { let cancel = load_cancel.clone(); let target = target.to_string(); - set.spawn(async move { - ping_beacon_continuously(cancel, target).await - }); + set.spawn( + ping_beacon_continuously(cancel, target).instrument(tracing::Span::current()), + ); } } } @@ -811,10 +815,10 @@ async fn beacon_simulation_test( tracing::info!("Starting general cluster requests..."); let cluster_cancel = sim_cancel.clone(); let cluster_target = target.to_string(); - let cluster_handle = - tokio::spawn( - async move { single_cluster_simulation(cluster_cancel, &cluster_target).await }, - ); + let cluster_handle = tokio::spawn( + async move { single_cluster_simulation(cluster_cancel, &cluster_target).await } + .instrument(tracing::Span::current()), + ); // Validator simulations let mut validator_set = tokio::task::JoinSet::new(); @@ -833,9 +837,10 @@ async fn beacon_simulation_test( let cancel = sim_cancel.clone(); let target = target.to_string(); let intensity = params.request_intensity; - validator_set.spawn(async move { - single_validator_simulation(cancel, &target, intensity, sync_duties).await - }); + validator_set.spawn( + async move { single_validator_simulation(cancel, &target, intensity, sync_duties).await } + .instrument(tracing::Span::current()), + ); } let proposal_duties = DutiesPerformed { @@ -852,9 +857,12 @@ async fn beacon_simulation_test( let cancel = sim_cancel.clone(); let target = target.to_string(); let intensity = params.request_intensity; - validator_set.spawn(async move { - single_validator_simulation(cancel, &target, intensity, proposal_duties).await - }); + validator_set.spawn( + async move { + single_validator_simulation(cancel, &target, intensity, proposal_duties).await + } + .instrument(tracing::Span::current()), + ); } let attester_duties = DutiesPerformed { @@ -871,9 +879,12 @@ async fn beacon_simulation_test( let cancel = sim_cancel.clone(); let target = target.to_string(); let intensity = params.request_intensity; - validator_set.spawn(async move { - single_validator_simulation(cancel, &target, intensity, attester_duties).await - }); + validator_set.spawn( + async move { + single_validator_simulation(cancel, &target, intensity, attester_duties).await + } + .instrument(tracing::Span::current()), + ); } tracing::info!("Waiting for simulation to complete..."); @@ -1094,9 +1105,10 @@ async fn single_validator_simulation( let att_handle = if duties.attestation { let cancel = cancel.clone(); let target = target.to_string(); - Some(tokio::spawn(async move { - attestation_duty(cancel, &target, intensity.attestation_duty).await - })) + Some(tokio::spawn( + async move { attestation_duty(cancel, &target, intensity.attestation_duty).await } + .instrument(tracing::Span::current()), + )) } else { None }; @@ -1105,9 +1117,10 @@ async fn single_validator_simulation( let agg_handle = if duties.aggregation { let cancel = cancel.clone(); let target = target.to_string(); - Some(tokio::spawn(async move { - aggregation_duty(cancel, &target, intensity.aggregator_duty).await - })) + Some(tokio::spawn( + async move { aggregation_duty(cancel, &target, intensity.aggregator_duty).await } + .instrument(tracing::Span::current()), + )) } else { None }; @@ -1116,9 +1129,10 @@ async fn single_validator_simulation( let prop_handle = if duties.proposal { let cancel = cancel.clone(); let target = target.to_string(); - Some(tokio::spawn(async move { - proposal_duty(cancel, &target, intensity.proposal_duty).await - })) + Some(tokio::spawn( + async move { proposal_duty(cancel, &target, intensity.proposal_duty).await } + .instrument(tracing::Span::current()), + )) } else { None }; @@ -1131,20 +1145,23 @@ async fn single_validator_simulation( if duties.sync_committee { let cancel = cancel.clone(); let target = target.to_string(); - tokio::spawn(async move { - sync_committee_duties( - cancel, - &target, - intensity.sync_committee_submit, - intensity.sync_committee_subscribe, - intensity.sync_committee_contribution, - sc_msg_tx, - sc_produce_tx, - sc_sub_tx, - sc_contrib_tx, - ) - .await; - }); + tokio::spawn( + async move { + sync_committee_duties( + cancel, + &target, + intensity.sync_committee_submit, + intensity.sync_committee_subscribe, + intensity.sync_committee_contribution, + sc_msg_tx, + sc_produce_tx, + sc_sub_tx, + sc_contrib_tx, + ) + .await; + } + .instrument(tracing::Span::current()), + ); } else { drop(sc_sub_tx); drop(sc_msg_tx); @@ -1451,16 +1468,28 @@ async fn sync_committee_duties( ) { let c1 = cancel.clone(); let t1 = target.to_string(); - tokio::spawn(async move { - sync_committee_contribution_duty(c1, &t1, tick_time_contribution, produce_tx, contrib_tx) + tokio::spawn( + async move { + sync_committee_contribution_duty( + c1, + &t1, + tick_time_contribution, + produce_tx, + contrib_tx, + ) .await; - }); + } + .instrument(tracing::Span::current()), + ); let c2 = cancel.clone(); let t2 = target.to_string(); - tokio::spawn(async move { - sync_committee_message_duty(c2, &t2, tick_time_submit, msg_tx).await; - }); + tokio::spawn( + async move { + sync_committee_message_duty(c2, &t2, tick_time_submit, msg_tx).await; + } + .instrument(tracing::Span::current()), + ); // Subscribe loop if cancel diff --git a/crates/cli/src/commands/test/mev.rs b/crates/cli/src/commands/test/mev.rs index 22e7c4fb1..ab1004cb6 100644 --- a/crates/cli/src/commands/test/mev.rs +++ b/crates/cli/src/commands/test/mev.rs @@ -5,7 +5,7 @@ use std::{collections::HashMap, io::Write, time::Duration}; use reqwest::{Method, StatusCode}; use tokio::{task::JoinSet, time::Instant}; use tokio_util::sync::CancellationToken; -use tracing::info; +use tracing::{Instrument as _, info}; use super::{ AllCategoriesResult, TestCategory, TestCategoryResult, TestConfigArgs, TestResult, TestVerdict, @@ -203,11 +203,14 @@ async fn test_all_mevs( let endpoint = endpoint.clone(); let token = token.clone(); - join_set.spawn(async move { - let results = test_single_mev(&queued_tests, &conf, &endpoint, token).await; - let relay_name = format_mev_relay_name(&endpoint); - (relay_name, results) - }); + join_set.spawn( + async move { + let results = test_single_mev(&queued_tests, &conf, &endpoint, token).await; + let relay_name = format_mev_relay_name(&endpoint); + (relay_name, results) + } + .instrument(tracing::Span::current()), + ); } let all_results = join_set.join_all().await; @@ -228,18 +231,21 @@ async fn test_single_mev( let conf = conf.clone(); let target = target.to_string(); - join_set.spawn(async move { - let tc_name = test_case.test_case_name(); - tokio::select! { - _ = token.cancelled() => { - let tr = TestResult::new(tc_name.name); - tr.fail(CliError::TimeoutInterrupted) - } - r = test_case.run(&target, &conf) => { - r + join_set.spawn( + async move { + let tc_name = test_case.test_case_name(); + tokio::select! { + _ = token.cancelled() => { + let tr = TestResult::new(tc_name.name); + tr.fail(CliError::TimeoutInterrupted) + } + r = test_case.run(&target, &conf) => { + r + } } } - }); + .instrument(tracing::Span::current()), + ); } join_set.join_all().await diff --git a/crates/cli/src/commands/test/peers.rs b/crates/cli/src/commands/test/peers.rs index ab449bd1e..7125284c6 100644 --- a/crates/cli/src/commands/test/peers.rs +++ b/crates/cli/src/commands/test/peers.rs @@ -32,6 +32,7 @@ use reqwest::Method; use sha2::{Digest, Sha256}; use tokio::task::JoinHandle; use tokio_util::sync::CancellationToken; +use tracing::Instrument as _; use super::{ AllCategoriesResult, TestCaseName, TestCategory, TestCategoryResult, TestConfigArgs, @@ -477,25 +478,29 @@ async fn run_relay_http_tests( let url = relay.to_string(); let ct = ct.clone(); let queued = queued.to_vec(); - tokio::spawn(async move { - let key = format!("relay {url}"); - let mut target_results = Vec::new(); - for test in &queued { - if ct.is_cancelled() { - target_results - .push(TestResult::new(test.name).fail(CliError::TimeoutInterrupted)); - continue; + tokio::spawn( + async move { + let key = format!("relay {url}"); + let mut target_results = Vec::new(); + for test in &queued { + if ct.is_cancelled() { + target_results.push( + TestResult::new(test.name).fail(CliError::TimeoutInterrupted), + ); + continue; + } + let result = match test.name { + "PingRelay" => relay_ping_test(&url, &ct).await, + "PingMeasureRelay" => relay_ping_measure_test(&url, &ct).await, + _ => TestResult::new(test.name) + .fail(TestResultError::from_string("unsupported relay test")), + }; + target_results.push(result); } - let result = match test.name { - "PingRelay" => relay_ping_test(&url, &ct).await, - "PingMeasureRelay" => relay_ping_measure_test(&url, &ct).await, - _ => TestResult::new(test.name) - .fail(TestResultError::from_string("unsupported relay test")), - }; - target_results.push(result); + (key, target_results) } - (key, target_results) - }) + .instrument(tracing::Span::current()), + ) }) .collect(); diff --git a/crates/cli/src/main.rs b/crates/cli/src/main.rs index 1f9aa47c5..add6f3f0f 100644 --- a/crates/cli/src/main.rs +++ b/crates/cli/src/main.rs @@ -9,7 +9,7 @@ use clap::FromArgMatches; use cli::{AlphaCommands, Cli, Commands, CreateCommands, TestCommands, UnsafeCommands}; use std::process::ExitCode; use tokio_util::sync::CancellationToken; -use tracing::error; +use tracing::{Instrument as _, error}; mod ascii; mod cli; @@ -36,16 +36,17 @@ async fn main() -> ExitCode { return ExitCode::FAILURE; } }; - cli.tracing.warn_unused(); + let cmd_span = tracing::debug_span!("cmd", topic = "cmd"); + cmd_span.in_scope(|| cli.tracing.warn_unused()); - let result = run(cli.command).await; + let result = run(cli.command).instrument(cmd_span.clone()).await; let exit = match &result { Ok(()) => ExitCode::SUCCESS, - Err(err) => { + Err(err) => cmd_span.in_scope(|| { error!(error = ?err, "command exited with error"); ExitCode::FAILURE - } + }), }; if let Some(loki) = loki { diff --git a/crates/consensus/src/qbft/component.rs b/crates/consensus/src/qbft/component.rs index ad3bf250e..1e9f161e6 100644 --- a/crates/consensus/src/qbft/component.rs +++ b/crates/consensus/src/qbft/component.rs @@ -13,6 +13,7 @@ use prost::{Message, Name}; use prost_types::Any; use tokio::{sync::mpsc, task::JoinHandle}; use tokio_util::sync::CancellationToken; +use tracing::Instrument as _; use crate::{ instance::InstanceIo, @@ -478,22 +479,26 @@ impl Consensus { .expect("start must be called exactly once"); let instances = Arc::clone(&self.instances); - tokio::spawn(async move { - loop { - tokio::select! { - () = ct.cancelled() => return, - duty = expired_rx.recv() => match duty { - Some(duty) => { - instances - .lock() - .unwrap_or_else(PoisonError::into_inner) - .remove(&duty); - } - None => return, - }, + let span = tracing::debug_span!("qbft", topic = "qbft"); + tokio::spawn( + async move { + loop { + tokio::select! { + () = ct.cancelled() => return, + duty = expired_rx.recv() => match duty { + Some(duty) => { + instances + .lock() + .unwrap_or_else(PoisonError::into_inner) + .remove(&duty); + } + None => return, + }, + } } } - }) + .instrument(span), + ) } /// Returns existing instance I/O for `duty`, or creates an empty one. diff --git a/crates/consensus/src/qbft/definition.rs b/crates/consensus/src/qbft/definition.rs index 2bbab6f5c..1cceaa084 100644 --- a/crates/consensus/src/qbft/definition.rs +++ b/crates/consensus/src/qbft/definition.rs @@ -56,7 +56,17 @@ pub(crate) fn new_definition(config: DefinitionConfig) -> qbft::Definition( consensus: &Consensus, duty: Duty, @@ -167,6 +169,7 @@ where } /// Starts participating in a duty without a local proposal value. +#[tracing::instrument(name = "qbft", level = "debug", skip_all, fields(topic = "qbft"))] pub(crate) async fn participate( consensus: &Consensus, duty: Duty, @@ -195,6 +198,7 @@ pub(crate) async fn participate( } /// Runs one consensus instance and publishes its completion result. +#[tracing::instrument(name = "qbft", level = "debug", skip_all, fields(topic = "qbft"))] pub(crate) async fn run_instance( consensus: &Consensus, duty: Duty, @@ -263,43 +267,53 @@ async fn run_instance_inner( Sniffer::new(i64::try_from(nodes).expect("node count fits i64"), peer_idx), )); + // `JoinSet::spawn` starts each task with an empty span stack, just like + // `tokio::spawn`, so the `qbft` topic opened by this function would not + // reach any of the instance's subtasks. Charon gets this for free by + // passing the instance `ctx` into every goroutine; here the span is + // re-attached to each spawned future by hand. + let qbft_span = tracing::Span::current(); + let mut tasks = JoinSet::new(); - tasks.spawn(bridge_mpsc_to_crossbeam( - instance_ct.clone(), - inner_recv_rx, - core_recv_tx, - )); - tasks.spawn(bridge_mpsc_to_crossbeam( - instance_ct.clone(), - hash_rx, - core_hash_tx, - )); - tasks.spawn(bridge_mpsc_to_crossbeam( - instance_ct.clone(), - verify_rx, - core_verify_tx, - )); + tasks.spawn( + bridge_mpsc_to_crossbeam(instance_ct.clone(), inner_recv_rx, core_recv_tx) + .instrument(qbft_span.clone()), + ); + tasks.spawn( + bridge_mpsc_to_crossbeam(instance_ct.clone(), hash_rx, core_hash_tx) + .instrument(qbft_span.clone()), + ); + tasks.spawn( + bridge_mpsc_to_crossbeam(instance_ct.clone(), verify_rx, core_verify_tx) + .instrument(qbft_span.clone()), + ); { let transport = Arc::clone(&transport); let instance_ct = instance_ct.clone(); let transport_error = Arc::clone(&transport_error); - tasks.spawn(async move { - if let Err(err) = transport.process_receives(instance_ct, outer_rx).await { - *transport_error - .lock() - .unwrap_or_else(PoisonError::into_inner) = Some(err.to_string()); + tasks.spawn( + async move { + if let Err(err) = transport.process_receives(instance_ct, outer_rx).await { + *transport_error + .lock() + .unwrap_or_else(PoisonError::into_inner) = Some(err.to_string()); + } } - }); + .instrument(qbft_span.clone()), + ); } { let instance_ct = instance_ct.clone(); let core_cts = Arc::clone(&core_cts); - tasks.spawn(async move { - instance_ct.cancelled().await; - core_cts.cancel(); - }); + tasks.spawn( + async move { + instance_ct.cancelled().await; + core_cts.cancel(); + } + .instrument(qbft_span.clone()), + ); } let decide_callback: DecideCallback = { @@ -377,7 +391,11 @@ async fn run_instance_inner( let core_ct_for_run = core_ct.clone(); let core_duty = duty.clone(); + // The blocking core runs the `Definition` callbacks, which log their own + // warnings; entering the span keeps those on `topic="qbft"`. + let core_span = qbft_span.clone(); let core_result = tokio::task::spawn_blocking(move || { + let _entered = core_span.enter(); qbft::run( &core_ct_for_run, &def, diff --git a/crates/core/src/bcast/mod.rs b/crates/core/src/bcast/mod.rs index 0b3d46382..79cb58495 100644 --- a/crates/core/src/bcast/mod.rs +++ b/crates/core/src/bcast/mod.rs @@ -222,6 +222,7 @@ impl Broadcaster { /// success record the broadcast count and submission delay. Internal-only /// duties (randao, prepare-aggregator, prepare-sync-contribution) are /// no-ops; deprecated and unknown duty types return an error. + #[tracing::instrument(name = "bcast", level = "debug", skip_all, fields(topic = "bcast"))] pub async fn broadcast(&self, mut duty: Duty, set: SignedDataSet) -> Result<()> { match duty.duty_type { DutyType::Attester => self.broadcast_attester(&duty, &set).await?, diff --git a/crates/core/src/bcast/recast.rs b/crates/core/src/bcast/recast.rs index be4d2ee3b..595ff6576 100644 --- a/crates/core/src/bcast/recast.rs +++ b/crates/core/src/bcast/recast.rs @@ -92,6 +92,7 @@ impl Recaster { } /// Called when new slots tick. + #[tracing::instrument(name = "bcast", level = "debug", skip_all, fields(topic = "bcast"))] pub async fn slot_ticked(&self, slot: Slot) -> Result<()> { if !slot.first_in_epoch() { return Ok(()); diff --git a/crates/core/src/deadline/mod.rs b/crates/core/src/deadline/mod.rs index c4f569656..915cceb59 100644 --- a/crates/core/src/deadline/mod.rs +++ b/crates/core/src/deadline/mod.rs @@ -48,6 +48,7 @@ use tokio::{ time::sleep, }; use tokio_util::sync::CancellationToken; +use tracing::Instrument as _; /// A safe far-future duration (~10 years) for timeout calculations. /// Using Duration::MAX can cause panics when computing Instant::now() + @@ -214,7 +215,7 @@ impl DeadlinerTask { curr_duty: Duty::new(SlotNumber::new(0), DutyType::Unknown), curr_deadline: DateTime::::MAX_UTC, }; - tokio::spawn(task.run_task()); + tokio::spawn(task.run_task().instrument(tracing::Span::current())); let handle = DeadlinerHandle { cancel_token, diff --git a/crates/core/src/scheduler.rs b/crates/core/src/scheduler.rs index 431e9217e..370e77743 100644 --- a/crates/core/src/scheduler.rs +++ b/crates/core/src/scheduler.rs @@ -6,6 +6,7 @@ use std::{ use backon::{BackoffBuilder, Retryable}; use tokio::{sync, task::JoinHandle}; use tokio_util::{future::FutureExt, sync::CancellationToken}; +use tracing::Instrument as _; use crate::{scheduler::metrics::SCHEDULER_METRICS, types}; use pluto_eth2api::{v1, valcache}; @@ -122,7 +123,7 @@ impl SchedulerBuilder { // TODO: We might want to return a handle so clients can `.abort()` them // to drop the subscription let label: Arc = Arc::from(label.as_ref()); - tokio::spawn(async move { + let pump = async move { loop { match rx.recv().await { Ok(slot) => { @@ -130,11 +131,12 @@ impl SchedulerBuilder { // delay later slots for this subscriber. let fut = f(&slot); let label = Arc::clone(&label); - tokio::spawn(async move { + let emit = async move { if let Err(err) = fut.await { tracing::error!(err = ?err, slot = %slot.slot, label = &*label, "Emit scheduled slot event"); } - }); + }; + tokio::spawn(emit.instrument(sched_span())); } // NOTE: Handlers are spawned per event above, so the // receive loop drains immediately. Lag therefore no longer @@ -158,7 +160,8 @@ impl SchedulerBuilder { Err(sync::broadcast::error::RecvError::Closed) => break, } } - }); + }; + tokio::spawn(pump.instrument(sched_span())); } /// Subscribes a callback function for triggered duties. @@ -171,7 +174,7 @@ impl SchedulerBuilder { let mut rx = self.duty_broadcast.subscribe(); let label: Arc = Arc::from(label.as_ref()); - tokio::spawn(async move { + let pump = async move { loop { match rx.recv().await { Ok((duty, set)) => { @@ -182,11 +185,12 @@ impl SchedulerBuilder { // subscriber. let fut = f(&duty, &set); let label = Arc::clone(&label); - tokio::spawn(async move { + let trigger = async move { if let Err(err) = fut.await { tracing::error!(err = ?err, %duty, label = &*label, "Trigger duty subscriber error"); } - }); + }; + tokio::spawn(trigger.instrument(sched_span())); } // NOTE: Same as in `subscribe_slot` Err(sync::broadcast::error::RecvError::Lagged(skipped)) => { @@ -196,7 +200,8 @@ impl SchedulerBuilder { Err(sync::broadcast::error::RecvError::Closed) => break, } } - }); + }; + tokio::spawn(pump.instrument(sched_span())); } /// Add a source of chain reorgs to the scheduler. @@ -298,6 +303,18 @@ impl SchedulerHandle { } } +/// The scheduler's `sched` topic span. +/// +/// Charon opens it once in `Scheduler.Run` and every goroutine started from +/// there inherits it through `context.Context`, so subscriber and slot-ticker +/// errors are all reported under `sched`. `tokio::spawn` starts a task with an +/// empty span stack instead, and several of these tasks are started from +/// wiring code rather than from the actor loop, so each opens the span itself +/// rather than inheriting one. +fn sched_span() -> tracing::Span { + tracing::debug_span!("sched", topic = "sched") +} + struct SchedulerActor { client: pluto_eth2api::EthBeaconNodeApiClient, validator_cache: valcache::ValidatorCache, @@ -315,6 +332,7 @@ struct SchedulerActor { } impl SchedulerActor { + #[tracing::instrument(name = "sched", level = "debug", skip_all, fields(topic = "sched"))] async fn run( mut self, mut slot_rx: sync::mpsc::Receiver, @@ -429,23 +447,26 @@ impl SchedulerActor { let ct = ct.clone(); let slot = slot.clone(); let broadcast = self.duty_broadcast.clone(); - tokio::spawn(async move { - if delay_slot_offset(&slot, &duty) - .with_cancellation_token_owned(ct) - .await - .is_none() - { - // Cancelled early - return; - } + tokio::spawn( + async move { + if delay_slot_offset(&slot, &duty) + .with_cancellation_token_owned(ct) + .await + .is_none() + { + // Cancelled early + return; + } - SCHEDULER_METRICS.duty_total[&duty.duty_type.to_string()] - .inc_by(def_set.len() as u64); + SCHEDULER_METRICS.duty_total[&duty.duty_type.to_string()] + .inc_by(def_set.len() as u64); - // NOTE: Ignore send errors, it means that there are no - // subscribers. - let _ = broadcast.send((duty.clone(), def_set.clone())); - }); + // NOTE: Ignore send errors, it means that there are no + // subscribers. + let _ = broadcast.send((duty.clone(), def_set.clone())); + } + .instrument(sched_span()), + ); } if slot.last_in_epoch() @@ -657,7 +678,7 @@ async fn new_slot_ticker( }; let (tx, rx) = sync::mpsc::channel(CHANNEL_BUFFER_SIZE); - tokio::spawn(async move { + let ticker = async move { let mut slot = current_slot(); loop { @@ -695,7 +716,8 @@ async fn new_slot_ticker( slot = next_slot; } - }); + }; + tokio::spawn(ticker.instrument(sched_span())); Ok(rx) } diff --git a/crates/core/src/sigagg.rs b/crates/core/src/sigagg.rs index d46f8366b..d51fac489 100644 --- a/crates/core/src/sigagg.rs +++ b/crates/core/src/sigagg.rs @@ -144,6 +144,7 @@ impl Aggregator { /// /// If aggregation fails for any validator the entire call returns that /// error immediately — no partial results are emitted. + #[tracing::instrument(name = "sigagg", level = "debug", skip_all, fields(topic = "sigagg"))] pub async fn aggregate( &self, duty: &Duty, diff --git a/crates/core/src/tracker/inclusion.rs b/crates/core/src/tracker/inclusion.rs index 8d675bd8b..9214022c2 100644 --- a/crates/core/src/tracker/inclusion.rs +++ b/crates/core/src/tracker/inclusion.rs @@ -685,6 +685,10 @@ impl InclusionChecker { /// Drives inclusion checking until `cancel` fires: once per due slot, ask /// the beacon node whether that slot produced a block, feed the verdict to /// the core, then trim submissions old enough to count as missed. + /// + /// Runs under the `tracker` topic, matching charon's + /// `InclusionChecker.Run`. + #[tracing::instrument(name = "tracker", level = "debug", skip_all, fields(topic = "tracker"))] pub async fn run(self: Arc, cancel: CancellationToken) { let mut ticker = tokio::time::interval(Duration::from_secs(1)); let mut checked_slot: Option = None; diff --git a/crates/core/src/tracker/mod.rs b/crates/core/src/tracker/mod.rs index 02e5532bb..859aaefb2 100644 --- a/crates/core/src/tracker/mod.rs +++ b/crates/core/src/tracker/mod.rs @@ -462,6 +462,7 @@ impl TrackerService { ); } + #[tracing::instrument(name = "tracker", level = "debug", skip_all, fields(topic = "tracker"))] async fn run(mut self) { let mut events: HashMap> = HashMap::new(); diff --git a/crates/core/src/validatorapi/router.rs b/crates/core/src/validatorapi/router.rs index 33208fccf..498115c61 100644 --- a/crates/core/src/validatorapi/router.rs +++ b/crates/core/src/validatorapi/router.rs @@ -250,9 +250,21 @@ pub fn new_router( ) .route("/eth/v1/node/version", get(node_version)) .fallback(proxy_handler) + // Attach the `vapi` topic to every request so warn/error logs emitted + // while handling it are counted under `app_log_{warn,error}_total{topic="vapi"}`. + .layer(middleware::from_fn(with_vapi_topic)) .with_state(state) } +/// Middleware that runs each request handler inside a `vapi` topic span so log +/// metrics are attributed to the validator API component. +async fn with_vapi_topic(req: Request, next: Next) -> Response { + use tracing::Instrument as _; + + let span = tracing::debug_span!("vapi", topic = "vapi"); + next.run(req).instrument(span).await +} + async fn attester_duties( State(state): State>, Path(epoch): Path, diff --git a/crates/dkg/src/dkg.rs b/crates/dkg/src/dkg.rs index 3b8602261..8a700a6a1 100644 --- a/crates/dkg/src/dkg.rs +++ b/crates/dkg/src/dkg.rs @@ -7,7 +7,7 @@ use pluto_app::{privkeylock, utils::UtilsError}; use pluto_core::version; use tokio::select; use tokio_util::sync::CancellationToken; -use tracing::{debug, error, info, warn}; +use tracing::{Instrument as _, Span, debug, error, info, warn}; pub use crate::{ aggregate::{AggregateError, agg_deposit_data, agg_lock_hash_sig, agg_validator_registrations}, @@ -364,6 +364,7 @@ fn default_p2p_config() -> P2PConfig { } /// Runs the DKG entrypoint. +#[tracing::instrument(name = "dkg", level = "debug", skip_all, fields(topic = "dkg"))] pub async fn run(conf: Config, ct: CancellationToken) -> Result<(), DkgError> { if ct.is_cancelled() { return Err(DkgError::ShutdownRequestedBeforeStartup); @@ -388,18 +389,21 @@ async fn start_private_key_lock( ); let lock_ct = CancellationToken::new(); let task_ct = lock_ct.clone(); - let task = tokio::spawn(async move { - let run_svc = lock_svc.clone(); - let mut run_task = tokio::spawn(async move { run_svc.run().await }); - - select! { - _ = task_ct.cancelled() => { - lock_svc.close().await; - log_private_key_lock_result(run_task.await); + let task = tokio::spawn( + async move { + let run_svc = lock_svc.clone(); + let mut run_task = tokio::spawn(async move { run_svc.run().await }); + + select! { + _ = task_ct.cancelled() => { + lock_svc.close().await; + log_private_key_lock_result(run_task.await); + } + result = &mut run_task => log_private_key_lock_result(result), } - result = &mut run_task => log_private_key_lock_result(result), } - }); + .instrument(Span::current()), + ); Ok((lock_ct, task)) } @@ -581,7 +585,13 @@ async fn run_inner(conf: Config, ct: CancellationToken) -> Result<(), DkgError> let sync_clients = handlers.sync.clone(); let sync_server = handlers.sync_server.clone(); let network_ct = ct.child_token(); - let network_task = tokio::spawn(drive_dkg_network(node, network_ct.clone())); + // A bare `tokio::spawn` starts the driver with an empty span stack, which + // would drop the `dkg` topic set by `run` and count the driver's warnings + // on `app_log_warn_total{topic=""}`. Re-attach the current span so the + // subtask keeps it, mirroring charon passing `context.Context` into the + // goroutine. + let network_task = + tokio::spawn(drive_dkg_network(node, network_ct.clone()).instrument(Span::current())); let result = run_ceremony() .conf(&conf) @@ -902,14 +912,17 @@ async fn start_sync_protocol( let client = client.clone(); let client_ct = cancellation.child_token(); let cancel_on_error = cancellation.clone(); - tasks.push(tokio::spawn(async move { - if let Err(error) = client.run(client_ct).await - && !matches!(error, crate::sync::Error::Canceled) - { - error!(?error, "Sync failed to peer"); - cancel_on_error.cancel(); + tasks.push(tokio::spawn( + async move { + if let Err(error) = client.run(client_ct).await + && !matches!(error, crate::sync::Error::Canceled) + { + error!(?error, "Sync failed to peer"); + cancel_on_error.cancel(); + } } - })); + .instrument(Span::current()), + )); } let mut ticker = tokio::time::interval(Duration::from_millis(250)); diff --git a/crates/eth2util/src/keystore/load.rs b/crates/eth2util/src/keystore/load.rs index 41cecdc98..4e0f10be5 100644 --- a/crates/eth2util/src/keystore/load.rs +++ b/crates/eth2util/src/keystore/load.rs @@ -5,6 +5,7 @@ use std::{ use pluto_crypto::types::PrivateKey; use regex::Regex; +use tracing::Instrument as _; use super::{ error::{KeystoreError, Result}, @@ -112,29 +113,35 @@ pub async fn load_files_unordered(dir: impl AsRef) -> Result { .await .expect("semaphore not closed"); - set.spawn(async move { - let _permit = permit; // released when this task completes - - let b = tokio::fs::read_to_string(&path).await?; - let store: Keystore = serde_json::from_str(&b)?; - let password = super::store::load_password(&path).await?; - let file_index = extract_file_index(path.to_string_lossy())?; - - // `decrypt` runs scrypt/PBKDF2 (CPU- and memory-heavy); run it on - // the blocking pool so it never blocks an async reactor - // thread. - let (private_key, path) = tokio::task::spawn_blocking(move || { - let key = super::store::decrypt(&store, &password)?; - Ok::<_, KeystoreError>((key, path)) - }) - .await??; - - Ok::(KeyFile { - private_key, - filename: path, - file_index, - }) - }); + let span = tracing::Span::current(); + set.spawn( + async move { + let _permit = permit; // released when this task completes + + let b = tokio::fs::read_to_string(&path).await?; + let store: Keystore = serde_json::from_str(&b)?; + let password = super::store::load_password(&path).await?; + let file_index = extract_file_index(path.to_string_lossy())?; + + // `decrypt` runs scrypt/PBKDF2 (CPU- and memory-heavy); run it + // on the blocking pool so it never blocks an async reactor + // thread. + let decrypt_span = tracing::Span::current(); + let (private_key, path) = tokio::task::spawn_blocking(move || { + let _entered = decrypt_span.enter(); + let key = super::store::decrypt(&store, &password)?; + Ok::<_, KeystoreError>((key, path)) + }) + .await??; + + Ok::(KeyFile { + private_key, + filename: path, + file_index, + }) + } + .instrument(span), + ); } if set.is_empty() { @@ -235,7 +242,11 @@ pub async fn load_files_recursively(dir: impl AsRef) -> Result { // `decrypt` is CPU-intensive (key derivation), so use `spawn_blocking` // to avoid blocking the async runtime. The closure has no // `.await` calls. + let span = tracing::Span::current(); set.spawn_blocking(move || { + // The blocking pool starts with an empty span stack; enter the + // caller's span so `decrypt`'s KDF warnings keep their topic. + let _entered = span.enter(); let _permit = permit; // released when this blocking task finishes // First try the password file that matches the keystore file. let mut err = None; diff --git a/crates/p2p/src/bootnode.rs b/crates/p2p/src/bootnode.rs index b1cea2c1a..ea95e1797 100644 --- a/crates/p2p/src/bootnode.rs +++ b/crates/p2p/src/bootnode.rs @@ -6,7 +6,7 @@ use backon::Retryable; use libp2p::Multiaddr; use pluto_eth2util::enr::Record; use tokio_util::sync::CancellationToken; -use tracing::{info, warn}; +use tracing::{Instrument as _, info, warn}; use url::Url; use crate::{ @@ -128,9 +128,13 @@ pub async fn new_relays( let mutable_clone = mutable.clone(); let cancel_clone = cancel.child_token(); - tokio::spawn(async move { - resolve_relay(cancel_clone, url, hash, mutable_clone).await; - }); + let span = tracing::debug_span!("relay", topic = "relay"); + tokio::spawn( + async move { + resolve_relay(cancel_clone, url, hash, mutable_clone).await; + } + .instrument(span), + ); resp.push(mutable); } diff --git a/crates/p2p/src/p2p.rs b/crates/p2p/src/p2p.rs index a3ae5e7f8..604e0c473 100644 --- a/crates/p2p/src/p2p.rs +++ b/crates/p2p/src/p2p.rs @@ -600,6 +600,7 @@ impl Node { } /// Handles a swarm event to update metrics and logging. + #[tracing::instrument(name = "p2p", level = "debug", skip_all, fields(topic = "p2p"))] fn handle_event(&mut self, event: &SwarmEvent>) { match event { // Identify - update peer addresses in the peer store. diff --git a/crates/parsigex/src/behaviour.rs b/crates/parsigex/src/behaviour.rs index 9df714c13..c676d0d32 100644 --- a/crates/parsigex/src/behaviour.rs +++ b/crates/parsigex/src/behaviour.rs @@ -21,6 +21,7 @@ use libp2p::{ }, }; use tokio::sync::{RwLock, mpsc, oneshot}; +use tracing::Instrument as _; use pluto_core::{ eth2signeddata, @@ -211,6 +212,12 @@ impl Handle { result_rx.await.map_err(|_| Error::Closed)? } + #[tracing::instrument( + name = "parsigex", + level = "debug", + skip_all, + fields(topic = "parsigex") + )] async fn enqueue( &self, duty: Duty, @@ -506,12 +513,16 @@ impl Behaviour { /// subscribers async). fn notify_subscribers(&self, duty: Duty, data_set: ParSignedDataSet) { let shared_subs = self.shared_subs.clone(); - tokio::spawn(async move { - let subs = shared_subs.subs.read().await.clone(); - for sub in &subs { - sub(duty.clone(), data_set.clone()).await; + let span = tracing::debug_span!("parsigex", topic = "parsigex"); + tokio::spawn( + async move { + let subs = shared_subs.subs.read().await.clone(); + for sub in &subs { + sub(duty.clone(), data_set.clone()).await; + } } - }); + .instrument(span), + ); } } diff --git a/crates/peerinfo/src/protocol.rs b/crates/peerinfo/src/protocol.rs index 44d2f8d90..beea06966 100644 --- a/crates/peerinfo/src/protocol.rs +++ b/crates/peerinfo/src/protocol.rs @@ -288,6 +288,12 @@ impl ProtocolState { /// Sends a peer info request and waits for a response. /// /// Returns the response `PeerInfo` on success. + #[tracing::instrument( + name = "peerinfo", + level = "debug", + skip_all, + fields(topic = "peerinfo") + )] pub async fn send_peer_info( &self, mut stream: Stream, @@ -308,6 +314,12 @@ impl ProtocolState { /// Receives a peer info request and sends a response. /// /// Returns the stream for potential reuse after successfully responding. + #[tracing::instrument( + name = "peerinfo", + level = "debug", + skip_all, + fields(topic = "peerinfo") + )] pub async fn recv_peer_info( &self, mut stream: Stream, diff --git a/crates/priority/src/prioritiser.rs b/crates/priority/src/prioritiser.rs index 39c754efa..714fa1940 100644 --- a/crates/priority/src/prioritiser.rs +++ b/crates/priority/src/prioritiser.rs @@ -29,6 +29,7 @@ use pluto_core::{ use pluto_p2p::p2p_context::P2PContext; use tokio::sync::{mpsc, oneshot}; use tokio_util::sync::CancellationToken; +use tracing::Instrument as _; use crate::{ calculate, @@ -474,31 +475,34 @@ fn exchange( let responses = responses.clone(); let own = own.clone(); - tokio::spawn(async move { - let send = sender.send_receive(peer, own); - let response = tokio::select! { - () = ct.cancelled() => return, - res = send => match res { - Ok(resp) => resp, - Err(_) => return, // Transport already logged. - }, - }; + tokio::spawn( + async move { + let send = sender.send_receive(peer, own); + let response = tokio::select! { + () = ct.cancelled() => return, + res = send => match res { + Ok(resp) => resp, + Err(_) => return, // Transport already logged. + }, + }; - if peer.to_string() != response.peer_id { - tracing::warn!(%peer, "Invalid priority message peer id"); - return; - } + if peer.to_string() != response.peer_id { + tracing::warn!(%peer, "Invalid priority message peer id"); + return; + } - if let Err(err) = validator(&response) { - tracing::warn!(%peer, %err, "Invalid priority message from peer"); - return; - } + if let Err(err) = validator(&response) { + tracing::warn!(%peer, %err, "Invalid priority message from peer"); + return; + } - tokio::select! { - () = ct.cancelled() => {} - _ = responses.send(response) => {} + tokio::select! { + () = ct.cancelled() => {} + _ = responses.send(response) => {} + } } - }); + .instrument(tracing::Span::current()), + ); } } @@ -518,14 +522,20 @@ fn start_consensus( let consensus = inner.consensus.clone(); let duty = duty.clone(); let ct = ct.clone(); - tokio::spawn(async move { - // Fire-and-forget so the instance keeps servicing peer requests while - // consensus runs. The instance token reaches consensus, so cancellation - // tears the proposal down; a propose failure is unexpected. - if let Err(err) = consensus.propose_priority(duty, result, &ct).await { - tracing::warn!(%err, "Priority protocol consensus"); + tokio::spawn( + async move { + // Fire-and-forget so the instance keeps servicing peer requests + // while consensus runs. The instance token reaches consensus, so + // cancellation tears the proposal down; a propose failure is + // unexpected. + if let Err(err) = consensus.propose_priority(duty, result, &ct).await { + tracing::warn!(%err, "Priority protocol consensus"); + } } - }); + // `tokio::spawn` starts the task with an empty span stack; re-attach + // the caller's span so the warning above keeps its topic. + .instrument(tracing::Span::current()), + ); Ok(()) } diff --git a/crates/relay-server/src/web.rs b/crates/relay-server/src/web.rs index 06aa8c0d0..8721c46f0 100644 --- a/crates/relay-server/src/web.rs +++ b/crates/relay-server/src/web.rs @@ -17,7 +17,7 @@ use libp2p::{Multiaddr, PeerId, multiaddr}; use pluto_eth2util::enr::{EnrEntry, Record}; use tokio::{net::TcpListener, sync::RwLock}; use tokio_util::sync::CancellationToken; -use tracing::{debug, info, instrument, warn}; +use tracing::{Instrument as _, Span, debug, info, instrument, warn}; use vise_exporter::{MetricsExporter, MetricsServer}; use crate::{ @@ -123,11 +123,20 @@ pub async fn enr_server( ) -> Result<()> { info!("Starting ENR server"); - // Start external host resolver task if configured + // Start external host resolver task if configured. + // + // `tokio::spawn` gives the new task an empty span stack, so the resolver + // would lose the `relay` topic this server runs under and its warnings + // would land on `app_log_warn_total{topic=""}`. Re-attaching the current + // span restores what charon gets for free by handing the goroutine its + // `context.Context`. let resolver_handle = state.p2p_config.external_host.clone().map(|external_host| { let state = state.clone(); let ct = ct.child_token(); - tokio::spawn(resolve_external_host_periodically(state, external_host, ct)) + tokio::spawn( + resolve_external_host_periodically(state, external_host, ct) + .instrument(Span::current()), + ) }); info!( diff --git a/crates/testutil/src/validatormock/component.rs b/crates/testutil/src/validatormock/component.rs index 34fdf2913..2d0c5f430 100644 --- a/crates/testutil/src/validatormock/component.rs +++ b/crates/testutil/src/validatormock/component.rs @@ -24,7 +24,7 @@ use tokio::{ task::JoinHandle, }; use tokio_util::sync::CancellationToken; -use tracing::warn; +use tracing::{Instrument as _, warn}; use super::{ SignFunc, @@ -138,6 +138,7 @@ impl Component { } /// Called externally each slot. Mirrors Go's `Component.SlotTicked`. + #[tracing::instrument(name = "vmock", level = "debug", skip_all, fields(topic = "vmock"))] pub async fn slot_ticked(&self, slot: u64) -> Result<()> { if self.delay_on_startup().await { return Ok(()); @@ -271,6 +272,7 @@ impl Drop for Component { } } +#[tracing::instrument(name = "vmock", level = "debug", skip_all, fields(topic = "vmock"))] async fn run_scheduler( inner: Arc, cancel: CancellationToken, @@ -288,7 +290,7 @@ async fn run_scheduler( let Some(scheduled) = maybe else { break }; let inner_for_task = Arc::clone(&inner); let cancel_for_task = cancel.clone(); - duties.spawn(async move { + let duty_task = async move { let start_time = scheduled.start_time; let slot = scheduled.slot; let duty_label = scheduled.duty_type.clone(); @@ -313,7 +315,10 @@ async fn run_scheduler( } } } - }); + }; + // `JoinSet::spawn` starts the task with an empty span stack, + // so re-attach the `vmock` span opened by this function. + duties.spawn(duty_task.instrument(tracing::Span::current())); } // Reap finished duties to keep the JoinSet bounded. Disabled when // empty — `Some(_)` does not match `None`. diff --git a/crates/tracing/src/init.rs b/crates/tracing/src/init.rs index f4484147b..5788ee2cc 100644 --- a/crates/tracing/src/init.rs +++ b/crates/tracing/src/init.rs @@ -2,6 +2,7 @@ use std::{str::FromStr, time::Duration}; use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64}; use percent_encoding::percent_decode_str; +use tracing::Instrument as _; use tracing_loki::{BackgroundTaskController, url::Url}; use tracing_subscriber::{ EnvFilter, Registry, layer::SubscriberExt as _, util::SubscriberInitExt as _, @@ -121,7 +122,7 @@ pub fn init(config: &TracingConfig) -> Result> { Ok(Some(LokiWorker { controller, - handle: tokio::spawn(task), + handle: tokio::spawn(task.instrument(tracing::debug_span!("loki", topic = "loki"))), })) } else { registry.try_init()?; diff --git a/crates/tracing/src/layers/metrics.rs b/crates/tracing/src/layers/metrics.rs index 2137ca4e9..5fa3834b3 100644 --- a/crates/tracing/src/layers/metrics.rs +++ b/crates/tracing/src/layers/metrics.rs @@ -82,6 +82,7 @@ where #[cfg(test)] mod tests { use super::*; + use tracing::Instrument as _; use tracing_subscriber::layer::SubscriberExt as _; #[test] @@ -102,6 +103,51 @@ mod tests { assert_eq!(TRACING_METRICS.warn_total[&topic.to_owned()].get(), 1); } + #[tokio::test] + async fn instrumented_spawn_keeps_topic_across_task_boundary() { + // `tokio::spawn` starts a task with an empty span stack, so a subtask + // only keeps its parent's topic when the future is explicitly + // re-attached to the current span. Callers that spawn from inside a + // topic span must do this by hand; the two assertions below pin both + // halves of that contract. + let topic = "metrics_layer_spawn_topic"; + let subscriber = tracing_subscriber::registry().with(MetricsLayer); + + let before = TRACING_METRICS.error_total[&topic.to_owned()].get(); + + // `Instrument` captures the dispatcher as well as the span, so the + // default subscriber set here applies inside the spawned task. + let guard = tracing::subscriber::set_default(subscriber); + let span = tracing::info_span!("component", topic); + + let instrumented = { + let _enter = span.enter(); + tokio::spawn( + async { + tracing::error!("boom from instrumented task"); + } + .instrument(tracing::Span::current()), + ) + }; + instrumented.await.unwrap(); + + let bare = { + let _enter = span.enter(); + tokio::spawn(async { + tracing::error!("boom from bare task"); + }) + }; + bare.await.unwrap(); + + drop(guard); + + assert_eq!( + TRACING_METRICS.error_total[&topic.to_owned()].get(), + before.saturating_add(1), + "only the instrumented spawn should be counted under the topic" + ); + } + #[test] fn events_without_topic_use_empty_label() { let subscriber = tracing_subscriber::registry().with(MetricsLayer);