espresso_telemetry/
lifecycle.rs

1//! In-process OTel logs export plus a periodic Prometheus remote-write push for
2//! metrics. Both pipelines share one six-month BLS-BN254 JWT minted from the
3//! staking key. Tracing events go through an OTLP/HTTP batch exporter; an
4//! externally supplied `prometheus::Registry` is scraped on a tokio interval and
5//! POSTed to `/api/v1/write` as snappy-compressed protobuf.
6//!
7//! Failure modes:
8//! - Proxy down: `BatchLogProcessor` queue fills (~2k records) then drops;
9//!   metrics push `warn!`s and retries next tick. Neither uses disk nor blocks
10//!   consensus.
11//! - JWT misconfig: [`init`] returns `Err`; the caller continues without telemetry.
12//! - Token TTL expiry mid-process is not handled; the six-month TTL outlasts
13//!   typical restart cadence.
14
15use std::{
16    collections::HashMap,
17    sync::{Arc, OnceLock, atomic::AtomicBool},
18    time::Duration,
19};
20
21use anyhow::Context;
22use clap::Parser;
23use derivative::Derivative;
24use jf_signature::bls_over_bn254::{SignKey, VerKey};
25use opentelemetry::KeyValue;
26use opentelemetry_appender_tracing::layer::OpenTelemetryTracingBridge;
27use opentelemetry_otlp::{Compression, LogExporter, Protocol, WithExportConfig, WithHttpConfig};
28use opentelemetry_sdk::{Resource, logs::SdkLoggerProvider};
29use prometheus::Registry;
30use tokio::sync::oneshot;
31use tracing::Subscriber;
32use tracing_subscriber::{EnvFilter, Layer, registry::LookupSpan};
33use url::Url;
34
35use crate::{
36    UnauthenticatedToken, push_task, redact_exporter::RedactingLogExporter, remote_write::Label,
37};
38
39const SERVICE_NAME: &str = "espresso-node";
40const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(10);
41
42const PUBKEY_SLUG_LEN: usize = 24;
43
44fn pubkey_slug(staking_key: &SignKey) -> String {
45    VerKey::from(staking_key)
46        .to_string()
47        .chars()
48        .take(PUBKEY_SLUG_LEN)
49        .collect()
50}
51
52fn segment(value: Option<&str>) -> String {
53    match value.unwrap_or("").trim() {
54        "" => "unk".to_owned(),
55        v => v.replace(' ', "-"),
56    }
57}
58
59fn instance_id(company: Option<&str>, node: Option<&str>, staking_key: &SignKey) -> String {
60    format!(
61        "{}-{}-{}",
62        segment(company),
63        segment(node),
64        pubkey_slug(staking_key)
65    )
66    .to_lowercase()
67}
68
69/// Join `thread`, detaching it after [`SHUTDOWN_TIMEOUT`] so a wedged exporter
70/// can't hang shutdown on the main thread.
71fn join_bounded(thread: std::thread::JoinHandle<()>, what: &str) {
72    let (done_tx, done_rx) = std::sync::mpsc::sync_channel::<()>(0);
73    if let Err(e) = std::thread::Builder::new()
74        .name("espresso-telemetry-join".into())
75        .spawn(move || {
76            let _ = thread.join();
77            let _ = done_tx.send(());
78        })
79    {
80        tracing::warn!(error = %e, "telemetry: cannot spawn {what} join watcher");
81        return;
82    }
83    if done_rx.recv_timeout(SHUTDOWN_TIMEOUT).is_err() {
84        tracing::warn!(
85            timeout_secs = SHUTDOWN_TIMEOUT.as_secs(),
86            "telemetry: {what} shutdown timed out; detaching thread",
87        );
88    }
89}
90
91/// Global handoff for the prometheus `Registry` populated by HotShot.
92///
93/// The API setup builds the `Registry` after [`init`] and can't reach the
94/// telemetry wiring, so it deposits a clone here for the run path to read back.
95/// Single writer, single reader, ordered by node startup. Tests pass the
96/// registry directly into [`init`] instead.
97static REGISTRY: OnceLock<Arc<Registry>> = OnceLock::new();
98
99/// Deposit the `Registry`. Idempotent: subsequent calls are no-ops, since
100/// `OnceLock::set` returns `Err`.
101pub fn set_registry(registry: Arc<Registry>) {
102    let _ = REGISTRY.set(registry);
103}
104
105/// Read the registry deposited by [`set_registry`]. `None` if no API setup has
106/// run yet (tests, CLI tools without the HTTP module).
107pub fn registry() -> Option<Arc<Registry>> {
108    REGISTRY.get().cloned()
109}
110
111/// Default OTel log filter: warnings and above, plus the `announce` target.
112const DEFAULT_LOG_FILTER: &str = "warn,announce=info";
113
114/// Operator-facing telemetry configuration.
115#[derive(Parser, Clone, Derivative)]
116#[derivative(Debug)]
117pub struct TelemetryOptions {
118    /// Enable the OTel logs pipeline.
119    #[clap(
120        long,
121        env = "ESPRESSO_NODE_TELEMETRY_LOGS_ENABLE",
122        default_value = "false"
123    )]
124    pub logs_enable: bool,
125
126    /// Enable the Prometheus remote-write metrics pipeline.
127    #[clap(
128        long,
129        env = "ESPRESSO_NODE_TELEMETRY_METRICS_ENABLE",
130        default_value = "false"
131    )]
132    pub metrics_enable: bool,
133
134    /// OTLP/HTTP base URL override. When unset, the caller selects the default
135    /// endpoint (the node picks it by chain ID).
136    #[clap(long, env = "ESPRESSO_NODE_TELEMETRY_ENDPOINT")]
137    pub endpoint: Option<Url>,
138
139    /// `EnvFilter` for the OTel log layer only; the local stderr layer is
140    /// unaffected. Per-target syntax works (e.g. `warn,hotshot=info`).
141    ///
142    /// The default admits the `announce` target at INFO, for milestones that
143    /// belong in telemetry without being warnings.
144    #[clap(long, env = "ESPRESSO_NODE_TELEMETRY_LOG", default_value = DEFAULT_LOG_FILTER)]
145    pub log_filter: String,
146
147    /// Seconds between Prometheus remote-write pushes.
148    #[clap(
149        long,
150        env = "ESPRESSO_NODE_TELEMETRY_METRICS_INTERVAL",
151        default_value = "60"
152    )]
153    pub metrics_interval_secs: u64,
154}
155
156impl Default for TelemetryOptions {
157    fn default() -> Self {
158        Self {
159            logs_enable: false,
160            metrics_enable: false,
161            endpoint: None,
162            log_filter: DEFAULT_LOG_FILTER.to_owned(),
163            metrics_interval_secs: 60,
164        }
165    }
166}
167
168/// Handle to the metrics push thread. The thread owns a dedicated
169/// single-threaded runtime so a slow proxy can't starve a consensus worker.
170struct MetricsPushHandle {
171    shutdown: oneshot::Sender<()>,
172    thread: std::thread::JoinHandle<()>,
173}
174
175/// Owns the OTel logger provider and, optionally, the metrics push task, so
176/// both flush on graceful shutdown. Stashes the JWT and endpoint so the push
177/// task can be spawned later via [`TelemetryHandle::attach_metrics_push`] once
178/// the API setup has built the `Registry`.
179pub struct TelemetryHandle {
180    /// `None` when only metrics are enabled (logs pipeline disabled).
181    logger_provider: Option<SdkLoggerProvider>,
182    log_filter: String,
183    jwt: String,
184    endpoint: String,
185    metrics_interval: Duration,
186    metrics_push: Option<MetricsPushHandle>,
187    /// When false, `attach_metrics_push` is a no-op.
188    metrics_enabled: bool,
189    /// Labels stamped onto every pushed TimeSeries (e.g. `service`, `instance`),
190    /// mirroring the OTel resource attributes so the aggregator partitions logs
191    /// and metrics consistently.
192    metrics_external_labels: Vec<Label>,
193    /// Latch flipped on the first HTTP 429 from the metrics push, so the
194    /// operator-facing ERROR is logged once per process across push ticks. The
195    /// logs pipeline retries 429s via opentelemetry-otlp's built-in
196    /// `experimental-http-retry` and does not surface them here.
197    rate_limit_warned: Arc<AtomicBool>,
198    /// `log_filter` embedded verbatim in the rate-limit ERROR so operators see
199    /// their active filter.
200    telemetry_log_filter: Arc<String>,
201}
202
203impl std::fmt::Debug for TelemetryHandle {
204    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
205        f.debug_struct("TelemetryHandle")
206            .field("log_filter", &self.log_filter)
207            .field("metrics_push_active", &self.metrics_push.is_some())
208            .finish()
209    }
210}
211
212impl TelemetryHandle {
213    /// Layer bridging tracing events into the OTLP exporter, filtered by
214    /// `log_filter`. `None` when the logs pipeline is disabled.
215    pub fn tracing_layer<S>(&self) -> Option<impl Layer<S> + Send + Sync + 'static>
216    where
217        S: Subscriber + for<'a> LookupSpan<'a>,
218    {
219        let provider = self.logger_provider.as_ref()?;
220        let bridge = OpenTelemetryTracingBridge::new(provider);
221        Some(bridge.with_filter(EnvFilter::new(self.log_filter.clone())))
222    }
223
224    /// Wrapper over [`attach_metrics_push_buffered`] that emits setup warnings
225    /// directly, for callers that already have a subscriber installed.
226    pub fn attach_metrics_push(&mut self, registry: Arc<Registry>) {
227        let mut deferred = Vec::new();
228        self.attach_metrics_push_buffered(registry, &mut deferred);
229        for w in deferred {
230            tracing::warn!("{w}");
231        }
232    }
233
234    /// Spawn the periodic metrics push on its own thread. Idempotent: no-op if
235    /// already attached or if metrics are disabled. Setup warnings buffer into
236    /// `deferred` so [`init`] can replay them after a subscriber is installed.
237    fn attach_metrics_push_buffered(
238        &mut self,
239        registry: Arc<Registry>,
240        deferred: &mut Vec<String>,
241    ) {
242        if !self.metrics_enabled {
243            return;
244        }
245        if self.metrics_push.is_some() {
246            return;
247        }
248        let push_endpoint: Url = match self.endpoint.parse() {
249            Ok(u) => u,
250            Err(e) => {
251                deferred.push(format!(
252                    "telemetry: cannot parse endpoint {endpoint:?} as URL ({e}); skipping metrics \
253                     push",
254                    endpoint = self.endpoint,
255                ));
256                return;
257            },
258        };
259        let (shutdown_tx, shutdown_rx) = oneshot::channel();
260        let jwt = self.jwt.clone();
261        let interval = self.metrics_interval;
262        let rate_limit_warned = self.rate_limit_warned.clone();
263        let telemetry_log_filter = self.telemetry_log_filter.clone();
264        let external_labels = self.metrics_external_labels.clone();
265        let thread = match std::thread::Builder::new()
266            .name("espresso-telemetry-metrics".into())
267            .spawn(move || {
268                let rt = match tokio::runtime::Builder::new_current_thread()
269                    .enable_all()
270                    .build()
271                {
272                    Ok(rt) => rt,
273                    Err(e) => {
274                        tracing::warn!(
275                            error = %e,
276                            "telemetry: cannot build dedicated runtime; metrics push disabled"
277                        );
278                        return;
279                    },
280                };
281                rt.block_on(push_task::run(
282                    registry,
283                    push_endpoint,
284                    jwt,
285                    interval,
286                    external_labels,
287                    rate_limit_warned,
288                    telemetry_log_filter,
289                    shutdown_rx,
290                ));
291            }) {
292            Ok(t) => t,
293            Err(e) => {
294                deferred.push(format!("telemetry: cannot spawn metrics push thread: {e}"));
295                return;
296            },
297        };
298        self.metrics_push = Some(MetricsPushHandle {
299            shutdown: shutdown_tx,
300            thread,
301        });
302    }
303
304    /// Flush the push thread, then shut down the OTel logger provider.
305    /// Best-effort; failures are logged, never bubbled. Both joins run through
306    /// [`join_bounded`] (the provider shutdown can deadlock on a current-thread
307    /// runtime).
308    pub fn shutdown(self) {
309        if let Some(MetricsPushHandle { shutdown, thread }) = self.metrics_push {
310            let _ = shutdown.send(());
311            join_bounded(thread, "metrics push");
312        }
313        if let Some(provider) = self.logger_provider {
314            match std::thread::Builder::new()
315                .name("espresso-telemetry-shutdown".into())
316                .spawn(move || {
317                    if let Err(e) = provider.shutdown() {
318                        tracing::warn!(error = %e, "telemetry: logger provider shutdown error");
319                    }
320                }) {
321                Ok(thread) => join_bounded(thread, "logger provider"),
322                Err(e) => tracing::warn!(error = %e, "telemetry: cannot spawn shutdown thread"),
323            }
324        }
325    }
326
327    /// True once the metrics push task is spawned.
328    #[doc(hidden)]
329    pub fn metrics_push_active(&self) -> bool {
330        self.metrics_push.is_some()
331    }
332
333    /// True when the metrics pipeline is enabled.
334    #[doc(hidden)]
335    pub fn metrics_enabled(&self) -> bool {
336        self.metrics_enabled
337    }
338}
339
340/// Initialize the OTel logger pipeline and, when `registry` is `Some`, spawn the
341/// metrics push immediately. Production passes `None` (the `Registry` isn't built
342/// yet) and later calls [`TelemetryHandle::attach_metrics_push`]; tests pass it
343/// directly.
344///
345/// `Ok((None, _))` when telemetry is disabled; `Err` on misconfig (bad endpoint,
346/// JWT mint failure). The returned warnings are produced before any subscriber is
347/// installed; callers MUST replay them via `tracing::warn!` afterward or they are
348/// lost.
349pub fn init(
350    opts: &TelemetryOptions,
351    staking_key: &SignKey,
352    node_name: Option<&str>,
353    company_name: Option<&str>,
354    endpoint: &Url,
355    registry: Option<Arc<Registry>>,
356) -> anyhow::Result<(Option<TelemetryHandle>, Vec<String>)> {
357    let logs_on = opts.logs_enable;
358    let metrics_on = opts.metrics_enable;
359
360    if !logs_on && !metrics_on {
361        return Ok((None, Vec::new()));
362    }
363
364    let mut deferred: Vec<String> = Vec::new();
365
366    if logs_on && let Err(e) = EnvFilter::try_new(&opts.log_filter) {
367        deferred.push(format!(
368            "telemetry: invalid log_filter {filter:?} ({e}); falling back to lossy parse, some \
369             directives may be ignored",
370            filter = opts.log_filter,
371        ));
372    }
373
374    let jwt = UnauthenticatedToken::generate_with(staking_key, node_name, company_name)
375        .context("mint telemetry JWT")?
376        .encode();
377
378    // Both signals share this base URL. Reject non-http(s) so the metrics push
379    // can't silently no-op later.
380    let scheme = endpoint.scheme();
381    if scheme != "http" && scheme != "https" {
382        anyhow::bail!(
383            "telemetry endpoint must use http or https scheme, got {scheme:?}: {endpoint}"
384        );
385    }
386    let endpoint = endpoint.as_str().to_owned();
387
388    let telemetry_log_filter: Arc<String> = Arc::new(opts.log_filter.clone());
389    let rate_limit_warned: Arc<AtomicBool> = Arc::new(AtomicBool::new(false));
390
391    let instance = instance_id(company_name, node_name, staking_key);
392
393    let logger_provider = if logs_on {
394        Some(build_logger_provider(jwt.clone(), &endpoint, &instance)?)
395    } else {
396        None
397    };
398
399    let metrics_interval = Duration::from_secs(opts.metrics_interval_secs.max(1));
400    let metrics_external_labels = vec![
401        Label {
402            name: "service".to_owned(),
403            value: SERVICE_NAME.to_owned(),
404        },
405        Label {
406            name: "instance".to_owned(),
407            value: instance,
408        },
409    ];
410    let mut handle = TelemetryHandle {
411        logger_provider,
412        log_filter: opts.log_filter.clone(),
413        jwt,
414        endpoint,
415        metrics_interval,
416        metrics_push: None,
417        metrics_enabled: metrics_on,
418        metrics_external_labels,
419        rate_limit_warned,
420        telemetry_log_filter,
421    };
422
423    if let Some(registry) = registry {
424        handle.attach_metrics_push_buffered(registry, &mut deferred);
425    }
426
427    Ok((Some(handle), deferred))
428}
429
430fn build_logger_provider(
431    jwt: String,
432    endpoint: &str,
433    instance: &str,
434) -> anyhow::Result<SdkLoggerProvider> {
435    let logs_endpoint = format!("{}/v1/logs", endpoint.trim_end_matches('/'));
436
437    let mut headers = HashMap::new();
438    headers.insert("authorization".to_string(), format!("Bearer {jwt}"));
439
440    // Logs are highly compressible; gzip keeps flush payloads small.
441    let exporter = LogExporter::builder()
442        .with_http()
443        .with_protocol(Protocol::HttpBinary)
444        .with_compression(Compression::Gzip)
445        .with_endpoint(logs_endpoint)
446        .with_headers(headers)
447        .build()
448        .context("build OTLP log exporter")?;
449
450    let resource = Resource::builder()
451        .with_service_name(SERVICE_NAME)
452        .with_attribute(KeyValue::new("service.instance.id", instance.to_owned()));
453
454    Ok(SdkLoggerProvider::builder()
455        .with_resource(resource.build())
456        .with_batch_exporter(RedactingLogExporter::new(exporter))
457        .build())
458}
459
460#[cfg(test)]
461mod tests {
462    use jf_signature::{SignatureScheme, bls_over_bn254::BLSOverBN254CurveSignatureScheme};
463
464    use super::*;
465
466    #[test]
467    fn instance_id_appends_pubkey_slug() {
468        let key = BLSOverBN254CurveSignatureScheme::key_gen(&(), &mut rand::thread_rng())
469            .unwrap()
470            .0;
471        let slug = pubkey_slug(&key);
472        assert_eq!(slug.len(), PUBKEY_SLUG_LEN);
473        assert!(slug.starts_with("BLS_VER_KEY~"));
474
475        let slug = slug.to_lowercase();
476        assert_eq!(
477            instance_id(Some("Acme Corp"), Some("Node 42"), &key),
478            format!("acme-corp-node-42-{slug}")
479        );
480        assert_eq!(instance_id(None, None, &key), format!("unk-unk-{slug}"));
481        assert_eq!(
482            instance_id(Some("  "), Some("node-1"), &key),
483            format!("unk-node-1-{slug}")
484        );
485    }
486}
espresso_telemetry/lifecycle.rs

espresso_telemetry/
lifecycle.rs
