micromegas_telemetry_sink/
lib.rs

1//! Telemetry HTTP sink library
2//!
3//! Provides logging, metrics, memory and performance profiling
4
5// crate-specific lint exceptions:
6#![allow(
7    unsafe_code,
8    missing_docs,
9    clippy::missing_errors_doc,
10    clippy::new_without_default
11)]
12
13// ---- Native-only modules ----
14#[cfg(not(target_arch = "wasm32"))]
15pub mod api_key_decorator;
16#[cfg(not(target_arch = "wasm32"))]
17pub mod composite_event_sink;
18#[cfg(not(target_arch = "wasm32"))]
19mod env_config;
20#[cfg(not(target_arch = "wasm32"))]
21pub mod http_event_sink;
22#[cfg(not(target_arch = "wasm32"))]
23pub mod local_event_sink;
24#[cfg(not(target_arch = "wasm32"))]
25pub mod log_interop;
26#[cfg(not(target_arch = "wasm32"))]
27pub mod oidc_client_credentials_decorator;
28#[cfg(not(target_arch = "wasm32"))]
29pub mod request_decorator;
30#[cfg(not(target_arch = "wasm32"))]
31pub mod stream_block;
32#[cfg(not(target_arch = "wasm32"))]
33pub mod stream_info;
34#[cfg(not(target_arch = "wasm32"))]
35pub mod system_monitor;
36#[cfg(not(target_arch = "wasm32"))]
37pub mod tracing_interop;
38
39// ---- Feature-gated, both targets ----
40// tikv-jemalloc-ctl is only a dependency for `cfg(all(not(wasm32), not(windows)))`
41// (see Cargo.toml), but the module itself needs no wasm32 gate: it declares no
42// dependency of its own, so it compiles to a no-op on wasm32 too.
43#[cfg(feature = "jemalloc")]
44pub mod jemalloc_conf;
45
46// ---- Wasm-only modules ----
47#[cfg(target_arch = "wasm32")]
48mod console_event_sink;
49#[cfg(target_arch = "wasm32")]
50pub use console_event_sink::*;
51
52// ---- Native implementation ----
53#[cfg(not(target_arch = "wasm32"))]
54mod native {
55    use std::any::TypeId;
56    use std::collections::HashMap;
57    use std::str::FromStr;
58    use std::sync::{Arc, Mutex, Weak};
59
60    use crate::env_config::{env_override, resolve};
61    use crate::log_interop::install_log_interop;
62    use crate::request_decorator::RequestDecorator;
63    use crate::tracing_interop::install_tracing_interop;
64    use micromegas_telemetry::property_names::RESERVED_PROPERTY_PREFIX;
65    use micromegas_tracing::event::BoxedEventSink;
66    use micromegas_tracing::info;
67    use micromegas_tracing::{
68        event::EventSink,
69        guards::{TracingSystemGuard, TracingThreadGuard},
70        prelude::*,
71    };
72
73    use crate::composite_event_sink::CompositeSink;
74    use crate::http_event_sink::{HttpEventSink, HttpSinkConfig};
75    use crate::local_event_sink::LocalEventSink;
76    use crate::system_monitor::spawn_system_monitor;
77
78    pub mod tokio_retry {
79        pub use tokio_retry2::*;
80    }
81
82    pub mod reqwest {
83        pub use reqwest::*;
84    }
85
86    /// Environment variable carrying deployment-supplied process properties as a
87    /// comma-separated `key=value` list, e.g. `cluster=prod,role=cache`. Read by
88    /// [`TelemetryGuardBuilder::build`], so every micromegas process honors it.
89    pub const PROCESS_PROPERTIES_ENV_VAR: &str = "MICROMEGAS_PROCESS_PROPERTIES";
90
91    /// Environment variable capping the local (stdout) sink's max level.
92    /// Read once by [`TelemetryGuardBuilder::build`]. Precedence is
93    /// *[`TelemetryGuardBuilder::with_local_sink_max_level`] (including the
94    /// `#[micromegas_main]` macro's `local_sink_max_level` attribute) > this
95    /// variable > the [`DEFAULT_LOCAL_SINK_MAX_LEVEL`] default*. Accepts `off`,
96    /// `fatal`, `error`, `warn`, `info`, `debug`, or `trace`, case-insensitively;
97    /// an unparseable value fails `build()`.
98    pub const LOCAL_SINK_MAX_LEVEL_ENV_VAR: &str = "MICROMEGAS_LOCAL_SINK_MAX_LEVEL";
99
100    /// Environment variable capping the HTTP telemetry sink's max level.
101    /// Read once by [`TelemetryGuardBuilder::build`]. Precedence is
102    /// *[`TelemetryGuardBuilder::with_telemetry_sink_max_level`] > this
103    /// variable > the [`DEFAULT_TELEMETRY_SINK_MAX_LEVEL`] default*. Accepts
104    /// the same values as [`LOCAL_SINK_MAX_LEVEL_ENV_VAR`]; an unparseable
105    /// value fails `build()`, even if no telemetry URL ends up configured.
106    pub const TELEMETRY_SINK_MAX_LEVEL_ENV_VAR: &str = "MICROMEGAS_TELEMETRY_SINK_MAX_LEVEL";
107
108    /// Default max level for the local (stdout) sink, applied when neither an
109    /// explicit builder call nor [`LOCAL_SINK_MAX_LEVEL_ENV_VAR`] sets one.
110    pub const DEFAULT_LOCAL_SINK_MAX_LEVEL: LevelFilter = LevelFilter::Info;
111
112    /// Default max level for the HTTP telemetry sink, applied when neither an
113    /// explicit builder call nor [`TELEMETRY_SINK_MAX_LEVEL_ENV_VAR`] sets one.
114    pub const DEFAULT_TELEMETRY_SINK_MAX_LEVEL: LevelFilter = LevelFilter::Debug;
115
116    const MAX_QUEUE_BYTES_ENV_VAR: &str = "MICROMEGAS_TELEMETRY_MAX_QUEUE_BYTES";
117    const HARD_QUEUE_BYTES_ENV_VAR: &str = "MICROMEGAS_TELEMETRY_HARD_QUEUE_BYTES";
118    const MAX_IN_FLIGHT_REQUESTS_ENV_VAR: &str = "MICROMEGAS_TELEMETRY_MAX_IN_FLIGHT_REQUESTS";
119    const REQUEST_TIMEOUT_SECS_ENV_VAR: &str = "MICROMEGAS_TELEMETRY_REQUEST_TIMEOUT_SECS";
120
121    /// Merges a [`PROCESS_PROPERTIES_ENV_VAR`]-formatted list into `properties`,
122    /// leaving keys already present untouched. Precedence is therefore
123    /// *explicit [`TelemetryGuardBuilder::with_process_property`] calls > environment >
124    /// `populate_default_system_properties` defaults*: an operator
125    /// can add deployment tags but cannot spoof the `version` that `micromegas_main`
126    /// stamps nor the identity properties derived from the host, both of which
127    /// dashboards filter on. Within one list the first occurrence of a key wins,
128    /// consistent with that rule.
129    ///
130    /// The whole list is parsed before anything is merged, so a malformed entry
131    /// fails the call with `properties` untouched rather than applying half a
132    /// config. Blank entries are dropped and each entry is trimmed: a k8s manifest
133    /// rendering an unset optional var produces `""`, hand-written lists routinely
134    /// carry a trailing comma, and a space after a comma would otherwise create a
135    /// key no query could match.
136    ///
137    /// Keys in the server-reserved [`RESERVED_PROPERTY_PREFIX`] namespace are
138    /// rejected here, turning the ingestion service's silent strip into a startup
139    /// failure the operator can see.
140    pub fn merge_process_properties(
141        properties: &mut HashMap<String, String>,
142        raw: &str,
143    ) -> anyhow::Result<()> {
144        use anyhow::Context;
145
146        let mut parsed = Vec::new();
147        for entry in raw.split(',').map(str::trim).filter(|e| !e.is_empty()) {
148            let (key, value) = entry.split_once('=').with_context(|| {
149                format!("invalid {PROCESS_PROPERTIES_ENV_VAR} entry {entry:?}: expected key=value")
150            })?;
151            let (key, value) = (key.trim(), value.trim());
152            if key.is_empty() {
153                anyhow::bail!("invalid {PROCESS_PROPERTIES_ENV_VAR} entry {entry:?}: empty key");
154            }
155            if key.starts_with(RESERVED_PROPERTY_PREFIX) {
156                anyhow::bail!(
157                    "invalid {PROCESS_PROPERTIES_ENV_VAR} entry {entry:?}: the \
158                     {RESERVED_PROPERTY_PREFIX} namespace is server-written only and \
159                     would be stripped on ingestion"
160                );
161            }
162            parsed.push((key.to_string(), value.to_string()));
163        }
164        for (key, value) in parsed {
165            properties.entry(key).or_insert(value);
166        }
167        Ok(())
168    }
169
170    pub struct TelemetryGuardBuilder {
171        logs_buffer_size: usize,
172        metrics_buffer_size: usize,
173        threads_buffer_size: usize,
174        target_max_levels: HashMap<String, String>,
175        max_level_override: Option<LevelFilter>,
176        interop_max_level_override: Option<LevelFilter>,
177        install_log_capture: bool,
178        install_tracing_capture: bool,
179        local_sink_enabled: bool,
180        local_sink_max_level: Option<LevelFilter>,
181        telemetry_sink_url: Option<String>,
182        telemetry_sink_max_level: Option<LevelFilter>,
183        telemetry_max_queue_bytes: Option<usize>,
184        telemetry_hard_queue_bytes: Option<usize>,
185        telemetry_max_in_flight_requests: Option<usize>,
186        telemetry_request_timeout: Option<std::time::Duration>,
187        telemetry_retry_by_priority:
188            Option<[core::iter::Take<tokio_retry::strategy::ExponentialBackoff>; 4]>,
189        telemetry_make_request_decorator: Box<dyn FnOnce() -> Arc<dyn RequestDecorator> + Send>,
190        extra_sinks: HashMap<TypeId, (LevelFilter, BoxedEventSink)>,
191        system_metrics_enabled: bool,
192        default_system_properties_enabled: bool,
193        process_properties: HashMap<String, String>,
194    }
195
196    impl Default for TelemetryGuardBuilder {
197        fn default() -> Self {
198            Self {
199                logs_buffer_size: 10 * 1024 * 1024,
200                metrics_buffer_size: 1024 * 1024,
201                threads_buffer_size: 10 * 1024 * 1024,
202                local_sink_enabled: true,
203                local_sink_max_level: None,
204                telemetry_sink_url: None,
205                telemetry_sink_max_level: None,
206                telemetry_max_queue_bytes: None,
207                telemetry_hard_queue_bytes: None,
208                telemetry_max_in_flight_requests: None,
209                telemetry_request_timeout: None,
210                telemetry_retry_by_priority: None,
211                telemetry_make_request_decorator: Box::new(|| {
212                    Arc::new(crate::request_decorator::TrivialRequestDecorator {})
213                }),
214                target_max_levels: HashMap::default(),
215                max_level_override: None,
216                interop_max_level_override: None,
217                install_log_capture: false,
218                install_tracing_capture: true,
219                extra_sinks: HashMap::default(),
220                system_metrics_enabled: true,
221                default_system_properties_enabled: true,
222                process_properties: HashMap::default(),
223            }
224        }
225    }
226
227    impl TelemetryGuardBuilder {
228        // Only one sink per type ?
229        #[must_use]
230        pub fn add_sink<Sink>(mut self, max_level: LevelFilter, sink: Sink) -> Self
231        where
232            Sink: EventSink + 'static,
233        {
234            let type_id = TypeId::of::<Sink>();
235
236            self.extra_sinks
237                .entry(type_id)
238                .or_insert_with(|| (max_level, Box::new(sink)));
239
240            self
241        }
242
243        /// Programmatic override
244        #[must_use]
245        pub fn with_max_level_override(mut self, level_filter: LevelFilter) -> Self {
246            self.max_level_override = Some(level_filter);
247            self
248        }
249
250        #[must_use]
251        pub fn with_local_sink_enabled(mut self, enabled: bool) -> Self {
252            self.local_sink_enabled = enabled;
253            self
254        }
255
256        #[must_use]
257        pub fn with_interop_max_level_override(mut self, level_filter: LevelFilter) -> Self {
258            self.interop_max_level_override = Some(level_filter);
259            self
260        }
261
262        #[must_use]
263        pub fn with_install_log_capture(mut self, enabled: bool) -> Self {
264            self.install_log_capture = enabled;
265            self
266        }
267
268        #[must_use]
269        pub fn with_install_tracing_capture(mut self, enabled: bool) -> Self {
270            self.install_tracing_capture = enabled;
271            self
272        }
273
274        /// Max level for the local (stdout) sink. Pins the value in code, so
275        /// [`LOCAL_SINK_MAX_LEVEL_ENV_VAR`] no longer applies. Falls back to
276        /// that variable, then [`DEFAULT_LOCAL_SINK_MAX_LEVEL`]; an
277        /// unparseable value for the variable fails `build()`.
278        #[must_use]
279        pub fn with_local_sink_max_level(mut self, level_filter: LevelFilter) -> Self {
280            self.local_sink_max_level = Some(level_filter);
281            self
282        }
283
284        /// Max level for the HTTP telemetry sink. Pins the value in code, so
285        /// [`TELEMETRY_SINK_MAX_LEVEL_ENV_VAR`] no longer applies. Falls back
286        /// to that variable, then [`DEFAULT_TELEMETRY_SINK_MAX_LEVEL`]; an
287        /// unparseable value for the variable fails `build()`.
288        #[must_use]
289        pub fn with_telemetry_sink_max_level(mut self, level_filter: LevelFilter) -> Self {
290            self.telemetry_sink_max_level = Some(level_filter);
291            self
292        }
293
294        #[must_use]
295        pub fn with_ctrlc_handling(self) -> Self {
296            ctrlc::set_handler(move || {
297                info!("Ctrl+C was hit!");
298                micromegas_tracing::guards::shutdown_telemetry();
299                std::process::exit(1);
300            })
301            .expect("Error setting Ctrl+C handler");
302            self
303        }
304
305        /// Soft byte cap for the telemetry upload queue: once reached, new
306        /// `Traces` items (thread and image blocks) are dropped first.
307        /// Falls back to `MICROMEGAS_TELEMETRY_MAX_QUEUE_BYTES`, then
308        /// [`HttpSinkConfig::DEFAULT_MAX_QUEUE_BYTES`]. An unparseable value
309        /// for that variable fails `build()`.
310        #[must_use]
311        pub fn with_max_queue_bytes(mut self, bytes: usize) -> Self {
312            self.telemetry_max_queue_bytes = Some(bytes);
313            self
314        }
315
316        /// Hard byte cap for the telemetry upload queue: once reached, `Logs`
317        /// and `Metrics` items are dropped too (`Metadata` is never
318        /// dropped). Falls back to `MICROMEGAS_TELEMETRY_HARD_QUEUE_BYTES`,
319        /// then [`HttpSinkConfig::DEFAULT_HARD_QUEUE_BYTES`]. An unparseable
320        /// value for that variable fails `build()`.
321        #[must_use]
322        pub fn with_hard_queue_bytes(mut self, bytes: usize) -> Self {
323            self.telemetry_hard_queue_bytes = Some(bytes);
324            self
325        }
326
327        /// Maximum number of `insert_*` HTTP requests in flight at once.
328        /// Falls back to `MICROMEGAS_TELEMETRY_MAX_IN_FLIGHT_REQUESTS`, then
329        /// [`HttpSinkConfig::DEFAULT_MAX_IN_FLIGHT_REQUESTS`]. An unparseable
330        /// value for that variable fails `build()`.
331        #[must_use]
332        pub fn with_max_in_flight_requests(mut self, max_in_flight_requests: usize) -> Self {
333            self.telemetry_max_in_flight_requests = Some(max_in_flight_requests);
334            self
335        }
336
337        /// Per-request timeout (covers connect + send + receive for one
338        /// attempt). Bounds how long a single retry attempt can hang against
339        /// an ingestion service that accepts connections but never responds,
340        /// which otherwise would make shutdown block indefinitely (`Drop for
341        /// HttpEventSink` joins the worker thread). Falls back to
342        /// `MICROMEGAS_TELEMETRY_REQUEST_TIMEOUT_SECS`, then
343        /// [`HttpSinkConfig::DEFAULT_REQUEST_TIMEOUT`]. An unparseable value
344        /// for that variable fails `build()`.
345        #[must_use]
346        pub fn with_request_timeout(mut self, timeout: std::time::Duration) -> Self {
347            self.telemetry_request_timeout = Some(timeout);
348            self
349        }
350
351        /// Retry strategy per upload priority (indexed by
352        /// `UploadPriority as usize`: Metadata, Logs, Metrics, Traces).
353        /// Defaults to [`HttpSinkConfig::default_retry_by_priority`].
354        #[must_use]
355        pub fn with_retry_by_priority(
356            mut self,
357            retry_by_priority: [core::iter::Take<tokio_retry::strategy::ExponentialBackoff>; 4],
358        ) -> Self {
359            self.telemetry_retry_by_priority = Some(retry_by_priority);
360            self
361        }
362
363        #[must_use]
364        pub fn with_request_decorator(
365            mut self,
366            make_decorator: Box<dyn FnOnce() -> Arc<dyn RequestDecorator> + Send>,
367        ) -> Self {
368            self.telemetry_make_request_decorator = make_decorator;
369            self
370        }
371
372        /// Automatically configure authentication from environment variables.
373        ///
374        /// Checks for authentication configuration in this order:
375        /// 1. API key authentication via `MICROMEGAS_INGESTION_API_KEY`
376        /// 2. OIDC client credentials via `MICROMEGAS_OIDC_TOKEN_ENDPOINT`,
377        ///    `MICROMEGAS_OIDC_CLIENT_ID`, and `MICROMEGAS_OIDC_CLIENT_SECRET`
378        /// 3. Falls back to no authentication (TrivialRequestDecorator)
379        ///
380        /// # Example
381        ///
382        /// ```rust,no_run
383        /// use micromegas_telemetry_sink::TelemetryGuardBuilder;
384        ///
385        /// // Set environment variable
386        /// unsafe {
387        ///     std::env::set_var("MICROMEGAS_INGESTION_API_KEY", "secret-key-123");
388        /// }
389        ///
390        /// // Builder automatically configures API key authentication
391        /// let _guard = TelemetryGuardBuilder::default()
392        ///     .with_auth_from_env()
393        ///     .build()
394        ///     .expect("Failed to build telemetry guard");
395        /// ```
396        #[must_use]
397        pub fn with_auth_from_env(mut self) -> Self {
398            use crate::api_key_decorator::ApiKeyRequestDecorator;
399            use crate::oidc_client_credentials_decorator::OidcClientCredentialsDecorator;
400
401            // Try API key authentication first
402            if let Ok(decorator) = ApiKeyRequestDecorator::from_env() {
403                info!("Configured telemetry sink with API key authentication");
404                self.telemetry_make_request_decorator = Box::new(move || Arc::new(decorator));
405                return self;
406            }
407
408            // Try OIDC client credentials authentication
409            if let Ok(decorator) = OidcClientCredentialsDecorator::from_env() {
410                info!("Configured telemetry sink with OIDC client credentials authentication");
411                self.telemetry_make_request_decorator = Box::new(move || Arc::new(decorator));
412                return self;
413            }
414
415            // No authentication configured - use trivial decorator (no-op)
416            info!(
417                "Telemetry sink authentication not configured - sending unauthenticated requests"
418            );
419            self
420        }
421
422        #[must_use]
423        pub fn with_system_metrics_enabled(mut self, enabled: bool) -> Self {
424            self.system_metrics_enabled = enabled;
425            self
426        }
427
428        #[must_use]
429        pub fn with_default_system_properties_enabled(mut self, enabled: bool) -> Self {
430            self.default_system_properties_enabled = enabled;
431            self
432        }
433
434        /// Set the URL of telemetry sink.
435        ///
436        /// If not explicitly set, the URL will be read from the `MICROMEGAS_TELEMETRY_URL` environment
437        /// variable.
438        #[must_use]
439        pub fn with_telemetry_sink_url(mut self, url: String) -> Self {
440            self.telemetry_sink_url = Some(url);
441            self
442        }
443
444        /// Add a single property to the process info.
445        ///
446        /// # Warning
447        ///
448        /// This will override any existing properties.
449        #[must_use]
450        pub fn with_process_property(mut self, key: String, value: String) -> Self {
451            self.process_properties.insert(key, value);
452            self
453        }
454
455        /// Add multiple properties to the process info.
456        ///
457        /// # Warning
458        ///
459        /// This will override any existing properties.
460        #[must_use]
461        pub fn with_process_properties(
462            mut self,
463            process_properties: HashMap<String, String>,
464        ) -> Self {
465            self.process_properties.extend(process_properties);
466            self
467        }
468
469        fn populate_default_system_properties(&mut self) {
470            let props = &mut self.process_properties;
471            // Process identity (duplicates ProcessInfo fields into properties)
472            type DefaultProperties = Vec<(&'static str, Box<dyn FnOnce() -> String>)>;
473            let defaults: DefaultProperties = vec![
474                (
475                    "exe",
476                    Box::new(|| {
477                        std::env::current_exe()
478                            .unwrap_or_default()
479                            .to_string_lossy()
480                            .into_owned()
481                    }),
482                ),
483                ("username", Box::new(whoami::username)),
484                ("realname", Box::new(whoami::realname)),
485                ("computer", Box::new(whoami::devicename)),
486                ("distro", Box::new(whoami::distro)),
487                (
488                    "cpu_brand",
489                    Box::new(|| {
490                        #[cfg(target_arch = "x86_64")]
491                        {
492                            raw_cpuid::CpuId::new()
493                                .get_processor_brand_string()
494                                .map_or_else(|| "unknown".to_owned(), |b| b.as_str().to_owned())
495                        }
496                        #[cfg(not(target_arch = "x86_64"))]
497                        {
498                            String::from(std::env::consts::ARCH)
499                        }
500                    }),
501                ),
502                (
503                    "physical_core_count",
504                    Box::new(|| {
505                        sysinfo::System::physical_core_count()
506                            .map(|c: usize| c.to_string())
507                            .unwrap_or_default()
508                    }),
509                ),
510                (
511                    "logical_cpu_count",
512                    Box::new(|| {
513                        use sysinfo::{CpuRefreshKind, RefreshKind};
514                        let system = sysinfo::System::new_with_specifics(
515                            RefreshKind::nothing().with_cpu(CpuRefreshKind::nothing()),
516                        );
517                        system.cpus().len().to_string()
518                    }),
519                ),
520                (
521                    "total_memory",
522                    Box::new(|| {
523                        use sysinfo::{MemoryRefreshKind, RefreshKind};
524                        let system = sysinfo::System::new_with_specifics(
525                            RefreshKind::nothing()
526                                .with_memory(MemoryRefreshKind::nothing().with_ram()),
527                        );
528                        system.total_memory().to_string()
529                    }),
530                ),
531            ];
532            for (key, make_value) in defaults {
533                props.entry(key.to_string()).or_insert_with(make_value);
534            }
535        }
536
537        pub fn build(mut self) -> anyhow::Result<TelemetryGuard> {
538            // Merged before the defaults below, which use `or_insert_with`: that
539            // orders precedence as explicit builder calls > env > derived defaults.
540            // `var_os` rather than `var`: a set-but-non-UTF-8 value would otherwise be
541            // indistinguishable from an unset one and silently ignored, which is the
542            // one shape of bad input the parsing below is here to make loud.
543            if let Some(raw) = std::env::var_os(PROCESS_PROPERTIES_ENV_VAR) {
544                merge_process_properties(&mut self.process_properties, &raw.to_string_lossy())?;
545            }
546            if self.default_system_properties_enabled {
547                self.populate_default_system_properties();
548            }
549            // Resolved up front, before the global guard is created below, so a
550            // bad value fails `build()` even when the sink it feeds ends up
551            // unused (e.g. a telemetry knob set with no telemetry URL
552            // configured). A typo should not hide until the URL is set.
553            const LEVEL_EXPECTED: &str = "one of off, fatal, error, warn, info, debug, trace";
554            let local_sink_max_level = resolve(
555                self.local_sink_max_level,
556                LOCAL_SINK_MAX_LEVEL_ENV_VAR,
557                DEFAULT_LOCAL_SINK_MAX_LEVEL,
558                LEVEL_EXPECTED,
559            )?;
560            let telemetry_sink_max_level = resolve(
561                self.telemetry_sink_max_level,
562                TELEMETRY_SINK_MAX_LEVEL_ENV_VAR,
563                DEFAULT_TELEMETRY_SINK_MAX_LEVEL,
564                LEVEL_EXPECTED,
565            )?;
566            const NON_NEGATIVE_INTEGER: &str = "a non-negative integer";
567            let max_queue_bytes = resolve(
568                self.telemetry_max_queue_bytes,
569                MAX_QUEUE_BYTES_ENV_VAR,
570                HttpSinkConfig::DEFAULT_MAX_QUEUE_BYTES,
571                NON_NEGATIVE_INTEGER,
572            )?;
573            let hard_queue_bytes = resolve(
574                self.telemetry_hard_queue_bytes,
575                HARD_QUEUE_BYTES_ENV_VAR,
576                HttpSinkConfig::DEFAULT_HARD_QUEUE_BYTES,
577                NON_NEGATIVE_INTEGER,
578            )?;
579            let max_in_flight_requests = resolve(
580                self.telemetry_max_in_flight_requests,
581                MAX_IN_FLIGHT_REQUESTS_ENV_VAR,
582                HttpSinkConfig::DEFAULT_MAX_IN_FLIGHT_REQUESTS,
583                NON_NEGATIVE_INTEGER,
584            )?;
585            // `Duration` isn't `FromStr`, so this can't go through `resolve`.
586            let request_timeout = match self.telemetry_request_timeout {
587                Some(timeout) => timeout,
588                None => env_override::<u64>(REQUEST_TIMEOUT_SECS_ENV_VAR, NON_NEGATIVE_INTEGER)?
589                    .map(std::time::Duration::from_secs)
590                    .unwrap_or(HttpSinkConfig::DEFAULT_REQUEST_TIMEOUT),
591            };
592            let target_max_level: Vec<_> = self
593                .target_max_levels
594                .into_iter()
595                .filter(|(key, _val)| key != "MAX_LEVEL")
596                .map(|(key, val)| {
597                    (
598                        key,
599                        LevelFilter::from_str(val.as_str()).unwrap_or(LevelFilter::Off),
600                    )
601                })
602                .collect();
603
604            let guard = {
605                lazy_static::lazy_static! {
606                    static ref GLOBAL_WEAK_GUARD: Mutex<Weak<TracingSystemGuard>> = Mutex::new(Weak::new());
607                }
608                let mut weak_guard = GLOBAL_WEAK_GUARD.lock().unwrap();
609                let weak = &mut *weak_guard;
610
611                if let Some(arc) = weak.upgrade() {
612                    arc
613                } else {
614                    let mut sinks: Vec<(LevelFilter, BoxedEventSink)> = vec![];
615                    let telemetry_sink_url = self
616                        .telemetry_sink_url
617                        .or_else(|| std::env::var("MICROMEGAS_TELEMETRY_URL").ok())
618                        .filter(|url| !url.trim().is_empty());
619
620                    if let Some(url) = telemetry_sink_url {
621                        let retry_by_priority = self
622                            .telemetry_retry_by_priority
623                            .unwrap_or_else(HttpSinkConfig::default_retry_by_priority);
624                        let config = HttpSinkConfig {
625                            max_queue_bytes,
626                            hard_queue_bytes,
627                            max_in_flight_requests,
628                            request_timeout,
629                            retry_by_priority,
630                        };
631                        sinks.push((
632                            telemetry_sink_max_level,
633                            Box::new(HttpEventSink::new(
634                                &url,
635                                config,
636                                self.telemetry_make_request_decorator,
637                            )),
638                        ));
639                    }
640                    if self.local_sink_enabled {
641                        sinks.push((local_sink_max_level, Box::new(LocalEventSink::new())));
642                    }
643                    let mut extra_sinks = self.extra_sinks.into_values().collect();
644                    sinks.append(&mut extra_sinks);
645
646                    let sink: BoxedEventSink = Box::new(CompositeSink::new(
647                        sinks,
648                        target_max_level,
649                        self.max_level_override,
650                    ));
651
652                    // the composite sink inits micromegas_tracing::levels::set_max_level, which install_log_interop needs
653                    if self.install_log_capture {
654                        install_log_interop(self.interop_max_level_override);
655                    }
656                    if self.install_tracing_capture {
657                        install_tracing_interop(self.interop_max_level_override);
658                    }
659
660                    let arc = Arc::<TracingSystemGuard>::new(TracingSystemGuard::new(
661                        self.logs_buffer_size,
662                        self.metrics_buffer_size,
663                        self.threads_buffer_size,
664                        sink.into(),
665                        self.process_properties,
666                        std::env::var("MICROMEGAS_ENABLE_CPU_TRACING")
667                            .map(|v| v == "true")
668                            .unwrap_or(false), // Default to disabled for minimal overhead
669                    )?);
670
671                    if self.system_metrics_enabled {
672                        spawn_system_monitor();
673                    }
674
675                    *weak = Arc::<TracingSystemGuard>::downgrade(&arc);
676                    arc
677                }
678            };
679            // order here is important
680            Ok(TelemetryGuard {
681                _guard: guard,
682                _thread_guard: TracingThreadGuard::new(),
683            })
684        }
685    }
686
687    pub struct TelemetryGuard {
688        // note we rely here on the drop order being the same as the declaration order
689        _thread_guard: TracingThreadGuard,
690        _guard: Arc<TracingSystemGuard>,
691    }
692
693    impl TelemetryGuard {
694        pub fn new() -> anyhow::Result<Self> {
695            TelemetryGuardBuilder::default().build()
696        }
697    }
698}
699
700#[cfg(not(target_arch = "wasm32"))]
701pub use native::*;
702
703// ---- Wasm implementation ----
704#[cfg(target_arch = "wasm32")]
705mod wasm {
706    use std::collections::HashMap;
707    use std::sync::Arc;
708
709    use micromegas_tracing::guards::TracingSystemGuard;
710
711    use crate::ConsoleEventSink;
712
713    pub struct TelemetryGuard {
714        _guard: Arc<TracingSystemGuard>,
715    }
716
717    pub fn init_telemetry() -> anyhow::Result<TelemetryGuard> {
718        let guard = Arc::new(TracingSystemGuard::new(
719            0,
720            0,
721            0,
722            Arc::new(ConsoleEventSink),
723            HashMap::new(),
724            false,
725        )?);
726        Ok(TelemetryGuard { _guard: guard })
727    }
728}
729
730#[cfg(target_arch = "wasm32")]
731pub use wasm::*;
micromegas_telemetry_sink/lib.rs

micromegas_telemetry_sink/
lib.rs
