diff --git a/AGENTS.md b/AGENTS.md index 187d4463c6..020313a5b5 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -325,6 +325,48 @@ until a runtime path actually reads it. Operator-facing documentation (encryption model, migration matrix, `freenet secrets` CLI) lives in [`docs/secrets-at-rest.md`](docs/secrets-at-rest.md). +## Two independent telemetry pipelines + +`telemetry-enabled` / `telemetry-endpoint` feed the project's central dashboard +(`tracing/telemetry.rs`). `otel-telemetry-enabled` / `otel-endpoint` are a +**separate, unrelated** OpenTelemetry SDK metrics pipeline (`tracing/otel.rs`): +no shared config, no shared endpoint, no fallback in either direction, and +`otel-endpoint` must never default to the dashboard collector. + +Rules when touching `tracing/otel.rs`: + +- Observable instruments must read **cumulative, never-reset** values. + `TransportSnapshot` fields are period accumulators that `take_snapshot` + zeroes for the legacy telemetry worker, so observing one as a counter yields + a non-monotonic series whenever `telemetry-enabled` is also on. +- Never export a `PeerId`, socket address, or any attribute identifying the + remote end of a connection. `PeerId` renders as `{pub_key}@{addr}`, which + leaks our address and re-identifies the node whenever it changes. This node's + identity is two resource attributes (`freenet.node.*`, one per export batch), + not a per-datapoint attribute. "Peer" means the *other* end of a connection. +- Histograms get their base-2 exponential aggregation from one `with_view` in + `build_provider_blocking`. Do not add explicit bucket boundaries per + instrument. +- Export outcomes must be logged by `OtlpHttpClient::send_bytes`. The SDK will + not do it: `opentelemetry-otlp` logs network errors and non-2xx at DEBUG on + the stated grounds that `PeriodicReader` re-logs them at error level, which + is true for the batch log/span processors and false for metrics. Deleting + that logging makes a dead collector produce no output at all. +- The `freenet.node.*` resource attributes are always emitted and must never + be made deferrable to `OTEL_RESOURCE_ATTRIBUTES`. They are what the + collector checks the bearer-token signature against; an override would + export an identity that does not match the signing key. +- The exporter always installs its own `HttpClient`, so `opentelemetry-otlp` + needs none of its `reqwest-*`/TLS features — enabling one pulls a second + reqwest major, a second TLS stack, and a C/asm aws-lc build into every + release target. + +Everything else — the bearer-token format, endpoint precedence, the +`OTEL_*` variables honored, and per-instrument notes — is in +[`docs/design/otel-metrics-exporter.md`](docs/design/otel-metrics-exporter.md), +and operator-facing configuration is in +[`docs/otel-metrics.md`](docs/otel-metrics.md). + ## External Resources - API docs: https://docs.rs/freenet diff --git a/Cargo.lock b/Cargo.lock index a9bb0ee823..90dad3498a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -881,6 +881,15 @@ version = "0.4.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3d52eff69cd5e647efe296129160853a42795992097e8af39800e1060caeea9b" +[[package]] +name = "convert_case" +version = "0.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "633458d4ef8c78b72454de2d54fd6ab2e60f9e02be22f3c6104cdc8a4e0fceb9" +dependencies = [ + "unicode-segmentation", +] + [[package]] name = "cookie" version = "0.18.1" @@ -1514,10 +1523,12 @@ version = "2.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "799a97264921d8623a957f6c3b9011f3b5492f557bbb7a5a19b7fa6d06ba8dcb" dependencies = [ + "convert_case", "proc-macro2", "quote", "rustc_version", "syn 2.0.119", + "unicode-xid", ] [[package]] @@ -2035,6 +2046,7 @@ dependencies = [ "anyhow", "arbitrary", "argon2", + "async-trait", "axum", "bincode", "blake3", @@ -2050,6 +2062,7 @@ dependencies = [ "cookie", "criterion", "ctrlc", + "curve25519-dalek 4.1.3", "dashmap", "delegate", "directories", @@ -2069,6 +2082,7 @@ dependencies = [ "hickory-resolver", "hkdf", "hostname", + "http 1.5.0", "httptest", "ipnet", "keyring", @@ -2078,6 +2092,7 @@ dependencies = [ "muda", "notify", "opentelemetry 0.32.0", + "opentelemetry-http 0.32.0", "opentelemetry-jaeger", "opentelemetry-otlp", "opentelemetry_sdk 0.32.1", @@ -2085,11 +2100,13 @@ dependencies = [ "pav_regression", "pin-project", "proptest", + "rand 0.10.1", "rand 0.9.4", + "rand_core 0.10.1", "redb", "regex", "renegade-ml", - "reqwest 0.12.28", + "reqwest", "rpassword", "rstest", "semver", @@ -2133,6 +2150,7 @@ dependencies = [ "winres", "wry", "x25519-dalek 3.0.0", + "xeddsa", "xz2", "zeroize", "zip", @@ -2225,7 +2243,7 @@ dependencies = [ "clap", "hex", "hmac", - "reqwest 0.12.28", + "reqwest", "semver", "serde", "serde_json", @@ -4153,7 +4171,7 @@ dependencies = [ "freenet-stdlib 0.8.5", "hex", "rand 0.9.4", - "reqwest 0.12.28", + "reqwest", "serde", "serde_json", "tempfile", @@ -4655,7 +4673,6 @@ dependencies = [ "bytes 1.12.1", "http 1.5.0", "opentelemetry 0.32.0", - "reqwest 0.13.3", ] [[package]] @@ -4689,7 +4706,6 @@ dependencies = [ "opentelemetry-proto", "opentelemetry_sdk 0.32.1", "prost", - "reqwest 0.13.3", "thiserror 2.0.19", ] @@ -5643,6 +5659,7 @@ checksum = "eddd3ca559203180a307f12d114c268abf583f59b03cb906fd0b3ff8646c1147" dependencies = [ "base64 0.22.1", "bytes 1.12.1", + "futures-channel", "futures-core", "futures-util", "http 1.5.0", @@ -5676,37 +5693,6 @@ dependencies = [ "webpki-roots 1.0.7", ] -[[package]] -name = "reqwest" -version = "0.13.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "62e0021ea2c22aed41653bc7e1419abb2c97e038ff2c33d0e1309e49a97deec0" -dependencies = [ - "base64 0.22.1", - "bytes 1.12.1", - "futures-channel", - "futures-core", - "futures-util", - "http 1.5.0", - "http-body", - "http-body-util", - "hyper", - "hyper-util", - "js-sys", - "log", - "percent-encoding", - "pin-project-lite", - "sync_wrapper", - "tokio", - "tower", - "tower-http 0.6.11", - "tower-service", - "url", - "wasm-bindgen", - "wasm-bindgen-futures", - "web-sys", -] - [[package]] name = "resolv-conf" version = "0.7.6" @@ -8991,6 +8977,22 @@ dependencies = [ "rustix", ] +[[package]] +name = "xeddsa" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4de76539e20b51353b2653cbf575bf987ab9fcef550fa59a2ff64562ed5b14cb" +dependencies = [ + "curve25519-dalek 4.1.3", + "derive_more", + "ed25519", + "ed25519-dalek", + "rand 0.10.1", + "sha2 0.11.0", + "x25519-dalek 2.0.1", + "zeroize", +] + [[package]] name = "xz2" version = "0.1.7" diff --git a/Cargo.toml b/Cargo.toml index 954db07222..ee7f1c8cab 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -87,10 +87,21 @@ chrono = { version = "0.4", default-features = true } # Random rand = "0.9" +# XEdDSA (Signal's construction): signs with the x25519 transport key so the +# OTel collector can verify node identity against the SAME public key peers +# and UIs see. Its rand 0.10 dep is exposed as `rand10` for the sign nonce. +xeddsa = "1.1" +rand10 = { package = "rand", version = "0.10", default-features = false, features = ["sys_rng"] } +rand_core10 = { package = "rand_core", version = "0.10" } + # Observability +async-trait = "0.1" opentelemetry = "0.32" +# Trait/types only (HttpClient for the per-request auth wrapper); its reqwest +# integration targets reqwest 0.13 and the workspace is on 0.12, so no features. +opentelemetry-http = "0.32" opentelemetry-jaeger = "0.22" -opentelemetry-otlp = "0.32.0" +opentelemetry-otlp = { version = "0.32.0", default-features = false } opentelemetry_sdk = { version = "0.32", features = ["rt-tokio"] } tracing = "0.1" # Bumping this? Re-read `rolling.rs`'s `impl io::Write for diff --git a/crates/core/Cargo.toml b/crates/core/Cargo.toml index a97212b004..92b23aa7fa 100644 --- a/crates/core/Cargo.toml +++ b/crates/core/Cargo.toml @@ -104,7 +104,9 @@ wasmtime = { workspace = true, optional = true } # PUT time must run regardless of which WASM execution backend is compiled in. wasmparser = { workspace = true } xz2 = { workspace = true } -reqwest = { workspace = true } +# `blocking` named explicitly (not just inherited via feature unification from +# opentelemetry-http) because tracing::otel constructs reqwest::blocking::Client. +reqwest = { workspace = true, features = ["blocking"] } # Hidden (no-echo) prompt for the `freenet secrets export/import` passphrase, so # secrets don't have to be passed on argv (where they leak via `ps`/history). rpassword = "7" @@ -126,8 +128,38 @@ tracing-core = { workspace = true } tracing-opentelemetry = { workspace = true, optional = true } tracing-subscriber = { workspace = true, features = ["json"], optional = true } tracing-appender = { workspace = true, optional = true } -opentelemetry-otlp = { workspace = true, optional = true } -opentelemetry_sdk = { workspace = true, optional = true } +# Non-optional: the SDK metrics pipeline (tracing::otel) ships in every build. +# default-features = false + explicit list drops the logs exporter; only metrics +# is a goal here (see docs/design/otel-metrics-exporter.md, non-goals). It does +# NOT drop trace: `http-proto` mandates `trace`, `prost` and +# opentelemetry-proto, so those stay in the graph regardless. +# NO reqwest-*-client / reqwest-rustls feature on purpose: tracing::otel always +# supplies its own HttpClient, so the exporter never builds one (it only does +# when none was given — opentelemetry-otlp src/exporter/http/mod.rs). Enabling +# them would pull opentelemetry-http's reqwest 0.13, whose `rustls` feature is +# hardwired to aws-lc-rs + rustls-platform-verifier: a second reqwest major, a +# second TLS root store, a C/asm build (aws-lc-sys via cc+cmake) on every +# release target, and two crypto providers in one rustls, which makes +# `CryptoProvider::get_default_or_install_from_crate_features` panic. +# Our client rides the workspace reqwest 0.12, which already has rustls-tls, so +# https:// collector endpoints work with nothing new in the graph. +# Direct dep so tracing::otel can implement HttpClient (per-request auth +# headers); the crate is already in the graph via opentelemetry-otlp. +opentelemetry-http = { workspace = true } +async-trait = { workspace = true } +http = { workspace = true } +# XEdDSA collector auth: sign with the x25519 transport key (see +# transport::crypto::TransportKeypair::auth_token_signer). rand10/rand_core10 +# exist only because xeddsa's rng bound is rand 0.10 (workspace rand is 0.9). +xeddsa = { workspace = true } +rand10 = { workspace = true } +rand_core10 = { workspace = true } +opentelemetry-otlp = { workspace = true, features = [ + "http-proto", + "metrics", + "internal-logs", +] } +opentelemetry_sdk = { workspace = true } hkdf = { workspace = true } keyring = { workspace = true } @@ -172,6 +204,9 @@ criterion = { workspace = true } freenet-stdlib = { features = ["net", "testing"], workspace = true } freenet-macros = { path = "../freenet-macros" } httptest = { workspace = true } +# Test-only: proves the collector can verify XEdDSA tokens with a stock +# ed25519 library after Montgomery->Edwards conversion (4.1.3 = xeddsa's). +curve25519-dalek = "4.1.3" libc = { workspace = true } # For sendmmsg syscall batching benchmarks parking_lot = { workspace = true, features = ["deadlock_detection"] } # Used by tests/in_process_restart.rs to directly probe the redb file lock as a @@ -230,7 +265,7 @@ winres = "0.1" default = ["redb", "trace", "websocket", "wasmtime-backend"] sqlite = ["sqlx"] trace = ["tracing-subscriber", "tracing-appender"] -trace-ot = ["opentelemetry-jaeger", "trace", "tracing-opentelemetry", "opentelemetry-otlp"] +trace-ot = ["opentelemetry-jaeger", "trace", "tracing-opentelemetry"] websocket = ["axum/ws"] testing = ["freenet-stdlib/testing", "parking_lot/deadlock_detection"] console-subscriber = ["dep:console-subscriber"] diff --git a/crates/core/src/config.rs b/crates/core/src/config.rs index f1501185b6..10f49033ce 100644 --- a/crates/core/src/config.rs +++ b/crates/core/src/config.rs @@ -234,6 +234,9 @@ pub struct ConfigArgs { #[command(flatten)] pub telemetry: TelemetryArgs, + + #[command(flatten)] + pub otel: OtelArgs, } impl Default for ConfigArgs { @@ -302,6 +305,7 @@ impl Default for ConfigArgs { shutdown_drain_secs: None, disable_auto_update: false, telemetry: Default::default(), + otel: Default::default(), } } } @@ -967,6 +971,17 @@ impl ConfigArgs { if cfg.telemetry.iface_tx_enabled { self.telemetry.iface_tx_enabled = true; } + // Kept separate from the telemetry merge above on purpose: the two + // features are independent. Unlike reference-ping/iface-tx this + // merge is bidirectional — `--otel-telemetry-enabled=false` parses + // to `Some(false)` and must override a config.toml that says true. + self.otel.enabled.get_or_insert(cfg.otel.enabled); + if let Some(endpoint) = cfg.otel.endpoint { + self.otel.endpoint.get_or_insert(endpoint); + } + // Always emitted (non-Option in OtelConfig), so merge + // unconditionally; the CLI value still wins via get_or_insert. + self.otel.auth_mode.get_or_insert(cfg.otel.auth_mode); } // Validate the effective config (CLI + values merged from config.toml). @@ -1473,6 +1488,14 @@ impl ConfigArgs { reference_ping_enabled: self.telemetry.reference_ping_enabled, iface_tx_enabled: self.telemetry.iface_tx_enabled, }, + otel: OtelConfig { + enabled: self.otel.enabled.unwrap_or(false), + endpoint: self.otel.endpoint, + auth_mode: self.otel.auth_mode.unwrap_or_default(), + // Same --id rule as telemetry: simulated networks and + // integration tests must not ship data to a collector. + is_test_environment: self.id.is_some(), + }, }; fs::create_dir_all(this.config_dir())?; @@ -1707,6 +1730,12 @@ pub struct Config { /// Telemetry configuration #[serde(flatten)] pub telemetry: TelemetryConfig, + + /// OpenTelemetry SDK metrics exporter settings. Strictly isolated from + /// `telemetry` above — see `docs/design/otel-metrics-exporter.md`. + #[serde(flatten)] + pub otel: OtelConfig, + /// Maximum seconds to wait on graceful shutdown for in-flight /// client-originated operations (PUT/UPDATE/GET/SUBSCRIBE) to /// finish before tearing down peer connections. @@ -2809,6 +2838,110 @@ fn default_iface_tx_enabled() -> bool { false } +/// How the OTel exporter authenticates to the collector. +/// +/// `freenet` sends a per-request `Authorization: Bearer +/// freenet////` token — an XEdDSA +/// signature over the preceding fields, signed with the node's x25519 +/// transport secret — see `tracing::otel::bearer_token`. Future methods get +/// new variants. +/// +/// `disabled` is the DEFAULT and sends no `Authorization` header: pointing the +/// exporter at your own collector must not ship a signed assertion of this +/// node's identity somewhere it was never asked to. Operators exporting to a +/// collector that verifies freenet tokens opt in explicitly; anyone else +/// carries their own auth in `OTEL_EXPORTER_OTLP_HEADERS`, which the exporter +/// never overwrites. +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, clap::ValueEnum)] +#[serde(rename_all = "lowercase")] +pub enum OtelAuthMode { + Freenet, + #[default] + Disabled, +} + +/// CLI/file args for the OpenTelemetry SDK metrics exporter. +/// +/// Strictly independent of [`TelemetryArgs`]: no shared field, no shared +/// default, no fallback in either direction. +#[derive(clap::Parser, Debug, Clone, Default, Serialize, Deserialize)] +pub struct OtelArgs { + /// Enable the OpenTelemetry SDK metrics exporter. Independent of + /// `telemetry-enabled`; enabling or disabling one has no effect on the + /// other. + /// + /// `num_args`/`default_missing_value` rather than a bare flag: with an + /// `env` binding, clap's `SetTrue` action treats ANY value of the variable + /// as true, so `FREENET_OTEL_TELEMETRY_ENABLED=false` would silently turn + /// the exporter ON. This form accepts `--otel-telemetry-enabled`, + /// `--otel-telemetry-enabled=false`, and a properly parsed env value. + /// + /// `Option` and NO `default_value`, unlike the sibling telemetry flags: + /// with a default, "unset" and "explicitly false" are indistinguishable + /// after parsing, so `build()` cannot let `--otel-telemetry-enabled=false` + /// override a `config.toml` that says true — i.e. the off switch would not + /// work. `None` means "not given"; `build()` resolves it to `false`. + #[arg( + id = "otel_telemetry_enabled", + long = "otel-telemetry-enabled", + env = "FREENET_OTEL_TELEMETRY_ENABLED", + num_args = 0..=1, + default_missing_value = "true", + action = clap::ArgAction::Set + )] + #[serde( + rename = "otel-telemetry-enabled", + skip_serializing_if = "Option::is_none" + )] + pub enabled: Option, + + /// OTLP/HTTP collector base URL (e.g. `http://collector:4318`). + /// + /// No clap `env =` binding on purpose. The standard + /// `OTEL_EXPORTER_OTLP_ENDPOINT` / `OTEL_EXPORTER_OTLP_METRICS_ENDPOINT` + /// variables must take priority over this file-level value, and binding + /// them here would merge them into the config layer and invert that + /// precedence. They are resolved in `tracing::otel` instead. + #[arg(id = "otel_endpoint", long = "otel-endpoint")] + #[serde(rename = "otel-endpoint", skip_serializing_if = "Option::is_none")] + pub endpoint: Option, + + /// Collector authentication method. `Option` so `build()` can tell "not + /// given on the CLI" from an explicit choice and merge the config-file + /// value; resolves to [`OtelAuthMode::default`] (`disabled`) when neither + /// sets it. + #[arg(id = "otel_auth_mode", long = "otel-auth-mode", value_enum)] + #[serde(rename = "otel-auth-mode", skip_serializing_if = "Option::is_none")] + pub auth_mode: Option, +} + +/// Resolved configuration for the OpenTelemetry SDK metrics exporter. +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct OtelConfig { + /// Whether the SDK metrics exporter is enabled. + #[serde(default, rename = "otel-telemetry-enabled")] + pub enabled: bool, + + /// Operator-configured OTLP/HTTP collector base URL, if any. `None` means + /// "let the SDK resolve it" — see `tracing::otel::resolve_metrics_endpoint`. + #[serde( + default, + rename = "otel-endpoint", + skip_serializing_if = "Option::is_none" + )] + pub endpoint: Option, + + /// Collector authentication method. + #[serde(default, rename = "otel-auth-mode")] + pub auth_mode: OtelAuthMode, + + /// Whether this is a test environment (detected via `--id`). Mirrors + /// [`TelemetryConfig::is_test_environment`]; suppresses export so test + /// networks can't ship data to a collector. + #[serde(skip)] + pub is_test_environment: bool, +} + impl Default for TelemetryConfig { fn default() -> Self { Self { @@ -6420,6 +6553,164 @@ shutdown-drain-secs = 42 } } + #[test] + fn otel_args_default_is_off_and_endpointless() { + // The new pipeline exports nothing yet, so shipping it on would be a + // behavior change. Operators opt in explicitly. + let args = OtelArgs::default(); + assert_eq!( + args.enabled, None, + "otel-telemetry-enabled unset must stay None so an explicit \ + --otel-telemetry-enabled=false can override config.toml" + ); + assert_eq!(args.endpoint, None, "no implicit collector"); + assert_eq!( + args.auth_mode.unwrap_or_default(), + OtelAuthMode::Disabled, + "auth must default off: pointing the exporter at a collector must \ + not ship a signed assertion of this node's identity unasked" + ); + } + + #[test] + fn otel_auth_mode_parses_from_cli_and_file() { + use clap::Parser; + let none = ConfigArgs::try_parse_from(["freenet"]).expect("bare parse"); + assert_eq!(none.otel.auth_mode, None, "unset on the CLI stays None"); + let off = ConfigArgs::try_parse_from(["freenet", "--otel-auth-mode", "disabled"]) + .expect("disabled parse"); + assert_eq!(off.otel.auth_mode, Some(OtelAuthMode::Disabled)); + let on = ConfigArgs::try_parse_from(["freenet", "--otel-auth-mode", "freenet"]) + .expect("freenet parse"); + assert_eq!(on.otel.auth_mode, Some(OtelAuthMode::Freenet)); + + // The file spelling is the lowercase variant name. + let cfg: OtelConfig = toml::from_str("otel-auth-mode = \"disabled\"").unwrap(); + assert_eq!(cfg.auth_mode, OtelAuthMode::Disabled); + let cfg: OtelConfig = toml::from_str("").unwrap(); + assert_eq!( + cfg.auth_mode, + OtelAuthMode::Disabled, + "absent key -> default" + ); + } + + #[test] + fn otel_flag_parses_from_cli() { + use clap::Parser; + let none = ConfigArgs::try_parse_from(["freenet"]).expect("bare parse"); + assert_eq!(none.otel.enabled, None, "no flag -> unset, not false"); + let set = ConfigArgs::try_parse_from(["freenet", "--otel-telemetry-enabled"]) + .expect("flag parse"); + assert_eq!( + set.otel.enabled, + Some(true), + "--otel-telemetry-enabled -> on" + ); + // Explicit `=false` must parse and mean false. Without this form the flag + // would be a bare ArgAction::SetTrue, and clap turns ANY value of the bound + // env var — including "false" — into true. + let off = ConfigArgs::try_parse_from(["freenet", "--otel-telemetry-enabled=false"]) + .expect("explicit false parse"); + assert_eq!( + off.otel.enabled, + Some(false), + "--otel-telemetry-enabled=false -> off" + ); + let with_ep = ConfigArgs::try_parse_from([ + "freenet", + "--otel-endpoint", + "http://collector.example:4318", + ]) + .expect("endpoint parse"); + assert_eq!( + with_ep.otel.endpoint.as_deref(), + Some("http://collector.example:4318") + ); + } + + /// C1 regression: the round-trip guard test above only round-trips the + /// serializer's OWN output, so a key-shape mismatch (nested `[otel]` + /// table vs. the flat keys the design spec and AGENTS.md document) is + /// invisible to it. Write the literal documented `config.toml` text and + /// confirm the flat keys actually parse into `Config::otel`. + #[tokio::test] + async fn otel_flat_config_toml_keys_are_honored() { + let temp_dir = tempfile::tempdir().unwrap(); + + // Base build to create the on-disk secrets + a valid config.toml for + // every OTHER field (all of them are `#[serde(flatten)]`d scalars, so + // this baseline has no `[table]` headers at all). + clap_bare_args(temp_dir.path()).build().await.unwrap(); + let base = tokio::fs::read_to_string(temp_dir.path().join("config.toml")) + .await + .unwrap(); + + // Strip whatever otel shape build() just wrote (pre-fix: a nested + // `[otel]` header + its two keys; post-fix: the two flat keys) so the + // literal lines appended below are unambiguous root-level keys. + let base: String = base + .lines() + .filter(|line| { + *line != "[otel]" + && !line.starts_with("otel-telemetry-enabled") + && !line.starts_with("otel-endpoint") + }) + .map(|line| format!("{line}\n")) + .collect(); + + // The literal config.toml the design spec (Configuration table) and + // AGENTS.md document: flat keys at the file root, no `[otel]` table. + let literal = format!( + "{base}otel-telemetry-enabled = true\notel-endpoint = \"http://collector.example:4318\"\n" + ); + std::fs::write(temp_dir.path().join("config.toml"), literal).unwrap(); + + let rebuilt = clap_bare_args(temp_dir.path()).build().await.unwrap(); + assert!( + rebuilt.otel.enabled, + "documented flat `otel-telemetry-enabled` key must be honored" + ); + assert_eq!( + rebuilt.otel.endpoint.as_deref(), + Some("http://collector.example:4318"), + "documented flat `otel-endpoint` key must be honored" + ); + } + + #[tokio::test] + async fn otel_cli_false_overrides_a_config_file_that_says_true() { + // The off switch has to work: an operator who exports to a collector + // and then needs it stopped must be able to do it from the command + // line without editing config.toml. `Some(false)` from the CLI beats + // the file; `None` (flag absent) lets the file's `true` through. + let temp_dir = tempfile::tempdir().unwrap(); + clap_bare_args(temp_dir.path()).build().await.unwrap(); + let path = temp_dir.path().join("config.toml"); + let base = tokio::fs::read_to_string(&path).await.unwrap(); + let base: String = base + .lines() + .filter(|line| !line.starts_with("otel-telemetry-enabled")) + .map(|line| format!("{line}\n")) + .collect(); + std::fs::write(&path, format!("{base}otel-telemetry-enabled = true\n")).unwrap(); + + // Flag absent first: build() rewrites config.toml, so the negative + // case has to run last or it would overwrite the seed. + let inherited = clap_bare_args(temp_dir.path()).build().await.unwrap(); + assert!( + inherited.otel.enabled, + "with the flag absent, config.toml's `true` must still win" + ); + + let mut args = clap_bare_args(temp_dir.path()); + args.otel.enabled = Some(false); + assert!( + !args.build().await.unwrap().otel.enabled, + "--otel-telemetry-enabled=false must override config.toml" + ); + } + #[tokio::test] async fn test_serde_config_args() { // Use tempfile for a guaranteed-writable directory (avoids CI permission issues on /tmp) @@ -7421,6 +7712,7 @@ shutdown-drain-secs = 42 shutdown_drain_secs: None, disable_auto_update: false, telemetry: Default::default(), + otel: Default::default(), } } @@ -7578,6 +7870,12 @@ shutdown-drain-secs = 42 reference_ping_enabled: true, iface_tx_enabled: true, }, + otel: OtelConfig { + enabled: true, + endpoint: Some("http://example.invalid:4319".to_string()), + auth_mode: OtelAuthMode::Freenet, // non-default: default is Disabled + is_test_environment: false, // #[serde(skip)] — derived from --id + }, shutdown_drain_secs: 77, disable_auto_update: true, // #[serde(skip)] — see destructure below } @@ -7631,6 +7929,7 @@ shutdown-drain-secs = 42 module_cache_budget_bytes, enable_event_log, telemetry, + otel, shutdown_drain_secs, // #[serde(skip)] runtime CLI/env flag — set from --disable-auto-update // at build() time, intentionally not persisted, so it does not @@ -7677,6 +7976,23 @@ shutdown-drain-secs = 42 shutdown_drain_secs, seed.shutdown_drain_secs, "shutdown_drain_secs" ); + let OtelConfig { + enabled: otel_enabled, + endpoint: otel_endpoint, + auth_mode: otel_auth_mode, + is_test_environment: _, // serde-skip, derived from --id + } = otel; + assert_eq!(otel_enabled, seed.otel.enabled, "otel.enabled"); + assert_eq!( + otel_endpoint, seed.otel.endpoint, + "otel.endpoint — an operator's collector URL must survive the \ + config.toml merge" + ); + assert_eq!( + otel_auth_mode, seed.otel.auth_mode, + "otel.auth_mode — an operator's explicit choice must survive the \ + config.toml merge, or auth silently reverts on restart" + ); let NetworkApiConfig { address, diff --git a/crates/core/src/contract/fair_queue.rs b/crates/core/src/contract/fair_queue.rs index 0f797cad11..ff48f52f5c 100644 --- a/crates/core/src/contract/fair_queue.rs +++ b/crates/core/src/contract/fair_queue.rs @@ -429,10 +429,18 @@ impl FairEventQueue { gauge::DEPTH_CLIENT_LOCAL.store(s.depth_client_local, Ordering::Relaxed); gauge::DEPTH_NETWORK_RELAY.store(s.depth_network_relay, Ordering::Relaxed); gauge::DEPTH_BACKGROUND.store(s.depth_background, Ordering::Relaxed); + // `fetch_max`, not `store`, for the same reason as HIGH_WATER: these + // four are exported as OTel observable *counters* + // (`tracing::otel::register_queue_metrics`), where a value that ever + // decreases is read as a counter reset. `self.counters` only grows, so + // this is equivalent today — there is one `FairEventQueue` per process + // — but a second queue, or a respawned `contract_handling` loop, would + // publish its own counters from zero and drag the global series + // backwards. gauge::HIGH_WATER.fetch_max(s.high_water, Ordering::Relaxed); - gauge::REJECTED_GLOBAL_CAPACITY.store(s.rejected_global_capacity, Ordering::Relaxed); - gauge::REJECTED_PER_CONTRACT.store(s.rejected_per_contract, Ordering::Relaxed); - gauge::BACKGROUND_SHED.store(s.background_shed, Ordering::Relaxed); + gauge::REJECTED_GLOBAL_CAPACITY.fetch_max(s.rejected_global_capacity, Ordering::Relaxed); + gauge::REJECTED_PER_CONTRACT.fetch_max(s.rejected_per_contract, Ordering::Relaxed); + gauge::BACKGROUND_SHED.fetch_max(s.background_shed, Ordering::Relaxed); for band in DEPTH_WARN_BANDS { if previous_high_water < band && s.high_water >= band { diff --git a/crates/core/src/node.rs b/crates/core/src/node.rs index d899fb75aa..b2429eb43a 100644 --- a/crates/core/src/node.rs +++ b/crates/core/src/node.rs @@ -818,6 +818,18 @@ impl NodeConfig { registers.push(Box::new(telemetry)); } + // Independent of the TelemetryReporter above: a separate opt-in + // (`otel-telemetry-enabled`), a separate endpoint, and a separate + // collector. It is not a NetEventRegister — it installs a global + // meter provider that instrumentation reaches via + // `opentelemetry::global::meter`. + // The transport keypair, NOT a `PeerId`: it yields both identity + // resource attributes and, in `freenet` auth mode, the token + // signing key. A PeerId renders as `{pub_key}@{addr}`, which would + // put this node's socket address on every exported batch and + // re-identify the node on every address change. + crate::tracing::otel::init(&self.config.otel, &self.key_pair); + (DynamicRegister::new(registers), flush_handle) }; let cfg = self.config.clone(); diff --git a/crates/core/src/node/network_status.rs b/crates/core/src/node/network_status.rs index 54d55d9326..c12717d64d 100644 --- a/crates/core/src/node/network_status.rs +++ b/crates/core/src/node/network_status.rs @@ -111,6 +111,64 @@ pub struct RingStatsSnapshot { pub lattice_probe_improvements: u64, } +/// Scalar-only view for the OTel metrics callbacks. +/// +/// Deliberately NOT [`get_snapshot`]: that builds per-peer and per-contract +/// vectors and formats failure HTML, and the SDK has no batch-callback API in +/// 0.32 — every observable instrument gets its own callback, so the exporter +/// would pay that cost once per instrument per collection cycle. +#[derive(Debug, Clone, Default)] +pub(crate) struct OtelMetricsSnapshot { + pub connection_attempts: u32, + pub ring: RingStatsSnapshot, + pub fair_queue: crate::contract::FairQueueStats, + /// Hosted contracts partitioned by why they are held. Deliberately NOT a + /// `RingStatsSnapshot` field: that provider runs on every dashboard HTTP + /// request, and this is an O(hosted) walk under the hosting-cache read + /// lock. Its own provider keeps the cost on the OTel collection cadence. + pub hosting_reasons: crate::ring::HostingReasonStats, +} + +/// Read the scalars the OTel exporter observes, or `None` before the node has +/// registered its status (metrics simply report nothing until then). +/// +/// Every unavailable source yields `None` for the whole snapshot rather than a +/// default: an observable instrument that skips a collection cycle exports +/// nothing, which reads as "not known yet", while a zero is a real datapoint — +/// `freenet.ring.connections = 0` before the ring provider registers is +/// indistinguishable from a node that has lost every connection. +pub(crate) fn otel_metrics_snapshot() -> Option { + let connection_attempts = NETWORK_STATUS.get()?.read().ok()?.connection_attempts; + let ring = RING_STATS_PROVIDER + .read() + .as_ref() + .map(|provider| provider())?; + let hosting_reasons = HOSTING_REASON_PROVIDER + .read() + .as_ref() + .map(|provider| provider())?; + Some(OtelMetricsSnapshot { + connection_attempts, + ring, + fair_queue: crate::contract::fair_queue_stats(), + hosting_reasons, + }) +} + +/// Source of the per-reason hosted-contract breakdown +/// (`Ring::hosted_by_reason`). OTel-only; see [`OtelMetricsSnapshot`]. +pub type HostingReasonProvider = + Arc crate::ring::HostingReasonStats + Send + Sync + 'static>; + +static HOSTING_REASON_PROVIDER: parking_lot::RwLock> = + parking_lot::RwLock::new(None); + +/// Register the hosting-reason data source. Replaces any previously-registered +/// provider. +pub fn set_hosting_reason_provider(provider: HostingReasonProvider) { + *HOSTING_REASON_PROVIDER.write() = Some(provider); +} + static GOVERNANCE_PROVIDER: parking_lot::RwLock> = parking_lot::RwLock::new(None); @@ -826,6 +884,15 @@ pub fn record_peer_disconnected(addr: SocketAddr) { /// Audit: `grep -rn "record_op_result" crates/core/src/operations/` /// must show coverage for every op type with a driver. pub fn record_op_result(op_type: OpType, success: bool) { + crate::tracing::otel::record_op_result( + match op_type { + OpType::Get => "get", + OpType::Put => "put", + OpType::Update => "update", + OpType::Subscribe => "subscribe", + }, + success, + ); if let Some(status) = NETWORK_STATUS.get() { if let Ok(mut s) = status.write() { let counter = match op_type { diff --git a/crates/core/src/node/p2p_impl.rs b/crates/core/src/node/p2p_impl.rs index fdab84c584..7a2804aba2 100644 --- a/crates/core/src/node/p2p_impl.rs +++ b/crates/core/src/node/p2p_impl.rs @@ -524,6 +524,15 @@ impl NodeP2P { hosting_ring.dashboard_hosting_snapshot() })); + // Per-reason hosted-contract breakdown for the OTel exporter only + // (count + state bytes per `HostingReason`). Kept off the ring-stats + // provider above because that one runs on every dashboard HTTP + // request and this is an O(hosted) walk under the cache read lock. + let reason_ring = self.op_manager.ring.clone(); + super::network_status::set_hosting_reason_provider(std::sync::Arc::new(move || { + reason_ring.hosted_by_reason() + })); + // Wire live ring stats for the dashboard: connection count + // hosted contracts + own public key, read on every homepage // request. diff --git a/crates/core/src/ring.rs b/crates/core/src/ring.rs index 9bc4dbce60..715548ec67 100644 --- a/crates/core/src/ring.rs +++ b/crates/core/src/ring.rs @@ -25,8 +25,8 @@ use parking_lot::{Mutex, RwLock}; use tokio_util::sync::CancellationToken; pub use hosting::{ - AddClientSubscriptionResult, AddSubscriberOutcome, ClientDisconnectResult, SubscribeResult, - SubscribedContractSnapshot, + AddClientSubscriptionResult, AddSubscriberOutcome, ClientDisconnectResult, HostingReason, + HostingReasonStats, SubscribeResult, SubscribedContractSnapshot, }; use crate::message::TransactionType; @@ -4189,6 +4189,14 @@ impl Ring { self.hosting_manager.hosting_contracts_count() } + /// The same hosted set as [`Self::hosting_contracts_count`], partitioned by + /// WHY each contract is held, with state bytes per bucket. Backs the + /// `freenet.node.contracts.hosted{,.bytes}` OTel gauges. See + /// [`HostingReason`]. + pub fn hosted_by_reason(&self) -> HostingReasonStats { + self.hosting_manager.hosted_by_reason() + } + /// Number of active network subscription leases this node currently holds. /// /// Together with [`hosting_contracts_count`](Self::hosting_contracts_count) diff --git a/crates/core/src/ring/hosting.rs b/crates/core/src/ring/hosting.rs index c712495b23..d365d65231 100644 --- a/crates/core/src/ring/hosting.rs +++ b/crates/core/src/ring/hosting.rs @@ -202,6 +202,93 @@ pub(crate) enum PhantomRepair { Drop(ContractKey), } +/// Why this node is holding a contract it hosts. +/// +/// This is a PARTITION, not a set of flags: the classifier in +/// [`HostingManager::hosted_by_reason`] evaluates the variants in declaration +/// order and assigns each hosted contract to the FIRST one that matches, so +/// the per-reason counts sum to the hosting-cache size and the per-reason +/// bytes sum to its used bytes. That is the whole point — the underlying +/// signals overlap (a contract can be locally accessed AND have downstream +/// subscribers), and an overlapping breakdown makes `sum by (reason)` lie. +/// +/// Ordering is strongest-claim-first: a reason further down the list only +/// applies when every reason above it is absent. `LocalClient` outranks +/// `Downstream` for the same reason eviction does (`local_and_downstream_counts` +/// — this node's own user beats forwarded demand), and everything outranks +/// `Routed`, which is the residual "no demand signal at all" bucket. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum HostingReason { + /// A local client (WebSocket/HTTP) holds a subscription. This node's own + /// user wants the contract. + LocalClient, + /// A downstream peer subscribes to us for this contract — we are a relay + /// in someone else's update mesh. + Downstream, + /// We hold an unexpired network subscription but nothing local or + /// downstream reads it: hosted on the network's behalf. + Subscribed, + /// No subscription of any kind, but a local client GET/PUT touched it + /// (`local_client_access`). The read-only / PUT-only local-demand class + /// (River UI containers and friends). + LocalAccess, + /// Was in use and no longer is (`abandoned_at`) — the eviction candidate + /// pool. Distinguished from `Routed` because a rising `abandoned` count is + /// churn, while a rising `routed` count is ordinary transit caching. + Abandoned, + /// Residual: arrived through a routed GET/PUT and never acquired any + /// demand signal. + Routed, +} + +impl HostingReason { + /// Every variant, in classifier (and export) order. + pub const ALL: [HostingReason; 6] = [ + HostingReason::LocalClient, + HostingReason::Downstream, + HostingReason::Subscribed, + HostingReason::LocalAccess, + HostingReason::Abandoned, + HostingReason::Routed, + ]; + + /// Stable attribute value. These strings are a metrics contract — a + /// collector-side dashboard filters on them, so renaming one silently + /// empties a panel. Add variants rather than repurposing these. + pub fn as_str(self) -> &'static str { + match self { + HostingReason::LocalClient => "local_client", + HostingReason::Downstream => "downstream", + HostingReason::Subscribed => "subscribed", + HostingReason::LocalAccess => "local_access", + HostingReason::Abandoned => "abandoned", + HostingReason::Routed => "routed", + } + } +} + +/// Hosted-contract count and state bytes per [`HostingReason`], indexed by +/// `reason as usize`. Both arrays partition the hosting cache (see +/// [`HostingReason`]), so `counts.iter().sum()` is the hosted-contract count +/// and `bytes.iter().sum()` is the cache's used bytes. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct HostingReasonStats { + counts: [u64; HostingReason::ALL.len()], + bytes: [u64; HostingReason::ALL.len()], +} + +impl HostingReasonStats { + /// Contracts held for `reason`. + pub fn count(&self, reason: HostingReason) -> u64 { + self.counts[reason as usize] + } + + /// Contract state bytes held for `reason`. + pub fn bytes(&self, reason: HostingReason) -> u64 { + self.bytes[reason as usize] + } +} + /// Result of adding a client subscription. #[derive(Debug)] pub struct AddClientSubscriptionResult { @@ -897,6 +984,40 @@ impl HostingManager { Some(tracker.stats()) } + /// Count and state bytes of hosted contracts, partitioned by WHY each one + /// is held (see [`HostingReason`]). Fixed cardinality — six buckets, no + /// contract identity survives the walk — so it is safe to export as + /// metric attributes. + /// + /// One O(hosted) pass under the hosting-cache read lock. The subscription + /// lookups inside the closure read only the `client_subscriptions` / + /// `downstream_subscribers` / `active_subscriptions` DashMaps, never the + /// hosting cache, so there is no re-lock — the same discipline + /// [`Self::cost_eligibility_stats`] relies on. + pub(crate) fn hosted_by_reason(&self) -> HostingReasonStats { + let mut stats = HostingReasonStats::default(); + self.hosting_cache.read().for_each_reason_row(|key, entry| { + let (local, downstream) = self.local_and_downstream_counts(key); + let reason = if local > 0 { + HostingReason::LocalClient + } else if downstream > 0 { + HostingReason::Downstream + } else if self.is_subscribed(key) { + HostingReason::Subscribed + } else if entry.local_client_access { + HostingReason::LocalAccess + } else if entry.abandoned_at.is_some() { + HostingReason::Abandoned + } else { + HostingReason::Routed + }; + let bucket = reason as usize; + stats.counts[bucket] = stats.counts[bucket].saturating_add(1); + stats.bytes[bucket] = stats.bytes[bucket].saturating_add(entry.size_bytes); + }); + stats + } + pub(crate) fn cost_eligibility_stats( &self, cost_axes: &[CostAxisPressure], @@ -4252,6 +4373,83 @@ mod tests { assert!(used.in_use, "a client subscription is real demand → in_use"); } + /// `hosted_by_reason` must PARTITION the hosting cache: one bucket per + /// contract, counts summing to the cache size and bytes to its used bytes. + /// The classification is priority-ordered, so each case below is set up + /// with every HIGHER-priority signal deliberately absent — a contract with + /// both a local client subscription and downstream subscribers must land in + /// `local_client` only, never be counted twice. + #[test] + fn hosted_by_reason_partitions_the_hosting_cache() { + let manager = HostingManager::new(DEFAULT_HOSTING_BUDGET_BYTES); + + // Empty cache: every bucket zero (a real datapoint, not absence). + let empty = manager.hosted_by_reason(); + for reason in HostingReason::ALL { + assert_eq!(empty.count(reason), 0, "{reason:?} on an empty cache"); + assert_eq!(empty.bytes(reason), 0, "{reason:?} on an empty cache"); + } + + // One contract per reason, distinct sizes so a mis-bucketed contract + // shows up in the bytes assertions too. + let local_client = make_contract_key(1); + let downstream = make_contract_key(2); + let subscribed = make_contract_key(3); + let local_access = make_contract_key(4); + let abandoned = make_contract_key(5); + let routed = make_contract_key(6); + for (key, size) in [ + (local_client, 100), + (downstream, 200), + (subscribed, 400), + (local_access, 800), + (abandoned, 1_600), + (routed, 3_200), + ] { + manager.record_contract_access(key, size, AccessType::Get); + } + + // `local_client` ALSO gets a downstream subscriber and a network + // subscription: priority must keep it in exactly one bucket. + manager.add_client_subscription(local_client.id(), crate::client_events::ClientId::next()); + manager.add_downstream_subscriber(&local_client, make_peer_key(10)); + manager.subscribe(local_client); + + // `downstream` also holds a network subscription — downstream wins. + manager.add_downstream_subscriber(&downstream, make_peer_key(11)); + manager.subscribe(downstream); + + manager.subscribe(subscribed); + manager.mark_local_client_access(&local_access); + + // Abandonment is a transition, not a flag: subscribe a downstream peer + // and take it away again. + manager.add_downstream_subscriber(&abandoned, make_peer_key(12)); + manager.remove_downstream_subscriber(&abandoned, &make_peer_key(12)); + + // `routed` gets nothing beyond the GET that seeded it. + + let stats = manager.hosted_by_reason(); + for (reason, size) in [ + (HostingReason::LocalClient, 100), + (HostingReason::Downstream, 200), + (HostingReason::Subscribed, 400), + (HostingReason::LocalAccess, 800), + (HostingReason::Abandoned, 1_600), + (HostingReason::Routed, 3_200), + ] { + assert_eq!(stats.count(reason), 1, "{reason:?} count"); + assert_eq!(stats.bytes(reason), size, "{reason:?} bytes"); + } + + // The partition property itself — what makes `sum by (reason)` valid. + let total_count: u64 = HostingReason::ALL.iter().map(|r| stats.count(*r)).sum(); + let total_bytes: u64 = HostingReason::ALL.iter().map(|r| stats.bytes(*r)).sum(); + let cache = manager.hosting_cache_stats(); + assert_eq!(total_count, cache.contract_count, "counts must partition"); + assert_eq!(total_bytes, cache.current_bytes, "bytes must partition"); + } + /// `is_eviction_eligible` gates the dashboard's "next to evict" badge on the /// sweep skip filter, which since 2026-07-08 is `!contract_in_use` ONLY (the /// `min_ttl` age gate was dropped — invariant 3). A freshly-accessed, not-in- diff --git a/crates/core/src/ring/hosting/cache.rs b/crates/core/src/ring/hosting/cache.rs index ed9a50c321..c24e2f35e1 100644 --- a/crates/core/src/ring/hosting/cache.rs +++ b/crates/core/src/ring/hosting/cache.rs @@ -1787,6 +1787,17 @@ impl HostingCache { } } + /// Current hosted rows for the fixed-cardinality hosting-REASON telemetry + /// (`HostingManager::hosted_by_reason`). Same contract-identity discipline + /// as [`Self::for_each_cost_eligibility_row`]: keys are visible to the + /// manager's classifier (which needs them to look up subscription state) + /// but are aggregated away before anything is exported. + pub(crate) fn for_each_reason_row(&self, mut visit: impl FnMut(&ContractKey, &HostedContract)) { + for (key, entry) in &self.contracts { + visit(key, entry); + } + } + /// Per-contract rows for the local dashboard, in the cache-side EVICTION /// order: ascending `(recency_seq, key)` — least-recent real GET/PUT first, the /// same order [`Self::keys_eviction_order`] uses. This reflects the order diff --git a/crates/core/src/tracing.rs b/crates/core/src/tracing.rs index 3c964931d0..bf2fd42f40 100644 --- a/crates/core/src/tracing.rs +++ b/crates/core/src/tracing.rs @@ -43,6 +43,10 @@ pub mod event_aggregator; pub mod telemetry; pub use telemetry::TelemetryReporter; +/// Standards-configured OpenTelemetry SDK metrics pipeline. Strictly isolated +/// from `telemetry` above — see `docs/design/otel-metrics-exporter.md`. +pub mod otel; + /// Automatic state verification through telemetry linearization. pub mod state_verifier; diff --git a/crates/core/src/tracing/otel.rs b/crates/core/src/tracing/otel.rs new file mode 100644 index 0000000000..8a4dbb191e --- /dev/null +++ b/crates/core/src/tracing/otel.rs @@ -0,0 +1,1803 @@ +//! Standards-configured OpenTelemetry SDK metrics pipeline. +//! +//! Strictly isolated from [`super::telemetry`]: nothing here reads +//! `TelemetryConfig`, and the endpoint never falls back to +//! `DEFAULT_TELEMETRY_ENDPOINT`. The two features are independent by design — +//! see `docs/design/otel-metrics-exporter.md`. + +use std::sync::OnceLock; +use std::sync::atomic::{AtomicBool, Ordering}; + +use opentelemetry::metrics::{Counter, Histogram}; +use opentelemetry::{KeyValue, global}; +use opentelemetry_http::{Bytes, HttpClient, HttpError, Request, Response}; +use opentelemetry_otlp::{ExporterBuildError, MetricExporter, WithExportConfig, WithHttpConfig}; +use opentelemetry_sdk::{ + Resource, + metrics::{Aggregation, Instrument, InstrumentKind, SdkMeterProvider, Stream}, +}; + +use crate::config::OtelConfig; + +/// Why the OTel metrics exporter was not started. +/// +/// Mirrors `telemetry::TelemetrySuppression` so both pipelines refuse to ship +/// data from a test process, but the decision is computed from `OtelConfig` +/// alone — the two flags never consult each other. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum OtelSuppression { + /// Operator left `otel-telemetry-enabled` off (the default). + Disabled, + /// `--id` test environment (integration/CLI harness sets `is_test_environment`). + TestEnvironmentFlag, + /// A `cfg(test)` build or a binary running under a cargo test/bench harness. + TestHarness, +} + +/// Decide whether the metrics exporter should be suppressed. +/// +/// Pure and side-effect free: callers pass `cfg!(test)` and the result of +/// `telemetry::running_under_cargo_test()` so this is testable for a +/// production release binary (must NOT suppress) from inside a test process, +/// which by construction trips both test signals. +/// +/// Suppression is keyed only on signals a real release binary never matches, +/// and deliberately NOT on `cfg!(feature = "testing")` — that flag leaks onto +/// the shipped binary through Cargo feature unification with `fdev` and +/// silently disabled telemetry across the fleet once already (#4366, the +/// 0.2.81 blackout). See `telemetry::telemetry_suppression_reason`. +pub(crate) fn otel_suppression_reason( + config: &OtelConfig, + is_test_build: bool, + running_under_cargo_test: bool, +) -> Option { + if !config.enabled { + return Some(OtelSuppression::Disabled); + } + if config.is_test_environment { + return Some(OtelSuppression::TestEnvironmentFlag); + } + if is_test_build || running_under_cargo_test { + return Some(OtelSuppression::TestHarness); + } + None +} + +/// Endpoint to hand to `MetricExporter`'s builder, or `None` to let the SDK +/// resolve it. +/// +/// Required precedence is env > config file > SDK default, but +/// `opentelemetry-otlp` 0.32 inverts the first two: `resolve_http_endpoint` +/// (`src/exporter/http/mod.rs`) checks the programmatic value FIRST and only +/// then `OTEL_EXPORTER_OTLP_METRICS_ENDPOINT` / `OTEL_EXPORTER_OTLP_ENDPOINT`. +/// So whenever either variable is set we return `None` and stay out of the +/// way. It also appends the `/v1/metrics` signal path only on the env-var +/// path, so a config-file value gets the path appended here. +pub(crate) fn resolve_metrics_endpoint( + cfg_endpoint: Option<&str>, + metrics_env: Option<&str>, + generic_env: Option<&str>, +) -> Option { + let is_set = |v: Option<&str>| v.is_some_and(|s| !s.trim().is_empty()); + if is_set(metrics_env) || is_set(generic_env) { + return None; + } + let base = cfg_endpoint.map(str::trim).filter(|s| !s.is_empty())?; + Some(format!("{}/v1/metrics", base.trim_end_matches('/'))) +} + +/// The endpoint the standard environment declares, in SDK precedence order, +/// or `None` when neither variable is set to a non-blank value. +/// +/// Blank-filtered on purpose: `env::var` returns `Ok("")` for a variable set +/// to the empty string, which [`resolve_metrics_endpoint`] correctly treats as +/// unset — so reading it raw would report an override that is not happening. +fn env_endpoint() -> Option { + [ + opentelemetry_otlp::OTEL_EXPORTER_OTLP_METRICS_ENDPOINT, + opentelemetry_otlp::OTEL_EXPORTER_OTLP_ENDPOINT, + ] + .iter() + .find_map(|var| { + let value = std::env::var(var).ok()?; + let value = value.trim(); + (!value.is_empty()).then(|| value.to_owned()) + }) +} + +/// Why an OTLP endpoint will not work, or `None` when it is usable. +/// +/// Both failure modes are otherwise near-invisible. `http::Uri` accepts +/// `collector:4318` as an authority with no scheme, so the exporter builds and +/// then every export dies converting to a `reqwest::Request` ("relative URL"); +/// and an endpoint the SDK cannot parse at all is swallowed with `.ok()`, +/// falling back to `http://localhost:4318` while the startup log still names +/// the operator's URL. +fn endpoint_problem(endpoint: &str) -> Option<&'static str> { + match endpoint.parse::() { + Err(_) => Some( + "not a valid URL; the SDK will silently fall back to http://localhost:4318. \ + Include the scheme, e.g. http://collector:4318", + ), + Ok(uri) if !matches!(uri.scheme_str(), Some("http" | "https")) => { + Some("missing an http:// or https:// scheme; every export will fail to build a request") + } + Ok(_) => None, + } +} + +/// Build one `freenet`-mode bearer token: +/// `freenet////`, where `` +/// is the XEdDSA signature over everything preceding it. +/// +/// `` is the base58 full x25519 transport public key — the node's one +/// real identity, the same key peers see and whose truncated fingerprint UIs +/// display. `` names the collector the request is going to (see +/// [`audience_of`]), so a token is only valid there: without it, any collector +/// we export to could replay the token to any other collector accepting this +/// scheme and impersonate this node. `` is seconds since the Unix +/// epoch, `` is base58 too. +/// Freshly built per export request so the timestamp stays current. +/// +/// Collector-side verification needs no exotic library: convert the +/// Montgomery pubkey to Edwards (sign bit 0), then standard Ed25519 verify — +/// see `node_pubkey_is_verifiable_with_stock_ed25519` below. The collector +/// must additionally check `` against the hash of each URL it +/// answers at (see [`audience_of`]) and `` against its own clock. +pub(crate) fn bearer_token( + signer: &xeddsa::xed25519::PrivateKey, + pubkey_b58: &str, + audience: &str, +) -> String { + use xeddsa::xeddsa::Sign; + // Wall-clock epoch seconds on purpose: the collector checks it against + // ITS clock, so simulation time would be meaningless here. + let timestamp = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or_default(); + let signed_payload = format!("freenet/{pubkey_b58}/{audience}/{timestamp}"); + // OS entropy (SysRng), not GlobalRng: XEdDSA's Z randomness hedges the + // signature nonce, which is cryptographic material — the same exception + // documented in .claude/rules/code-style.md for keys/nonces. UnwrapErr is + // required because xeddsa's bound is the infallible rand 0.10 CryptoRng. + let signature: [u8; 64] = signer.sign( + signed_payload.as_bytes(), + rand_core10::UnwrapErr(rand10::rngs::SysRng), + ); + let signature = bs58::encode(signature).into_string(); + format!("{signed_payload}/{signature}") +} + +/// The `` field of a bearer token: base58 of the first 16 bytes of +/// `SHA-256(canonical target URL)`. +/// +/// A hash rather than the URL itself for two reasons: a URL contains `/`, +/// which is the token's field separator, and the full URL is longer than the +/// binding needs to be. 16 bytes is 128 bits — an attacker looking to reuse a +/// token elsewhere needs a *second meaningful collector URL* colliding with +/// the first, which this is far past sufficient for. +/// +/// The collector recomputes this from the URL(s) it expects to be reached at +/// and compares, so both sides must canonicalize identically. The rules, +/// exactly: +/// +/// - `{host}:{port}{path}`, e.g. `collector.example:4318/v1/metrics`. +/// - host lowercased (case-insensitive). +/// - port always explicit, defaulting to 80 for an `http` URL and 443 for an +/// `https` one, so `https://c.example/x` and `https://c.example:443/x` agree. +/// - path verbatim — no trailing-slash or dot-segment normalization. +/// - **userinfo stripped**, and query/fragment dropped (an OTLP export URL has +/// neither). Stripping userinfo is not cosmetic: hashing an endpoint of +/// `https://user:secret@collector/` would make the value unreproducible for +/// a collector that does not know the password, and it keeps credentials out +/// of the signed input entirely. +/// +/// The scheme is deliberately NOT part of the hashed string. It identifies a +/// transport, not a party, so binding it would not narrow "which collector may +/// use this token" at all — while forcing every collector reachable over both +/// http and https to configure the same URL twice. Note it still leaks in +/// indirectly through the default-port rule above, which is why +/// `http://c.example/x` and `https://c.example/x` do not collide. +/// +/// Cost of hashing: a rejected token tells the collector nothing about what +/// the sender aimed at. The node logs its resolved endpoint at startup, and +/// `docs/otel-metrics.md` documents this computation so a mismatch can be +/// worked out by hand. +fn audience_of(uri: &http::Uri) -> String { + use sha2::{Digest, Sha256}; + + let Some(host) = uri.host() else { + return String::new(); + }; + let port = uri.port_u16().unwrap_or(match uri.scheme_str() { + Some("https") => 443, + _ => 80, + }); + let canonical = format!("{}:{port}{}", host.to_ascii_lowercase(), uri.path()); + bs58::encode(&Sha256::digest(canonical.as_bytes())[..16]).into_string() +} + +/// The exporter's only HTTP transport, in every auth mode. +/// +/// Always installed, so `opentelemetry-otlp` never builds a client of its own +/// and none of its `reqwest-*` features have to be enabled — see the comment +/// on the dependency in `crates/core/Cargo.toml` for the dependency-graph +/// reason that matters. +/// +/// With `signer` set (`otel-auth-mode = "freenet"`) it adds a fresh +/// `Authorization: Bearer` token (see [`bearer_token`]) to each request; +/// with it unset it is a plain sender. +struct OtlpHttpClient { + inner: reqwest::blocking::Client, + /// `None` in `disabled` auth mode. + signer: Option, + pubkey_b58: String, +} + +// Manual impl: never print the signing key. +impl std::fmt::Debug for OtlpHttpClient { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("OtlpHttpClient").finish_non_exhaustive() + } +} + +/// Whether the last export attempt failed, so failures log on the first +/// occurrence and on each failing<->recovering transition rather than every +/// 60s forever. +static EXPORT_FAILING: AtomicBool = AtomicBool::new(false); + +/// Report a failed export at WARN, once per failing streak. +/// +/// The SDK will not do this for us, despite `opentelemetry-otlp` logging both +/// network errors and non-2xx responses at DEBUG on the stated grounds that +/// "PeriodicReader already logs the returned error via `otel_error!`". That is +/// true for the batch log/span processors and FALSE for metrics: the metrics +/// `PeriodicReader` logs its export result via `otel_debug!` +/// ("PeriodReaderInvokedExport"), and the only `otel_error!` in that file is +/// thread-creation failure. Without this, a collector that is down, rejecting +/// our token, or 413-ing our batches produces NO output at the default log +/// level while startup still says "OTel metrics exporter started". +fn report_export_failure(uri: &http::Uri, detail: &str) { + if !EXPORT_FAILING.swap(true, Ordering::Relaxed) { + tracing::warn!( + %uri, + detail, + "OTel metrics export is failing; metrics are not reaching the collector" + ); + } +} + +/// Report a successful export, logging only the failing -> recovered edge. +fn report_export_success() { + if EXPORT_FAILING.swap(false, Ordering::Relaxed) { + tracing::info!("OTel metrics export recovered"); + } +} + +/// First 256 bytes of an error body — OTLP puts rejection detail there, but a +/// collector may echo back headers (including our bearer token), so it is +/// truncated rather than logged whole. +fn truncated_body(body: &[u8]) -> String { + String::from_utf8_lossy(&body[..body.len().min(256)]).into_owned() +} + +#[async_trait::async_trait] +impl HttpClient for OtlpHttpClient { + async fn send_bytes(&self, mut request: Request) -> Result, HttpError> { + // Never clobber an operator-supplied header: the exporter applies + // OTEL_EXPORTER_OTLP_HEADERS before calling us, so a hosted collector + // configured with `Authorization: Basic ...` there must win. Our + // bearer token is one auth scheme among several, not the only one. + if let Some(signer) = self.signer.as_ref() { + if !request.headers().contains_key(http::header::AUTHORIZATION) { + let token = bearer_token(signer, &self.pubkey_b58, &audience_of(request.uri())); + request.headers_mut().insert( + http::header::AUTHORIZATION, + http::HeaderValue::from_str(&format!("Bearer {token}"))?, + ); + } + } + let uri = request.uri().clone(); + // Hand-rolled send, mirroring opentelemetry-http's blocking impl: + // that impl is on reqwest 0.13's client (opentelemetry-http's own + // dep), while the workspace is on 0.12, so we can't delegate to it. + // Blocking inside async is fine here for the same reason the SDK's + // default client is blocking: PeriodicReader exports via block_on on + // a dedicated thread. Fold both in when the workspace moves to 0.13. + // + // Every outcome is reported through `report_export_*`: the SDK logs + // them at DEBUG only, so this is the only place a failing export + // becomes visible to an operator. + let request: reqwest::blocking::Request = request + .map(|body| body.to_vec()) + .try_into() + .inspect_err(|error| { + // Reached when the endpoint has no scheme: reqwest requires an + // absolute URL, while `http::Uri` accepts `host:port` as an + // authority. `init` warns about that at startup too. + report_export_failure(&uri, &format!("invalid request URL: {error}")) + })?; + // Deliberately no `error_for_status()`: it discards the body, which is + // where OTLP puts rejection detail. Hand the whole response back and + // the SDK turns a non-2xx into an export error itself. + let mut response = self + .inner + .execute(request) + .inspect_err(|error| report_export_failure(&uri, &format!("{error}")))?; + let status = response.status(); + let headers = std::mem::take(response.headers_mut()); + let body = response + .bytes() + .inspect_err(|error| report_export_failure(&uri, &format!("{error}")))?; + if status.is_success() { + report_export_success(); + } else { + report_export_failure(&uri, &format!("HTTP {status}: {}", truncated_body(&body))); + } + let mut http_response = Response::builder().status(status).body(body)?; + *http_response.headers_mut() = headers; + Ok(http_response) + } +} + +/// Instrumentation scope name for every instrument this crate registers. +const METER_NAME: &str = "freenet"; + +/// Start the OpenTelemetry SDK metrics pipeline and install it as the +/// process-global meter provider. +/// +/// No-op when suppressed (see [`otel_suppression_reason`]) and best-effort +/// otherwise: an exporter that cannot be built logs a warning and the node +/// starts anyway. Metrics export must never be a startup dependency. +/// +/// `keypair` is the node's transport keypair: it yields the +/// `freenet.node.pubkey` / `freenet.node.fingerprint` resource attributes +/// (see [`build_provider`]) and, when `otel-auth-mode = "freenet"`, its +/// derived signing key authenticates every export request (see +/// [`bearer_token`]). +/// +/// Returns `Some(reason)` when the exporter was SUPPRESSED and `None` +/// otherwise — including when the pipeline failed to build, which is logged +/// but is not a suppression. Callers ignore the value; it exists so a test can +/// assert that `init` consults [`otel_suppression_reason`] and returns before +/// building anything — deleting the check would make an enabled config return +/// `None` under `cfg(test)`, i.e. ship a test network's metrics to a collector. +pub(crate) fn init( + config: &OtelConfig, + keypair: &crate::transport::TransportKeypair, +) -> Option { + if let Some(reason) = otel_suppression_reason( + config, + cfg!(test), + super::telemetry::running_under_cargo_test(), + ) { + // Being off by default is unremarkable; being switched ON and then + // suppressed anyway is something the operator has to be told about, + // or the exporter looks enabled and silently ships nothing. + if config.enabled && reason != OtelSuppression::Disabled { + tracing::warn!( + ?reason, + "otel-telemetry-enabled is set but the OTel metrics exporter was suppressed" + ); + } else { + tracing::debug!(?reason, "OTel metrics exporter not started"); + } + return Some(reason); + } + + let env_endpoint = env_endpoint(); + let endpoint = resolve_metrics_endpoint( + config.endpoint.as_deref(), + std::env::var(opentelemetry_otlp::OTEL_EXPORTER_OTLP_METRICS_ENDPOINT) + .ok() + .as_deref(), + std::env::var(opentelemetry_otlp::OTEL_EXPORTER_OTLP_ENDPOINT) + .ok() + .as_deref(), + ); + + // Log where this node's signed identity is actually going, including when + // an inherited OTEL_* variable overrode the configured endpoint — an + // operator who cannot see that from the logs cannot tell their collector + // was bypassed. + if let (Some(env_endpoint), Some(cfg_endpoint)) = + (env_endpoint.as_deref(), config.endpoint.as_deref()) + { + tracing::warn!( + %env_endpoint, + %cfg_endpoint, + "OTEL_EXPORTER_OTLP_* overrides the configured otel-endpoint" + ); + } + // Validate before building: an endpoint the SDK accepts but reqwest does + // not produces a per-export failure rather than a build error, and an + // unparseable env value is swallowed by the SDK, which then silently + // exports to localhost while the log line below names the operator's URL. + if let Some(effective) = endpoint.as_deref().or(env_endpoint.as_deref()) { + if let Some(problem) = endpoint_problem(effective) { + tracing::warn!(endpoint = effective, problem, "OTLP endpoint is unusable"); + } + } + + let (pubkey, fingerprint) = identity_attributes(keypair); + let auth_signer = match config.auth_mode { + crate::config::OtelAuthMode::Freenet => Some(keypair.auth_token_signer()), + crate::config::OtelAuthMode::Disabled => None, + }; + + match build_provider(endpoint.as_deref(), pubkey, fingerprint, auth_signer) { + Ok(provider) => { + // NOTE: no shutdown hook. `set_meter_provider` holds a + // reference for the process lifetime and PeriodicReader exports + // every 60s (OTEL_METRIC_EXPORT_INTERVAL), so at most one partial + // interval is lost at exit. If that tail ever matters, keep the + // provider in a OnceLock and call `shutdown()` from the graceful + // shutdown path in `bin/freenet.rs`. + global::set_meter_provider(provider); + register_metrics(); + tracing::info!( + endpoint = endpoint + .as_deref() + .or(env_endpoint.as_deref()) + .unwrap_or("http://localhost:4318 (SDK default)"), + auth_mode = ?config.auth_mode, + "OTel metrics exporter started" + ); + } + Err(error) => { + tracing::warn!( + %error, + "OTel metrics exporter failed to start; node continues without metrics" + ); + } + } + None +} + +/// The two resource attributes that identify THIS node, both derived from the +/// one transport keypair: `(freenet.node.pubkey, freenet.node.fingerprint)`. +/// +/// A function rather than two inline expressions in [`init`] so the guards in +/// this module's tests assert on what production actually attaches — building +/// the same strings in a test body would pass no matter what `init` does. +fn identity_attributes(keypair: &crate::transport::TransportKeypair) -> (String, String) { + // The full base58 x25519 transport public key. Byte-equal to the bearer + // token's `` field, so the collector self-validates the node id + // against the signing key after verifying the signature. Derived from the + // keypair even when auth is disabled, so the id is stable across auth-mode + // changes. NEVER a `PeerId`: its Display is `{pub_key}@{addr}`. + ( + bs58::encode(keypair.public_key_bytes()).into_string(), + keypair.public().to_string(), + ) +} + +/// Build the OTLP/HTTP exporter and meter provider. +/// +/// `endpoint` is `None` when the standard env vars should win — see +/// [`resolve_metrics_endpoint`] for why calling `with_endpoint` at all would +/// override them. +/// +/// Two identity resource attributes, both computed by [`init`] from the one +/// transport keypair: +/// +/// - `freenet.node.pubkey` — the base58 full x25519 transport public key, +/// byte-equal to the bearer token's `` field. The collector +/// verifies the token's XEdDSA signature against this key, so the identity +/// is self-validating and unforgeable. +/// - `freenet.node.fingerprint` — base58 of the FIRST 12 BYTES of the same +/// key (`TransportPublicKey::Display`, what UIs show). A pure public +/// function of `pubkey`, so the collector recomputes and checks it rather +/// than trusting it. +/// +/// Neither may ever be a `PeerId`: its `Display` is `{pub_key}@{addr}`, so +/// using it would put this node's socket address in every exported batch AND +/// make the identity churn on every address change. +pub(crate) fn build_provider( + endpoint: Option<&str>, + pubkey: String, + fingerprint: String, + auth_signer: Option, +) -> Result { + // The blocking reqwest clients below (ours and the exporter's default) + // each own a private tokio runtime. Creating one — or dropping one on the + // error path — inside an async context panics with "Cannot drop a runtime + // in a context where blocking is not allowed", and `init` runs inside the + // node's async build path. Hop to a plain thread so the whole build is + // async-context-free regardless of the caller. + std::thread::scope(|scope| { + scope + .spawn(move || build_provider_blocking(endpoint, pubkey, fingerprint, auth_signer)) + .join() + // A panic in the builder must surface as a build error, not + // propagate: `init` runs on node startup and metrics export must + // never be a startup dependency. + .unwrap_or_else(|panic| { + let msg = panic + .downcast_ref::<&str>() + .map(ToString::to_string) + .or_else(|| panic.downcast_ref::().cloned()) + .unwrap_or_else(|| "non-string panic payload".to_owned()); + Err(ExporterBuildError::InternalFailure(format!( + "otel provider build thread panicked: {msg}" + ))) + }) + }) +} + +/// Per-export HTTP timeout, resolved exactly like `opentelemetry-otlp` would +/// (`OTEL_EXPORTER_OTLP_METRICS_TIMEOUT` > `OTEL_EXPORTER_OTLP_TIMEOUT` > +/// 10s, in milliseconds). The SDK applies its own resolution only to a client +/// it builds itself, and we always supply one, so it has to happen here. +fn export_timeout() -> std::time::Duration { + for var in [ + opentelemetry_otlp::OTEL_EXPORTER_OTLP_METRICS_TIMEOUT, + opentelemetry_otlp::OTEL_EXPORTER_OTLP_TIMEOUT, + ] { + let Ok(raw) = std::env::var(var) else { + continue; + }; + let raw = raw.trim(); + if raw.is_empty() { + continue; + } + match raw.parse::() { + Ok(ms) => { + // The spec unit is MILLISECONDS, so `=10` meaning "10 seconds" + // yields a 10ms timeout and every export times out instead. + if ms < 100 { + tracing::warn!( + var, + ms, + "OTLP export timeout is in MILLISECONDS and this value is very small; \ + a seconds value was probably intended" + ); + } + return std::time::Duration::from_millis(ms); + } + // Fall through to the next variable, as the SDK's own `.ok()` + // resolution would — but say so rather than ignoring it silently. + Err(error) => tracing::warn!( + var, + raw, + %error, + "ignoring unparseable OTLP export timeout (expected whole milliseconds)" + ), + } + } + opentelemetry_otlp::OTEL_EXPORTER_OTLP_TIMEOUT_DEFAULT +} + +/// Resource attribute keys the operator declared through the environment — +/// see [`resource_attributes`] for which of them win. +fn env_declared_resource_keys() -> Vec { + let mut keys = Vec::new(); + if std::env::var("OTEL_SERVICE_NAME").is_ok_and(|v| !v.trim().is_empty()) { + keys.push("service.name".to_owned()); + } + if let Ok(attrs) = std::env::var("OTEL_RESOURCE_ATTRIBUTES") { + keys.extend(attrs.split(',').filter_map(|pair| { + let key = pair.split('=').next()?.trim(); + (!key.is_empty()).then(|| key.to_owned()) + })); + } + keys +} + +/// The resource attributes to attach, given the keys the operator declared +/// through `OTEL_SERVICE_NAME` / `OTEL_RESOURCE_ATTRIBUTES`. +/// +/// Descriptive attributes defer to the environment: `Resource::builder` seeds +/// from those variables and `with_attribute` merges OVER that seed, so setting +/// a literal unconditionally would silently discard the operator's value — two +/// nodes on one host with distinct `OTEL_SERVICE_NAME`s would both export +/// `service.name=freenet-node`. +/// +/// The two `freenet.node.*` identity attributes are the exception and are +/// ALWAYS emitted. They are not description, they are the collector's proof of +/// which node sent the batch: `freenet.node.pubkey` must equal the bearer +/// token's `` field, which is verified against the signature. Letting +/// `OTEL_RESOURCE_ATTRIBUTES=freenet.node.pubkey=...` shadow them would export +/// an identity that does not match the key everything was signed with, and +/// break that self-validation silently on both sides. +fn resource_attributes( + pubkey: String, + fingerprint: String, + declared: &[String], +) -> Vec<(&'static str, String)> { + let mut attributes = vec![ + ("freenet.node.pubkey", pubkey), + ("freenet.node.fingerprint", fingerprint), + ]; + for (key, _) in &attributes { + if declared.iter().any(|declared| declared == key) { + tracing::warn!( + key, + "the node identity resource attribute cannot be overridden by \ + OTEL_RESOURCE_ATTRIBUTES; the signed value is exported instead" + ); + } + } + attributes.extend( + [ + ("service.name", "freenet-node".to_owned()), + ("service.version", env!("CARGO_PKG_VERSION").to_owned()), + ("os.type", std::env::consts::OS.to_owned()), + ("host.arch", std::env::consts::ARCH.to_owned()), + ] + .into_iter() + .filter(|(key, _)| !declared.iter().any(|declared| declared == key)), + ); + attributes +} + +fn build_provider_blocking( + endpoint: Option<&str>, + pubkey: String, + fingerprint: String, + auth_signer: Option, +) -> Result { + let mut builder = MetricExporter::builder().with_http(); + if let Some(endpoint) = endpoint { + builder = builder.with_endpoint(endpoint); + } + // Always ours, in every auth mode — see `OtlpHttpClient`. Blocking on + // purpose: PeriodicReader exports off-runtime (see Cargo.toml). The + // timeout is the one the SDK would have applied to its own client, which + // it does not apply to a supplied one. + // + // A build failure is propagated rather than falling back to + // `Client::new()`: that fallback has NO timeout, so a collector that + // accepts the connection and never answers would block the PeriodicReader + // thread indefinitely and stop all metric collection. + let http_client = reqwest::blocking::Client::builder() + .timeout(export_timeout()) + .build() + .map_err(|error| { + ExporterBuildError::InternalFailure(format!("otel http client build failed: {error}")) + })?; + builder = builder.with_http_client(OtlpHttpClient { + inner: http_client, + signer: auth_signer, + pubkey_b58: pubkey.clone(), + }); + let exporter = builder.build()?; + + // Resource attributes ride once per export batch, not per datapoint, so + // identifying THIS node here costs nothing per series — unlike a + // per-datapoint attribute, which is why no instrument below carries one + // identifying the remote end of a connection. + // Which of these defer to the environment and which do not is + // [`resource_attributes`]'s decision. + let mut resource = Resource::builder(); + for (key, value) in resource_attributes(pubkey, fingerprint, &env_declared_resource_keys()) { + resource = resource.with_attribute(KeyValue::new(key, value)); + } + let resource = resource.build(); + + Ok(SdkMeterProvider::builder() + .with_periodic_exporter(exporter) + .with_resource(resource) + // Every histogram this crate records is base-2 exponential rather than + // explicit-bucket: the SDK's default boundaries are tuned for + // millisecond latency and are useless for byte-scale instruments, and + // exponential buckets self-adjust instead of needing a hand-picked + // boundary set per instrument. + .with_view(|instrument: &Instrument| { + (instrument.kind() == InstrumentKind::Histogram) + .then(|| { + Stream::builder() + .with_aggregation(Aggregation::Base2ExponentialHistogram { + max_size: 160, + max_scale: 20, + record_min_max: true, + }) + .build() + .ok() + }) + .flatten() + }) + .build()) +} + +/// Synchronous instruments, recorded from the code paths they measure. +/// +/// These need a handle held somewhere, unlike the observable instruments below +/// whose callbacks the pipeline owns. Kept behind a `OnceLock` set at the end +/// of [`init`] for two reasons: instruments built before +/// `global::set_meter_provider` would bind to the no-op provider forever, and +/// when the exporter is disabled the record helpers collapse to one relaxed +/// atomic load and a branch. +struct Instruments { + rtt: Histogram, + cwnd: Histogram, + transfers: Counter, + nat_traversal: Counter, + operations: Counter, +} + +static INSTRUMENTS: OnceLock = OnceLock::new(); + +/// Record a transport RTT sample. No-op until [`init`] installs the pipeline. +pub(crate) fn record_rtt_ms(rtt_ms: f64) { + if let Some(i) = INSTRUMENTS.get() { + i.rtt.record(rtt_ms, &[]); + } +} + +/// Record a congestion-window sample. +pub(crate) fn record_cwnd(cwnd_bytes: u64) { + if let Some(i) = INSTRUMENTS.get() { + i.cwnd.record(cwnd_bytes, &[]); + } +} + +/// Record a stream transfer outcome (`completed` / `failed`). +pub(crate) fn record_transfer(result: &'static str) { + if let Some(i) = INSTRUMENTS.get() { + i.transfers.add(1, &[KeyValue::new("result", result)]); + } +} + +/// Record a NAT traversal outcome (`attempt` / `established` / +/// `failed_error` / `failed_version`). +pub(crate) fn record_nat_traversal(result: &'static str) { + if let Some(i) = INSTRUMENTS.get() { + i.nat_traversal.add(1, &[KeyValue::new("result", result)]); + } +} + +/// Record an operation outcome. `op` is one of get/put/update/subscribe. +/// +/// NOTE: outcome only, no duration histogram — no driver measures its own +/// elapsed time today, and adding one means threading `TimeSource` through +/// every `op_ctx_task` (raw `Instant::now()` is banned in this crate). Add the +/// histogram when someone needs operation latency percentiles. +pub(crate) fn record_op_result(op: &'static str, success: bool) { + if let Some(i) = INSTRUMENTS.get() { + i.operations.add( + 1, + &[ + KeyValue::new("op", op), + KeyValue::new("result", if success { "success" } else { "failure" }), + ], + ); + } +} + +/// Register the instruments this crate owns. +/// +/// Must run AFTER `global::set_meter_provider`: `global::meter` binds to +/// whatever provider is installed at call time. +/// +/// Observable handles are dropped on purpose — the callback is registered into +/// the pipeline at `build()` and observed on every collection cycle regardless. +/// The SDK has no batch-callback API, so each one reads +/// [`network_status::otel_metrics_snapshot`] independently; that is why the +/// accessor is a cheap scalar read rather than the dashboard's `get_snapshot`. +fn register_metrics() { + let meter = global::meter(METER_NAME); + + let registered = INSTRUMENTS.set(Instruments { + rtt: meter + .f64_histogram("freenet.transport.rtt") + .with_unit("ms") + .with_description("Round-trip time observed on transport connections") + .build(), + cwnd: meter + .u64_histogram("freenet.transport.cwnd") + .with_unit("By") + .with_description("Congestion window samples") + .build(), + transfers: meter + .u64_counter("freenet.transport.transfers") + .with_description("Stream transfers by outcome") + .build(), + nat_traversal: meter + .u64_counter("freenet.transport.nat_traversal") + .with_description("Outbound NAT traversal attempts by outcome") + .build(), + operations: meter + .u64_counter("freenet.operation.results") + .with_description("Completed operations by type and outcome") + .build(), + }); + if registered.is_err() { + // A second `init` would leave the sync instruments bound to the first + // provider while the observable ones move to the new one — loud rather + // than silently half-migrated. + tracing::warn!("OTel instruments already registered; keeping the first set"); + } + + let _rss = meter + .u64_observable_gauge("freenet.process.memory.rss") + .with_unit("By") + .with_description("Resident set size of the freenet process") + .with_callback(|observer| { + if let Some(rss) = crate::node::resource_metrics::rss_bytes() { + observer.observe(rss, &[]); + } + }) + .build(); + + register_transport_metrics(&meter); + register_ring_metrics(&meter); + register_queue_metrics(&meter); +} + +/// Wire-level counters, read from the cumulative (never-reset) transport +/// totals. +/// +/// Deliberately NOT read from `TransportSnapshot`: those fields are period +/// accumulators that `take_snapshot` zeroes for the legacy telemetry worker, so +/// observing them as counters would report a non-monotonic series whenever +/// `telemetry-enabled` is also on. +fn register_transport_metrics(meter: &opentelemetry::metrics::Meter) { + use crate::transport::TRANSPORT_METRICS; + + let _bytes = meter + .u64_observable_counter("freenet.transport.bytes") + .with_unit("By") + .with_description( + "Wire bytes. Sent is metered at the socket (includes keep-alives, ACKs and \ + NAT probes); received is metered post-authentication, so the two directions \ + are deliberately not symmetric.", + ) + .with_callback(|observer| { + observer.observe( + TRANSPORT_METRICS.cumulative_bytes_sent(), + &[KeyValue::new("direction", "sent")], + ); + observer.observe( + TRANSPORT_METRICS.cumulative_bytes_received(), + &[KeyValue::new("direction", "received")], + ); + }) + .build(); + + let _packets = meter + .u64_observable_counter("freenet.transport.packets") + .with_description("UDP datagrams, metered at the same sites as freenet.transport.bytes") + .with_callback(|observer| { + let (sent, received) = TRANSPORT_METRICS.cumulative_packets(); + observer.observe(sent, &[KeyValue::new("direction", "sent")]); + observer.observe(received, &[KeyValue::new("direction", "received")]); + }) + .build(); +} + +/// Ring / topology state, mirroring the dashboard's connection-status tiles. +fn register_ring_metrics(meter: &opentelemetry::metrics::Meter) { + use crate::node::network_status::otel_metrics_snapshot as snapshot; + use crate::ring::HostingReason; + + let _connections = meter + .u64_observable_gauge("freenet.ring.connections") + .with_description("Active ring connections") + .with_callback(|observer| { + if let Some(s) = snapshot() { + observer.observe(s.ring.connection_count as u64, &[]); + } + }) + .build(); + + // Both hosted-contract gauges are attributed by `reason` and carry NO + // un-attributed total: emitting both on one instrument would make + // `sum by (reason)` double-count. `HostingReason` partitions the hosted + // set, so the total is `sum(freenet.node.contracts.hosted)`. + let _hosted = meter + .u64_observable_gauge("freenet.node.contracts.hosted") + .with_description( + "Contracts currently hosted by this node, partitioned by why each one is held", + ) + .with_callback(|observer| { + if let Some(s) = snapshot() { + for reason in HostingReason::ALL { + observer.observe( + s.hosting_reasons.count(reason), + &[KeyValue::new("reason", reason.as_str())], + ); + } + } + }) + .build(); + + let _hosted_bytes = meter + .u64_observable_gauge("freenet.node.contracts.hosted.bytes") + .with_unit("By") + .with_description( + "Contract state bytes hosted by this node, partitioned by why each contract is \ + held. State only — WASM code blobs and database overhead are excluded, matching \ + what the hosting cache's byte budget measures.", + ) + .with_callback(|observer| { + if let Some(s) = snapshot() { + for reason in HostingReason::ALL { + observer.observe( + s.hosting_reasons.bytes(reason), + &[KeyValue::new("reason", reason.as_str())], + ); + } + } + }) + .build(); + + let _attempts = meter + .u64_observable_counter("freenet.connect.attempts") + .with_description("Connection attempts made since startup") + .with_callback(|observer| { + if let Some(s) = snapshot() { + observer.observe(s.connection_attempts as u64, &[]); + } + }) + .build(); + + let _lattice = meter + .u64_observable_gauge("freenet.ring.lattice.neighbor") + .with_description( + "1 when this node holds its closest connected ring neighbor on that side. \ + Held does not mean tight — compare distances across nodes.", + ) + .with_callback(|observer| { + if let Some(s) = snapshot() { + observer.observe( + s.ring.lattice_has_successor as u64, + &[KeyValue::new("position", "successor")], + ); + observer.observe( + s.ring.lattice_has_predecessor as u64, + &[KeyValue::new("position", "predecessor")], + ); + } + }) + .build(); + + let _distance = meter + .f64_observable_gauge("freenet.ring.lattice.neighbor.distance") + .with_description("Ring distance to each held lattice edge; absent when unheld") + .with_callback(|observer| { + if let Some(s) = snapshot() { + if let Some(d) = s.ring.lattice_successor_distance { + observer.observe(d, &[KeyValue::new("position", "successor")]); + } + if let Some(d) = s.ring.lattice_predecessor_distance { + observer.observe(d, &[KeyValue::new("position", "predecessor")]); + } + } + }) + .build(); + + let _probes = meter + .u64_observable_counter("freenet.ring.lattice.probes") + .with_description( + "Route-to-self probes fired, and lattice improvements observed. Counted \ + independently — an improvement lands some ticks after the probe that caused \ + it, so the ratio is a convergence gauge, not a success rate.", + ) + .with_callback(|observer| { + if let Some(s) = snapshot() { + observer.observe( + s.ring.lattice_probes_issued, + &[KeyValue::new("result", "issued")], + ); + observer.observe( + s.ring.lattice_probe_improvements, + &[KeyValue::new("result", "improvement")], + ); + } + }) + .build(); + + let _updates = meter + .u64_observable_counter("freenet.contract.updates") + .with_description("Relayed UPDATEs by admission outcome") + .with_callback(|observer| { + if let Some(s) = snapshot() { + observer.observe( + s.ring.updates_accepted, + &[KeyValue::new("result", "accepted")], + ); + observer.observe( + s.ring.updates_rate_limited, + &[KeyValue::new("result", "rate_limited")], + ); + observer.observe( + s.ring.updates_capacity_dropped, + &[KeyValue::new("result", "capacity_dropped")], + ); + } + }) + .build(); +} + +/// Executor fair-queue occupancy and admission outcomes. +fn register_queue_metrics(meter: &opentelemetry::metrics::Meter) { + use crate::node::network_status::otel_metrics_snapshot as snapshot; + + let _depth = meter + .u64_observable_gauge("freenet.contract.queue.depth") + .with_description( + "Current fair-queue occupancy, per tier. No `total` series: it would \ + double-count under `sum by (queue)` — sum the tiers instead.", + ) + .with_callback(|observer| { + if let Some(s) = snapshot() { + let q = &s.fair_queue; + for (tier, depth) in [ + ("client_local", q.depth_client_local), + ("network_relay", q.depth_network_relay), + ("background", q.depth_background), + ] { + observer.observe(depth as u64, &[KeyValue::new("queue", tier)]); + } + } + }) + .build(); + + // A gauge, not a counter: `high_water` is a running maximum, and a + // collector that saw it as a counter would read a plateau as "no traffic". + // It exists because a burst between two 60s collections leaves no trace in + // the instantaneous depth. + let _high_water = meter + .u64_observable_gauge("freenet.contract.queue.depth.high_water") + .with_description("Highest fair-queue occupancy reached since startup") + .with_callback(|observer| { + if let Some(s) = snapshot() { + observer.observe(s.fair_queue.high_water as u64, &[]); + } + }) + .build(); + + let _rejected = meter + .u64_observable_counter("freenet.contract.queue.rejected") + .with_description( + "Fair-queue admission rejections. global_capacity is node-wide saturation; \ + per_contract is one noisy contract hitting its own cap.", + ) + .with_callback(|observer| { + if let Some(s) = snapshot() { + observer.observe( + s.fair_queue.rejected_global_capacity, + &[KeyValue::new("reason", "global_capacity")], + ); + observer.observe( + s.fair_queue.rejected_per_contract, + &[KeyValue::new("reason", "per_contract")], + ); + } + }) + .build(); + + let _shed = meter + .u64_observable_counter("freenet.contract.queue.background_shed") + .with_description("Background events shed to make room for higher-priority work") + .with_callback(|observer| { + if let Some(s) = snapshot() { + observer.observe(s.fair_queue.background_shed, &[]); + } + }) + .build(); +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::config::OtelConfig; + + /// The collector this module's tests mint tokens for, as the audience + /// hash of `http://collector.example:4318/v1/metrics`. + fn test_audience() -> String { + audience_of( + &"http://collector.example:4318/v1/metrics" + .parse::() + .unwrap(), + ) + } + + fn enabled_config() -> OtelConfig { + OtelConfig { + enabled: true, + endpoint: None, + auth_mode: Default::default(), + is_test_environment: false, + } + } + + /// One keypair plus its token, pre-split, for the verification tests. + fn token_fixture() -> (crate::transport::TransportKeypair, String) { + let keypair = crate::transport::TransportKeypair::new(); + let pubkey_b58 = bs58::encode(keypair.public_key_bytes()).into_string(); + let token = bearer_token(&keypair.auth_token_signer(), &pubkey_b58, &test_audience()); + (keypair, token) + } + + #[test] + fn bearer_token_has_the_documented_shape_and_verifies() { + use xeddsa::xeddsa::Verify; + + let (keypair, token) = token_fixture(); + + let parts: Vec<&str> = token.split('/').collect(); + let [scheme, pubkey, audience, timestamp, signature] = parts[..] else { + panic!("expected 5 /-separated parts, got {token}"); + }; + assert_eq!(scheme, "freenet"); + assert_eq!( + pubkey, + bs58::encode(keypair.public_key_bytes()).into_string(), + "pubkey part must be the full base58 x25519 transport public key" + ); + assert_eq!( + audience, + test_audience(), + "the token must name the collector it was minted for, or it can be \ + replayed to any other collector accepting this scheme" + ); + let ts: u64 = timestamp.parse().expect("timestamp is epoch seconds"); + assert!( + ts > 1_700_000_000, + "timestamp must be current epoch seconds" + ); + // The signature covers everything before its own slash, and verifies + // against the token's OWN pubkey — the transport key itself. + let signed_payload = format!("freenet/{pubkey}/{audience}/{timestamp}"); + let sig_bytes: [u8; 64] = bs58::decode(signature) + .into_vec() + .unwrap() + .try_into() + .expect("64-byte signature"); + xeddsa::xed25519::PublicKey(keypair.public_key_bytes()) + .verify(signed_payload.as_bytes(), &sig_bytes) + .expect("XEdDSA signature must verify against the transport pubkey"); + + // A forged payload with the same signature must fail. + assert!( + xeddsa::xed25519::PublicKey(keypair.public_key_bytes()) + .verify(b"freenet/forged", &sig_bytes) + .is_err() + ); + } + + #[test] + fn node_pubkey_is_verifiable_with_stock_ed25519() { + // The collector-side contract, spelled out: no xeddsa dependency + // needed there. Convert the Montgomery (x25519) pubkey to an Edwards + // point with sign bit 0, then run ordinary Ed25519 verification. + use ed25519_dalek::{Signature, Verifier, VerifyingKey}; + + let (keypair, token) = token_fixture(); + let (payload, sig_b58) = token.rsplit_once('/').unwrap(); + let sig_bytes: [u8; 64] = bs58::decode(sig_b58) + .into_vec() + .unwrap() + .try_into() + .unwrap(); + + let edwards = curve25519_dalek::montgomery::MontgomeryPoint(keypair.public_key_bytes()) + .to_edwards(0) + .expect("transport pubkey must map to Edwards") + .compress() + .to_bytes(); + VerifyingKey::from_bytes(&edwards) + .unwrap() + .verify(payload.as_bytes(), &Signature::from_bytes(&sig_bytes)) + .expect("stock ed25519 verify after Montgomery->Edwards conversion"); + } + + /// One-shot collector on a real socket: returns its address and a handle + /// yielding the raw request text it received. + fn oneshot_collector() -> (std::net::SocketAddr, std::thread::JoinHandle) { + use std::io::{Read, Write}; + + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + let server = std::thread::spawn(move || { + let (mut stream, _) = listener.accept().unwrap(); + let mut raw = Vec::new(); + let mut buf = [0u8; 1024]; + // Read until the body payload arrives; the client writes the + // whole request before waiting on the response. + while !raw.windows(14).any(|w| w == b"export-payload") { + let n = stream.read(&mut buf).unwrap(); + if n == 0 { + break; + } + raw.extend_from_slice(&buf[..n]); + } + stream + .write_all(b"HTTP/1.1 200 OK\r\ncontent-length: 0\r\n\r\n") + .unwrap(); + String::from_utf8_lossy(&raw).into_owned() + }); + (addr, server) + } + + /// The `Authorization` header value as the collector saw it, if any. + fn wire_auth_header(raw: &str) -> Option { + raw.lines() + .find(|l| l.to_ascii_lowercase().starts_with("authorization:")) + .map(|l| l.split_once(':').unwrap().1.trim().to_owned()) + } + + fn post_export(client: &OtlpHttpClient, addr: std::net::SocketAddr, auth: Option<&str>) { + let mut request = Request::builder() + .method("POST") + .uri(format!("http://{addr}/v1/metrics")); + if let Some(auth) = auth { + request = request.header(http::header::AUTHORIZATION, auth); + } + let request = request.body(Bytes::from_static(b"export-payload")).unwrap(); + // futures' executor, not tokio: send_bytes blocks internally. + let response = futures::executor::block_on(client.send_bytes(request)).unwrap(); + assert!(response.status().is_success()); + } + + fn test_client(keypair: &crate::transport::TransportKeypair, signed: bool) -> OtlpHttpClient { + OtlpHttpClient { + inner: reqwest::blocking::Client::new(), + signer: signed.then(|| keypair.auth_token_signer()), + pubkey_b58: bs58::encode(keypair.public_key_bytes()).into_string(), + } + } + + #[test] + fn send_bytes_puts_a_verifiable_bearer_header_on_the_wire() { + use ed25519_dalek::{Signature, Verifier, VerifyingKey}; + + // The full auth transport path: header injection plus the + // http::Request -> reqwest::blocking::Request conversion, observed + // from the collector's side of a real socket. Plain #[test], not + // tokio: the blocking client must stay out of async contexts. + let (addr, server) = oneshot_collector(); + let keypair = crate::transport::TransportKeypair::new(); + let pubkey_b58 = bs58::encode(keypair.public_key_bytes()).into_string(); + post_export(&test_client(&keypair, true), addr, None); + + let raw = server.join().unwrap(); + let token = wire_auth_header(&raw) + .expect("Authorization header must reach the wire") + .strip_prefix("Bearer ") + .expect("Bearer scheme") + .to_owned(); + let (payload, sig_b58) = token.rsplit_once('/').unwrap(); + // What the collector computes from the URL it answers at. + let audience = audience_of( + &format!("http://{addr}/v1/metrics") + .parse::() + .unwrap(), + ); + assert!( + payload.starts_with(&format!("freenet/{pubkey_b58}/{audience}/")), + "wire token must carry this node's pubkey and the audience hash of \ + the URL it was actually sent to: {token}" + ); + // Verify exactly like a collector would — see + // node_pubkey_is_verifiable_with_stock_ed25519. + let sig_bytes: [u8; 64] = bs58::decode(sig_b58) + .into_vec() + .unwrap() + .try_into() + .unwrap(); + let edwards = curve25519_dalek::montgomery::MontgomeryPoint(keypair.public_key_bytes()) + .to_edwards(0) + .unwrap() + .compress() + .to_bytes(); + VerifyingKey::from_bytes(&edwards) + .unwrap() + .verify(payload.as_bytes(), &Signature::from_bytes(&sig_bytes)) + .expect("wire bearer token must verify with stock ed25519"); + assert!( + raw.contains("export-payload"), + "body must survive the http -> reqwest conversion" + ); + } + + #[test] + fn an_operator_supplied_authorization_header_is_never_overwritten() { + // OTEL_EXPORTER_OTLP_HEADERS is applied by the exporter before it + // calls us. An operator pointing at a hosted collector that wants + // `Authorization: Basic ...` would otherwise get a 401 on every + // export with no way to turn our token off but `otel-auth-mode`. + let (addr, server) = oneshot_collector(); + let keypair = crate::transport::TransportKeypair::new(); + post_export( + &test_client(&keypair, true), + addr, + Some("Basic b3BlcmF0b3I="), + ); + assert_eq!( + wire_auth_header(&server.join().unwrap()).as_deref(), + Some("Basic b3BlcmF0b3I="), + "the operator's own credentials must reach the collector" + ); + } + + #[test] + fn disabled_auth_mode_sends_no_authorization_header() { + // The default mode: exporting to your own collector must not ship a + // signed assertion of this node's identity there. + let (addr, server) = oneshot_collector(); + let keypair = crate::transport::TransportKeypair::new(); + post_export(&test_client(&keypair, false), addr, None); + assert_eq!( + wire_auth_header(&server.join().unwrap()), + None, + "auth-mode disabled must put no Authorization header on the wire" + ); + } + + #[test] + fn fingerprint_attr_is_recomputable_from_the_pubkey_attr() { + // Requirement: a node cannot fake the UI-facing fingerprint. The + // collector derives it from the verified pubkey instead of trusting + // it: b58-decode pubkey, take the first 12 bytes, b58-encode. + // Asserts on `identity_attributes` — the function production uses — + // not on strings rebuilt here, which would pass whatever init does. + let keypair = crate::transport::TransportKeypair::new(); + let (pubkey_attr, fingerprint_attr) = identity_attributes(&keypair); + + let decoded = bs58::decode(&pubkey_attr).into_vec().unwrap(); + assert_eq!( + bs58::encode(&decoded[..12]).into_string(), + fingerprint_attr, + "fingerprint must be a pure function of pubkey, or the collector \ + cannot validate the UI-facing id" + ); + } + + #[test] + fn bearer_tokens_are_unique_per_request() { + // XEdDSA's random Z makes each signature distinct even over an + // identical payload (same pubkey, same second). + let keypair = crate::transport::TransportKeypair::new(); + let pubkey = bs58::encode(keypair.public_key_bytes()).into_string(); + let signer = keypair.auth_token_signer(); + assert_ne!( + bearer_token(&signer, &pubkey, &test_audience()), + bearer_token(&signer, &pubkey, &test_audience()) + ); + } + + #[test] + fn a_token_for_one_collector_does_not_verify_at_another() { + // The replay bound: a collector we export to must not be able to + // present our token at a different collector and impersonate us. + use xeddsa::xeddsa::Verify; + + let keypair = crate::transport::TransportKeypair::new(); + let (pubkey, _) = identity_attributes(&keypair); + let token = bearer_token(&keypair.auth_token_signer(), &pubkey, &test_audience()); + let (payload, sig_b58) = token.rsplit_once('/').unwrap(); + let sig_bytes: [u8; 64] = bs58::decode(sig_b58) + .into_vec() + .unwrap() + .try_into() + .unwrap(); + let other = audience_of( + &"http://other-collector:4318/v1/metrics" + .parse::() + .unwrap(), + ); + let replayed = payload.replace(&test_audience(), &other); + assert_ne!(replayed, payload, "audience must appear in the payload"); + assert!( + xeddsa::xed25519::PublicKey(keypair.public_key_bytes()) + .verify(replayed.as_bytes(), &sig_bytes) + .is_err(), + "a token re-aimed at another collector must fail verification" + ); + } + + #[test] + fn the_audience_hash_is_reproducible_from_the_documented_canonical_url() { + // The collector recomputes this from the URL it expects, so the + // canonicalization is a wire contract. Recompute it here the way the + // doc comment (and docs/otel-metrics.md) describe, independently of + // audience_of's own string building. + use sha2::{Digest, Sha256}; + let expect = |canonical: &str| { + bs58::encode(&Sha256::digest(canonical.as_bytes())[..16]).into_string() + }; + + for (uri, canonical) in [ + ( + "http://collector.example:4318/v1/metrics", + "collector.example:4318/v1/metrics", + ), + // Default port filled in from the scheme, host lowercased. + ( + "https://Collector.Example/v1/metrics", + "collector.example:443/v1/metrics", + ), + ( + "http://collector.example/v1/metrics", + "collector.example:80/v1/metrics", + ), + // Credentials stripped: a collector that does not know the + // password must still be able to reproduce the hash. + ( + "https://user:secret@collector.example:4318/v1/metrics", + "collector.example:4318/v1/metrics", + ), + ("http://[::1]:4318/v1/metrics", "[::1]:4318/v1/metrics"), + // Everything at once: credentials, port, multi-segment path. + ( + "http://user:pass@host:1234/path/here", + "host:1234/path/here", + ), + ] { + let audience = audience_of(&uri.parse::().unwrap()); + assert_eq!(audience, expect(canonical), "audience of {uri}"); + assert!( + !audience.contains('/'), + "audience must not contain the token's field separator" + ); + } + + // The binding has to be sensitive to the parts it claims to cover. + let of = |u: &str| audience_of(&u.parse::().unwrap()); + assert_ne!( + of("http://c.example:4318/v1/metrics"), + of("http://c.example:4319/v1/metrics"), + "port is bound" + ); + assert_ne!( + of("http://c.example:4318/tenant-a/v1/metrics"), + of("http://c.example:4318/tenant-b/v1/metrics"), + "path is bound — two collectors behind one host must differ" + ); + // ...and NOT to the scheme, which names a transport rather than a + // party: one collector reachable both ways must not need two + // audience entries. Deliberate, so pin it rather than leave it to + // be "fixed" later. + assert_eq!( + of("https://c.example:4318/v1/metrics"), + of("http://c.example:4318/v1/metrics"), + "scheme is deliberately not bound" + ); + // It does still reach the hash through the default-port rule, so + // these two do not collide. + assert_ne!( + of("https://c.example/v1/metrics"), + of("http://c.example/v1/metrics"), + "an omitted port defaults per scheme: 443 vs 80" + ); + } + + #[test] + fn init_refuses_to_start_from_a_test_process() { + // The suppression check lives in `init`, and `init` is unreachable + // under cfg(test) — so without this, deleting the whole block leaves + // every test green and a `--id` test network ships to a collector. + // Under cfg(test) an ENABLED config must still come back suppressed; + // if the check were gone, init would build a pipeline and return None. + assert_eq!( + init( + &enabled_config(), + &crate::transport::TransportKeypair::new() + ), + Some(OtelSuppression::TestHarness), + "init must consult otel_suppression_reason and return before \ + building anything" + ); + assert!( + INSTRUMENTS.get().is_none(), + "a suppressed init must not have registered instruments" + ); + } + + /// Cross-file pin (a same-file scrape can be satisfied by its own literal + /// — see .claude/rules/bug-prevention-patterns.md): every hot-path mirror + /// that feeds a synchronous instrument must still exist. Delete one and + /// its counter reports zero forever with nothing else failing. + /// + /// Ceiling: presence in the file, not in the right function. A mirror + /// moved to the wrong call site still passes; a deleted one does not. + #[test] + fn every_sync_instrument_still_has_its_hot_path_mirror() { + for (source, call) in [ + ( + include_str!("../transport/metrics.rs"), + "crate::tracing::otel::record_rtt_ms(", + ), + ( + include_str!("../transport/metrics.rs"), + "crate::tracing::otel::record_cwnd(", + ), + ( + include_str!("../transport/metrics.rs"), + "crate::tracing::otel::record_transfer(\"completed\")", + ), + ( + include_str!("../transport/metrics.rs"), + "crate::tracing::otel::record_transfer(\"failed\")", + ), + ( + include_str!("../transport/metrics.rs"), + "crate::tracing::otel::record_nat_traversal(\"attempt\")", + ), + ( + include_str!("../transport/metrics.rs"), + "crate::tracing::otel::record_nat_traversal(\"established\")", + ), + ( + include_str!("../transport/metrics.rs"), + "crate::tracing::otel::record_nat_traversal(\"failed_error\")", + ), + ( + include_str!("../transport/metrics.rs"), + "crate::tracing::otel::record_nat_traversal(\"failed_version\")", + ), + ( + include_str!("../node/network_status.rs"), + "crate::tracing::otel::record_op_result(", + ), + ] { + assert!( + source.contains(call), + "missing hot-path mirror `{call}`: the instrument it feeds \ + would report zero forever" + ); + } + } + + #[test] + fn production_shaped_input_is_not_suppressed() { + assert_eq!( + otel_suppression_reason(&enabled_config(), false, false), + None, + "a real release binary with the flag on must export" + ); + } + + #[test] + fn every_test_signal_suppresses() { + let disabled = OtelConfig { + enabled: false, + ..enabled_config() + }; + assert_eq!( + otel_suppression_reason(&disabled, false, false), + Some(OtelSuppression::Disabled) + ); + + let test_env = OtelConfig { + is_test_environment: true, + ..enabled_config() + }; + assert_eq!( + otel_suppression_reason(&test_env, false, false), + Some(OtelSuppression::TestEnvironmentFlag) + ); + + assert_eq!( + otel_suppression_reason(&enabled_config(), true, false), + Some(OtelSuppression::TestHarness), + "cfg(test) build" + ); + assert_eq!( + otel_suppression_reason(&enabled_config(), false, true), + Some(OtelSuppression::TestHarness), + "running from a cargo deps/ harness" + ); + } + + #[test] + fn metrics_env_wins_over_generic_env_and_config() { + // Both env forms mean "let the SDK resolve it", because + // opentelemetry-otlp gives a programmatic endpoint priority over the + // env vars — passing one would invert the required precedence. + assert_eq!( + resolve_metrics_endpoint( + Some("http://from-config:4318"), + Some("http://from-metrics-env:4318/v1/metrics"), + Some("http://from-generic-env:4318"), + ), + None + ); + assert_eq!( + resolve_metrics_endpoint( + Some("http://from-config:4318"), + None, + Some("http://from-generic-env:4318"), + ), + None + ); + } + + #[test] + fn config_endpoint_gets_the_signal_path_appended() { + // The SDK appends /v1/metrics only on the env-var path; a programmatic + // endpoint is used verbatim, so we append it ourselves. + assert_eq!( + resolve_metrics_endpoint(Some("http://collector:4318"), None, None), + Some("http://collector:4318/v1/metrics".to_string()) + ); + assert_eq!( + resolve_metrics_endpoint(Some("http://collector:4318/"), None, None), + Some("http://collector:4318/v1/metrics".to_string()), + "trailing slash must not double up" + ); + } + + #[test] + fn nothing_configured_defers_to_the_sdk_default() { + assert_eq!(resolve_metrics_endpoint(None, None, None), None); + assert_eq!( + resolve_metrics_endpoint(Some(" "), None, None), + None, + "a blank endpoint is not a configuration" + ); + } + + #[test] + fn node_pubkey_attr_matches_the_bearer_token_pubkey() { + // The collector's self-validation contract: after verifying the token + // signature, `` must equal `freenet.node.pubkey` exactly. + let (keypair, token) = token_fixture(); + let pubkey_attr = bs58::encode(keypair.public_key_bytes()).into_string(); + assert_eq!( + token.split('/').nth(1), + Some(pubkey_attr.as_str()), + "token must equal freenet.node.pubkey, or the collector \ + cannot self-validate the node id against the signing key" + ); + } + + #[test] + fn instance_id_carries_no_network_address() { + // `PeerId` renders as `{pub_key}@{addr}`, so using it — as the + // exporter originally did — leaks our socket address into every + // batch and re-identifies the node whenever the address changes. + // Both identity attributes must stay address-free. Asserts on + // `identity_attributes` — the function production uses — because + // rebuilding the strings here would pass whatever `init` actually + // attaches, including a `PeerId`. + let keypair = crate::transport::TransportKeypair::new(); + let (pubkey_attr, fingerprint_attr) = identity_attributes(&keypair); + for instance_id in [pubkey_attr, fingerprint_attr] { + assert!(!instance_id.is_empty()); + assert!( + !instance_id.contains('@') && !instance_id.contains(':'), + "identity attribute must not embed an address, got {instance_id}" + ); + } + + let peer_id = crate::node::PeerId::new( + keypair.public().clone(), + "203.0.113.7:31337".parse().expect("valid addr"), + ); + assert!( + peer_id.to_string().contains("203.0.113.7"), + "guard is meaningless if PeerId stops embedding the address" + ); + } + + #[test] + fn record_helpers_are_inert_without_a_pipeline() { + // Every record helper is called from production paths that run whether + // or not the exporter is enabled, so an unset OnceLock must be a no-op + // rather than a panic or an implicit no-op-provider binding. + record_rtt_ms(12.5); + record_cwnd(4096); + record_transfer("completed"); + record_nat_traversal("attempt"); + record_op_result("get", true); + assert!( + INSTRUMENTS.get().is_none(), + "recording must not lazily bind instruments to the no-op provider" + ); + } + + #[test] + fn env_declared_attributes_cannot_shadow_the_node_identity() { + // The collector verifies the bearer token's signature and then trusts + // `freenet.node.pubkey` as the sender's identity. If + // OTEL_RESOURCE_ATTRIBUTES could override that attribute, a node would + // export an identity that does not match the key it signed with, and + // neither side would notice. + let declared = [ + "freenet.node.pubkey".to_owned(), + "freenet.node.fingerprint".to_owned(), + "service.name".to_owned(), + ]; + let attributes = resource_attributes("REAL-PK".into(), "REAL-FP".into(), &declared); + let value = |key: &str| { + attributes + .iter() + .find(|(k, _)| *k == key) + .map(|(_, v)| v.as_str()) + }; + + assert_eq!( + value("freenet.node.pubkey"), + Some("REAL-PK"), + "the signed pubkey must be exported even when the environment declares it" + ); + assert_eq!(value("freenet.node.fingerprint"), Some("REAL-FP")); + // Descriptive attributes still defer: the operator's own service.name + // has to survive, or two nodes on one host collapse into one series. + assert_eq!( + value("service.name"), + None, + "a declared service.name must be left to the environment" + ); + assert!( + value("service.version").is_some(), + "undeclared descriptive attributes are still filled in" + ); + } + + #[test] + fn unusable_endpoints_are_diagnosed() { + // `http::Uri` accepts `host:port` as an authority with no scheme, so + // the exporter builds and every export then dies converting to a + // reqwest request — the failure is per-export, not at startup. + assert!( + endpoint_problem("collector.example:4318").is_some(), + "a schemeless authority must be reported" + ); + // Not parseable at all: the SDK swallows this and falls back to + // localhost while the startup log names the operator's URL. + assert!(endpoint_problem("collector.example:4318/v1/metrics").is_some()); + assert!(endpoint_problem("not a url").is_some()); + // A non-HTTP scheme parses fine but reqwest cannot send it. + assert!(endpoint_problem("ftp://collector.example:4318/v1/metrics").is_some()); + + assert_eq!( + endpoint_problem("http://collector.example:4318/v1/metrics"), + None + ); + assert_eq!( + endpoint_problem("https://collector.example/v1/metrics"), + None + ); + } + + #[tokio::test] + async fn provider_builds_with_auth_disabled() { + // The default auth mode, and the only path where no signer is + // installed. It still gets our HttpClient — the exporter has no + // reqwest feature enabled and would fail with NoHttpClient otherwise. + let provider = build_provider( + Some("http://127.0.0.1:1/v1/metrics"), + "pubkey-under-test".to_string(), + "fingerprint-under-test".to_string(), + None, + ) + .expect("exporter build must succeed with auth disabled"); + tokio::task::spawn_blocking(move || provider.shutdown().expect("clean shutdown")) + .await + .expect("shutdown thread panicked"); + } + + #[tokio::test] + async fn provider_builds_inside_a_tokio_runtime() { + // Two things under test. First, exporter construction must not panic + // when invoked from an async context — the blocking reqwest client + // owns a private tokio runtime, so `build_provider` hops to a plain + // thread internally; this asserts that hop works (on Linux, building + // inline panics with "Cannot drop a runtime in a context where + // blocking is not allowed"). Second, an unreachable collector must not + // surface as a build error: export failures are asynchronous and must + // never fail node startup. Port 1 is chosen because nothing can be + // listening there. + let provider = build_provider( + Some("http://127.0.0.1:1/v1/metrics"), + "pubkey-under-test".to_string(), + "fingerprint-under-test".to_string(), + // Auth on: the signing client path must not panic in async + // context either. + Some(crate::transport::TransportKeypair::new().auth_token_signer()), + ) + .expect("exporter build must succeed against an unreachable collector"); + // Shutdown drops the exporter's blocking client and with it that + // private runtime — same hazard as construction, so it must happen + // where blocking is allowed, not on this async test thread. + tokio::task::spawn_blocking(move || provider.shutdown().expect("clean shutdown")) + .await + .expect("shutdown thread panicked"); + } +} diff --git a/crates/core/src/tracing/telemetry.rs b/crates/core/src/tracing/telemetry.rs index c286f32e28..7850c4b546 100644 --- a/crates/core/src/tracing/telemetry.rs +++ b/crates/core/src/tracing/telemetry.rs @@ -612,7 +612,7 @@ fn try_enqueue_event( /// /// Best-effort: if `current_exe()` fails we return `false` and fall back to the /// compile-time guards, which already cover unit tests and CI. -fn running_under_cargo_test() -> bool { +pub(crate) fn running_under_cargo_test() -> bool { std::env::current_exe() .ok() .and_then(|exe| { diff --git a/crates/core/src/transport/crypto.rs b/crates/core/src/transport/crypto.rs index 054b91fe5a..c93b45fbf9 100644 --- a/crates/core/src/transport/crypto.rs +++ b/crates/core/src/transport/crypto.rs @@ -99,6 +99,26 @@ impl TransportKeypair { } } +impl TransportKeypair { + /// XEdDSA signer for authenticating this node out-of-band (the OTel + /// collector bearer token — `tracing::otel::bearer_token`). + /// + /// XEdDSA (Signal's construction) signs with the x25519 transport secret + /// itself, so signatures verify against the SAME public key peers and + /// UIs already know: the verifier converts the Montgomery public key to + /// Edwards (sign bit 0) and runs stock Ed25519 verification. No second + /// identity, nothing to cross-certify. + pub(crate) fn auth_token_signer(&self) -> xeddsa::xed25519::PrivateKey { + xeddsa::xed25519::PrivateKey::from(self.secret.0.as_bytes()) + } + + /// The full 32-byte public key, e.g. for the base58 identity the OTel + /// exporter publishes (`Display` is a truncated fingerprint). + pub(crate) fn public_key_bytes(&self) -> [u8; 32] { + *self.public.0.as_bytes() + } +} + impl Default for TransportKeypair { fn default() -> Self { Self::new() diff --git a/crates/core/src/transport/metrics.rs b/crates/core/src/transport/metrics.rs index fd61c77ca2..dd374dfa68 100644 --- a/crates/core/src/transport/metrics.rs +++ b/crates/core/src/transport/metrics.rs @@ -73,6 +73,15 @@ pub struct TransportMetrics { cumulative_bytes_sent: AtomicU64, cumulative_bytes_received: AtomicU64, + // Cumulative wire-packet counters (never reset). Metered at exactly the + // same two call sites as the cumulative byte counters above, so the same + // send-at-socket / receive-post-auth asymmetry applies. Kept here rather + // than incremented as an OTel counter at the call site because these fire + // once per UDP datagram — an observable counter reading these costs + // nothing on the hot path. + cumulative_packets_sent: AtomicU64, + cumulative_packets_received: AtomicU64, + // Timing accumulators (for computing averages) total_transfer_time_ms: AtomicU64, @@ -163,6 +172,8 @@ impl TransportMetrics { bytes_received: AtomicU64::new(0), cumulative_bytes_sent: AtomicU64::new(0), cumulative_bytes_received: AtomicU64::new(0), + cumulative_packets_sent: AtomicU64::new(0), + cumulative_packets_received: AtomicU64::new(0), total_transfer_time_ms: AtomicU64::new(0), peak_throughput_bps: AtomicU64::new(0), peak_cwnd_bytes: AtomicU32::new(0), @@ -185,6 +196,7 @@ impl TransportMetrics { /// Record a completed outbound transfer. pub fn record_transfer_completed(&self, stats: &super::TransferStats) { + crate::tracing::otel::record_transfer("completed"); // Use saturating arithmetic to prevent overflow (though extremely unlikely // in practice - would require billions of transfers or exabytes of data) self.transfers_completed @@ -269,6 +281,7 @@ impl TransportMetrics { /// and the bytes it did put on the wire are already counted at the socket /// layer by `record_packet_sent`. pub fn record_transfer_failed(&self) { + crate::tracing::otel::record_transfer("failed"); // Saturating, matching `record_transfer_completed` — a pinned u32::MAX // is a better failure mode than wrapping to 0 and reporting a healthy // node. @@ -281,11 +294,13 @@ impl TransportMetrics { /// Record the start of an outbound NAT traversal attempt. pub fn record_nat_traversal_attempt(&self) { + crate::tracing::otel::record_nat_traversal("attempt"); self.nat_traversal_attempts.fetch_add(1, Ordering::Relaxed); } /// Record a NAT traversal attempt that established a connection. pub fn record_nat_traversal_established(&self) { + crate::tracing::otel::record_nat_traversal("established"); self.nat_traversal_established .fetch_add(1, Ordering::Relaxed); } @@ -293,6 +308,7 @@ impl TransportMetrics { /// Record a NAT traversal attempt that failed for a non-version reason /// (unreachable / symmetric-NAT / generic transport error). pub fn record_nat_traversal_failed_error(&self) { + crate::tracing::otel::record_nat_traversal("failed_error"); self.nat_traversal_failed_error .fetch_add(1, Ordering::Relaxed); } @@ -300,12 +316,14 @@ impl TransportMetrics { /// Record a NAT traversal attempt that failed due to a protocol version /// mismatch. pub fn record_nat_traversal_failed_version(&self) { + crate::tracing::otel::record_nat_traversal("failed_version"); self.nat_traversal_failed_version .fetch_add(1, Ordering::Relaxed); } /// Record a cwnd sample (called periodically or on transfer completion). pub(crate) fn record_cwnd_sample(&self, cwnd_bytes: u32) { + crate::tracing::otel::record_cwnd(cwnd_bytes as u64); self.cwnd_sum .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |v| { Some(v.saturating_add(cwnd_bytes as u64)) @@ -337,6 +355,7 @@ impl TransportMetrics { /// keep-alive path is what keeps RTT statistics populated for quiet, /// long-lived connections that rarely complete a stream transfer (#4000). pub(crate) fn record_rtt_sample(&self, rtt_us: u64) { + crate::tracing::otel::record_rtt_ms(rtt_us as f64 / 1000.0); self.rtt_sum_us .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |v| { Some(v.saturating_add(rtt_us)) @@ -412,6 +431,7 @@ impl TransportMetrics { pub fn record_packet_sent(&self, remote_addr: SocketAddr, bytes: u64) { self.cumulative_bytes_sent .fetch_add(bytes, Ordering::Relaxed); + self.cumulative_packets_sent.fetch_add(1, Ordering::Relaxed); self.record_per_peer(remote_addr, bytes, |s| &s.bytes_sent); } @@ -428,6 +448,8 @@ impl TransportMetrics { pub fn record_packet_received(&self, remote_addr: SocketAddr, bytes: u64) { self.cumulative_bytes_received .fetch_add(bytes, Ordering::Relaxed); + self.cumulative_packets_received + .fetch_add(1, Ordering::Relaxed); self.record_per_peer(remote_addr, bytes, |s| &s.bytes_received); } @@ -436,6 +458,14 @@ impl TransportMetrics { self.cumulative_bytes_received.load(Ordering::Relaxed) } + /// Read cumulative packets sent/received without resetting counters. + pub(crate) fn cumulative_packets(&self) -> (u64, u64) { + ( + self.cumulative_packets_sent.load(Ordering::Relaxed), + self.cumulative_packets_received.load(Ordering::Relaxed), + ) + } + /// Record per-peer bytes for the given direction. /// /// Bounded to [`MAX_TRACKED_PEERS`] entries with LRU eviction: when the diff --git a/docs/design/otel-metrics-exporter.md b/docs/design/otel-metrics-exporter.md new file mode 100644 index 0000000000..e8e511be45 --- /dev/null +++ b/docs/design/otel-metrics-exporter.md @@ -0,0 +1,409 @@ +# Design: OpenTelemetry metrics exporter (isolated from existing telemetry) + +Status: implemented (`crates/core/src/tracing/otel.rs`). +Operator-facing configuration: [`docs/otel-metrics.md`](../otel-metrics.md). + +## Problem + +`TelemetryReporter` (`crates/core/src/tracing/telemetry.rs`) hand-builds OTLP-JSON +log records and POSTs them to `{telemetry-endpoint}/v1/logs` with `reqwest` +(`telemetry.rs:1415-1500`). It is the feed for the project's central dashboard and +it works. It is not an OpenTelemetry SDK pipeline: no meter provider, no +instruments, no standard `OTEL_*` env-var handling, no metrics or traces — so +there is nowhere to hang actual metrics. + +## Goal + +Add a real OpenTelemetry SDK metrics pipeline that can be pointed at any OTLP +collector and configured the standard way. The existing telemetry path is +untouched. + +## Non-goals + +- Any change to `TelemetryReporter`, `to_otlp_logs`, the `/v1/logs` wire format, + `telemetry-enabled`, or `telemetry-endpoint`. +- Migrating existing events onto the new pipeline. +- Logs and traces exporters. The endpoint resolution is designed to serve all + three signals; only metrics ships now. +- Per-connection metrics. Identifying the remote end of a connection is a + per-datapoint attribute, multiplied by bucket count on histograms; the + aggregate signals below answer the operational questions without it. Transfer + and connection *events*, if wanted, belong on a log pipeline rather than as + metric series. +- Operation and contract-execution latency histograms. No driver measures its + own elapsed time today, and raw `Instant::now()` is banned in `crates/core/` + (`.claude/rules/testing.md`), so adding them means threading `TimeSource` + through every `op_ctx_task`. `freenet.operation.results` ships the outcome + counter now; the duration histogram is deferred until someone needs + percentiles. + +## Instruments + +Registered in `tracing/otel.rs::register_metrics`. + +| Instrument | Kind | Attributes | Source | +|---|---|---|---| +| `freenet.process.memory.rss` | gauge | — | `node::resource_metrics::rss_bytes` (Linux only) | +| `freenet.transport.bytes` | counter | `direction` | `cumulative_bytes_{sent,received}` | +| `freenet.transport.packets` | counter | `direction` | `cumulative_packets_{sent,received}` | +| `freenet.transport.transfers` | counter | `result` | `record_transfer_{completed,failed}` | +| `freenet.transport.nat_traversal` | counter | `result` | `record_nat_traversal_*` | +| `freenet.transport.rtt` | histogram | — | `record_rtt_sample` | +| `freenet.transport.cwnd` | histogram | — | `record_cwnd_sample` | +| `freenet.operation.results` | counter | `op`, `result` | `network_status::record_op_result` | +| `freenet.ring.connections` | gauge | — | `RingStatsSnapshot` | +| `freenet.node.contracts.hosted` | gauge | `reason` | `HostingReasonStats` | +| `freenet.node.contracts.hosted.bytes` | gauge | `reason` | `HostingReasonStats` | +| `freenet.connect.attempts` | counter | — | `NetworkStatus` | +| `freenet.ring.lattice.neighbor` | gauge | `position` | `RingStatsSnapshot` | +| `freenet.ring.lattice.neighbor.distance` | gauge | `position` | `RingStatsSnapshot` | +| `freenet.ring.lattice.probes` | counter | `result` | `RingStatsSnapshot` | +| `freenet.contract.updates` | counter | `result` | `RingStatsSnapshot` | +| `freenet.contract.queue.depth` | gauge | `queue` | `FairQueueStats` | +| `freenet.contract.queue.depth.high_water` | gauge | — | `FairQueueStats` | +| `freenet.contract.queue.rejected` | counter | `reason` | `FairQueueStats` | +| `freenet.contract.queue.background_shed` | counter | — | `FairQueueStats` | + +Everything but the two histograms and the three synchronous counters is an +observable callback over state that already existed for the local dashboard. + +### `reason` on the hosted-contract gauges + +`freenet.node.contracts.hosted` and `.hosted.bytes` answer "how much are we +holding, and *why*". The `reason` values come from `ring::HostingReason` and +are a **partition**: the classifier assigns each hosted contract to the first +matching bucket in priority order, so `sum by (reason)` is the hosted-contract +count and the byte gauge sums to the hosting cache's used bytes. Neither gauge +emits an un-attributed total — that would double-count under `sum`. + +| `reason` | held because | +|---|---| +| `local_client` | a local client (WebSocket/HTTP) holds a subscription | +| `downstream` | a downstream peer subscribes to us — we relay its updates | +| `subscribed` | unexpired network subscription, no local or downstream reader | +| `local_access` | no subscription, but a local client GET/PUT touched it | +| `abandoned` | was in use and no longer is — the eviction-candidate pool | +| `routed` | residual: arrived via a routed GET/PUT, no demand signal | + +The strings are a metrics contract — collector-side dashboards filter on them, +so add variants rather than repurpose existing values. + +The breakdown has its own provider (`set_hosting_reason_provider`) rather than +riding `RingStatsSnapshot`: that provider runs on every dashboard HTTP request, +and this is an O(hosted) walk under the hosting-cache read lock. Its own +provider confines the cost to the OTel collection cadence. The bytes gauge +counts contract **state** only — no WASM blobs, no database overhead — matching +what the hosting cache's byte budget measures, so it is comparable against the +budget but not against on-disk usage. + +Resource attributes: `freenet.node.pubkey` (the full base58 **x25519** transport +public key — byte-equal to the bearer token's `` field, so a collector +that verified the signature has also verified the node id), +`freenet.node.fingerprint` (the truncated form UIs show, recomputable from +`freenet.node.pubkey`, so the collector derives rather than trusts it), +`service.name`, `service.version`, `os.type`, `host.arch`. Never a `PeerId`, +which renders as `{pub_key}@{addr}` and would export our socket address. + +The literals are applied only for keys the operator did **not** declare through +`OTEL_SERVICE_NAME` / `OTEL_RESOURCE_ATTRIBUTES`: `ResourceBuilder` seeds from +the environment and then merges `with_attribute` over that seed, so setting one +unconditionally would silently discard the operator's value. This deference +applies to the descriptive attributes only — the two `freenet.node.*` identity +attributes are always emitted, because they are what the collector checks the +bearer-token signature against, and an operator-supplied override would export +an identity that does not match the signing key. + +Not instrumented: `TransportMetrics::slowdowns_triggered` is a period +accumulator that `take_snapshot` zeroes for the legacy telemetry worker, so +observing it as a counter would report a non-monotonic series whenever +`telemetry-enabled` is also on — the same hazard as the transport byte and +packet counters, which is why those are read from the cumulative totals +instead. + +## Isolation requirement (hard) + +`otel-telemetry-enabled` and `telemetry-enabled` are strictly independent +features. The two pipelines are not expected to share a backend. Concretely: + +- Separate config structs. `OtelConfig` is a sibling of `TelemetryConfig`, never a + field on it. +- `otel::init` takes `&OtelConfig` only and must never read `TelemetryConfig`. +- No endpoint fallback between them. `otel-endpoint` never defaults to + `DEFAULT_TELEMETRY_ENDPOINT` (nova). +- Enabling or disabling one has no effect on the other. +- The only shared code is the test-harness detection helper, a free function. + +## Configuration + +New `OtelArgs` (clap + serde) and `OtelConfig` (resolved), beside +`TelemetryArgs`/`TelemetryConfig` in `crates/core/src/config.rs`: + +| Key | Env | Default | Meaning | +|---|---|---|---| +| `otel-telemetry-enabled` | `FREENET_OTEL_TELEMETRY_ENABLED` | `false` | Enable the SDK metrics exporter | +| `otel-endpoint` | (see below) | none | OTLP/HTTP collector base URL | +| `otel-auth-mode` | — | `disabled` | `freenet` (bearer token) or `disabled` (no header) | + +Default is `false`: nothing is exported yet, so off is the no-behavior-change +default. Operators opt in. + +`otel-telemetry-enabled` is `Option` in `OtelArgs` and has no clap +`default_value`, unlike its `telemetry-*` siblings. With a default, "unset" and +"explicitly false" are indistinguishable after parsing, so +`--otel-telemetry-enabled=false` could not override a `config.toml` that says +`true` — the off switch would not work. + +### Collector authentication + +`otel-auth-mode = "freenet"` puts a per-request + +``` +Authorization: Bearer freenet//// +``` + +on every export. `` is an XEdDSA (Signal construction, `xeddsa` +crate) signature over everything preceding it, made with the x25519 transport +secret itself (`TransportKeypair::auth_token_signer`) — the node's one +identity, no second key to cross-certify. `` and `` are +base58, `` is epoch seconds. + +`` binds the token to the collector it was minted for: without it, +any collector we export to could replay our token at any other collector +accepting this scheme and impersonate the node. It is +**base58 of the first 16 bytes of `SHA-256(canonical target URL)`**, hashed +rather than sent literally because a URL contains `/`, the token's field +separator. The canonical form both sides must agree on: + +- `{host}:{port}{path}`, e.g. `collector.example:4318/v1/metrics` +- host lowercased +- port always explicit, defaulting to 80 for an `http` URL and 443 for an `https` one +- path verbatim, no normalization +- userinfo stripped, query and fragment dropped +- **scheme not included**: it names a transport, not a party, so binding it + would not narrow which collector may use the token, while forcing every + collector reachable over both http and https to be configured twice. It does + still reach the hash through the default-port rule, so `http://c/x` and + `https://c/x` differ. + +Stripping userinfo is load-bearing twice over: it keeps operator credentials +out of a signed, wire-visible field, and a collector that does not know the +password could not otherwise reproduce the hash. + +Hashing the full URL rather than just the authority means two collectors +behind one hostname on different paths (`/tenant-a` vs `/tenant-b`) get +distinct audiences. The cost is diagnosability — a rejected token tells the +collector nothing about where the sender thought it was pointing — so the node +logs its resolved endpoint at startup and `docs/otel-metrics.md` documents the +computation for hand-checking a mismatch. + +A collector verifies with a stock Ed25519 library after converting the +Montgomery public key to Edwards with sign bit 0, then checks `` +against the hash of each URL it answers at and `` against its own +clock. + +The default is `disabled` — no `Authorization` header. Pointing the exporter at +your own collector must not ship a signed assertion of this node's identity +somewhere that never asked for one; `freenet` mode is for collectors that +actually verify these tokens. Either way, an `Authorization` header supplied +through `OTEL_EXPORTER_OTLP_HEADERS` is never overwritten. + +### Endpoint precedence + +`OTEL_EXPORTER_OTLP_METRICS_ENDPOINT` > `OTEL_EXPORTER_OTLP_ENDPOINT` > +`otel-endpoint` in config.toml > `http://localhost:4318` (the SDK's own default). + +This is deliberately *not* what the SDK does by itself. In +`opentelemetry-otlp` 0.32, `resolve_http_endpoint` +(`src/exporter/http/mod.rs:720-750`) gives a programmatic `with_endpoint` value +**priority over both env vars**, and uses it **verbatim** — `build_endpoint_uri` +appends `/v1/metrics` only on the env-var path. So to get env-wins precedence the +code must: + +- call `with_endpoint` only when neither env var is set, and +- append `/v1/metrics` itself when passing the config-file value. + +`otel-endpoint` therefore has no clap `env =` binding — binding it would merge the +standard variable into the config layer and invert the precedence. + +Note the two endpoint variables are NOT interchangeable. `resolve_http_endpoint` +uses the signal-specific `OTEL_EXPORTER_OTLP_METRICS_ENDPOINT` **verbatim** +(`// per signal env var is not modified`); only the generic +`OTEL_EXPORTER_OTLP_ENDPOINT` and the built-in default go through +`build_endpoint_uri`. So the signal-specific variable must carry the full +`/v1/metrics` path and the generic one must not. + +Most other standard variables — `OTEL_SERVICE_NAME`, `OTEL_RESOURCE_ATTRIBUTES`, +`OTEL_EXPORTER_OTLP_HEADERS`, `OTEL_METRIC_EXPORT_INTERVAL` (default 60s, +`opentelemetry_sdk/src/metrics/periodic_reader.rs:24-43`) — are read by the SDK. +No code for them. Two exceptions: + +- `OTEL_EXPORTER_OTLP_TIMEOUT` / `OTEL_EXPORTER_OTLP_METRICS_TIMEOUT` are + resolved by `otel::export_timeout`, not the SDK: the SDK applies its resolved + timeout only to a client it builds itself, and we always supply one. +- `OTEL_EXPORTER_OTLP_COMPRESSION` is **not supported**. The exporter validates + it at build time and hard-errors unless the `gzip-http` / `zstd-http` feature + is enabled, which we deliberately do not enable. Setting it therefore fails + the exporter build and the node runs with no metrics (with a WARN naming the + cause). Enable the matching feature if compression is ever wanted. + +## Dependencies + +`opentelemetry` is already non-optional in `crates/core/Cargo.toml`. +`opentelemetry_sdk` and `opentelemetry-otlp` were optional and reachable only +through the `trace-ot` feature; both were already in `Cargo.lock`. Making them +non-optional, with `opentelemetry-otlp` trimmed to `http-proto` + `metrics` + +`internal-logs`, is the dependency change. Net new packages in `Cargo.lock`: +`xeddsa` and its `convert_case`. + +**No `reqwest-*-client` and no `reqwest-rustls` feature.** `tracing::otel` +always supplies its own `HttpClient`, and `opentelemetry-otlp` builds a client +only when none was given, so those features would be dead weight with a real +cost: they pull `opentelemetry-http`'s reqwest **0.13**, whose `rustls` feature +is hardwired to `aws-lc-rs` + `rustls-platform-verifier`. That is a second +reqwest major, a second TLS root store, a C/assembly `aws-lc-sys` build (via +`cc`+`cmake`, on musl and Windows release targets that install neither), and +two crypto providers inside one rustls — which makes +`CryptoProvider::get_default_or_install_from_crate_features` return `None` and +panic for any caller that does not pass a provider explicitly. Note that +switching to `reqwest-rustls-webpki-roots` does **not** avoid this: it still +enables `reqwest/default-tls`, i.e. the same aws-lc stack, and merely adds +webpki roots on top. + +Our client is blocking on purpose: `PeriodicReader` runs exports on a dedicated +thread through `futures_executor::block_on` (`periodic_reader.rs:419`), where +an async reqwest client has no tokio reactor. It rides the workspace reqwest +0.12, which already carries `rustls-tls`, so `https://` endpoints work with +nothing new in the graph. It also applies the export timeout +(`OTEL_EXPORTER_OTLP_METRICS_TIMEOUT` > `OTEL_EXPORTER_OTLP_TIMEOUT` > 10s) +itself, because the SDK only resolves that for a client it built. + +Because a feature array may not name a non-optional dependency, `trace-ot` drops +`"opentelemetry-otlp"` from its list. + +## Suppression + +`otel::init` returns `Some(reason)` — no provider, no exporter, no global +registration — when any of these hold, mirroring `telemetry_suppression_reason` +(`telemetry.rs:660-679`): + +1. `!cfg.enabled` +2. `cfg.is_test_environment` (the `--id` flag) +3. `cfg!(test)` +4. `running_under_cargo_test()` (executable's parent dir is `deps/`) + +Same rationale as #4366: keyed only on signals a real release binary never trips, +and deliberately not on `cfg!(feature = "testing")`, which leaks onto the shipped +binary via Cargo feature unification with `fdev`. + +The decision lives in a pure function so both directions are unit-testable from +inside a test process, which by construction trips signals 3 and 4. + +## Pipeline + +`crates/core/src/tracing/otel.rs`: + +``` +MetricExporter::builder().with_http()[.with_endpoint(resolved)] + .with_http_client(OtlpHttpClient { .. }) // always ours; signs when auth is on + .build() + → SdkMeterProvider::builder() + .with_periodic_exporter(exporter) + .with_resource(Resource::builder() + .with_service_name("freenet-node") // only if OTEL_* did not set it + .with_attribute(KeyValue::new("freenet.node.pubkey", pubkey)) + .with_attribute(KeyValue::new("freenet.node.fingerprint", fingerprint)) + .build()) + .with_view(/* base-2 exponential for every histogram */) + .build() + → opentelemetry::global::set_meter_provider(provider) +``` + +Exporter build failure logs a WARN and the node starts anyway. Metrics export +must never fail node startup. + +The whole build runs on a plain `std::thread`: `reqwest::blocking::Client` owns +a private tokio runtime, and creating or dropping one inside an async context +panics with "Cannot drop a runtime in a context where blocking is not allowed". +`init` is called from the node's async build path. + +No wrapper type, no registry, no facade. Future instrumentation is +`opentelemetry::global::meter("freenet").u64_counter(…)` at the call site. + +### Proof-of-life metric + +One observable gauge, `freenet.process.memory.rss`, over +`crate::node::resource_metrics::rss_bytes()` (already exists, +`node/resource_metrics.rs:79`). A real datapoint end to end, and the thing to look +at in a collector to confirm the pipeline works. + +### Shutdown + +Not wired. `global::set_meter_provider` holds a reference for the process +lifetime, and `PeriodicReader` exports every 60s, so at most one partial interval +is lost at exit. Flushing on the signal path is plumbing this does not need yet; +the code carries a `NOTE:` comment naming the ceiling and the upgrade path. + +## Wire-up + +`crates/core/src/node.rs`, in `build_with_flush_handle` beside the +`TelemetryReporter::new` call (`node.rs:815`): + +```rust +crate::tracing::otel::init(&self.config.otel, &self.key_pair); +``` + +The transport keypair, not a `PeerId`: it yields both identity resource +attributes and, in `freenet` auth mode, the token signing key. The provider is +registered globally; nothing is stored on `Node`. `init` returns the +suppression reason (or `None` when it started) so a test can prove it consults +that check before building anything — `init` is otherwise unreachable under +`cfg(test)`. + +## Testing + +- Endpoint precedence: metrics env > generic env > config > default. Pure + function, no network. +- Suppression: the pure decision function returns "export" only for a + production-shaped input, and a reason for each of disabled / `--id` / + `cfg(test)` / cargo-`deps` harness. Both directions. +- Isolation: enforced structurally — the otel decision function takes + `&OtelConfig` and cannot reach `TelemetryConfig` — plus a review check that + the diff touches `telemetry.rs` in exactly one line (the `pub(crate)` + widening of the harness detector). +- CLI/env parsing: `--otel-telemetry-enabled=false` parses as false. A bare + clap flag with an `env` binding treats any value of the variable as true, + which would turn the exporter on for an operator trying to turn it off. It + also overrides a `config.toml` that says `true`, asserted through + `ConfigArgs::build()` rather than at parse level. +- Auth: token shape and stock-Ed25519 verification; a token re-aimed at another + collector fails to verify; the header reaches a real socket in `freenet` + mode, is absent in `disabled` mode, and never replaces an operator-supplied + `Authorization`. +- Pins that would otherwise be vacuous: `init` returns a suppression reason + under `cfg(test)` (deleting the check makes it build a pipeline instead), and + a cross-file scrape asserts every hot-path `record_*` mirror still exists in + `transport/metrics.rs` / `node/network_status.rs`. +- Provider construction succeeds inside a tokio runtime against an unreachable + endpoint (guards the reqwest-blocking-in-async-context concern; export failure + is asynchronous and must not surface at build time). +- Config round-trip through `ConfigArgs::build()` — mandatory per + `.claude/rules/code-style.md`. + +## Risks + +- Making the two crates non-optional grows the default build: the trim drops + the logs exporter only, since `http-proto` mandates `trace`, `prost` and + `opentelemetry-proto`. No new native toolchain requirement — see + Dependencies — so the musl/Windows release targets in + `.github/workflows/cross-compile.yml` (which never runs on PRs) build the + same way they did before. +- `global::set_meter_provider` is process-global. A `trace-ot` build also sets + OTel globals (tracer provider, not meter provider). +- `reqwest/blocking` gets enabled workspace-wide by feature unification, which + pulls in a background runtime thread for blocking clients. + +## Process note + +This is a feature, not a bug fix. Per +[CONTRIBUTING.md](../../CONTRIBUTING.md) it needs a maintainer-approved issue +before implementation starts — see #5046. diff --git a/docs/otel-metrics.md b/docs/otel-metrics.md new file mode 100644 index 0000000000..a32e3f9671 --- /dev/null +++ b/docs/otel-metrics.md @@ -0,0 +1,142 @@ +# Exporting node metrics to an OpenTelemetry collector + +A freenet node can export its own metrics — transport, ring, contract queue, +process RSS — to any OTLP/HTTP collector. It is **off by default** and +completely separate from `telemetry-enabled`, which feeds the project's central +dashboard: turning one on or off has no effect on the other. + +Design notes and the full instrument list are in +[`docs/design/otel-metrics-exporter.md`](design/otel-metrics-exporter.md). + +## Turning it on + +In `config.toml`: + +```toml +otel-telemetry-enabled = true +otel-endpoint = "http://collector.example:4318" +``` + +or on the command line: + +```bash +freenet network --otel-telemetry-enabled --otel-endpoint http://collector.example:4318 +``` + +`--otel-telemetry-enabled=false` turns it off again without editing the file. +`FREENET_OTEL_TELEMETRY_ENABLED` works too, and unlike a plain flag it honors +`=false`. + +Nodes started with `--id` (test networks, the integration harness) never +export, regardless of configuration. + +## Settings + +| `config.toml` key | Default | Meaning | +|---|---|---| +| `otel-telemetry-enabled` | `false` | Enable the exporter | +| `otel-endpoint` | none | Collector base URL, e.g. `http://collector:4318`. `/v1/metrics` is appended for you | +| `otel-auth-mode` | `disabled` | `disabled` sends no `Authorization` header; `freenet` sends a signed bearer token (below) | + +The standard OpenTelemetry environment variables take priority over +`otel-endpoint`: `OTEL_EXPORTER_OTLP_METRICS_ENDPOINT`, +`OTEL_EXPORTER_OTLP_ENDPOINT`, `OTEL_EXPORTER_OTLP_HEADERS`, +`OTEL_EXPORTER_OTLP_TIMEOUT`, `OTEL_EXPORTER_OTLP_METRICS_TIMEOUT`, +`OTEL_SERVICE_NAME`, `OTEL_RESOURCE_ATTRIBUTES`, `OTEL_METRIC_EXPORT_INTERVAL` +(default 60s). With none of them and no `otel-endpoint`, the exporter uses +`http://localhost:4318`. + +**The two endpoint variables are not interchangeable.** The signal-specific +`OTEL_EXPORTER_OTLP_METRICS_ENDPOINT` is used exactly as written, so it must +include the full path: + +```bash +OTEL_EXPORTER_OTLP_METRICS_ENDPOINT=http://collector:4318/v1/metrics # full path +OTEL_EXPORTER_OTLP_ENDPOINT=http://collector:4318 # base URL only +``` + +Getting this backwards produces a 404 on every export — and, in `freenet` auth +mode, an audience hash over the wrong path. Include the scheme in either form: +a bare `collector:4318` is accepted as a URL but cannot be sent, and the node +warns about it at startup. + +The timeout variables are in **milliseconds** (`OTEL_EXPORTER_OTLP_TIMEOUT=10` +is 10ms, not 10 seconds, and every export will time out). The node warns on a +suspiciously small value. + +`OTEL_EXPORTER_OTLP_COMPRESSION` is **not supported** — the compression +features are deliberately not compiled in, and setting the variable makes the +exporter fail to start, leaving the node with no metrics at all. The startup +warning names the cause. + +When an environment variable overrides a configured `otel-endpoint`, the node +logs a warning at startup naming both, and the "OTel metrics exporter started" +line always names the endpoint actually in use. + +## Authentication + +For most setups leave `otel-auth-mode` at `disabled` and carry whatever +credentials your collector wants in `OTEL_EXPORTER_OTLP_HEADERS`: + +```bash +OTEL_EXPORTER_OTLP_HEADERS="Authorization=Basic $(printf 'user:pass' | base64)" +``` + +An `Authorization` header set that way is never overwritten by the node. + +`otel-auth-mode = "freenet"` is for collectors that verify freenet node +identities. It adds + +``` +Authorization: Bearer freenet//// +``` + +to each export, where `` is an XEdDSA signature over the preceding +fields made with the node's transport key, `` is that key in base58, +and `` identifies the exact URL the export was sent to. It proves the +metrics came from the node they claim to, and the audience field means a token +sent to one collector cannot be replayed at another. Do not enable it for a +collector that does not check these tokens — it ships a signed assertion of +your node's identity to whatever it is pointed at. + +`` is base58 of the first 16 bytes of `SHA-256` over the canonical +target, which is `{host}:{port}{path}` — host lowercased, port always explicit +(filled in from the scheme as 80 or 443 when the URL omits it), path verbatim, +any `user:password@` stripped, and no scheme. So an endpoint of +`http://collector.example:4318` produces the audience for +`collector.example:4318/v1/metrics`, which you can reproduce with: + +```bash +# base58 has no standard CLI; this uses Python, which is universally available. +printf 'collector.example:4318/v1/metrics' \ + | openssl dgst -sha256 -binary | head -c 16 \ + | python3 -c 'import sys;d=sys.stdin.buffer.read();n=int.from_bytes(d,"big");a="123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz";s="" +while n: n,r = divmod(n,58); s = a[r]+s +print("1"*(len(d)-len(d.lstrip(b"\0")))+s)' +``` + +If your collector rejects tokens with an audience mismatch, compare that value +against the endpoint in the node's "OTel metrics exporter started" log line — +the two must name the same URL, including the path your ingress finally +delivers to. + +## Identifying a node + +Every export batch carries two resource attributes: + +- `freenet.node.pubkey` — the node's full transport public key, base58. This is + the value a collector verifies the bearer token against. +- `freenet.node.fingerprint` — the short form shown in UIs and the local + dashboard, for cross-referencing. + +Neither contains an address, so a node keeps the same identity across IP +changes. + +## Notes + +- `freenet.process.memory.rss` is Linux-only. On macOS and Windows the series + is empty; that is expected, not a broken pipeline. +- Export failures never affect the node: a collector that is down or rejecting + batches produces a `WARN` naming the endpoint and the reason, and nothing + else. The warning is logged once per failing streak, with an `INFO` when + exports recover — not once per 60s interval.