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

espresso_telemetry/
lifecycle.rs
