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"))]
19pub mod http_event_sink;
20#[cfg(not(target_arch = "wasm32"))]
21pub mod local_event_sink;
22#[cfg(not(target_arch = "wasm32"))]
23pub mod log_interop;
24#[cfg(not(target_arch = "wasm32"))]
25pub mod oidc_client_credentials_decorator;
26#[cfg(not(target_arch = "wasm32"))]
27pub mod request_decorator;
28#[cfg(not(target_arch = "wasm32"))]
29pub mod stream_block;
30#[cfg(not(target_arch = "wasm32"))]
31pub mod stream_info;
32#[cfg(not(target_arch = "wasm32"))]
33pub mod system_monitor;
34#[cfg(not(target_arch = "wasm32"))]
35pub mod tracing_interop;
36
37// ---- Wasm-only modules ----
38#[cfg(target_arch = "wasm32")]
39mod console_event_sink;
40#[cfg(target_arch = "wasm32")]
41pub use console_event_sink::*;
42
43// ---- Native implementation ----
44#[cfg(not(target_arch = "wasm32"))]
45mod native {
46    use std::any::TypeId;
47    use std::collections::HashMap;
48    use std::str::FromStr;
49    use std::sync::{Arc, Mutex, Weak};
50
51    use crate::log_interop::install_log_interop;
52    use crate::request_decorator::RequestDecorator;
53    use crate::tracing_interop::install_tracing_interop;
54    use micromegas_tracing::event::BoxedEventSink;
55    use micromegas_tracing::info;
56    use micromegas_tracing::{
57        event::EventSink,
58        guards::{TracingSystemGuard, TracingThreadGuard},
59        prelude::*,
60    };
61
62    use crate::composite_event_sink::CompositeSink;
63    use crate::http_event_sink::{HttpEventSink, HttpSinkConfig};
64    use crate::local_event_sink::LocalEventSink;
65    use crate::system_monitor::spawn_system_monitor;
66
67    pub mod tokio_retry {
68        pub use tokio_retry2::*;
69    }
70
71    pub mod reqwest {
72        pub use reqwest::*;
73    }
74
75    pub struct TelemetryGuardBuilder {
76        logs_buffer_size: usize,
77        metrics_buffer_size: usize,
78        threads_buffer_size: usize,
79        target_max_levels: HashMap<String, String>,
80        max_level_override: Option<LevelFilter>,
81        interop_max_level_override: Option<LevelFilter>,
82        install_log_capture: bool,
83        install_tracing_capture: bool,
84        local_sink_enabled: bool,
85        local_sink_max_level: LevelFilter,
86        telemetry_sink_url: Option<String>,
87        telemetry_sink_max_level: LevelFilter,
88        telemetry_max_queue_bytes: Option<usize>,
89        telemetry_hard_queue_bytes: Option<usize>,
90        telemetry_max_in_flight_requests: Option<usize>,
91        telemetry_request_timeout: Option<std::time::Duration>,
92        telemetry_retry_by_priority:
93            Option<[core::iter::Take<tokio_retry::strategy::ExponentialBackoff>; 4]>,
94        telemetry_make_request_decorator: Box<dyn FnOnce() -> Arc<dyn RequestDecorator> + Send>,
95        extra_sinks: HashMap<TypeId, (LevelFilter, BoxedEventSink)>,
96        system_metrics_enabled: bool,
97        default_system_properties_enabled: bool,
98        process_properties: HashMap<String, String>,
99    }
100
101    impl Default for TelemetryGuardBuilder {
102        fn default() -> Self {
103            Self {
104                logs_buffer_size: 10 * 1024 * 1024,
105                metrics_buffer_size: 1024 * 1024,
106                threads_buffer_size: 10 * 1024 * 1024,
107                local_sink_enabled: true,
108                local_sink_max_level: LevelFilter::Info,
109                telemetry_sink_url: None,
110                telemetry_sink_max_level: LevelFilter::Debug,
111                telemetry_max_queue_bytes: None,
112                telemetry_hard_queue_bytes: None,
113                telemetry_max_in_flight_requests: None,
114                telemetry_request_timeout: None,
115                telemetry_retry_by_priority: None,
116                telemetry_make_request_decorator: Box::new(|| {
117                    Arc::new(crate::request_decorator::TrivialRequestDecorator {})
118                }),
119                target_max_levels: HashMap::default(),
120                max_level_override: None,
121                interop_max_level_override: None,
122                install_log_capture: false,
123                install_tracing_capture: true,
124                extra_sinks: HashMap::default(),
125                system_metrics_enabled: true,
126                default_system_properties_enabled: true,
127                process_properties: HashMap::default(),
128            }
129        }
130    }
131
132    impl TelemetryGuardBuilder {
133        // Only one sink per type ?
134        #[must_use]
135        pub fn add_sink<Sink>(mut self, max_level: LevelFilter, sink: Sink) -> Self
136        where
137            Sink: EventSink + 'static,
138        {
139            let type_id = TypeId::of::<Sink>();
140
141            self.extra_sinks
142                .entry(type_id)
143                .or_insert_with(|| (max_level, Box::new(sink)));
144
145            self
146        }
147
148        /// Programmatic override
149        #[must_use]
150        pub fn with_max_level_override(mut self, level_filter: LevelFilter) -> Self {
151            self.max_level_override = Some(level_filter);
152            self
153        }
154
155        #[must_use]
156        pub fn with_local_sink_enabled(mut self, enabled: bool) -> Self {
157            self.local_sink_enabled = enabled;
158            self
159        }
160
161        #[must_use]
162        pub fn with_interop_max_level_override(mut self, level_filter: LevelFilter) -> Self {
163            self.interop_max_level_override = Some(level_filter);
164            self
165        }
166
167        #[must_use]
168        pub fn with_install_log_capture(mut self, enabled: bool) -> Self {
169            self.install_log_capture = enabled;
170            self
171        }
172
173        #[must_use]
174        pub fn with_install_tracing_capture(mut self, enabled: bool) -> Self {
175            self.install_tracing_capture = enabled;
176            self
177        }
178
179        #[must_use]
180        pub fn with_local_sink_max_level(mut self, level_filter: LevelFilter) -> Self {
181            self.local_sink_max_level = level_filter;
182            self
183        }
184
185        #[must_use]
186        pub fn with_ctrlc_handling(self) -> Self {
187            ctrlc::set_handler(move || {
188                info!("Ctrl+C was hit!");
189                micromegas_tracing::guards::shutdown_telemetry();
190                std::process::exit(1);
191            })
192            .expect("Error setting Ctrl+C handler");
193            self
194        }
195
196        /// Soft byte cap for the telemetry upload queue: once reached, new
197        /// `Traces` items (thread and image blocks) are dropped first.
198        /// Falls back to `MICROMEGAS_TELEMETRY_MAX_QUEUE_BYTES`, then
199        /// [`HttpSinkConfig::DEFAULT_MAX_QUEUE_BYTES`].
200        #[must_use]
201        pub fn with_max_queue_bytes(mut self, bytes: usize) -> Self {
202            self.telemetry_max_queue_bytes = Some(bytes);
203            self
204        }
205
206        /// Hard byte cap for the telemetry upload queue: once reached, `Logs`
207        /// and `Metrics` items are dropped too (`Metadata` is never
208        /// dropped). Falls back to `MICROMEGAS_TELEMETRY_HARD_QUEUE_BYTES`,
209        /// then [`HttpSinkConfig::DEFAULT_HARD_QUEUE_BYTES`].
210        #[must_use]
211        pub fn with_hard_queue_bytes(mut self, bytes: usize) -> Self {
212            self.telemetry_hard_queue_bytes = Some(bytes);
213            self
214        }
215
216        /// Maximum number of `insert_*` HTTP requests in flight at once.
217        /// Falls back to `MICROMEGAS_TELEMETRY_MAX_IN_FLIGHT_REQUESTS`, then
218        /// [`HttpSinkConfig::DEFAULT_MAX_IN_FLIGHT_REQUESTS`].
219        #[must_use]
220        pub fn with_max_in_flight_requests(mut self, max_in_flight_requests: usize) -> Self {
221            self.telemetry_max_in_flight_requests = Some(max_in_flight_requests);
222            self
223        }
224
225        /// Per-request timeout (covers connect + send + receive for one
226        /// attempt). Bounds how long a single retry attempt can hang against
227        /// an ingestion service that accepts connections but never responds,
228        /// which otherwise would make shutdown block indefinitely (`Drop for
229        /// HttpEventSink` joins the worker thread). Falls back to
230        /// `MICROMEGAS_TELEMETRY_REQUEST_TIMEOUT_SECS`, then
231        /// [`HttpSinkConfig::DEFAULT_REQUEST_TIMEOUT`].
232        #[must_use]
233        pub fn with_request_timeout(mut self, timeout: std::time::Duration) -> Self {
234            self.telemetry_request_timeout = Some(timeout);
235            self
236        }
237
238        /// Retry strategy per upload priority (indexed by
239        /// `UploadPriority as usize`: Metadata, Logs, Metrics, Traces).
240        /// Defaults to [`HttpSinkConfig::default_retry_by_priority`].
241        #[must_use]
242        pub fn with_retry_by_priority(
243            mut self,
244            retry_by_priority: [core::iter::Take<tokio_retry::strategy::ExponentialBackoff>; 4],
245        ) -> Self {
246            self.telemetry_retry_by_priority = Some(retry_by_priority);
247            self
248        }
249
250        #[must_use]
251        pub fn with_request_decorator(
252            mut self,
253            make_decorator: Box<dyn FnOnce() -> Arc<dyn RequestDecorator> + Send>,
254        ) -> Self {
255            self.telemetry_make_request_decorator = make_decorator;
256            self
257        }
258
259        /// Automatically configure authentication from environment variables.
260        ///
261        /// Checks for authentication configuration in this order:
262        /// 1. API key authentication via `MICROMEGAS_INGESTION_API_KEY`
263        /// 2. OIDC client credentials via `MICROMEGAS_OIDC_TOKEN_ENDPOINT`,
264        ///    `MICROMEGAS_OIDC_CLIENT_ID`, and `MICROMEGAS_OIDC_CLIENT_SECRET`
265        /// 3. Falls back to no authentication (TrivialRequestDecorator)
266        ///
267        /// # Example
268        ///
269        /// ```rust,no_run
270        /// use micromegas_telemetry_sink::TelemetryGuardBuilder;
271        ///
272        /// // Set environment variable
273        /// unsafe {
274        ///     std::env::set_var("MICROMEGAS_INGESTION_API_KEY", "secret-key-123");
275        /// }
276        ///
277        /// // Builder automatically configures API key authentication
278        /// let _guard = TelemetryGuardBuilder::default()
279        ///     .with_auth_from_env()
280        ///     .build()
281        ///     .expect("Failed to build telemetry guard");
282        /// ```
283        #[must_use]
284        pub fn with_auth_from_env(mut self) -> Self {
285            use crate::api_key_decorator::ApiKeyRequestDecorator;
286            use crate::oidc_client_credentials_decorator::OidcClientCredentialsDecorator;
287
288            // Try API key authentication first
289            if let Ok(decorator) = ApiKeyRequestDecorator::from_env() {
290                info!("Configured telemetry sink with API key authentication");
291                self.telemetry_make_request_decorator = Box::new(move || Arc::new(decorator));
292                return self;
293            }
294
295            // Try OIDC client credentials authentication
296            if let Ok(decorator) = OidcClientCredentialsDecorator::from_env() {
297                info!("Configured telemetry sink with OIDC client credentials authentication");
298                self.telemetry_make_request_decorator = Box::new(move || Arc::new(decorator));
299                return self;
300            }
301
302            // No authentication configured - use trivial decorator (no-op)
303            info!(
304                "Telemetry sink authentication not configured - sending unauthenticated requests"
305            );
306            self
307        }
308
309        #[must_use]
310        pub fn with_system_metrics_enabled(mut self, enabled: bool) -> Self {
311            self.system_metrics_enabled = enabled;
312            self
313        }
314
315        #[must_use]
316        pub fn with_default_system_properties_enabled(mut self, enabled: bool) -> Self {
317            self.default_system_properties_enabled = enabled;
318            self
319        }
320
321        /// Set the URL of telemetry sink.
322        ///
323        /// If not explicitly set, the URL will be read from the `MICROMEGAS_TELEMETRY_URL` environment
324        /// variable.
325        #[must_use]
326        pub fn with_telemetry_sink_url(mut self, url: String) -> Self {
327            self.telemetry_sink_url = Some(url);
328            self
329        }
330
331        /// Add a single property to the process info.
332        ///
333        /// # Warning
334        ///
335        /// This will override any existing properties.
336        #[must_use]
337        pub fn with_process_property(mut self, key: String, value: String) -> Self {
338            self.process_properties.insert(key, value);
339            self
340        }
341
342        /// Add multiple properties to the process info.
343        ///
344        /// # Warning
345        ///
346        /// This will override any existing properties.
347        #[must_use]
348        pub fn with_process_properties(
349            mut self,
350            process_properties: HashMap<String, String>,
351        ) -> Self {
352            self.process_properties.extend(process_properties);
353            self
354        }
355
356        fn populate_default_system_properties(&mut self) {
357            let props = &mut self.process_properties;
358            // Process identity (duplicates ProcessInfo fields into properties)
359            type DefaultProperties = Vec<(&'static str, Box<dyn FnOnce() -> String>)>;
360            let defaults: DefaultProperties = vec![
361                (
362                    "exe",
363                    Box::new(|| {
364                        std::env::current_exe()
365                            .unwrap_or_default()
366                            .to_string_lossy()
367                            .into_owned()
368                    }),
369                ),
370                ("username", Box::new(whoami::username)),
371                ("realname", Box::new(whoami::realname)),
372                ("computer", Box::new(whoami::devicename)),
373                ("distro", Box::new(whoami::distro)),
374                (
375                    "cpu_brand",
376                    Box::new(|| {
377                        #[cfg(target_arch = "x86_64")]
378                        {
379                            raw_cpuid::CpuId::new()
380                                .get_processor_brand_string()
381                                .map_or_else(|| "unknown".to_owned(), |b| b.as_str().to_owned())
382                        }
383                        #[cfg(not(target_arch = "x86_64"))]
384                        {
385                            String::from(std::env::consts::ARCH)
386                        }
387                    }),
388                ),
389                (
390                    "physical_core_count",
391                    Box::new(|| {
392                        sysinfo::System::physical_core_count()
393                            .map(|c: usize| c.to_string())
394                            .unwrap_or_default()
395                    }),
396                ),
397                (
398                    "logical_cpu_count",
399                    Box::new(|| {
400                        use sysinfo::{CpuRefreshKind, RefreshKind};
401                        let system = sysinfo::System::new_with_specifics(
402                            RefreshKind::nothing().with_cpu(CpuRefreshKind::nothing()),
403                        );
404                        system.cpus().len().to_string()
405                    }),
406                ),
407                (
408                    "total_memory",
409                    Box::new(|| {
410                        use sysinfo::{MemoryRefreshKind, RefreshKind};
411                        let system = sysinfo::System::new_with_specifics(
412                            RefreshKind::nothing()
413                                .with_memory(MemoryRefreshKind::nothing().with_ram()),
414                        );
415                        system.total_memory().to_string()
416                    }),
417                ),
418            ];
419            for (key, make_value) in defaults {
420                props.entry(key.to_string()).or_insert_with(make_value);
421            }
422        }
423
424        pub fn build(mut self) -> anyhow::Result<TelemetryGuard> {
425            if self.default_system_properties_enabled {
426                self.populate_default_system_properties();
427            }
428            let target_max_level: Vec<_> = self
429                .target_max_levels
430                .into_iter()
431                .filter(|(key, _val)| key != "MAX_LEVEL")
432                .map(|(key, val)| {
433                    (
434                        key,
435                        LevelFilter::from_str(val.as_str()).unwrap_or(LevelFilter::Off),
436                    )
437                })
438                .collect();
439
440            let guard = {
441                lazy_static::lazy_static! {
442                    static ref GLOBAL_WEAK_GUARD: Mutex<Weak<TracingSystemGuard>> = Mutex::new(Weak::new());
443                }
444                let mut weak_guard = GLOBAL_WEAK_GUARD.lock().unwrap();
445                let weak = &mut *weak_guard;
446
447                if let Some(arc) = weak.upgrade() {
448                    arc
449                } else {
450                    let mut sinks: Vec<(LevelFilter, BoxedEventSink)> = vec![];
451                    let telemetry_sink_url = self
452                        .telemetry_sink_url
453                        .or_else(|| std::env::var("MICROMEGAS_TELEMETRY_URL").ok())
454                        .filter(|url| !url.trim().is_empty());
455
456                    if let Some(url) = telemetry_sink_url {
457                        let max_queue_bytes = self
458                            .telemetry_max_queue_bytes
459                            .or_else(|| {
460                                std::env::var("MICROMEGAS_TELEMETRY_MAX_QUEUE_BYTES")
461                                    .ok()
462                                    .and_then(|v| v.parse().ok())
463                            })
464                            .unwrap_or(HttpSinkConfig::DEFAULT_MAX_QUEUE_BYTES);
465                        let hard_queue_bytes = self
466                            .telemetry_hard_queue_bytes
467                            .or_else(|| {
468                                std::env::var("MICROMEGAS_TELEMETRY_HARD_QUEUE_BYTES")
469                                    .ok()
470                                    .and_then(|v| v.parse().ok())
471                            })
472                            .unwrap_or(HttpSinkConfig::DEFAULT_HARD_QUEUE_BYTES);
473                        let max_in_flight_requests = self
474                            .telemetry_max_in_flight_requests
475                            .or_else(|| {
476                                std::env::var("MICROMEGAS_TELEMETRY_MAX_IN_FLIGHT_REQUESTS")
477                                    .ok()
478                                    .and_then(|v| v.parse().ok())
479                            })
480                            .unwrap_or(HttpSinkConfig::DEFAULT_MAX_IN_FLIGHT_REQUESTS);
481                        let request_timeout = self
482                            .telemetry_request_timeout
483                            .or_else(|| {
484                                std::env::var("MICROMEGAS_TELEMETRY_REQUEST_TIMEOUT_SECS")
485                                    .ok()
486                                    .and_then(|v| v.parse().ok())
487                                    .map(std::time::Duration::from_secs)
488                            })
489                            .unwrap_or(HttpSinkConfig::DEFAULT_REQUEST_TIMEOUT);
490                        let retry_by_priority = self
491                            .telemetry_retry_by_priority
492                            .unwrap_or_else(HttpSinkConfig::default_retry_by_priority);
493                        let config = HttpSinkConfig {
494                            max_queue_bytes,
495                            hard_queue_bytes,
496                            max_in_flight_requests,
497                            request_timeout,
498                            retry_by_priority,
499                        };
500                        sinks.push((
501                            self.telemetry_sink_max_level,
502                            Box::new(HttpEventSink::new(
503                                &url,
504                                config,
505                                self.telemetry_make_request_decorator,
506                            )),
507                        ));
508                    }
509                    if self.local_sink_enabled {
510                        sinks.push((self.local_sink_max_level, Box::new(LocalEventSink::new())));
511                    }
512                    let mut extra_sinks = self.extra_sinks.into_values().collect();
513                    sinks.append(&mut extra_sinks);
514
515                    let sink: BoxedEventSink = Box::new(CompositeSink::new(
516                        sinks,
517                        target_max_level,
518                        self.max_level_override,
519                    ));
520
521                    // the composite sink inits micromegas_tracing::levels::set_max_level, which install_log_interop needs
522                    if self.install_log_capture {
523                        install_log_interop(self.interop_max_level_override);
524                    }
525                    if self.install_tracing_capture {
526                        install_tracing_interop(self.interop_max_level_override);
527                    }
528
529                    let arc = Arc::<TracingSystemGuard>::new(TracingSystemGuard::new(
530                        self.logs_buffer_size,
531                        self.metrics_buffer_size,
532                        self.threads_buffer_size,
533                        sink.into(),
534                        self.process_properties,
535                        std::env::var("MICROMEGAS_ENABLE_CPU_TRACING")
536                            .map(|v| v == "true")
537                            .unwrap_or(false), // Default to disabled for minimal overhead
538                    )?);
539
540                    if self.system_metrics_enabled {
541                        spawn_system_monitor();
542                    }
543
544                    *weak = Arc::<TracingSystemGuard>::downgrade(&arc);
545                    arc
546                }
547            };
548            // order here is important
549            Ok(TelemetryGuard {
550                _guard: guard,
551                _thread_guard: TracingThreadGuard::new(),
552            })
553        }
554    }
555
556    pub struct TelemetryGuard {
557        // note we rely here on the drop order being the same as the declaration order
558        _thread_guard: TracingThreadGuard,
559        _guard: Arc<TracingSystemGuard>,
560    }
561
562    impl TelemetryGuard {
563        pub fn new() -> anyhow::Result<Self> {
564            TelemetryGuardBuilder::default().build()
565        }
566    }
567}
568
569#[cfg(not(target_arch = "wasm32"))]
570pub use native::*;
571
572// ---- Wasm implementation ----
573#[cfg(target_arch = "wasm32")]
574mod wasm {
575    use std::collections::HashMap;
576    use std::sync::Arc;
577
578    use micromegas_tracing::guards::TracingSystemGuard;
579
580    use crate::ConsoleEventSink;
581
582    pub struct TelemetryGuard {
583        _guard: Arc<TracingSystemGuard>,
584    }
585
586    pub fn init_telemetry() -> anyhow::Result<TelemetryGuard> {
587        let guard = Arc::new(TracingSystemGuard::new(
588            0,
589            0,
590            0,
591            Arc::new(ConsoleEventSink),
592            HashMap::new(),
593            false,
594        )?);
595        Ok(TelemetryGuard { _guard: guard })
596    }
597}
598
599#[cfg(target_arch = "wasm32")]
600pub use wasm::*;
micromegas_telemetry_sink/lib.rs

micromegas_telemetry_sink/
lib.rs
