diff --git a/CLAUDE.md b/CLAUDE.md index a171704a8..fe9245d2e 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -43,6 +43,7 @@ crates/ ├─ crypto/ # XMSS sign/verify + aggregation (leanVM wrapper) ├─ ssz-tree/ # Persistent Merkle tree (List/Vector) behind the beacon registry and balances ├─ metrics/ # Prometheus re-exports, TimingGuard, gather utilities + ├─ telemetry/ # ClickHouse block_events rows (opt-in) └─ test-fixtures/ # Spec-fixture loading (prod dep of rpc's Hive test driver) net/ ├─ api/ # Actor protocol traits wiring BlockChain ↔ P2P @@ -590,6 +591,17 @@ default** on its two SSZ endpoints and bare integers in JSON, because See [`docs/rpc.md`](docs/rpc.md) for the full reference: CLI flags and defaults, the lean API endpoints (health, finalized state/block, justified checkpoint, blocks by root/slot, fork-choice tree + D3.js UI, runtime aggregator toggle), the Beacon API endpoints and the three ids they refuse, the metrics/debug endpoints (Prometheus `/metrics`, jemalloc heap profiling), the Hive test-driver endpoints, plus request/response shapes, status codes, and content types. +## Block telemetry (`crates/common/telemetry`) + +`--telemetry.clickhouse-url` writes Silver's `block_events` rows (one per point in a block's +life, per data column arrival) to ClickHouse over HTTP. A global sink, installed in `run_node` +before `P2P::spawn`; `ethlambda_telemetry::emit(|| ...)` builds nothing when it is off. Rows +come from `BlockImportReport::telemetry_events` (final pass only, `imported_here`), +`CascadeTelemetry` (arrival rows, and beacon `attestable` at the per-arrival head +recomputation), and p2p's column rows: `column_checks::column_rows` (after the check), and +`passed_rows` called directly by the gossip-accept path and by this node's own fulu columns. See +[`docs/telemetry.md`](docs/telemetry.md). + ## Beacon Chain types (`crates/common/types/src/beacon/`) `ethlambda-types` carries the **Ethereum Beacon Chain** containers (phase0 diff --git a/Cargo.lock b/Cargo.lock index 0bec6c9e0..352c3b832 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1994,6 +1994,7 @@ dependencies = [ "ethlambda-rpc", "ethlambda-state-transition", "ethlambda-storage", + "ethlambda-telemetry", "ethlambda-types", "ethlambda-validator", "eyre", @@ -2032,6 +2033,7 @@ dependencies = [ "ethlambda-network-api", "ethlambda-state-transition", "ethlambda-storage", + "ethlambda-telemetry", "ethlambda-test-fixtures", "ethlambda-types", "hex", @@ -2112,6 +2114,7 @@ dependencies = [ "ethlambda-network-api", "ethlambda-state-transition", "ethlambda-storage", + "ethlambda-telemetry", "ethlambda-types", "ethrex-common", "ethrex-p2p", @@ -2228,6 +2231,20 @@ dependencies = [ "xdelta3", ] +[[package]] +name = "ethlambda-telemetry" +version = "1.0.0" +dependencies = [ + "axum", + "ethlambda-metrics", + "hex", + "reqwest", + "serde", + "serde_json", + "tokio", + "tracing", +] + [[package]] name = "ethlambda-test-fixtures" version = "1.0.0" diff --git a/Cargo.toml b/Cargo.toml index e13475448..cb80e6e39 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -9,6 +9,7 @@ members = [ "crates/common/crypto", "crates/common/ssz-tree", "crates/common/metrics", + "crates/common/telemetry", "crates/common/test-fixtures", "crates/common/types", "crates/net/api", @@ -72,6 +73,7 @@ ethlambda-state-transition = { path = "crates/blockchain/state_transition" } ethlambda-crypto = { path = "crates/common/crypto" } ethlambda-ssz-tree = { path = "crates/common/ssz-tree" } ethlambda-metrics = { path = "crates/common/metrics" } +ethlambda-telemetry = { path = "crates/common/telemetry" } ethlambda-test-fixtures = { path = "crates/common/test-fixtures" } ethlambda-types = { path = "crates/common/types" } ethlambda-network-api = { path = "crates/net/api" } diff --git a/bin/ethlambda/Cargo.toml b/bin/ethlambda/Cargo.toml index 330940e28..45c6f3e72 100644 --- a/bin/ethlambda/Cargo.toml +++ b/bin/ethlambda/Cargo.toml @@ -28,6 +28,7 @@ ethlambda-state-transition.workspace = true ethlambda-types.workspace = true ethlambda-rpc.workspace = true ethlambda-storage.workspace = true +ethlambda-telemetry.workspace = true ethlambda-validator.workspace = true # Parallel XMSS keygen for the real-crypto benchmark corpus. diff --git a/bin/ethlambda/src/cli.rs b/bin/ethlambda/src/cli.rs index 4dbe91741..0388f88c9 100644 --- a/bin/ethlambda/src/cli.rs +++ b/bin/ethlambda/src/cli.rs @@ -5,6 +5,67 @@ use ethlambda_types::beacon::constants::SAFE_SLOTS_TO_IMPORT_OPTIMISTICALLY; use std::net::IpAddr; use std::path::PathBuf; +/// The ClickHouse endpoint, wrapped so the options' derived `Debug` cannot +/// print its credentials: `reqwest::Url`'s own `Debug` includes the password +/// and the query. +#[derive(Clone)] +pub(crate) struct ClickHouseUrl(reqwest::Url); + +impl ClickHouseUrl { + pub(crate) fn into_inner(self) -> reqwest::Url { + self.0 + } +} + +impl std::fmt::Debug for ClickHouseUrl { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("ClickHouseUrl") + .field("scheme", &self.0.scheme()) + .field("host", &self.0.host_str()) + .field("port", &self.0.port_or_known_default()) + .finish() + } +} + +/// Parses `--telemetry.clickhouse-url`. +/// +/// A plain `value_parser` function would leave clap to print +/// `invalid value ''` around its message, and the value may carry a +/// password. A `TypedValueParser` builds the whole `clap::Error` itself, so +/// the raw value never reaches the output. +#[derive(Clone)] +struct ClickHouseUrlParser; + +impl clap::builder::TypedValueParser for ClickHouseUrlParser { + type Value = ClickHouseUrl; + + fn parse_ref( + &self, + cmd: &clap::Command, + arg: Option<&clap::Arg>, + value: &std::ffi::OsStr, + ) -> Result { + let reject = |why: &str| { + let flag = arg.map_or_else(String::new, |arg| arg.to_string()); + cmd.clone().error( + clap::error::ErrorKind::ValueValidation, + format!( + "invalid {flag} (or ETHLAMBDA_TELEMETRY_CLICKHOUSE_URL): {why} \ + (the value is not shown, it may hold a password)" + ), + ) + }; + let raw = value.to_str().ok_or_else(|| reject("not valid UTF-8"))?; + let url = reqwest::Url::parse(raw).map_err(|_| reject("not a valid URL"))?; + if !matches!(url.scheme(), "http" | "https") { + return Err(reject( + "only http and https are supported (ClickHouse's HTTP interface, usually port 8123)", + )); + } + Ok(ClickHouseUrl(url)) + } +} + /// Flags every sub-command takes, with the same meaning on each. /// /// `--node-key`, `--bootnodes` and `--checkpoint-sync-url` used to be declared @@ -94,6 +155,28 @@ pub(crate) struct CommonOptions { /// directory, a URL, nor a loaded network's genesis, startup aborts. #[arg(long, value_delimiter = ',')] pub(crate) checkpoint_sync_url: Vec, + /// ClickHouse HTTP endpoint that per-block telemetry rows are written to + /// (Silver's `block_events` table), e.g. `http://127.0.0.1:8123`. + /// + /// Off when unset. Credentials go in the URL (`http://user:pass@host:8123`) + /// and are sent as basic auth; a `?database=` parameter passes through. + /// A URL with a password is better passed through the environment + /// variable, since any local user can read a process's command line. + /// See `docs/telemetry.md`. + #[arg( + long = "telemetry.clickhouse-url", + env = "ETHLAMBDA_TELEMETRY_CLICKHOUSE_URL", + hide_env_values = true, + value_parser = ClickHouseUrlParser + )] + pub(crate) telemetry_clickhouse_url: Option, + /// `meta_client_name` on every telemetry row. + /// + /// Defaults to `--node-id` on `node`, since devnets run several nodes per + /// host, and to `ethlambda@` on `beacon`, which keeps its rows + /// apart from another client writing the same table from the same host. + #[arg(long = "telemetry.client-name", requires = "telemetry_clickhouse_url")] + pub(crate) telemetry_client_name: Option, #[command(flatten)] pub(crate) discovery: DiscoveryConfig, } @@ -973,6 +1056,79 @@ mod tests { } } + /// A client name alone configures nothing, so it is a usage error rather + /// than a flag silently ignored. + /// + /// The first half is skipped when the URL variable is exported: clap reads + /// the environment itself, has no per-parse override, and `set_var` is + /// unsafe and racy across tests, so the flag would be satisfied by it. + #[test] + fn the_telemetry_client_name_needs_a_clickhouse_url() { + if std::env::var_os("ETHLAMBDA_TELEMETRY_CLICKHOUSE_URL").is_none() { + let mut args = beacon_args(); + args.extend(["--telemetry.client-name", "x"]); + let err = try_parse_from(args).expect_err("a client name needs a URL"); + assert_eq!(err.kind(), clap::error::ErrorKind::MissingRequiredArgument); + } + + let mut args = beacon_args(); + args.extend([ + "--telemetry.clickhouse-url", + "http://u:p@127.0.0.1:8123/?database=default", + "--telemetry.client-name", + "x", + ]); + let beacon = parse_beacon(args); + let url = beacon + .common + .telemetry_clickhouse_url + .expect("the URL parses") + .into_inner(); + assert_eq!(url.username(), "u"); + assert_eq!(beacon.common.telemetry_client_name.as_deref(), Some("x")); + } + + /// `reqwest::Url`'s `Debug` prints the password, and the options derive + /// `Debug`. + #[test] + fn the_clickhouse_url_keeps_its_credentials_out_of_debug() { + let mut args = beacon_args(); + args.extend([ + "--telemetry.clickhouse-url", + "http://user:hunter2@clickhouse.example:8123/?database=secret_db", + ]); + let beacon = parse_beacon(args); + let debug = format!("{beacon:?}"); + for secret in ["hunter2", "user:", "secret_db"] { + assert!(!debug.contains(secret), "Debug leaked {secret}: {debug}"); + } + assert!(debug.contains("clickhouse.example")); + } + + /// Only the HTTP interface is spoken, so another scheme is a startup + /// error rather than endless insert failures, and the message does not + /// echo the value, which may hold a password. + #[test] + fn the_clickhouse_url_must_be_http_and_errors_do_not_echo_it() { + for raw in [ + "tcp://user:hunter2@127.0.0.1:9000", + "http://user:hunter2@host:notaport", + ] { + let mut args = beacon_args(); + args.extend(["--telemetry.clickhouse-url", raw]); + let err = try_parse_from(args).expect_err("must be refused"); + assert_eq!(err.kind(), clap::error::ErrorKind::ValueValidation); + let shown = err.to_string(); + assert!( + !shown.contains("hunter2"), + "error echoed the value: {shown}" + ); + } + let mut args = beacon_args(); + args.extend(["--telemetry.clickhouse-url", "https://127.0.0.1:8443"]); + parse_beacon(args); + } + /// The invariant `From`'s `unreachable!` rests on. Each /// flag `requires` the other, so a half-configured Engine API is a usage /// error out of the parser rather than a panic several frames later. diff --git a/bin/ethlambda/src/main.rs b/bin/ethlambda/src/main.rs index 9110da5de..946fba7e3 100644 --- a/bin/ethlambda/src/main.rs +++ b/bin/ethlambda/src/main.rs @@ -349,6 +349,12 @@ async fn run_node(options: Options) -> eyre::Result<()> { let Options { common, network } = options; // Read before the chain match below moves `network`. let discovery_enabled = network.discovery_enabled(); + let lean_node_id = match &network { + Network::Lean(lean) => Some(lean.node_id.clone()), + Network::Mainnet { .. } => None, + }; + let telemetry_url = common.telemetry_clickhouse_url; + let telemetry_client_name = common.telemetry_client_name; #[cfg(feature = "shadow-integration")] if let Network::Lean(lean) = &network { @@ -777,6 +783,42 @@ async fn run_node(options: Options) -> eyre::Result<()> { // block production and the contribution endpoint. let sync_committee_pool = ethlambda_state_transition::beacon::sync_committee_pool::SharedSyncCommitteePool::default(); + // Before the P2P actor starts, so its first column rows are not lost. + if let Some(url) = telemetry_url { + let url = url.into_inner(); + let config = setup.store.config(); + let grid = config.time_grid(); + let (client_name, network_name) = telemetry_names( + lean_node_id.as_deref(), + telemetry_client_name, + setup.store.chain(), + config.config_name.as_str(), + ); + if sends_credentials_in_clear(&url) { + warn!( + host = url.host_str(), + "The ClickHouse URL carries credentials over plain http to a host that is not \ + loopback; they cross the network unencrypted. Use https, or a loopback address" + ); + } + // Only the host and port: the URL may carry credentials. + info!( + %client_name, + %network_name, + host = url.host_str(), + port = url.port_or_known_default(), + "Writing block telemetry to ClickHouse" + ); + ethlambda_telemetry::install(ethlambda_telemetry::TelemetryConfig { + url, + client_name, + network_name, + clock: ethlambda_telemetry::SlotClock::new( + grid.genesis_time_ms(), + grid.milliseconds_per_slot, + ), + }); + } let p2p = P2P::spawn( built, setup.store.clone(), @@ -934,6 +976,60 @@ async fn run_node(options: Options) -> eyre::Result<()> { Ok(()) } +/// `(meta_client_name, meta_network_name)` for the telemetry rows. +/// +/// An explicit `--telemetry.client-name` wins. Otherwise a lean node uses its +/// `--node-id` (devnets run several nodes per host) and a beacon node +/// `ethlambda@`. The hostname is read only on that last path, so a +/// node with telemetry off never asks. A loaded `--network` directory may have +/// no `CONFIG_NAME`, hence the beacon fallback. +fn telemetry_names( + lean_node_id: Option<&str>, + explicit: Option, + chain: ethlambda_storage::Chain, + config_name: &str, +) -> (String, String) { + let client_name = explicit.unwrap_or_else(|| match lean_node_id { + Some(node_id) => node_id.to_owned(), + None => format!("ethlambda@{}", hostname()), + }); + let network_name = match chain { + ethlambda_storage::Chain::Lean => "lean".to_owned(), + ethlambda_storage::Chain::Beacon if config_name.is_empty() => "unknown".to_owned(), + ethlambda_storage::Chain::Beacon => config_name.to_owned(), + }; + (client_name, network_name) +} + +/// `url` sends a username or password as basic auth, unencrypted, to a host +/// that is not this machine. +fn sends_credentials_in_clear(url: &reqwest::Url) -> bool { + let has_credentials = !url.username().is_empty() || url.password().is_some(); + // An IPv6 host comes back bracketed. + let host = url.host_str().unwrap_or_default(); + let host = host.trim_start_matches('[').trim_end_matches(']'); + let loopback = host.eq_ignore_ascii_case("localhost") + || host + .parse::() + .is_ok_and(|ip| ip.is_loopback()); + has_credentials && url.scheme() == "http" && !loopback +} + +/// This machine's hostname, for the default telemetry client name. +fn hostname() -> String { + let mut buf = [0u8; 256]; + // SAFETY: `buf` is valid for writes of `buf.len()` bytes, and + // `gethostname` writes at most that many. It may leave no NUL when the + // name fills the buffer; the `position` fallback below handles that. + let rc = unsafe { libc::gethostname(buf.as_mut_ptr().cast(), buf.len()) }; + if rc != 0 { + warn!("gethostname failed; telemetry client name falls back to \"unknown\""); + return "unknown".to_owned(); + } + let len = buf.iter().position(|&byte| byte == 0).unwrap_or(buf.len()); + String::from_utf8_lossy(&buf[..len]).into_owned() +} + /// The root of the anchor block when it is a gloas block whose execution /// payload envelope the store does not hold. /// @@ -2312,6 +2408,50 @@ mod tests { use ethlambda_types::genesis::GenesisValidatorEntry; use ethlambda_types::state::PUBLIC_KEY_SIZE; + #[test] + fn an_explicit_telemetry_name_wins() { + let (client, network) = telemetry_names( + Some("node-1"), + Some("mine".to_owned()), + ethlambda_storage::Chain::Lean, + "", + ); + assert_eq!((client.as_str(), network.as_str()), ("mine", "lean")); + } + + #[test] + fn a_lean_node_is_named_by_its_node_id() { + let (client, network) = + telemetry_names(Some("node-1"), None, ethlambda_storage::Chain::Lean, ""); + assert_eq!((client.as_str(), network.as_str()), ("node-1", "lean")); + } + + #[test] + fn a_beacon_node_is_named_by_host_and_config() { + let (client, network) = + telemetry_names(None, None, ethlambda_storage::Chain::Beacon, "mainnet"); + assert!(client.starts_with("ethlambda@"), "{client}"); + assert_eq!(network, "mainnet"); + } + + #[test] + fn credentials_in_clear_are_flagged_only_off_loopback_over_http() { + let flagged = |raw: &str| sends_credentials_in_clear(&raw.parse().unwrap()); + assert!(flagged("http://u:p@10.0.0.5:8123/")); + assert!(flagged("http://u@clickhouse.example:8123/")); + assert!(!flagged("https://u:p@clickhouse.example:8443/")); + assert!(!flagged("http://u:p@127.0.0.1:8123/")); + assert!(!flagged("http://u:p@[::1]:8123/")); + assert!(!flagged("http://u:p@localhost:8123/")); + assert!(!flagged("http://10.0.0.5:8123/")); + } + + #[test] + fn a_beacon_network_without_a_config_name_is_unknown() { + let (_, network) = telemetry_names(None, None, ethlambda_storage::Chain::Beacon, ""); + assert_eq!(network, "unknown"); + } + /// Emits one event per (target, level) of interest under a subscriber /// built from `node_log_filter(rust_log)` and returns what got written. fn logged_with(rust_log: Option<&str>) -> String { diff --git a/crates/blockchain/Cargo.toml b/crates/blockchain/Cargo.toml index f6aaa1136..6638eccc8 100644 --- a/crates/blockchain/Cargo.toml +++ b/crates/blockchain/Cargo.toml @@ -18,6 +18,7 @@ ethlambda-state-transition.workspace = true ethlambda-fork-choice.workspace = true ethlambda-crypto.workspace = true ethlambda-metrics.workspace = true +ethlambda-telemetry.workspace = true ethlambda-types.workspace = true ethlambda-test-fixtures.workspace = true diff --git a/crates/blockchain/src/import_timing.rs b/crates/blockchain/src/import_timing.rs index 6625a472f..fbffe94d0 100644 --- a/crates/blockchain/src/import_timing.rs +++ b/crates/blockchain/src/import_timing.rs @@ -66,6 +66,7 @@ use std::time::{Duration, Instant}; use ethlambda_network_api::BlockSource; +use ethlambda_telemetry::BlockEvent; use crate::metrics; use ethlambda_types::{ShortRoot, primitives::H256}; @@ -171,6 +172,11 @@ pub struct ImportTimings { /// Beacon: the custody-column availability check itself, not the wait. pub da_check_start: Option, pub da_check_end: Option, + /// Beacon: the check found columns to require (a block with commitments + /// inside the availability window). A gloas block or one with no + /// commitments is checked too, but has nothing to make available, so its + /// end must not be named as if it did. + pub da_columns_required: bool, // --- Hold: custody columns have not all arrived (beacon). --- pub columns_wait_start: Option, @@ -180,6 +186,10 @@ pub struct ImportTimings { /// I/O wait, not work. pub engine_start: Option, pub engine_end: Option, + /// Beacon: the execution client's answer, for the `el_sent` and + /// `el_verdict` rows. `None` when no client answered, which is what keeps + /// those rows off a node with no `--execution-endpoint`. + pub engine_answer: Option, /// Lean: participant bounds checks and pubkey resolution. pub verify_structural_start: Option, @@ -253,12 +263,7 @@ impl ImportTimings { /// moment the import began, and mixing those into a wire source's /// percentiles would understate it. pub fn source_label(&self) -> Option<&'static str> { - match self.source { - Some(BlockSource::Gossip) => Some("gossip"), - Some(BlockSource::Sync) => Some("sync"), - Some(BlockSource::Replay) => Some("replay"), - Some(BlockSource::Deferred) | None => None, - } + source_label(self.source) } /// What the log calls this block's source, which unlike the metric label @@ -462,6 +467,119 @@ impl BlockImportReport { } } + /// The block was imported by this pass's own state transition. + /// + /// Not merely `outcome == "imported"`: a block the store already holds + /// reports imported too (lean's `store::on_block` returns early with only + /// the guards marked), and a re-delivery is not a second import. + pub fn imported_here(&self) -> bool { + self.outcome == "imported" && self.timings.stf_start.is_some() + } + + /// This report as `block_events` rows (`docs/telemetry.md`). + /// + /// Only a final pass is written: one that imported the block here, or + /// failed it. A held block's timings travel with it, so writing every held + /// pass too would repeat its early rows. Not every mark is the final + /// pass's own: `received`, `decode_end`, `queue_end`, `defer_*`, `admit_*` + /// and the `*_wait_start` marks survive a hold, while `guards_*`, + /// `da_check_*` and everything after them are overwritten by each pass, so + /// they describe the last pass only. The population is the histogram's: a + /// source with no metric label writes no rows. + /// + /// Silver's stage names stand in for ours where the meaning matches; the + /// two that claim success are only used for an imported block. Beacon's + /// `attestable` is not written here at all: Silver's means the new head + /// is published, and beacon recomputes the head once per arrival, after + /// the cascade (see [`CascadeTelemetry`]). Lean updates its head per + /// block, so its `fc_head` end is its `attestable`. + pub fn telemetry_events(&self) -> Vec { + let Some(source) = self.timings.source_label() else { + return Vec::new(); + }; + let imported = match self.outcome { + "imported" if self.imported_here() => true, + "failed" => false, + _ => return Vec::new(), + }; + let t = &self.timings; + // Only a check that found columns to require made data available; for + // the others it ran and found nothing to require. + let da_check_end = if t.da_columns_required { + "da_available" + } else { + "da_check_end" + }; + let (stf_end, fc_head_end) = if imported { + ("stf_done", "attestable") + } else { + ("stf_end", "fc_head_end") + }; + // No `queue_start` row: it is always another row's instant, the + // hand-off that ends `decode` on gossip, or `received` itself on a + // block that was not decoded here. + let mut marks = vec![ + ("received", t.first_instant()), + ("decode_end", t.decode_end), + ("queue_end", t.queue_end), + ("defer_start", t.defer_start), + ("defer_end", t.defer_end), + ("admit_start", t.admit_start), + ("admit_end", t.admit_end), + ("guards_start", t.guards_start), + ("guards_end", t.guards_end), + ("preamble_start", t.preamble_start), + ("preamble_end", t.preamble_end), + ("parent_wait_start", t.parent_wait_start), + ("parent_wait_end", t.parent_wait_end), + ("cascade_wait_start", t.cascade_wait_start), + ("cascade_wait_end", t.cascade_wait_end), + ("da_check_start", t.da_check_start), + (da_check_end, t.da_check_end), + ("columns_wait_start", t.columns_wait_start), + ("columns_wait_end", t.columns_wait_end), + ("verify_struct_start", t.verify_structural_start), + ("verify_struct_end", t.verify_structural_end), + ("verify_crypto_start", t.verify_crypto_start), + ("verify_crypto_end", t.verify_crypto_end), + ("stf_start", t.stf_start), + (stf_end, t.stf_end), + ("writer_wait_start", t.writer_wait_start), + ("writer_wait_end", t.writer_wait_end), + ("db_write_start", t.db_write_start), + ("db_write_end", t.db_write_end), + ("fc_head_start", t.fc_head_start), + (fc_head_end, t.fc_head_end), + ("block_atts_start", t.block_atts_start), + ("block_atts_end", t.block_atts_end), + ]; + if !imported { + marks.push(("failed", t.last_instant())); + } + let mut events: Vec = marks + .into_iter() + .filter_map(|(stage, at)| Some(self.event(stage, at?, source, None))) + .collect(); + if let Some(answer) = t.engine_answer { + events.push(self.event("el_sent", answer.sent, source, None)); + events.push(self.event("el_verdict", answer.answered, source, Some(answer.verdict))); + } + events + } + + fn event( + &self, + stage: &'static str, + at: Instant, + source: &'static str, + verdict: Option<&'static str>, + ) -> BlockEvent { + BlockEvent { + verdict, + ..row(stage, at, self.slot, self.block_root, source) + } + } + /// The tree's child lines, one per section that ran. /// /// Split from [`Self::log`] so the rendering can be asserted without a @@ -593,6 +711,25 @@ pub struct CascadeTimings { pub head_end: Option, pub fcu_start: Option, pub fcu_end: Option, + /// The head computation succeeded. A failed one still spent the time, so + /// its instants are kept, but no block became attestable at it. + pub head_ok: bool, +} + +/// An `engine_newPayload` answer as the `el_sent` and `el_verdict` rows +/// record it. +/// +/// Its own instants rather than the `engine` pair: that pair is the time the +/// import waited for the answer, which only brackets the request while +/// nothing else runs during it. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct EngineAnswer { + /// The request went out. + pub sent: Instant, + /// The answer came back, retry ladder included. + pub answered: Instant, + /// The verdict, as the row's `verdict` column records it. + pub verdict: &'static str, } /// Timings taken while re-running beacon fork choice after an import. @@ -605,14 +742,22 @@ pub struct HeadTimings { pub head_end: Option, pub fcu_start: Option, pub fcu_end: Option, + pub head_ok: bool, } impl CascadeTimings { + /// The `source` this arrival reports under; the same two exclusions a + /// block's own sections make (see `ImportTimings::source_label`). + fn source_label(&self) -> Option<&'static str> { + source_label(self.source) + } + pub fn absorb_head(&mut self, head: HeadTimings) { self.head_start = head.head_start; self.head_end = head.head_end; self.fcu_start = head.fcu_start; self.fcu_end = head.fcu_end; + self.head_ok = head.head_ok; } pub fn starting_now() -> Self { @@ -629,13 +774,8 @@ impl CascadeTimings { /// arrival, and dividing one by the other's count is the mistake keeping /// them in separate metrics prevents. pub fn observe(&self, blocks: usize) { - let source = match self.source { - Some(BlockSource::Gossip) => "gossip", - Some(BlockSource::Sync) => "sync", - Some(BlockSource::Replay) => "replay", - // Same two exclusions a block's own sections make; see - // `ImportTimings::source_label`. - Some(BlockSource::Deferred) | None => return, + let Some(source) = self.source_label() else { + return; }; metrics::observe_block_import_cascade_blocks(blocks); @@ -691,6 +831,93 @@ impl CascadeTimings { } } +/// What one import cascade did, for its telemetry rows: filled while the +/// cascade runs, read once its head is recomputed. +/// +/// Lives on the chain actor rather than on [`CascadeTimings`] because the +/// arriving block's root is only computed inside `process_or_pend_block`, +/// and recording it there saves hashing the block a second time. +#[derive(Debug, Default)] +pub struct CascadeTelemetry { + /// Slot and root of the block whose arrival opened the cascade. + pub arrival: Option<(u64, H256)>, + /// Blocks this cascade's own state transitions imported, with the + /// `source` each reports under. + pub imported: Vec<(u64, H256, &'static str)>, +} + +impl CascadeTelemetry { + /// The arrival-scope sections, filed under the arriving block when the + /// arriving block was imported here, and an `attestable` row at the head + /// recomputation's end for every block imported here. + pub fn events(&self, cascade: &CascadeTimings) -> Vec { + let mut events = Vec::new(); + if let (Some((slot, root)), Some(source)) = (self.arrival, cascade.source_label()) + && self + .imported + .iter() + .any(|&(_, imported, _)| imported == root) + { + let marks = [ + ("cascade_start", cascade.cascade_start), + ("cascade_end", cascade.cascade_end), + ("prune_start", cascade.prune_start), + ("prune_end", cascade.prune_end), + ("get_head_start", cascade.head_start), + ("get_head_end", cascade.head_end), + ("fcu_start", cascade.fcu_start), + ("fcu_end", cascade.fcu_end), + ]; + events.extend( + marks + .into_iter() + .filter_map(|(stage, at)| Some(row(stage, at?, slot, root, source))), + ); + } + // Beacon only: lean recomputes no head per arrival, and its + // `attestable` is the block's own `fc_head` end. + if cascade.head_ok + && let Some(head_end) = cascade.head_end + { + events.extend( + self.imported + .iter() + .map(|&(slot, root, source)| row("attestable", head_end, slot, root, source)), + ); + } + events + } +} + +/// The `source` a block or arrival reports under, shared by both so the two +/// cannot disagree about which sources are measured. +fn source_label(source: Option) -> Option<&'static str> { + match source { + Some(BlockSource::Gossip) => Some("gossip"), + Some(BlockSource::Sync) => Some("sync"), + Some(BlockSource::Replay) => Some("replay"), + Some(BlockSource::Deferred) | None => None, + } +} + +fn row( + stage: &'static str, + at: Instant, + slot: u64, + root: H256, + source: &'static str, +) -> BlockEvent { + BlockEvent { + stage, + at, + slot, + block_root: root.0, + source, + verdict: None, + column_index: None, + } +} + fn span(start: Option, end: Option) -> Option { match (start, end) { (Some(start), Some(end)) => Some(end.saturating_duration_since(start)), @@ -1052,6 +1279,364 @@ mod tests { assert!(report(ImportTimings::default()).lines().is_empty()); } + fn at(base: Instant, ms: u64) -> Option { + Some(base + Duration::from_millis(ms)) + } + + fn stages(report: &BlockImportReport) -> Vec<&'static str> { + report + .telemetry_events() + .iter() + .map(|event| event.stage) + .collect() + } + + /// A gossip beacon block that waited for nothing, with an execution client. + fn beacon_gossip(base: Instant) -> ImportTimings { + ImportTimings { + source: Some(BlockSource::Gossip), + decode_start: Some(base), + decode_end: at(base, 1), + queue_start: at(base, 1), + queue_end: at(base, 2), + guards_start: at(base, 2), + guards_end: at(base, 3), + da_check_start: at(base, 3), + da_check_end: at(base, 4), + da_columns_required: true, + engine_start: at(base, 4), + engine_end: at(base, 40), + engine_answer: Some(EngineAnswer { + sent: base + Duration::from_millis(4), + answered: base + Duration::from_millis(40), + verdict: "Valid", + }), + stf_start: at(base, 40), + stf_end: at(base, 45), + block_atts_start: at(base, 45), + block_atts_end: at(base, 46), + ..ImportTimings::default() + } + } + + #[test] + fn an_imported_block_uses_silvers_names_where_they_match() { + let base = Instant::now(); + let report = report(beacon_gossip(base)); + + assert_eq!( + stages(&report), + [ + "received", + "decode_end", + "queue_end", + "guards_start", + "guards_end", + "da_check_start", + "da_available", + "stf_start", + "stf_done", + "block_atts_start", + // Beacon `attestable` waits for the head recomputation, which + // is per arrival: `CascadeTelemetry` writes it. + "block_atts_end", + "el_sent", + "el_verdict", + ] + ); + let events = report.telemetry_events(); + assert!(events.iter().all(|event| event.source == "gossip" + && event.slot == 12_345 + && event.column_index.is_none())); + assert_eq!(events[0].at, base, "arrival is the wire, not the hand-off"); + let verdict = events + .iter() + .find(|event| event.stage == "el_verdict") + .unwrap(); + assert_eq!(verdict.verdict, Some("Valid")); + assert_eq!(verdict.at, base + Duration::from_millis(40)); + } + + /// The rows bracket the request, not the `engine` section: that section + /// is the time the import waited, which need not be the request's own + /// round trip (one overlapped by other work waits less than it lasts). + #[test] + fn the_engine_rows_bracket_the_request_not_the_wait() { + let base = Instant::now(); + let report = report(ImportTimings { + engine_start: at(base, 30), + engine_end: at(base, 30), + engine_answer: Some(EngineAnswer { + sent: base + Duration::from_millis(4), + answered: base + Duration::from_millis(20), + verdict: "Valid", + }), + ..beacon_gossip(base) + }); + + let events = report.telemetry_events(); + let at_of = |stage| { + events + .iter() + .find(|event| event.stage == stage) + .map(|event| event.at) + }; + assert_eq!(at_of("el_sent"), at(base, 4)); + assert_eq!(at_of("el_verdict"), at(base, 20)); + } + + #[test] + fn a_block_with_nothing_to_make_available_does_not_claim_it_was() { + let base = Instant::now(); + let report = report(ImportTimings { + da_columns_required: false, + ..beacon_gossip(base) + }); + + let stages = stages(&report); + assert!(stages.contains(&"da_check_end")); + assert!(!stages.contains(&"da_available")); + } + + #[test] + fn a_fetched_block_is_received_at_its_hand_off() { + let base = Instant::now(); + let fetched = report(ImportTimings { + source: Some(BlockSource::Sync), + decode_start: None, + decode_end: None, + ..beacon_gossip(base) + }); + + assert_eq!(stages(&fetched)[..2], ["received", "queue_end"]); + let events = fetched.telemetry_events(); + assert_eq!(events[0].at, base + Duration::from_millis(1)); + assert!(events.iter().all(|event| event.source == "sync")); + } + + #[test] + fn a_failed_import_keeps_our_names_and_adds_failed() { + let base = Instant::now(); + let mut failed = report(beacon_gossip(base)); + failed.outcome = "failed"; + + let stages = stages(&failed); + for name in ["stf_end", "block_atts_end", "failed"] { + assert!(stages.contains(&name), "missing {name}"); + } + for name in ["stf_done", "attestable"] { + assert!(!stages.contains(&name), "{name} claims a success"); + } + let failed_at = failed + .telemetry_events() + .into_iter() + .find(|event| event.stage == "failed") + .unwrap() + .at; + assert_eq!( + failed_at, + base + Duration::from_millis(46), + "the last instant recorded" + ); + } + + #[test] + fn a_held_pass_writes_nothing() { + let mut held = report(beacon_gossip(Instant::now())); + held.outcome = "held"; + assert!(held.telemetry_events().is_empty()); + } + + /// Lean's store reports a block it already holds as imported, with only + /// the guards marked: a re-delivery, not a second import. + #[test] + fn a_block_already_imported_writes_nothing() { + let base = Instant::now(); + let redelivered = report(ImportTimings { + source: Some(BlockSource::Gossip), + decode_start: Some(base), + decode_end: at(base, 1), + queue_start: at(base, 1), + queue_end: at(base, 2), + guards_start: at(base, 2), + guards_end: at(base, 3), + // What `absorb_store` copies in from the store's early return. + preamble_start: at(base, 3), + preamble_end: at(base, 3), + ..ImportTimings::default() + }); + assert!(!redelivered.imported_here()); + assert!(redelivered.telemetry_events().is_empty()); + } + + #[test] + fn a_source_the_histogram_skips_writes_nothing() { + let local = report(ImportTimings { + source: None, + ..beacon_gossip(Instant::now()) + }); + assert!(local.telemetry_events().is_empty()); + } + + #[test] + fn no_execution_client_means_no_el_rows() { + let no_engine = report(ImportTimings { + engine_answer: None, + ..beacon_gossip(Instant::now()) + }); + let stages = stages(&no_engine); + assert!(!stages.contains(&"el_sent") && !stages.contains(&"el_verdict")); + } + + #[test] + fn a_lean_block_becomes_attestable_at_its_head_update() { + let base = Instant::now(); + let lean = report(ImportTimings { + source: Some(BlockSource::Gossip), + decode_start: Some(base), + decode_end: at(base, 1), + queue_start: at(base, 1), + queue_end: at(base, 2), + verify_crypto_start: at(base, 2), + verify_crypto_end: at(base, 30), + stf_start: at(base, 30), + stf_end: at(base, 35), + fc_head_start: at(base, 35), + fc_head_end: at(base, 36), + ..ImportTimings::default() + }); + + let events = lean.telemetry_events(); + let attestable = events + .iter() + .find(|event| event.stage == "attestable") + .unwrap(); + assert_eq!(attestable.at, base + Duration::from_millis(36)); + assert!(events.iter().all(|event| event.verdict.is_none())); + } + + fn arrival_cascade(base: Instant) -> CascadeTimings { + CascadeTimings { + source: Some(BlockSource::Gossip), + cascade_start: Some(base), + cascade_end: at(base, 50), + head_start: at(base, 50), + head_end: at(base, 52), + fcu_start: at(base, 52), + fcu_end: at(base, 60), + head_ok: true, + ..CascadeTimings::default() + } + } + + #[test] + fn an_imported_arrival_files_its_rows_and_every_import_becomes_attestable() { + let base = Instant::now(); + let arriving = H256::repeat_byte(3); + let child = H256::repeat_byte(4); + let telemetry = CascadeTelemetry { + arrival: Some((7, arriving)), + imported: vec![(7, arriving, "gossip"), (8, child, "sync")], + }; + + let events = telemetry.events(&arrival_cascade(base)); + let rows: Vec<_> = events + .iter() + .map(|event| (event.stage, event.slot, event.block_root, event.source)) + .collect(); + assert_eq!( + rows, + [ + ("cascade_start", 7, [3; 32], "gossip"), + ("cascade_end", 7, [3; 32], "gossip"), + ("get_head_start", 7, [3; 32], "gossip"), + ("get_head_end", 7, [3; 32], "gossip"), + ("fcu_start", 7, [3; 32], "gossip"), + ("fcu_end", 7, [3; 32], "gossip"), + ("attestable", 7, [3; 32], "gossip"), + ("attestable", 8, [4; 32], "sync"), + ] + ); + let head_end = base + Duration::from_millis(52); + assert!( + events + .iter() + .filter(|event| event.stage == "attestable") + .all(|event| event.at == head_end) + ); + } + + /// A re-delivered block the store already holds reports imported without + /// running anything: no arrival rows under it. A child it released and + /// that did import is still attestable. + #[test] + fn a_re_delivered_arrival_files_nothing_under_itself() { + let base = Instant::now(); + let telemetry = CascadeTelemetry { + arrival: Some((7, H256::repeat_byte(3))), + imported: vec![(8, H256::repeat_byte(4), "gossip")], + }; + + let events = telemetry.events(&arrival_cascade(base)); + assert_eq!( + events + .iter() + .map(|event| (event.stage, event.slot)) + .collect::>(), + [("attestable", 8)] + ); + } + + /// A head computation that failed still spent its time, so the arrival's + /// `get_head` rows stay, but nothing became attestable at it. + #[test] + fn a_failed_head_computation_makes_nothing_attestable() { + let base = Instant::now(); + let telemetry = CascadeTelemetry { + arrival: Some((7, H256::repeat_byte(3))), + imported: vec![(7, H256::repeat_byte(3), "gossip")], + }; + let failed = CascadeTimings { + head_ok: false, + ..arrival_cascade(base) + }; + + let stages: Vec<_> = telemetry + .events(&failed) + .iter() + .map(|event| event.stage) + .collect(); + assert!(stages.contains(&"get_head_end")); + assert!(!stages.contains(&"attestable")); + } + + /// Lean has no per-arrival head recomputation: its `attestable` comes + /// with the block's own rows, so none is written here. + #[test] + fn a_lean_cascade_writes_no_attestable() { + let base = Instant::now(); + let lean = CascadeTimings { + source: Some(BlockSource::Gossip), + cascade_start: Some(base), + cascade_end: at(base, 40), + prune_start: at(base, 40), + prune_end: at(base, 41), + ..CascadeTimings::default() + }; + let telemetry = CascadeTelemetry { + arrival: Some((7, H256::repeat_byte(3))), + imported: vec![(7, H256::repeat_byte(3), "gossip")], + }; + assert_eq!( + telemetry + .events(&lean) + .iter() + .map(|event| event.stage) + .collect::>(), + ["cascade_start", "cascade_end", "prune_start", "prune_end"] + ); + } + fn report(timings: ImportTimings) -> BlockImportReport { BlockImportReport { slot: 12_345, diff --git a/crates/blockchain/src/lib.rs b/crates/blockchain/src/lib.rs index 251048b7a..474099d6e 100644 --- a/crates/blockchain/src/lib.rs +++ b/crates/blockchain/src/lib.rs @@ -48,7 +48,9 @@ use crate::beacon_columns::{ColumnParking, ParkedColumn}; use crate::beacon_envelope::EnvelopeQueues; use crate::block_builder::ProposerConfig; use crate::events::ChainEvents; -use crate::import_timing::{BlockImportReport, CascadeTimings, HeadTimings, ImportTimings}; +use crate::import_timing::{ + BlockImportReport, CascadeTelemetry, CascadeTimings, EngineAnswer, HeadTimings, ImportTimings, +}; use crate::store::StoreError; pub use events::{ChainEvent, EventBus, Topic, UnknownTopic}; @@ -415,6 +417,7 @@ impl BlockChain { pending_block_parents: HashMap::new(), blocks_awaiting_columns: HashMap::new(), held_timings: HashMap::new(), + cascade_telemetry: CascadeTelemetry::default(), sidecars_awaiting_parent: HashMap::new(), column_parking: ColumnParking::default(), envelopes: EnvelopeQueues::default(), @@ -490,6 +493,10 @@ pub struct BlockChainServer { /// leave one behind. held_timings: HashMap, + /// The running import cascade's telemetry record: reset when `on_block` + /// opens a cascade, read once its head is recomputed. + cascade_telemetry: CascadeTelemetry, + /// Sidecars whose block's parent this node cannot yet transition from, /// keyed by that parent's root and replayed when it gains a post-state. /// @@ -1777,6 +1784,17 @@ impl BlockChainServer { let evidence = data_availability_for(&self.store, &beacon_block, &self.custody_columns); timings.da_check_end = Some(Instant::now()); + // Only a block with data to wait for has made data + // available; a gloas block or one with no commitments is + // answered `NotRequired`. Exhaustive, so a new kind of + // evidence has to say which it is. + timings.da_columns_required = match &evidence { + Some( + fork_choice::DataAvailability::Columns(_) + | fork_choice::DataAvailability::Blobs { .. }, + ) => true, + Some(fork_choice::DataAvailability::NotRequired) | None => false, + }; match evidence { Some(evidence) => evidence, None => { @@ -1796,12 +1814,20 @@ impl BlockChainServer { // `is_data_available` runs before `state_transition`, so a // block about to be held for its custody columns is never one // this node asks an execution client about. - timings.engine_start = Some(Instant::now()); + let sent = Instant::now(); + timings.engine_start = Some(sent); let validity = match &self.engine { None => fork_choice::PayloadValidity::NotRequired, Some(client) => match beacon_engine::ask(client, &beacon_block).await { Ok(None) => fork_choice::PayloadValidity::NotRequired, - Ok(Some(validity)) => validity, + Ok(Some(validity)) => { + timings.engine_answer = Some(EngineAnswer { + sent, + answered: Instant::now(), + verdict: verdict_label(&validity), + }); + validity + } // No answer after the whole ladder. // `optimistic-sync.md`: a consensus engine MUST NOT // import the block and MUST NOT apply it to the fork @@ -2010,6 +2036,7 @@ impl BlockChainServer { pending_block_parents: HashMap::new(), blocks_awaiting_columns: HashMap::new(), held_timings: HashMap::new(), + cascade_telemetry: CascadeTelemetry::default(), sidecars_awaiting_parent: HashMap::new(), column_parking: ColumnParking::default(), envelopes: EnvelopeQueues::default(), @@ -2077,6 +2104,7 @@ impl BlockChainServer { }; let mut queue = VecDeque::new(); queue.push_back((signed_block, timings)); + self.cascade_telemetry = CascadeTelemetry::default(); let (outcome, blocks) = self.run_import_cascade(queue, &mut cascade).await; if self.store.chain() == Chain::Beacon { @@ -2094,6 +2122,7 @@ impl BlockChainServer { cascade.observe(blocks); cascade.log(blocks); + ethlambda_telemetry::emit(|| self.cascade_telemetry.events(&cascade)); outcome } @@ -2212,7 +2241,7 @@ impl BlockChainServer { /// any of them is subtracted from another. `outcome` is borrowed rather /// than taken because the caller still has to act on it. fn log_import_report( - &self, + &mut self, slot: u64, block_root: H256, attestations: usize, @@ -2232,6 +2261,17 @@ impl BlockChainServer { timings, }; report.observe(); + ethlambda_telemetry::emit(|| report.telemetry_events()); + // `process_and_publish_block` also reaches here outside any + // `on_block`; harmless, since a locally built block has no source + // label and so is never recorded. + if report.imported_here() + && let Some(source) = report.timings.source_label() + { + self.cascade_telemetry + .imported + .push((report.slot, report.block_root, source)); + } report.log(); } @@ -2506,6 +2546,7 @@ impl BlockChainServer { return timings; } timings.head_end = Some(Instant::now()); + timings.head_ok = true; if let Some((head_slot, _)) = self.store.beacon_head() { metrics::update_head_slot(head_slot); } @@ -2677,6 +2718,10 @@ impl BlockChainServer { ) -> Option { let slot = signed_block.slot(); let block_root = signed_block.message_hash_tree_root(); + // The first block a cascade processes is the one that arrived. + self.cascade_telemetry + .arrival + .get_or_insert((slot, block_root)); let parent_root = signed_block.parent_root(); let proposer = signed_block.proposer_index(); timings.guards_start = Some(Instant::now()); @@ -4172,6 +4217,16 @@ impl Handler for BlockChainServer { } } +/// The `verdict` a `block_events` row records for an execution client's answer. +fn verdict_label(validity: &fork_choice::PayloadValidity) -> &'static str { + match validity { + fork_choice::PayloadValidity::NotRequired => "NotRequired", + fork_choice::PayloadValidity::Validated => "Valid", + fork_choice::PayloadValidity::Optimistic => "Optimistic", + fork_choice::PayloadValidity::Invalidated { .. } => "Invalid", + } +} + #[cfg(test)] mod tests { use std::sync::Arc; @@ -4480,6 +4535,7 @@ mod tests { pending_block_parents: HashMap::new(), blocks_awaiting_columns: HashMap::new(), held_timings: HashMap::new(), + cascade_telemetry: CascadeTelemetry::default(), sidecars_awaiting_parent: HashMap::new(), column_parking: ColumnParking::default(), envelopes: EnvelopeQueues::default(), @@ -5360,6 +5416,214 @@ mod tests { assert_eq!(outcome, None); } + /// The arriving block is recorded whatever becomes of it, but only a + /// block whose own state transition ran counts as imported: a held one + /// does not, and neither does a re-delivery of a stored one, so neither + /// gets arrival rows filed under it. + #[tokio::test] + async fn only_a_block_imported_here_gets_arrival_rows() { + let mut store = beacon_store_fulu_at_genesis(GENESIS_TIME, 0); + store + .insert_signed_block(H256::ZERO, bare_block(0, H256::repeat_byte(0xcc))) + .expect("insert"); + store + .insert_state(H256::ZERO, bare_state()) + .expect("insert"); + store + .set_time_ms(seconds_to_milliseconds( + GENESIS_TIME + 5 * Config::mainnet().seconds_per_slot, + )) + .unwrap(); + let mut server = beacon_server(store); + server.custody_columns = CUSTODY.to_vec(); + let gossip = || ImportTimings { + source: Some(BlockSource::Gossip), + ..ImportTimings::starting_now() + }; + let cascade = |base: Instant| CascadeTimings { + source: Some(BlockSource::Gossip), + cascade_start: Some(base), + cascade_end: Some(base), + head_start: Some(base), + head_end: Some(base), + head_ok: true, + ..CascadeTimings::default() + }; + + // Held for its columns: recorded as the arrival, not imported. + let held = fulu_block_with_commitments(&server.store, 2); + let held_root = held.message_hash_tree_root(); + server.on_block(held.clone(), gossip()).await; + assert_eq!( + server.cascade_telemetry.arrival, + Some((held.slot(), held_root)) + ); + assert!(server.cascade_telemetry.imported.is_empty()); + assert!( + server + .cascade_telemetry + .events(&cascade(Instant::now())) + .is_empty() + ); + + // A re-delivery of a stored beacon block returns early at the + // `has_state` check, before any report is logged: it files no arrival + // rows. (`imported_here` itself is pinned by import_timing's + // `a_block_already_imported_writes_nothing`.) + let stored = fulu_block(H256::ZERO, 1, 0); + let stored_root = stored.message_hash_tree_root(); + server + .store + .insert_signed_block(stored_root, stored.clone()) + .expect("insert"); + server + .store + .insert_state(stored_root, bare_state()) + .expect("insert"); + server.on_block(stored.clone(), gossip()).await; + assert_eq!( + server.cascade_telemetry.arrival, + Some((stored.slot(), stored_root)) + ); + assert!(server.cascade_telemetry.imported.is_empty()); + assert!( + server + .cascade_telemetry + .events(&cascade(Instant::now())) + .is_empty() + ); + } + + /// The arrival is the cascade's first block, not the latest one processed: + /// later blocks of the same cascade must not overwrite it. + #[tokio::test] + async fn the_arrival_is_the_first_block_a_cascade_processes() { + let mut store = beacon_store_fulu_at_genesis(GENESIS_TIME, 0); + store + .insert_signed_block(H256::ZERO, bare_block(0, H256::repeat_byte(0xcc))) + .expect("insert"); + store + .insert_state(H256::ZERO, bare_state()) + .expect("insert"); + store + .set_time_ms(seconds_to_milliseconds( + GENESIS_TIME + 5 * Config::mainnet().seconds_per_slot, + )) + .unwrap(); + let mut server = beacon_server(store); + server.custody_columns = CUSTODY.to_vec(); + + let first = fulu_block_with_commitments(&server.store, 2); + let first_root = first.message_hash_tree_root(); + let first_slot = first.slot(); + let child = fulu_block(first_root, first_slot + 1, 0); + + server.cascade_telemetry = CascadeTelemetry::default(); + for block in [first, child] { + let mut queue = VecDeque::new(); + server + .process_or_pend_block(block, ImportTimings::default(), &mut queue) + .await; + } + + assert_eq!( + server.cascade_telemetry.arrival, + Some((first_slot, first_root)) + ); + } + + /// Through `on_block`: the arrival is the block that arrived, even though + /// the cascade then processes a stored ancestor the walk queued behind it, + /// and the record is reset once per arrival, not once per block. + #[tokio::test] + async fn the_arrival_is_the_block_on_block_was_called_with() { + let mut store = beacon_store_fulu_at_genesis(GENESIS_TIME, 0); + store + .insert_signed_block(H256::ZERO, bare_block(0, H256::repeat_byte(0xcc))) + .expect("insert"); + store + .insert_state(H256::ZERO, bare_state()) + .expect("insert"); + store + .set_time_ms(seconds_to_milliseconds( + GENESIS_TIME + 5 * Config::mainnet().seconds_per_slot, + )) + .unwrap(); + let mut server = beacon_server(store); + server.custody_columns = CUSTODY.to_vec(); + + // Stored as pending from an earlier session: the child's walk finds + // it, its parent has a state, and it is queued behind the child. It is + // then held for its columns, so no state transition is needed. + let first = fulu_block_with_commitments(&server.store, 2); + let first_root = first.message_hash_tree_root(); + server + .store + .insert_pending_block(first_root, first.clone()) + .expect("insert"); + let child = fulu_block(first_root, first.slot() + 1, 0); + let child_root = child.message_hash_tree_root(); + let child_slot = child.slot(); + + server + .on_block( + child, + ImportTimings { + source: Some(BlockSource::Gossip), + ..ImportTimings::starting_now() + }, + ) + .await; + + assert!( + server.blocks_awaiting_columns.contains_key(&first_root), + "the walk queued the stored ancestor, which then held" + ); + assert_eq!( + server.cascade_telemetry.arrival, + Some((child_slot, child_root)) + ); + } + + /// `da_columns_required` says the check found columns to require, not + /// merely that it ran. + #[tokio::test] + async fn the_columns_required_flag_follows_the_commitments() { + let mut store = beacon_store_fulu_at_genesis(GENESIS_TIME, 0); + store + .insert_signed_block(H256::ZERO, bare_block(0, H256::repeat_byte(0xcc))) + .expect("insert"); + store + .insert_state(H256::ZERO, bare_state()) + .expect("insert"); + store + .set_time_ms(seconds_to_milliseconds( + GENESIS_TIME + 5 * Config::mainnet().seconds_per_slot, + )) + .unwrap(); + let mut server = beacon_server(store); + server.custody_columns = CUSTODY.to_vec(); + + let bare = fulu_block_with_commitments(&server.store, 0); + let (timings, _) = server.process_block(bare, ImportTimings::default()).await; + assert!(timings.da_check_end.is_some(), "inside the window"); + assert!(!timings.da_columns_required, "nothing to make available"); + + let with_columns = fulu_block_with_commitments(&server.store, 2); + let root = with_columns.message_hash_tree_root(); + for index in CUSTODY { + let sidecar = sidecar_for(&with_columns, index); + server + .store + .put_data_column_sidecar(with_columns.slot(), &root, index, sidecar.to_ssz()) + .unwrap(); + } + let (timings, _) = server + .process_block(with_columns, ImportTimings::default()) + .await; + assert!(timings.da_columns_required, "columns were required"); + } + #[tokio::test] async fn the_descendants_of_a_held_block_wait_for_it_without_re_importing_it() { // A range batch hands the actor a whole chain at once. When its first diff --git a/crates/common/telemetry/Cargo.toml b/crates/common/telemetry/Cargo.toml new file mode 100644 index 000000000..c543a2d96 --- /dev/null +++ b/crates/common/telemetry/Cargo.toml @@ -0,0 +1,21 @@ +[package] +name = "ethlambda-telemetry" +version.workspace = true +edition.workspace = true +rust-version.workspace = true +license.workspace = true + +[dependencies] +ethlambda-metrics.workspace = true +hex.workspace = true +reqwest.workspace = true +serde.workspace = true +serde_json.workspace = true +tokio = { workspace = true, features = ["macros", "rt", "sync", "time"] } +tracing.workspace = true + +[dev-dependencies] +# A stand-in ClickHouse HTTP endpoint for the writer tests. Already in the +# lockfile through ethlambda-rpc, so this adds no new crate. +axum = "0.8.1" +tokio = { workspace = true, features = ["macros", "net", "rt", "sync", "time"] } diff --git a/crates/common/telemetry/src/event.rs b/crates/common/telemetry/src/event.rs new file mode 100644 index 000000000..506eba1ce --- /dev/null +++ b/crates/common/telemetry/src/event.rs @@ -0,0 +1,282 @@ +//! A block event and its `block_events` row. + +use std::time::{Instant, SystemTime, UNIX_EPOCH}; + +use serde::Serialize; + +/// Slots after a slot's start within which an event's wall clock still says +/// something about that slot. Past it the event is catch-up or backfill, and +/// its offset is left NULL rather than reported as lateness. Silver uses the +/// same window. +const LIVE_SLOTS: u64 = 2; + +/// One point in a block's life on this node: a row of `block_events`. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct BlockEvent { + /// Silver's stage name where the meaning matches, `
_start` or + /// `
_end` for ethlambda's other sections. + pub stage: &'static str, + pub at: Instant, + pub slot: u64, + pub block_root: [u8; 32], + /// How the block or column reached this node. + pub source: &'static str, + /// The execution client's verdict, on `el_verdict` rows only. + pub verdict: Option<&'static str>, + /// The data column's index, on `column_*` rows only. + pub column_index: Option, +} + +/// Genesis and slot length: what turns a slot into the wall clock it starts at. +#[derive(Debug, Clone, Copy)] +pub struct SlotClock { + genesis_ms: u64, + slot_ms: u64, +} + +impl SlotClock { + pub fn new(genesis_ms: u64, slot_ms: u64) -> Self { + Self { + genesis_ms, + // A zero slot length would make the live window empty. + slot_ms: slot_ms.max(1), + } + } + + fn slot_start_ms(&self, slot: u64) -> u64 { + self.genesis_ms + .saturating_add(slot.saturating_mul(self.slot_ms)) + } + + /// Milliseconds from `slot`'s start to `unix_ns`, or `None` outside the + /// live window. + fn offset_ms(&self, unix_ns: u64, slot: u64) -> Option { + let start_ns = u128::from(self.slot_start_ms(slot)) * 1_000_000; + let window_ns = u128::from(self.slot_ms) * u128::from(LIVE_SLOTS) * 1_000_000; + let at = u128::from(unix_ns); + (start_ns..start_ns + window_ns) + .contains(&at) + .then(|| (at - start_ns) as f64 / 1e6) + } +} + +/// What every row carries besides its event, and where rows go. +/// +/// No `Debug`: the URL may carry credentials, and its `Debug` prints them. +#[derive(Clone)] +pub struct TelemetryConfig { + /// ClickHouse's HTTP interface. Credentials in the URL's userinfo are + /// sent as basic auth; query parameters such as `database` pass through. + pub url: reqwest::Url, + /// `meta_client_name` on every row. + pub client_name: String, + /// `meta_network_name` on every row. + pub network_name: String, + pub clock: SlotClock, +} + +/// A monotonic and a wall-clock reading taken together. +/// +/// Import timings are [`Instant`]s, which only subtract from one another, so +/// each is converted against a pair read at encoding time: the wall clock at +/// the anchor, minus how long before the anchor the instant was. +#[derive(Debug, Clone, Copy)] +pub(crate) struct Anchor { + instant: Instant, + unix_ns: u64, +} + +impl Anchor { + pub(crate) fn now() -> Self { + let unix_ns = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|since| since.as_nanos() as u64) + .unwrap_or_default(); + Self { + instant: Instant::now(), + unix_ns, + } + } + + fn unix_ns(&self, at: Instant) -> u64 { + match at.checked_duration_since(self.instant) { + Some(after) => self.unix_ns.saturating_add(after.as_nanos() as u64), + None => self + .unix_ns + .saturating_sub(self.instant.duration_since(at).as_nanos() as u64), + } + } +} + +/// One `block_events` row. Field names are the table's column names. +#[derive(Serialize)] +struct Row<'a> { + /// Unix nanoseconds, which `DateTime64(9)` reads from a `JSONEachRow` + /// integer as is. + event_date_time: u64, + stage: &'static str, + slot: u64, + /// `None` past what `DateTime` holds: ClickHouse wraps a larger value + /// instead of rejecting it. + slot_start_date_time: Option, + time_into_slot_ms: Option, + block_root: String, + source: &'static str, + verdict: Option<&'static str>, + column_index: Option, + meta_client_name: &'a str, + meta_network_name: &'a str, +} + +/// `event` as one `JSONEachRow` line, without the trailing newline. +pub(crate) fn encode(event: &BlockEvent, anchor: Anchor, config: &TelemetryConfig) -> String { + let unix_ns = anchor.unix_ns(event.at); + let row = Row { + event_date_time: unix_ns, + stage: event.stage, + slot: event.slot, + slot_start_date_time: Some(config.clock.slot_start_ms(event.slot) / 1_000) + .filter(|seconds| *seconds <= u64::from(u32::MAX)), + time_into_slot_ms: config.clock.offset_ms(unix_ns, event.slot), + block_root: format!("0x{}", hex::encode(event.block_root)), + source: event.source, + verdict: event.verdict, + column_index: event.column_index, + meta_client_name: &config.client_name, + meta_network_name: &config.network_name, + }; + serde_json::to_string(&row).expect("a row of plain fields always serializes") +} + +#[cfg(test)] +mod tests { + use std::time::{Duration, Instant}; + + use super::*; + + const GENESIS_MS: u64 = 1_000_000; + const SLOT_MS: u64 = 12_000; + + fn config() -> TelemetryConfig { + TelemetryConfig { + url: "http://127.0.0.1:8123".parse().unwrap(), + client_name: "node-a".into(), + network_name: "mainnet".into(), + clock: SlotClock::new(GENESIS_MS, SLOT_MS), + } + } + + /// An anchor whose instant sits `ms` milliseconds into `slot`. + fn anchor_at(slot: u64, ms: u64) -> Anchor { + Anchor { + instant: Instant::now(), + unix_ns: (GENESIS_MS + slot * SLOT_MS + ms) * 1_000_000, + } + } + + fn event(at: Instant, slot: u64) -> BlockEvent { + BlockEvent { + stage: "received", + at, + slot, + block_root: [0xab; 32], + source: "gossip", + verdict: None, + column_index: None, + } + } + + fn row(event: &BlockEvent, anchor: Anchor) -> serde_json::Value { + serde_json::from_str(&encode(event, anchor, &config())).unwrap() + } + + #[test] + fn a_row_carries_silvers_columns() { + let anchor = anchor_at(2, 300); + let row = row(&event(anchor.instant, 2), anchor); + + let mut keys: Vec<_> = row.as_object().unwrap().keys().cloned().collect(); + keys.sort(); + // The table definition is the source of truth for the columns. + let mut columns: Vec<_> = crate::writer::DDL + .lines() + .skip(1) + .take_while(|line| !line.starts_with(')')) + .map(|line| line.split_whitespace().next().unwrap().to_owned()) + .collect(); + columns.sort(); + assert_eq!(keys, columns); + assert_eq!( + row["event_date_time"], + (GENESIS_MS + 2 * SLOT_MS + 300) * 1_000_000 + ); + assert_eq!(row["stage"], "received"); + assert_eq!(row["slot"], 2); + assert_eq!( + row["slot_start_date_time"], + (GENESIS_MS + 2 * SLOT_MS) / 1_000 + ); + assert_eq!(row["time_into_slot_ms"], 300.0); + assert_eq!(row["block_root"], format!("0x{}", "ab".repeat(32))); + assert_eq!(row["source"], "gossip"); + assert!(row["verdict"].is_null()); + assert!(row["column_index"].is_null()); + assert_eq!(row["meta_client_name"], "node-a"); + assert_eq!(row["meta_network_name"], "mainnet"); + } + + #[test] + fn instants_convert_relative_to_the_anchor() { + let anchor = anchor_at(2, 300); + + let later = row( + &event(anchor.instant + Duration::from_millis(50), 2), + anchor, + ); + assert_eq!(later["time_into_slot_ms"], 350.0); + + // Rows are encoded after the fact, so most instants precede the anchor. + let earlier_at = anchor + .instant + .checked_sub(Duration::from_millis(100)) + .unwrap(); + let earlier = row(&event(earlier_at, 2), anchor); + assert_eq!(earlier["time_into_slot_ms"], 200.0); + } + + #[test] + fn the_slot_start_is_null_past_what_datetime_holds() { + // ClickHouse wraps a DateTime above u32::MAX seconds rather than + // rejecting it, so such a slot must not be sent as one. + let anchor = Anchor { + instant: Instant::now(), + unix_ns: 0, + }; + let row = row(&event(anchor.instant, 1 << 40), anchor); + assert!(row["slot_start_date_time"].is_null()); + assert!(row["time_into_slot_ms"].is_null()); + } + + #[test] + fn the_offset_is_null_outside_the_live_window() { + // Two whole slots after its own slot started: catch-up, not lateness. + let anchor = anchor_at(9, 0); + let catch_up = row(&event(anchor.instant, 7), anchor); + assert!(catch_up["time_into_slot_ms"].is_null()); + assert_eq!( + catch_up["slot_start_date_time"], + (GENESIS_MS + 7 * SLOT_MS) / 1_000 + ); + + // Before its slot started. + let anchor = anchor_at(6, 11_000); + assert!(row(&event(anchor.instant, 7), anchor)["time_into_slot_ms"].is_null()); + + // A slot late is still inside: late blocks are real. + let anchor = anchor_at(8, 500); + assert_eq!( + row(&event(anchor.instant, 7), anchor)["time_into_slot_ms"], + 12_500.0 + ); + } +} diff --git a/crates/common/telemetry/src/lib.rs b/crates/common/telemetry/src/lib.rs new file mode 100644 index 000000000..5ece97ca7 --- /dev/null +++ b/crates/common/telemetry/src/lib.rs @@ -0,0 +1,70 @@ +//! Per-block telemetry rows for ClickHouse, in Silver's `block_events` layout. +//! +//! One row per point in a block's life on this node, and per data column +//! arrival, written by a background task over ClickHouse's HTTP interface. +//! Off unless [`install`] ran: every [`emit`] without it is a no-op that does +//! not even build its rows. See `docs/telemetry.md`. + +mod event; +mod metrics; +mod writer; + +use std::sync::OnceLock; + +use tokio::sync::mpsc; + +pub use event::{BlockEvent, SlotClock, TelemetryConfig}; + +/// Rows waiting for the writer task, which drains it continuously. A full +/// channel drops the new row rather than blocking the emitting actor. +const CHANNEL_CAPACITY: usize = 16_384; + +static SINK: OnceLock> = OnceLock::new(); + +/// Start the writer task and route every later [`emit`] to it. +/// +/// Must run inside a tokio runtime. A second call is ignored: the first +/// writer already owns the destination. +pub fn install(config: TelemetryConfig) { + let (tx, rx) = mpsc::channel(CHANNEL_CAPACITY); + if SINK.set(tx).is_err() { + tracing::warn!("Telemetry is already installed; ignoring a second install"); + return; + } + metrics::init(); + tokio::spawn(writer::run(config, rx, writer::Limits::DEFAULT)); +} + +/// Queue the events `build` returns, if telemetry is installed. +/// +/// `build` runs only when it is, so an emit site pays nothing for its rows on +/// a node without a ClickHouse URL. +pub fn emit(build: impl FnOnce() -> Vec) { + let Some(tx) = SINK.get() else { + return; + }; + for event in build() { + if tx.try_send(event).is_err() { + metrics::inc_rows_dropped(1); + } + } +} + +#[cfg(test)] +mod tests { + use std::cell::Cell; + + use super::*; + + /// No test in this crate calls `install`, so the global stays empty here; + /// keep it that way, or this one would race it. + #[test] + fn emit_builds_nothing_when_telemetry_is_not_installed() { + let built = Cell::new(false); + emit(|| { + built.set(true); + Vec::new() + }); + assert!(!built.get()); + } +} diff --git a/crates/common/telemetry/src/metrics.rs b/crates/common/telemetry/src/metrics.rs new file mode 100644 index 000000000..32a58d79d --- /dev/null +++ b/crates/common/telemetry/src/metrics.rs @@ -0,0 +1,46 @@ +//! Counters for the telemetry writer, so a node whose rows are not arriving +//! says why on its own `/metrics`. + +use std::sync::LazyLock; + +use ethlambda_metrics::{ + IntCounter, IntCounterVec, register_int_counter, register_int_counter_vec, +}; + +static ROWS: LazyLock = LazyLock::new(|| { + register_int_counter_vec!( + "lean_telemetry_rows_total", + "Telemetry block_events rows, by what became of them", + &["outcome"] + ) + .unwrap() +}); + +static INSERT_FAILURES: LazyLock = LazyLock::new(|| { + register_int_counter!( + "lean_telemetry_insert_failures_total", + "Failed ClickHouse insert attempts, refusals included" + ) + .unwrap() +}); + +/// Publish every series at zero, so a node whose inserts never succeed shows +/// a rising failure count next to zero rows inserted, rather than no series. +pub(crate) fn init() { + for outcome in ["inserted", "dropped"] { + ROWS.with_label_values(&[outcome]); + } + LazyLock::force(&INSERT_FAILURES); +} + +pub(crate) fn inc_rows_inserted(rows: u64) { + ROWS.with_label_values(&["inserted"]).inc_by(rows); +} + +pub(crate) fn inc_rows_dropped(rows: u64) { + ROWS.with_label_values(&["dropped"]).inc_by(rows); +} + +pub(crate) fn inc_insert_failures() { + INSERT_FAILURES.inc(); +} diff --git a/crates/common/telemetry/src/writer.rs b/crates/common/telemetry/src/writer.rs new file mode 100644 index 000000000..d83c07bc2 --- /dev/null +++ b/crates/common/telemetry/src/writer.rs @@ -0,0 +1,739 @@ +//! The background task that moves rows from the channel into ClickHouse. +//! +//! Rows are read off the channel continuously, including while a request is +//! in flight, into a pending buffer; the channel's capacity therefore only +//! has to cover the gap between two polls of this task, not a slow request. +//! At most one request is in flight, and it starts on a tick, so a steady +//! trickle is one insert per tick: ClickHouse writes a part per insert. A +//! failed request leaves its rows at the front of the buffer for the next +//! tick, except a batch ClickHouse refuses as malformed (HTTP 400), which is +//! dropped since a retry would be refused again. The buffer drops its oldest +//! rows past a cap, so an unreachable ClickHouse costs bounded memory. + +use std::collections::VecDeque; +use std::error::Error as _; +use std::time::Duration; + +use reqwest::{Client, RequestBuilder, Url}; +use tokio::sync::mpsc; +use tokio::task::JoinHandle; +use tokio::time::MissedTickBehavior; +use tracing::{debug, error, info, warn}; + +use crate::event::{Anchor, BlockEvent, TelemetryConfig, encode}; +use crate::metrics; + +/// Silver's `block_events` table, verbatim (gattaca-com/silver, +/// `crates/telemetry/src/block_events.rs`, MIT), so both clients' rows share +/// one table and one set of queries. +pub(crate) const DDL: &str = "CREATE TABLE IF NOT EXISTS block_events ( + event_date_time DateTime64(9) COMMENT 'Node-local wall clock at the observation', + stage LowCardinality(String) COMMENT 'Point in the block path through the node this row observes', + slot Nullable(UInt64) COMMENT 'NULL when the root was first seen before the collector attached', + slot_start_date_time Nullable(DateTime) COMMENT 'Wall clock the slot started at', + time_into_slot_ms Nullable(Float64) COMMENT 'Milliseconds from slot start; NULL outside the live window (replay or backfill)', + block_root String COMMENT '0x-prefixed beacon block root; rows written before 2026-08-18 lack the prefix', + source LowCardinality(String) COMMENT 'How the node obtained the block', + verdict LowCardinality(Nullable(String)) COMMENT 'Execution payload status; set on el_verdict rows only', + column_index Nullable(UInt64) COMMENT 'Data-column index; set on column_* rows only', + meta_client_name LowCardinality(String) COMMENT 'Hostname of the node that produced the row', + meta_network_name LowCardinality(String) COMMENT 'Ethereum network the node is running' +) ENGINE = MergeTree +ORDER BY (meta_client_name, event_date_time)"; + +pub(crate) const INSERT: &str = "INSERT INTO block_events FORMAT JSONEachRow"; + +#[derive(Debug, Clone, Copy)] +pub(crate) struct Limits { + pub(crate) flush_interval: Duration, + pub(crate) max_rows_per_request: usize, + pub(crate) max_pending_rows: usize, + pub(crate) request_timeout: Duration, +} + +impl Limits { + pub(crate) const DEFAULT: Self = Self { + flush_interval: Duration::from_secs(1), + max_rows_per_request: 10_000, + max_pending_rows: 100_000, + request_timeout: Duration::from_secs(10), + }; +} + +/// Longest prefix of ClickHouse's answer kept in a failure message. +const MAX_BODY_IN_MESSAGE: usize = 512; + +/// Why a request failed. +struct Failure { + message: String, + /// ClickHouse refused the rows themselves (HTTP 400: a row it cannot + /// parse or a value of the wrong type). Retrying the same rows would be + /// refused again, and would hold back every row queued behind them. + rejected: bool, +} + +type Outcome = (Vec, Result<(), Failure>); + +/// A request in flight. Its rows come back with its result, so a failed one +/// can be retried; `rows` is what to count as dropped if its task panics and +/// the rows never come back. +struct InFlight { + rows: usize, + handle: JoinHandle, +} + +/// Run until every sender is gone. Production keeps the sender in a static, +/// so there the task runs for the life of the process; when the channel does +/// close, the rows still pending are counted as dropped. +pub(crate) async fn run( + config: TelemetryConfig, + mut rx: mpsc::Receiver, + limits: Limits, +) { + let client = match Client::builder().timeout(limits.request_timeout).build() { + Ok(client) => client, + Err(err) => { + error!(%err, "Telemetry disabled: could not build the HTTP client"); + return; + } + }; + // Credentials in the URL need no handling here: reqwest moves them into + // basic auth itself, percent-decoded, and strips them from the URL. + let urls = Urls::new(&config.url); + let mut pending: VecDeque = VecDeque::new(); + let mut in_flight: Option = None; + // Before the first insert and after every failure but a refusal: the statement is + // idempotent, and rerunning it recreates a table dropped under us. + let mut needs_schema = true; + let mut failing = false; + let mut rejecting = false; + let mut tick = tokio::time::interval(limits.flush_interval); + tick.set_missed_tick_behavior(MissedTickBehavior::Delay); + + loop { + tokio::select! { + event = rx.recv() => { + let Some(event) = event else { + metrics::inc_rows_dropped(pending.len() as u64); + break; + }; + pending.push_back(encode(&event, Anchor::now(), &config)); + drop_past_cap(&mut pending, limits.max_pending_rows); + } + // Polled every period, whatever there is to send: a tick switched + // off while idle would fire at once when switched back on, sending + // the first row of the next burst on its own. + _ = tick.tick() => { + if in_flight.is_none() && !pending.is_empty() { + let batch = take_batch(&mut pending, limits.max_rows_per_request); + in_flight = Some(send(&client, &urls, batch, needs_schema)); + } + } + (batch, result) = join(&mut in_flight) => { + in_flight = None; + match result { + Ok(()) => { + metrics::inc_rows_inserted(batch.len() as u64); + needs_schema = false; + rejecting = false; + if failing { + failing = false; + info!("Telemetry inserts recovered"); + } + // Only a full request goes ahead of the tick, so a + // backlog drains at once while a trickle still waits + // to be batched. + if pending.len() >= limits.max_rows_per_request { + let batch = take_batch(&mut pending, limits.max_rows_per_request); + in_flight = Some(send(&client, &urls, batch, needs_schema)); + } + } + Err(Failure { message, rejected: true }) => { + metrics::inc_insert_failures(); + metrics::inc_rows_dropped(batch.len() as u64); + let rows = batch.len(); + if rejecting { + debug!(rows, err = %message, "ClickHouse rejected a telemetry batch"); + } else { + rejecting = true; + warn!(rows, err = %message, "ClickHouse rejected a batch; dropping it"); + } + } + Err(Failure { message, rejected: false }) => { + metrics::inc_insert_failures(); + if !failing { + failing = true; + warn!(err = %message, "Telemetry insert failed; keeping the rows to retry"); + } + needs_schema = true; + for row in batch.into_iter().rev() { + pending.push_front(row); + } + drop_past_cap(&mut pending, limits.max_pending_rows); + } + } + } + } + } +} + +/// The in-flight request's outcome, or never when there is none. +async fn join(in_flight: &mut Option) -> Outcome { + match in_flight { + Some(InFlight { rows, handle }) => handle.await.unwrap_or_else(|err| { + metrics::inc_rows_dropped(*rows as u64); + let failure = Failure { + message: format!("insert task failed: {err}"), + rejected: false, + }; + (Vec::new(), Err(failure)) + }), + None => std::future::pending().await, + } +} + +fn take_batch(pending: &mut VecDeque, max_rows: usize) -> Vec { + let rows = pending.len().min(max_rows); + pending.drain(..rows).collect() +} + +fn drop_past_cap(pending: &mut VecDeque, max_rows: usize) { + let excess = pending.len().saturating_sub(max_rows); + if excess > 0 { + pending.drain(..excess); + metrics::inc_rows_dropped(excess as u64); + } +} + +fn send(client: &Client, urls: &Urls, batch: Vec, needs_schema: bool) -> InFlight { + let client = client.clone(); + let urls = urls.clone(); + let rows = batch.len(); + let handle = tokio::spawn(async move { + let result = insert(&client, &urls, &batch, needs_schema).await; + (batch, result) + }); + InFlight { rows, handle } +} + +/// The schema statement is best effort: a user that may insert but not create +/// still inserts into a table that exists. Only the insert's own failure counts, +/// with the schema statement's message attached when both failed. +async fn insert( + client: &Client, + urls: &Urls, + batch: &[String], + needs_schema: bool, +) -> Result<(), Failure> { + let schema_error = if needs_schema { + execute(client.post(urls.schema.clone()).body(DDL), false) + .await + .err() + } else { + None + }; + execute( + client.post(urls.insert.clone()).body(batch.join("\n")), + true, + ) + .await + .map_err(|mut failure| { + if let Some(schema) = schema_error.filter(|schema| schema.message != failure.message) { + failure.message = format!( + "{}; the schema statement also failed: {}", + failure.message, schema.message + ); + } + failure + }) +} + +/// A non-2xx answer is an error carrying ClickHouse's own message. With +/// `is_insert`, a 400 marks the rows as rejected. +async fn execute(request: RequestBuilder, is_insert: bool) -> Result<(), Failure> { + let response = request.send().await.map_err(|err| { + // Without the URL: it may carry credentials, and the error is logged. + let err = err.without_url(); + let mut message = err.to_string(); + let mut source = err.source(); + while let Some(cause) = source { + message.push_str(": "); + message.push_str(&cause.to_string()); + source = cause.source(); + } + Failure { + message, + rejected: false, + } + })?; + let status = response.status(); + if status.is_success() { + return Ok(()); + } + let body = response.text().await.unwrap_or_default(); + Err(Failure { + message: format!("{status}: {}", truncate(body.trim(), MAX_BODY_IN_MESSAGE)), + rejected: is_insert && status == reqwest::StatusCode::BAD_REQUEST, + }) +} + +/// At most `max_bytes` of `text`, cut on a character boundary. +fn truncate(text: &str, max_bytes: usize) -> &str { + let mut end = max_bytes.min(text.len()); + while !text.is_char_boundary(end) { + end -= 1; + } + &text[..end] +} + +/// Where the two statements go. The insert carries its statement as the +/// `query` parameter, since its POST body is the data; the schema statement +/// is the body itself. +#[derive(Clone)] +struct Urls { + schema: Url, + insert: Url, +} + +impl Urls { + fn new(url: &Url) -> Self { + let mut insert = url.clone(); + insert.query_pairs_mut().append_pair("query", INSERT); + Self { + schema: url.clone(), + insert, + } + } +} + +#[cfg(test)] +mod tests { + use std::collections::HashMap; + use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; + use std::sync::{Arc, Mutex}; + use std::time::Instant; + + use axum::Router; + use axum::extract::{Query, State}; + use axum::http::{HeaderMap, StatusCode}; + use axum::routing::post; + + use super::*; + use crate::event::SlotClock; + + const FAST: Limits = Limits { + flush_interval: Duration::from_millis(10), + max_rows_per_request: 10_000, + max_pending_rows: 100_000, + request_timeout: Duration::from_secs(5), + }; + + #[derive(Clone)] + struct Seen { + query: String, + params: HashMap, + authorization: Option, + body: String, + ok: bool, + at: Instant, + } + + /// A stand-in for ClickHouse's HTTP interface that records every request. + #[derive(Clone, Default)] + struct Mock { + requests: Arc>>, + /// Requests that reached the handler, counted before any delay. + started: Arc, + /// Fail every request. + fail: Arc, + /// Fail inserts only, letting the schema statement through. + fail_inserts: Arc, + /// Fail the schema statement only. + fail_schema: Arc, + delay_ms: Arc, + } + + impl Mock { + fn requests(&self) -> Vec { + self.requests.lock().unwrap().clone() + } + + fn failures(&self) -> usize { + self.requests().iter().filter(|seen| !seen.ok).count() + } + + /// The lines of every insert that succeeded, one `Vec` per request. + fn inserted(&self) -> Vec> { + self.requests() + .into_iter() + .filter(|seen| seen.ok && seen.query == INSERT) + .map(|seen| seen.body.lines().map(str::to_owned).collect()) + .collect() + } + } + + async fn handle( + State(mock): State, + Query(params): Query>, + headers: HeaderMap, + body: String, + ) -> StatusCode { + mock.started.fetch_add(1, Ordering::SeqCst); + let delay = mock.delay_ms.load(Ordering::SeqCst); + if delay > 0 { + tokio::time::sleep(Duration::from_millis(delay)).await; + } + let query = params.get("query").cloned().unwrap_or_default(); + let authorization = headers + .get("authorization") + .map(|value| value.to_str().unwrap().to_owned()); + // A row ClickHouse cannot take: the whole insert is refused. + let poisoned = query == INSERT && body.contains("\"slot\":666"); + let ok = !poisoned + && !mock.fail.load(Ordering::SeqCst) + && !(query == INSERT && mock.fail_inserts.load(Ordering::SeqCst)) + && !(query != INSERT && mock.fail_schema.load(Ordering::SeqCst)); + mock.requests.lock().unwrap().push(Seen { + query, + params, + authorization, + body, + ok, + at: Instant::now(), + }); + if poisoned { + StatusCode::BAD_REQUEST + } else if ok { + StatusCode::OK + } else { + StatusCode::INTERNAL_SERVER_ERROR + } + } + + async fn serve(mock: Mock) -> Url { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let app = Router::new().route("/", post(handle)).with_state(mock); + tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + format!("http://{addr}/").parse().unwrap() + } + + fn start_writer(url: Url, limits: Limits, capacity: usize) -> mpsc::Sender { + let (tx, rx) = mpsc::channel(capacity); + let config = TelemetryConfig { + url, + client_name: "test".into(), + network_name: "testnet".into(), + clock: SlotClock::new(0, 12_000), + }; + tokio::spawn(run(config, rx, limits)); + tx + } + + fn event(slot: u64) -> BlockEvent { + BlockEvent { + stage: "received", + at: Instant::now(), + slot, + block_root: [1; 32], + source: "gossip", + verdict: None, + column_index: None, + } + } + + fn slots(lines: &[String]) -> Vec { + lines + .iter() + .map(|line| { + serde_json::from_str::(line).unwrap()["slot"] + .as_u64() + .unwrap() + }) + .collect() + } + + async fn wait_until(what: &str, done: impl Fn() -> bool) { + let deadline = Instant::now() + Duration::from_secs(5); + while !done() { + assert!(Instant::now() < deadline, "timed out waiting for {what}"); + tokio::time::sleep(Duration::from_millis(5)).await; + } + } + + #[tokio::test] + async fn creates_the_table_then_inserts_json_each_row() { + let mock = Mock::default(); + let tx = start_writer(serve(mock.clone()).await, FAST, 64); + for slot in 0..3 { + tx.send(event(slot)).await.unwrap(); + } + + wait_until("three rows", || mock.inserted().concat().len() == 3).await; + let requests = mock.requests(); + assert!( + requests[0] + .body + .starts_with("CREATE TABLE IF NOT EXISTS block_events") + ); + assert_eq!(requests[1].query, INSERT); + assert_eq!(slots(&mock.inserted().concat()), [0, 1, 2]); + } + + #[tokio::test] + async fn sends_nothing_without_rows() { + let mock = Mock::default(); + let _tx = start_writer(serve(mock.clone()).await, FAST, 64); + tokio::time::sleep(Duration::from_millis(100)).await; + assert!(mock.requests().is_empty()); + } + + #[tokio::test] + async fn splits_a_backlog_into_bounded_requests() { + let mock = Mock::default(); + mock.fail.store(true, Ordering::SeqCst); + let limits = Limits { + max_rows_per_request: 2, + ..FAST + }; + let tx = start_writer(serve(mock.clone()).await, limits, 64); + for slot in 0..5 { + tx.send(event(slot)).await.unwrap(); + } + wait_until("two failed attempts", || mock.failures() >= 2).await; + + mock.fail.store(false, Ordering::SeqCst); + wait_until("every row", || mock.inserted().concat().len() == 5).await; + assert!(mock.inserted().iter().all(|request| request.len() <= 2)); + assert_eq!(slots(&mock.inserted().concat()), [0, 1, 2, 3, 4]); + } + + /// One insert succeeds first, so `needs_schema` is already false when the + /// next one fails: the retry only recreates the table if the failure + /// re-armed it. + #[tokio::test] + async fn retries_a_failed_insert_and_recreates_the_table_first() { + let mock = Mock::default(); + let tx = start_writer(serve(mock.clone()).await, FAST, 64); + tx.send(event(0)).await.unwrap(); + wait_until("the first insert", || mock.inserted().len() == 1).await; + + mock.fail_inserts.store(true, Ordering::SeqCst); + tx.send(event(1)).await.unwrap(); + wait_until("a failed insert", || mock.failures() >= 1).await; + mock.fail_inserts.store(false, Ordering::SeqCst); + wait_until("both rows", || mock.inserted().concat().len() == 2).await; + + let requests = mock.requests(); + let is_schema = |seen: &Seen| { + seen.body + .starts_with("CREATE TABLE IF NOT EXISTS block_events") + }; + let retry = requests + .iter() + .rposition(|seen| seen.ok && seen.query == INSERT) + .unwrap(); + assert!( + is_schema(&requests[retry - 1]) && requests[retry - 1].ok, + "after a failed insert the table is recreated before the retry" + ); + assert_eq!(slots(&mock.inserted().concat()), [0, 1]); + } + + /// One insert per tick, not one per round trip: ClickHouse writes a part + /// per insert, so a request per handful of rows would multiply the parts. + #[tokio::test] + async fn a_steady_trickle_is_one_insert_per_tick() { + let mock = Mock::default(); + let limits = Limits { + flush_interval: Duration::from_millis(200), + ..FAST + }; + let tx = start_writer(serve(mock.clone()).await, limits, 64); + let start = Instant::now(); + for slot in 0..500 { + tx.send(event(slot)).await.unwrap(); + tokio::time::sleep(Duration::from_millis(1)).await; + } + let ticks = start.elapsed().as_millis() / 200; + + wait_until("every row", || mock.inserted().concat().len() == 500).await; + let inserts = mock.inserted().len() as u128; + assert!(inserts <= ticks + 2, "{inserts} inserts over {ticks} ticks"); + } + + /// A tick switched off while there is nothing to send would fire at once + /// when re-enabled, sending the first row of a burst on its own. + #[tokio::test] + async fn a_burst_after_an_idle_spell_goes_out_as_one_insert() { + let mock = Mock::default(); + let limits = Limits { + flush_interval: Duration::from_millis(200), + ..FAST + }; + let tx = start_writer(serve(mock.clone()).await, limits, 64); + tx.send(event(0)).await.unwrap(); + wait_until("the first insert", || mock.inserted().len() == 1).await; + tokio::time::sleep(Duration::from_millis(500)).await; + + for slot in 1..=24 { + tx.try_send(event(slot)).unwrap(); + } + wait_until("the burst", || mock.inserted().concat().len() == 25).await; + assert_eq!(mock.inserted().len(), 2, "the burst went out as one insert"); + } + + #[tokio::test] + async fn drops_the_oldest_rows_past_the_cap() { + let mock = Mock::default(); + mock.fail.store(true, Ordering::SeqCst); + let limits = Limits { + max_pending_rows: 3, + ..FAST + }; + let tx = start_writer(serve(mock.clone()).await, limits, 64); + for slot in 0..5 { + tx.send(event(slot)).await.unwrap(); + } + wait_until("two failed attempts", || mock.failures() >= 2).await; + + mock.fail.store(false, Ordering::SeqCst); + wait_until("an insert", || !mock.inserted().is_empty()).await; + tokio::time::sleep(Duration::from_millis(50)).await; + assert_eq!(slots(&mock.inserted().concat()), [2, 3, 4]); + } + + #[tokio::test] + async fn keeps_reading_the_channel_while_a_request_is_in_flight() { + let mock = Mock::default(); + mock.delay_ms.store(300, Ordering::SeqCst); + // Room for four: the sends below only all fit if the writer keeps + // draining while its request waits. + let tx = start_writer(serve(mock.clone()).await, FAST, 4); + tx.send(event(0)).await.unwrap(); + wait_until("a request in flight", || { + mock.started.load(Ordering::SeqCst) >= 1 + }) + .await; + + for slot in 1..=20 { + tx.try_send(event(slot)) + .expect("the writer drains the channel while a request is in flight"); + tokio::time::sleep(Duration::from_millis(2)).await; + } + + mock.delay_ms.store(0, Ordering::SeqCst); + wait_until("every row", || mock.inserted().concat().len() == 21).await; + } + + /// A batch ClickHouse refuses as malformed is dropped, not retried: it + /// would be refused again and hold back every row queued behind it. + #[tokio::test] + async fn a_rejected_batch_is_dropped_not_retried() { + let mock = Mock::default(); + let tx = start_writer(serve(mock.clone()).await, FAST, 64); + tx.send(event(666)).await.unwrap(); + wait_until("the rejected insert", || mock.failures() >= 1).await; + + tx.send(event(2)).await.unwrap(); + tx.send(event(3)).await.unwrap(); + wait_until("the rows behind it", || mock.inserted().concat().len() == 2).await; + tokio::time::sleep(Duration::from_millis(100)).await; + + assert_eq!(slots(&mock.inserted().concat()), [2, 3]); + let attempts = mock + .requests() + .iter() + .filter(|seen| seen.query == INSERT && seen.body.contains("\"slot\":666")) + .count(); + assert_eq!(attempts, 1, "the rejected batch was sent again"); + } + + /// A full request goes out as soon as the previous one lands, without + /// waiting for the next tick, so a backlog drains at once. + #[tokio::test] + async fn a_backlog_drains_without_waiting_for_ticks() { + let mock = Mock::default(); + mock.fail.store(true, Ordering::SeqCst); + let limits = Limits { + flush_interval: Duration::from_millis(500), + max_rows_per_request: 2, + ..FAST + }; + let tx = start_writer(serve(mock.clone()).await, limits, 64); + for slot in 0..6 { + tx.send(event(slot)).await.unwrap(); + } + wait_until("two failed requests", || mock.failures() >= 2).await; + + mock.fail.store(false, Ordering::SeqCst); + wait_until("every row", || mock.inserted().concat().len() == 6).await; + let times: Vec = mock + .requests() + .iter() + .filter(|seen| seen.ok && seen.query == INSERT) + .map(|seen| seen.at) + .collect(); + let span = *times.last().unwrap() - times[0]; + assert!( + span < limits.flush_interval, + "the inserts took {span:?}, a tick apart" + ); + assert_eq!(slots(&mock.inserted().concat()), [0, 1, 2, 3, 4, 5]); + } + + #[tokio::test] + async fn credentials_in_the_url_become_basic_auth_percent_decoded() { + let mock = Mock::default(); + let url = serve(mock.clone()).await; + let url = url.as_str().replace("http://", "http://user:p%40ss@"); + let tx = start_writer(url.parse().unwrap(), FAST, 64); + tx.send(event(0)).await.unwrap(); + + wait_until("an insert", || !mock.inserted().is_empty()).await; + let requests = mock.requests(); + assert!(!requests.is_empty()); + for seen in requests { + // base64 of `user:p@ss` + assert_eq!(seen.authorization.as_deref(), Some("Basic dXNlcjpwQHNz")); + } + } + + #[tokio::test] + async fn an_existing_query_parameter_survives() { + let mock = Mock::default(); + let url = serve(mock.clone()).await.join("?database=x").unwrap(); + let tx = start_writer(url, FAST, 64); + tx.send(event(0)).await.unwrap(); + + wait_until("an insert", || !mock.inserted().is_empty()).await; + let requests = mock.requests(); + let insert = requests + .iter() + .find(|seen| seen.query == INSERT) + .expect("an insert request"); + assert_eq!(insert.params.get("database").map(String::as_str), Some("x")); + assert_eq!(insert.params.get("query").map(String::as_str), Some(INSERT)); + } + + /// The schema statement is best effort: a user allowed to insert but not + /// to create must still insert. + #[tokio::test] + async fn a_failed_schema_statement_does_not_stop_the_insert() { + let mock = Mock::default(); + mock.fail_schema.store(true, Ordering::SeqCst); + let tx = start_writer(serve(mock.clone()).await, FAST, 64); + tx.send(event(0)).await.unwrap(); + + wait_until("the row", || mock.inserted().concat().len() == 1).await; + assert!(mock.failures() >= 1); + } + + #[test] + fn truncation_cuts_on_a_character_boundary() { + assert_eq!(truncate("abcdef", 3), "abc"); + assert_eq!(truncate("ab", 3), "ab"); + // The second byte of a two-byte character is not a boundary. + assert_eq!(truncate("aé", 2), "a"); + } +} diff --git a/crates/net/p2p/Cargo.toml b/crates/net/p2p/Cargo.toml index 22154e141..2c78cff0a 100644 --- a/crates/net/p2p/Cargo.toml +++ b/crates/net/p2p/Cargo.toml @@ -13,6 +13,7 @@ version.workspace = true ethlambda-network-api.workspace = true ethlambda-storage.workspace = true ethlambda-metrics.workspace = true +ethlambda-telemetry.workspace = true ethlambda-types.workspace = true # For `verify_data_column_sidecar`, the one sidecar check cheap enough to run # in the gossip loop: pure structural validation, no KZG. `blst` and `c-kzg` diff --git a/crates/net/p2p/src/beacon/column_checks.rs b/crates/net/p2p/src/beacon/column_checks.rs index 91fbd1efe..d648ef86c 100644 --- a/crates/net/p2p/src/beacon/column_checks.rs +++ b/crates/net/p2p/src/beacon/column_checks.rs @@ -9,30 +9,138 @@ //! - a gossiped sidecar reported `Queue` or `Ignore(Overloaded)`, whose //! stateful checks stopped early or never ran; //! - every `DataColumnsByRoot` and `DataColumnsByRange` answer; -//! - sidecars the chain actor had parked, handed back once the block they -//! wait on imported (`BlockChainToP2P::check_data_column_sidecars`): a fulu -//! sidecar's parent, a gloas sidecar's own block. +//! - sidecars the chain actor was told to hold, handed back once the block +//! they wait on imported (`BlockChainToP2P::check_data_column_sidecars`): a +//! fulu sidecar's parent, a gloas sidecar's own block. //! //! Each sidecar is checked on a blocking thread, bounded by //! [`P2PServer::column_check_permits`]. Unlike a gossip verdict, nothing here //! has a deadline, so a sidecar with no free permit waits for one rather than //! being dropped. +use std::time::Instant; + use ethlambda_state_transition::beacon::gossip::column::{self, ChainVerdict}; use ethlambda_state_transition::beacon::gossip::{IgnoreReason, Outcome}; +use ethlambda_telemetry::BlockEvent; use ethlambda_types::beacon::containers::DataColumnSidecar; use ethlambda_types::time::unix_now_ms; use tracing::{error, warn}; use crate::{P2PServer, metrics}; +/// How a batch of sidecars reached [`check_and_forward`], for the telemetry +/// rows (`docs/telemetry.md`). +#[derive(Debug, Clone, Copy)] +pub(crate) enum ColumnArrival { + /// Sidecars that just reached this node. + Received { + /// The rows' `source`: how they reached this node. + source: &'static str, + /// When they arrived. For a by-root or by-range answer this is when + /// the whole answer was delivered to the handler, so every sidecar in + /// it shares one instant. + at: Instant, + }, + /// Sidecars judged to wait for their block, handed back once it imports: + /// their arrival was written then, so only their validation is new. + Recheck, +} + +impl ColumnArrival { + pub(crate) fn received(source: &'static str, at: Instant) -> Self { + Self::Received { source, at } + } + + fn source(self) -> &'static str { + match self { + Self::Received { source, .. } => source, + Self::Recheck => "recheck", + } + } +} + +/// One `block_events` row for `sidecar`: the one place its fields are named, +/// so the root (a header hash for a fulu sidecar) is computed per row built. +fn column_event( + stage: &'static str, + sidecar: &DataColumnSidecar, + source: &'static str, + at: Instant, +) -> BlockEvent { + BlockEvent { + stage, + at, + slot: sidecar.slot(), + block_root: sidecar.block_root().0, + source, + verdict: None, + column_index: Some(sidecar.index()), + } +} + +/// The rows of a sidecar that passed validation at `validated_at`: its arrival +/// (unless it is a re-check) and its validation. Gossip-accepted columns and +/// this node's own columns call this directly, as they need no check of their +/// own. +pub(crate) fn passed_rows( + sidecar: &DataColumnSidecar, + arrival: ColumnArrival, + validated_at: Instant, +) -> Vec { + let validated = column_event("column_validated", sidecar, arrival.source(), validated_at); + match arrival { + ColumnArrival::Received { at, .. } => vec![ + BlockEvent { + stage: "column_recv", + at, + ..validated + }, + validated, + ], + ColumnArrival::Recheck => vec![validated], + } +} + +/// The rows one checked sidecar writes (see [`check_and_forward`]): its +/// arrival unless already written, and its validation once kept. A dropped +/// sidecar writes none, as Silver records a column only once it passed +/// validation. +pub(crate) fn column_rows( + sidecar: &DataColumnSidecar, + arrival: ColumnArrival, + verdict: ChainVerdict, + checked_at: Instant, +) -> Vec { + match verdict { + ChainVerdict::Keep => passed_rows(sidecar, arrival, checked_at), + ChainVerdict::AwaitParent => match arrival { + ColumnArrival::Received { source, at } => { + vec![column_event("column_recv", sidecar, source, at)] + } + ColumnArrival::Recheck => Vec::new(), + }, + ChainVerdict::Drop(_) => Vec::new(), + } +} + /// Run the chain checks on `sidecars`, then send the chain actor the ones that /// passed as one batch and the ones waiting on a block as another. /// /// Returns at once: the checks run on a task of their own, so the p2p actor /// goes on handling swarm events while a range batch's KZG proofs are /// verified. -pub(crate) fn check_and_forward(server: &P2PServer, sidecars: Vec) { +/// +/// Each sidecar's `block_events` rows are written here once its check is done, +/// by [`column_rows`]: after the check, so an already stored re-delivery +/// writes none, and off the p2p actor. A column delivered again while it +/// waits for its block, or two copies passing at once, write rows each, so +/// rows are at least once per column and stage: a reader takes the earliest. +pub(crate) fn check_and_forward( + server: &P2PServer, + sidecars: Vec, + arrival: ColumnArrival, +) { if sidecars.is_empty() { return; } @@ -53,7 +161,7 @@ pub(crate) fn check_and_forward(server: &P2PServer, sidecars: Vec checked, Err(err) => { error!(%err, "A data column check panicked; dropping the sidecar"); continue; } }; + ethlambda_telemetry::emit(|| column_rows(&sidecar, arrival, verdict, checked_at)); match verdict { ChainVerdict::Keep => keep.push(sidecar), ChainVerdict::AwaitParent => awaiting_parent.push(sidecar), @@ -108,6 +217,7 @@ fn count_drop(outcome: Outcome) { #[cfg(test)] mod tests { use std::sync::Arc; + use std::time::Duration; use ethlambda_network_api::{ AggregateArrival, BlockAnnouncement, BlockArrival, BlockSource, P2PToBlockChain, @@ -192,7 +302,7 @@ mod tests { } /// A sidecar the checks cannot judge yet goes back to the chain actor to - /// be parked, and one they refuse goes nowhere: neither reaches the batch + /// be held for its block, and one they refuse goes nowhere: neither reaches the batch /// the chain actor stores unchecked. #[tokio::test] async fn only_what_passes_is_forwarded_as_checked() { @@ -211,6 +321,7 @@ mod tests { DataColumnSidecar::Fulu(orphan.clone()), DataColumnSidecar::Fulu(malformed), ], + ColumnArrival::Recheck, ); assert_eq!( @@ -227,10 +338,10 @@ mod tests { /// A gloas sidecar is judged against its own block, not a parent: one /// whose block this node has not seen goes back to the chain actor to be - /// parked, and one from a slot that has not started is dropped, so neither - /// reaches the batch the chain actor stores unchecked. + /// held for it, and one from a slot that has not started is dropped, so + /// neither reaches the batch the chain actor stores unchecked. #[tokio::test] - async fn a_gloas_sidecar_for_an_unknown_block_is_parked_and_a_future_one_dropped() { + async fn a_gloas_sidecar_for_an_unknown_block_is_held_and_a_future_one_dropped() { let mut server = unconnected_beacon_server(Config::mainnet(), 0).await; let (sender, mut received) = mpsc::unbounded_channel(); server.blockchain = Some(Arc::new(RecordingChain(sender))); @@ -248,7 +359,11 @@ mod tests { ..Default::default() }); - check_and_forward(&server, vec![unknown_block.clone(), future]); + check_and_forward( + &server, + vec![unknown_block.clone(), future], + ColumnArrival::Recheck, + ); assert_eq!( received.recv().await, @@ -257,4 +372,92 @@ mod tests { drop(server); assert_eq!(received.recv().await, None); } + + fn sidecar_48() -> DataColumnSidecar { + DataColumnSidecar::Gloas(gloas::DataColumnSidecar { + index: 48, + slot: 5, + beacon_block_root: Root::repeat_byte(9), + ..Default::default() + }) + } + + /// Silver records a column only once it passed validation: one judged to + /// wait for its block writes its arrival now and its validation when the + /// recheck keeps it, and a dropped one (an already stored re-delivery + /// included) writes none. + #[test] + fn a_column_writes_rows_only_once_judged_and_validation_only_once_it_passes() { + let sidecar = sidecar_48(); + let arrived = Instant::now(); + let checked = arrived + Duration::from_millis(3); + let fresh = ColumnArrival::received("by_root", arrived); + let rows = |rows: Vec| { + rows.into_iter() + .map(|row| (row.stage, row.at)) + .collect::>() + }; + + assert_eq!( + rows(column_rows(&sidecar, fresh, ChainVerdict::Keep, checked)), + [("column_recv", arrived), ("column_validated", checked)] + ); + let awaiting = column_rows(&sidecar, fresh, ChainVerdict::AwaitParent, checked); + assert_eq!( + awaiting, + [BlockEvent { + stage: "column_recv", + at: arrived, + slot: 5, + block_root: [9; 32], + source: "by_root", + verdict: None, + column_index: Some(48), + }] + ); + let stored = ChainVerdict::Drop(Outcome::Ignore(IgnoreReason::AlreadyStored)); + assert!(column_rows(&sidecar, fresh, stored, checked).is_empty()); + assert_eq!( + rows(column_rows( + &sidecar, + ColumnArrival::Recheck, + ChainVerdict::Keep, + checked + )), + [("column_validated", checked)] + ); + } + + /// A fulu sidecar names its block by its header's root, which the row + /// must carry. + #[test] + fn a_fulu_column_row_names_its_header_root() { + use ethlambda_types::beacon::primitives::HashTreeRoot as _; + + let fulu = valid_shaped_sidecar(5, 3); + let expected = fulu.signed_block_header.message.hash_tree_root(); + let sidecar = DataColumnSidecar::Fulu(fulu); + + let arrival = ColumnArrival::received("gossip", Instant::now()); + let event = passed_rows(&sidecar, arrival, Instant::now()).remove(0); + assert_eq!(event.block_root, expected.0); + assert_eq!(event.slot, 5); + assert_eq!(event.column_index, Some(3)); + } + + #[test] + fn a_column_row_names_its_block_slot_and_index() { + let sidecar = sidecar_48(); + let at = Instant::now(); + + let rows = passed_rows(&sidecar, ColumnArrival::received("by_root", at), at); + let event = rows[0]; + assert_eq!(event.stage, "column_recv"); + assert_eq!(event.slot, 5); + assert_eq!(event.block_root, [9; 32]); + assert_eq!(event.column_index, Some(48)); + assert_eq!(event.source, "by_root"); + assert_eq!(event.at, at); + assert_eq!(event.verdict, None); + } } diff --git a/crates/net/p2p/src/beacon/verdict.rs b/crates/net/p2p/src/beacon/verdict.rs index ab5651c26..76191f0b3 100644 --- a/crates/net/p2p/src/beacon/verdict.rs +++ b/crates/net/p2p/src/beacon/verdict.rs @@ -351,11 +351,26 @@ impl Validated { .inspect_err(|err| warn!(%err, "Failed to forward a gossip block")); } Self::Column(sidecar) if outcome == Outcome::Accept => { + // Accept means gossip already ran every rule. This instant is + // taken in `forward`, after the verdict's trip back through + // the p2p actor, where `check_and_forward` takes the instant + // its check finished. + ethlambda_telemetry::emit(|| { + column_checks::passed_rows( + &sidecar, + column_checks::ColumnArrival::received("gossip", received_at), + Instant::now(), + ) + }); let _ = blockchain .new_data_column_sidecars(vec![*sidecar]) .inspect_err(|err| warn!(%err, "Failed to forward a data column sidecar")); } - Self::Column(sidecar) => column_checks::check_and_forward(server, vec![*sidecar]), + Self::Column(sidecar) => column_checks::check_and_forward( + server, + vec![*sidecar], + column_checks::ColumnArrival::received("gossip", received_at), + ), Self::Aggregate { aggregate, attesting_indices, diff --git a/crates/net/p2p/src/gossipsub/handler.rs b/crates/net/p2p/src/gossipsub/handler.rs index d6b90bce0..d9f08f8ae 100644 --- a/crates/net/p2p/src/gossipsub/handler.rs +++ b/crates/net/p2p/src/gossipsub/handler.rs @@ -34,6 +34,7 @@ use super::{ attestation_subnet_topic, topic_kind, }, }; +use crate::beacon::column_checks::{ColumnArrival, check_and_forward, passed_rows}; use crate::beacon::constants::ATTESTATION_SUBNET_COUNT; use crate::beacon::decode::BeaconGossip; use crate::beacon::verdict::{self, Dispatch, GossipId, Validated}; @@ -981,6 +982,15 @@ pub async fn publish_beacon_block( .filter(|sidecar| beacon.custody_columns.contains(&sidecar.index())) .collect(); if !custody.is_empty() { + let now = Instant::now(); + // This node built these columns, so they are trusted without a check. + ethlambda_telemetry::emit(|| { + let arrival = ColumnArrival::received("local", now); + custody + .iter() + .flat_map(|sidecar| passed_rows(sidecar, arrival, now)) + .collect() + }); let _ = blockchain .new_data_column_sidecars(custody) .inspect_err(|err| error!(%err, "Failed to hand the custody columns to the chain")); @@ -1086,7 +1096,11 @@ pub async fn publish_execution_payload_envelope( .new_execution_payload_envelope(Box::new(envelope), BlockArrival::now()) .inspect_err(|err| error!(%err, "Failed to hand the published envelope to the chain")); } - crate::beacon::column_checks::check_and_forward(server, sidecars); + check_and_forward( + server, + sidecars, + ColumnArrival::received("local", Instant::now()), + ); } /// Gossip a payload attestation message a validator client signed, handed diff --git a/crates/net/p2p/src/lib.rs b/crates/net/p2p/src/lib.rs index cc9ca2bd1..50c532d24 100644 --- a/crates/net/p2p/src/lib.rs +++ b/crates/net/p2p/src/lib.rs @@ -1853,7 +1853,11 @@ impl Handler for P2PServer { impl Handler for P2PServer { async fn handle(&mut self, msg: CheckDataColumnSidecars, _ctx: &Context) { - beacon::column_checks::check_and_forward(self, msg.sidecars); + beacon::column_checks::check_and_forward( + self, + msg.sidecars, + beacon::column_checks::ColumnArrival::Recheck, + ); } } diff --git a/crates/net/p2p/src/req_resp/handlers.rs b/crates/net/p2p/src/req_resp/handlers.rs index 5801a84f2..eec647d53 100644 --- a/crates/net/p2p/src/req_resp/handlers.rs +++ b/crates/net/p2p/src/req_resp/handlers.rs @@ -2470,6 +2470,7 @@ async fn handle_data_column_sidecars_range_response( end_slot: u64, sidecars: Vec, ) { + let received_at = Instant::now(); let received = sidecars.len(); trace!(%peer, start_slot, end_slot, received, "Received DataColumnsByRange response"); @@ -2498,7 +2499,11 @@ async fn handle_data_column_sidecars_range_response( // delivery of, and it arrives precisely when the chain actor is busiest // draining a backlog, so a message per sidecar would put that many // mailbox hops between the answer and the imports waiting on it. - column_checks::check_and_forward(server, in_range); + column_checks::check_and_forward( + server, + in_range, + column_checks::ColumnArrival::received("by_range", received_at), + ); } /// Keep the sidecars of `sidecars` that answer the request: the requested @@ -2531,6 +2536,7 @@ async fn handle_data_column_sidecars_response( sidecars: Vec, ctx: &Context, ) { + let received_at = Instant::now(); let received = sidecars.len(); trace!(%peer, %block_root, received, "Received DataColumnsByRoot response"); @@ -2560,7 +2566,11 @@ async fn handle_data_column_sidecars_response( return; } - column_checks::check_and_forward(server, sidecars); + column_checks::check_and_forward( + server, + sidecars, + column_checks::ColumnArrival::received("by_root", received_at), + ); } /// Take delivery of a range answer, checking it against what was asked for, diff --git a/docs/SUMMARY.md b/docs/SUMMARY.md index 311798283..ba5d16acf 100644 --- a/docs/SUMMARY.md +++ b/docs/SUMMARY.md @@ -23,6 +23,7 @@ - [Command line](./cli.md) - [HTTP API](./rpc.md) - [Metrics](./metrics.md) +- [Block telemetry (ClickHouse)](./telemetry.md) - [Checkpoint Sync](./checkpoint_sync.md) - [Fork Choice Visualization](./fork_choice_visualization.md) - [Data Storage](./data_storage.md) diff --git a/docs/cli.md b/docs/cli.md index 2dcbf6703..fe7808e5b 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -63,6 +63,8 @@ one `CommonOptions` struct flattened into each. | `--discovery.port` | `9000` | discv5 UDP port; must differ from `--gossipsub-port`. See [Peer discovery](./discovery.md) | | `--discovery.advertise-ip` | bind address | IP published in the ENR | | `--discovery.target-peers` | `200` | Connected-peer count above which discovery stops dialing | +| `--telemetry.clickhouse-url` | off | ClickHouse HTTP endpoint for per-block `block_events` rows; also read from `ETHLAMBDA_TELEMETRY_CLICKHOUSE_URL`, the place for a URL with a password. See [Block telemetry](./telemetry.md) | +| `--telemetry.client-name` | `--node-id` / `ethlambda@` | `meta_client_name` on those rows; requires the URL | Three of those have no clap default, because the two chains answer them differently. A single `default_value` can only say one thing, so the flag diff --git a/docs/metrics.md b/docs/metrics.md index 7107778d1..edc0ba9ef 100644 --- a/docs/metrics.md +++ b/docs/metrics.md @@ -304,6 +304,15 @@ Per-arrival phases, charged once per arriving message however many blocks its ca | `get_head` | `fork_choice::get_head` (beacon) | | `fcu` | The `engine_forkchoiceUpdated` round trip (beacon) | +### Block Telemetry + +Present only with `--telemetry.clickhouse-url`; see [Block telemetry](./telemetry.md). Both series are published at zero at startup, so a node whose inserts never succeed shows a rising failure count next to zero inserted rows instead of no series. + +| Name | Type | Usage | Sample collection event | Labels | +|------|------|-------|-------------------------|--------| +| `lean_telemetry_rows_total` | Counter | `block_events` rows, by what became of them | On every successful insert, and on every row dropped (channel full, backlog past its cap, a batch ClickHouse refused, or rows still pending when the writer's channel closes) | outcome=inserted,dropped | +| `lean_telemetry_insert_failures_total` | Counter | Failed ClickHouse insert attempts, refusals included | On every failed insert; a failed best-effort schema statement alone is not counted | | + ### Gossip Arrival Timing These histograms record the absolute distance between a gossip message's arrival and the start of the interval it was due in, so an arrival that is early by some amount and one that is late by the same amount land in the same bucket; the counters' `position` label is what tells them apart. `inside` means the message arrived within the interval it was due in, not merely somewhere in the right slot: an attestation for slot 10 that lands during slot 10's interval 2 is `after`, not `inside`, since it missed the AttestationProduction interval it was actually due in. diff --git a/docs/telemetry.md b/docs/telemetry.md new file mode 100644 index 000000000..439fb821e --- /dev/null +++ b/docs/telemetry.md @@ -0,0 +1,144 @@ +# Block telemetry (ClickHouse) + +With `--telemetry.clickhouse-url`, a node writes one row per point in each +block's life, and per data column arrival, to a ClickHouse table named +`block_events`. The table layout is [Silver](https://github.com/gattaca-com/silver)'s, +so both clients' rows can share one table and one set of queries. + +Rows are batched and sent over ClickHouse's HTTP interface on a one second +tick, with one request in flight at a time, since ClickHouse writes a part per +insert. A full request's worth of backlog goes out without waiting for the next +tick. A failed request is retried on a later tick, with the backlog capped so an +unreachable ClickHouse costs bounded memory (the oldest rows are dropped first). +A batch ClickHouse rejects as malformed (HTTP 400) is dropped instead, since +retrying it would block every later row. + +Delivery is at least once: a request that times out after ClickHouse committed +it is sent again, so a query should take `min`/`any` per block, stage and client +rather than count rows. Rows still pending when the node exits are lost. +`lean_telemetry_rows_total` and `lean_telemetry_insert_failures_total` say what +happened to the rows. The table is created on first use; that statement is best +effort, so a user that may insert but not create still works against an existing +table. + +## Flags + +| Flag | Default | Meaning | +|---|---|---| +| `--telemetry.clickhouse-url` (or `ETHLAMBDA_TELEMETRY_CLICKHOUSE_URL`) | off | ClickHouse HTTP endpoint, e.g. `http://127.0.0.1:8123`. `user:pass@` in the URL is sent as basic auth; `?database=` passes through. Pass a URL with a password through the environment variable: a command line is readable by any local user | +| `--telemetry.client-name` | `--node-id` (`node`), `ethlambda@` (`beacon`) | `meta_client_name` on every row; requires the URL | + +`meta_network_name` is the network's `CONFIG_NAME` on `beacon` (`unknown` when a +loaded network directory has none) and `lean` on `node`. + +In a container, set `--telemetry.client-name` (or the container's hostname): +Docker's default hostname is the container ID, so the `beacon` default would +change on every recreate and split one node's rows across names. Only +`http`/`https` URLs are accepted; ClickHouse's native port (9000) is not this +interface. A rejected URL is never echoed back, since it may hold a password. + +## Columns + +| Column | Content | +|---|---| +| `event_date_time` | Wall clock of the event, nanoseconds | +| `stage` | Which point in the block's life (below) | +| `slot`, `slot_start_date_time` | The block's slot and when it started | +| `time_into_slot_ms` | Offset from the slot start; NULL outside the two slots after it (catch-up, backfill) | +| `block_root` | `0x`-prefixed | +| `source` | Blocks: `gossip`, `sync` (`replay` comes only from the offline benchmark and replay commands, which never install telemetry). Columns: `gossip`, `by_root`, `by_range`, `local`, `recheck` | +| `verdict` | `el_verdict` rows only: `Valid`, `Optimistic`, `Invalid` | +| `column_index` | `column_*` rows only | + +## Stages + +Block rows are written once a block's import finishes (imported by its own +state transition, or failed); a block that never leaves a hold has no rows, and +a re-delivery of a block already stored writes none. Silver's names are used +where the meaning matches; every other section of the +[import timing](./metrics.md#block-import-timing) is written as +`
_start` and `
_end`. The only block that writes no rows is +lean's locally built one, which crossed no wire and carries no source. A block +re-delivered after waiting for its slot reports under its original source and +adds `defer_start` and `defer_end`. On `beacon`, a proposal handed over by a +validator client enters as `gossip`, with `received` at the hand-off and no +`decode_end` row. + +| Stage | Meaning | +|---|---| +| `received` | The block came off the wire (gossip) or was handed to the chain actor (req/resp) | +| `da_available` | The data-availability check passed for a block that needed columns. A block with nothing to wait for (no blobs, or gloas, whose data gates the envelope instead) writes `da_check_end` instead | +| `el_sent`, `el_verdict` | `engine_newPayload` sent and answered; only when an execution client answered | +| `stf_done` | The state transition finished | +| `attestable` | The new head including the block is computed, as in Silver. Beacon recomputes the head once per arriving block, after the whole import cascade, so every block that cascade imported gets the same time, and only when that head computation succeeded (a failed one moves nothing, so writes no `attestable`); lean updates its head per block | +| `failed` | The import failed; `stf_done` and lean's `attestable` are then written as `stf_end` and `fc_head_end` | +| `column_recv`, `column_validated` | A data column arrived, and passed validation. Written only for a column that passed (or, for `column_recv`, was judged to wait for its block), as Silver does, so rejected columns and re-delivered stored ones write nothing. At least once per block, column and stage: a column delivered again while it waits, or two copies passing at once, write again, so take the earliest. A column that waited writes `column_validated` with source `recheck`, so join the two on block root and column index, not on source. By-root and by-range answers stamp every column with the moment the whole answer reached the handler. A gossip-accepted column's `column_validated` comes after the verdict's trip back through the p2p actor; a checked one's when its check finished | +| `cascade_*`, `prune_*`, `get_head_*`, `fcu_*` | Once per arrival, filed under the arriving block, when that block was imported in that cascade | + +ethlambda waits for the execution client before the state transition, so +`el_sent` to `stf_done` covers both; `stf_start` separates them. + +`source` and `verdict` are ethlambda's own values on every row, not Silver's +(`Gossip`, `Rpc`, `Syncing`, ...); a query shared with Silver's rows must +match both. + +## Every stage written + +| Group | Stages | +|---|---| +| Per block, in order | `received`, `decode_end`, `queue_end`, `defer_start`, `defer_end`, `admit_start`, `admit_end`, `guards_start`, `guards_end`, `preamble_start`, `preamble_end`, `parent_wait_start`, `parent_wait_end`, `cascade_wait_start`, `cascade_wait_end`, `da_check_start`, `da_available` (or `da_check_end`), `columns_wait_start`, `columns_wait_end`, `verify_struct_start`, `verify_struct_end`, `verify_crypto_start`, `verify_crypto_end`, `stf_start`, `stf_done`, `writer_wait_start`, `writer_wait_end`, `db_write_start`, `db_write_end`, `fc_head_start`, `attestable` (lean), `block_atts_start`, `block_atts_end` | +| Per block, with an execution client | `el_sent`, `el_verdict` | +| Per block, on failure | `stf_end` and `fc_head_end` in place of `stf_done` and `attestable`, plus `failed` | +| Per arrival | `cascade_start`, `cascade_end`, `prune_start`, `prune_end`, `get_head_start`, `get_head_end`, `fcu_start`, `fcu_end`, and (beacon) `attestable` | +| Per column | `column_recv`, `column_validated` | + +Only the stages a block actually crossed are written. `decode` and `queue` write +only their `_end` rows, since `received` is their start. `admit_*` is the +chain actor's admission section; it is written here and listed in no other doc. + +## Known gaps + +- On a node with a validator client, its own beacon proposals appear under + `source = 'gossip'` with near-zero arrival latency. Filter them out by their + missing `decode_end` row when comparing with another client. + +- A held block keeps its arrival marks (`received`, `decode_end`, + `queue_end`, `defer_*`, `admit_*`, the starts of its waits) across the + hold, but the sections after them (`guards_*`, `da_check_*`, the import + itself) are the final pass's: an earlier pass's failed check is not + written. +- A block re-delivered while held reports the last delivery as `received`, + not the first (the import timing histogram does the same). +- Some blocks released from a hold carry no timings and write no rows, e.g. + one held again by the execution client after its columns arrived. A block + the execution client could not judge, or that is not an optimistic + candidate, writes no rows either. +- An ancestor read back from storage (mainly after a restart) reports that + read as its `received`. +- On gloas the payload goes to the execution client with its envelope, not + the block, so gloas blocks have no `el_*` rows; a gloas block held for its + parent's payload has no wait section of its own. +- Lean's `attestable` is the end of its head update plus the bookkeeping an + import ends with (events, eviction, table sizes), sub-millisecond in + practice. + +## A query to start from + +Per-block arrival to attestable, in milliseconds: + +```sql +SELECT block_root, any(slot) AS slot, + (toUnixTimestamp64Nano(minIf(event_date_time, stage = 'attestable')) + - toUnixTimestamp64Nano(minIf(event_date_time, stage = 'received'))) / 1e6 AS total_ms +FROM block_events +WHERE meta_client_name = 'ethlambda@' AND event_date_time > now() - INTERVAL 1 HOUR +GROUP BY block_root +HAVING countIf(stage = 'attestable') > 0 +ORDER BY slot DESC +LIMIT 50 +``` + +## Retention + +The table has no TTL. Set one as an operator, e.g. +`ALTER TABLE block_events MODIFY TTL toDateTime(event_date_time) + INTERVAL 14 DAY`.