1use serde::{Deserialize, Serialize};
7
8use crate::config::{RuntimeOverrides, TelemetryConfig};
9use crate::errors::TelemetryError;
10use crate::otel::DrainOutcome;
11
12#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
13pub struct SignalStatus {
14 pub logs: bool,
15 pub traces: bool,
16 pub metrics: bool,
17}
18
19#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
20pub enum ProviderMode {
21 Owned,
22 Host,
23 Local,
24}
25
26#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
27pub enum RuntimeState {
28 Local,
29 Starting,
30 Ready,
31 Degraded,
32 Reconfiguring,
33 Stopping,
34 Stopped,
35}
36
37#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
38pub struct SignalFlushResult {
39 pub flushed: bool,
40 pub not_installed: bool,
41 pub not_owned: bool,
42 pub timed_out: bool,
43 pub failed: bool,
44}
45
46#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
47pub struct FlushResult {
48 pub logs: SignalFlushResult,
49 pub traces: SignalFlushResult,
50 pub metrics: SignalFlushResult,
51}
52
53#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
54pub struct ReconfigureResult {
55 pub applied: bool,
56 pub previous: Option<TelemetryConfig>,
57 pub current: Option<TelemetryConfig>,
58 pub error: Option<String>,
59 pub state: RuntimeState,
60}
61
62#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
63pub struct RuntimeStatus {
64 pub setup_done: bool,
65 pub signals: SignalStatus,
66 pub providers: SignalStatus,
67 pub fallback: SignalStatus,
68 pub setup_error: Option<String>,
69}
70
71#[allow(non_camel_case_types)]
72pub type provider_mode = ProviderMode;
73#[allow(non_camel_case_types)]
74pub type runtime_state = RuntimeState;
75#[allow(non_camel_case_types)]
76pub type signal_flush_result = SignalFlushResult;
77#[allow(non_camel_case_types)]
78pub type flush_result = FlushResult;
79#[allow(non_camel_case_types)]
80pub type reconfigure_result = ReconfigureResult;
81#[allow(non_camel_case_types)]
82pub type telemetry_config = TelemetryConfig;
83#[allow(non_camel_case_types)]
84pub type telemetry_runtime = TelemetryRuntime;
85#[allow(non_camel_case_types)]
86pub type runtime_status = RuntimeStatus;
87
88fn any_owned_drain_abandoned(result: &FlushResult) -> bool {
93 result.logs.timed_out || result.traces.timed_out || result.metrics.timed_out
94}
95
96fn any_owned_drain_rejected(result: &FlushResult) -> bool {
99 result.logs.failed || result.traces.failed || result.metrics.failed
100}
101
102fn signal_flush_result(installed: bool, owned: bool, outcome: DrainOutcome) -> SignalFlushResult {
118 if !installed {
119 return SignalFlushResult {
120 not_installed: true,
121 ..SignalFlushResult::default()
122 };
123 }
124 if !owned {
125 return SignalFlushResult {
126 not_owned: true,
127 ..SignalFlushResult::default()
128 };
129 }
130 SignalFlushResult {
131 flushed: outcome == DrainOutcome::Drained,
132 timed_out: outcome == DrainOutcome::TimedOut,
133 failed: outcome == DrainOutcome::Failed,
134 ..SignalFlushResult::default()
135 }
136}
137
138pub struct TelemetryRuntime {
139 provider_mode: ProviderMode,
140 state: RuntimeState,
141}
142
143impl Default for TelemetryRuntime {
144 fn default() -> Self {
145 Self::new()
146 }
147}
148
149impl TelemetryRuntime {
150 pub fn new() -> Self {
151 Self {
152 provider_mode: ProviderMode::Owned,
153 state: RuntimeState::Ready,
154 }
155 }
156
157 pub fn start(
162 &mut self,
163 config: Option<TelemetryConfig>,
164 ) -> Result<TelemetryConfig, TelemetryError> {
165 self.state = RuntimeState::Starting;
166 match crate::setup::setup_telemetry(config) {
167 Ok(cfg) => {
168 self.state = RuntimeState::Ready;
169 Ok(cfg)
170 }
171 Err(err) => {
172 self.state = RuntimeState::Degraded;
173 Err(err)
174 }
175 }
176 }
177
178 pub fn shutdown(&mut self, timeout_seconds: Option<f64>) -> Result<(), TelemetryError> {
180 self.state = RuntimeState::Stopping;
181 let result = crate::setup::shutdown_telemetry(timeout_seconds);
182 self.state = RuntimeState::Stopped;
183 result
184 }
185
186 pub fn flush(&self, timeout_seconds: Option<f64>) -> Result<FlushResult, TelemetryError> {
201 let providers = crate::runtime::get_runtime_status().providers;
202 let owned = crate::otel::owned_signals();
203 let drained = crate::otel::flush_otel_by_signal(timeout_seconds);
204 let result = FlushResult {
205 logs: signal_flush_result(providers.logs, owned.logs, drained.logs),
206 traces: signal_flush_result(providers.traces, owned.traces, drained.traces),
207 metrics: signal_flush_result(providers.metrics, owned.metrics, drained.metrics),
208 };
209 if any_owned_drain_abandoned(&result) {
210 return Err(TelemetryError::new(
211 "telemetry flush exceeded its deadline; records may not have been exported",
212 ));
213 }
214 if any_owned_drain_rejected(&result) {
215 return Err(TelemetryError::new(
216 "telemetry flush failed: an exporter rejected the drain; records may not have been exported",
217 ));
218 }
219 Ok(result)
220 }
221
222 pub fn get_logger(&self, name: Option<&str>) -> crate::logger::Logger {
223 crate::logger::get_logger(name)
224 }
225
226 pub fn get_tracer(&self, name: Option<&str>) -> crate::tracer::Tracer {
227 crate::tracing::get_tracer(name)
228 }
229
230 pub fn get_meter(&self, name: Option<&str>) -> crate::metrics::Meter {
231 crate::metrics::get_meter(name)
232 }
233
234 pub fn get_runtime_config(&self) -> Option<TelemetryConfig> {
235 crate::runtime::get_runtime_config()
236 }
237
238 pub fn get_runtime_status(&self) -> RuntimeStatus {
239 crate::runtime::get_runtime_status()
240 }
241
242 pub fn update_config(&mut self, overrides: RuntimeOverrides) -> ReconfigureResult {
243 let previous = crate::runtime::get_runtime_config();
244 let next = match crate::runtime::update_runtime_config(overrides) {
245 Ok(cfg) => Some(cfg),
246 Err(err) => {
247 return ReconfigureResult {
248 applied: false,
249 previous,
250 current: None,
251 error: Some(err.message),
252 state: self.state,
253 };
254 }
255 };
256 ReconfigureResult {
257 applied: true,
258 previous,
259 current: next,
260 error: None,
261 state: self.state,
262 }
263 }
264
265 pub fn reconfigure(
266 &mut self,
267 config: Option<TelemetryConfig>,
268 ) -> Result<TelemetryConfig, TelemetryError> {
269 crate::runtime::reconfigure_telemetry(config)
270 }
271
272 pub fn provider_mode(&self) -> ProviderMode {
273 self.provider_mode
274 }
275
276 pub fn state(&self) -> RuntimeState {
277 self.state
278 }
279}
280
281#[cfg(test)]
282mod signal_flush_result_tests {
283 use super::{
284 any_owned_drain_abandoned, any_owned_drain_rejected, signal_flush_result, DrainOutcome,
285 FlushResult, SignalFlushResult,
286 };
287
288 fn drained() -> SignalFlushResult {
289 SignalFlushResult {
290 flushed: true,
291 ..SignalFlushResult::default()
292 }
293 }
294
295 fn abandoned() -> SignalFlushResult {
296 SignalFlushResult {
297 timed_out: true,
298 ..SignalFlushResult::default()
299 }
300 }
301
302 #[test]
306 fn one_abandoned_signal_is_enough_to_report_an_abandoned_drain() {
307 let clean = FlushResult {
308 logs: drained(),
309 traces: drained(),
310 metrics: drained(),
311 };
312 assert!(!any_owned_drain_abandoned(&clean));
313
314 for signal in ["logs", "traces", "metrics"] {
315 let mut result = clean.clone();
316 match signal {
317 "logs" => result.logs = abandoned(),
318 "traces" => result.traces = abandoned(),
319 _ => result.metrics = abandoned(),
320 }
321 assert!(
322 any_owned_drain_abandoned(&result),
323 "an abandoned {signal} drain must be reported"
324 );
325 }
326 }
327
328 #[test]
331 fn not_installed_and_not_owned_signals_are_not_abandoned_drains() {
332 let result = FlushResult {
333 logs: SignalFlushResult {
334 not_installed: true,
335 ..SignalFlushResult::default()
336 },
337 traces: SignalFlushResult {
338 not_owned: true,
339 ..SignalFlushResult::default()
340 },
341 metrics: drained(),
342 };
343 assert!(!any_owned_drain_abandoned(&result));
344 }
345
346 #[test]
350 fn covers_every_combination() {
351 let cases: [(bool, bool, DrainOutcome, SignalFlushResult); 6] = [
352 (
353 false,
354 false,
355 DrainOutcome::Drained,
356 SignalFlushResult {
357 not_installed: true,
358 ..SignalFlushResult::default()
359 },
360 ),
361 (
362 false,
363 true,
364 DrainOutcome::Drained,
365 SignalFlushResult {
366 not_installed: true,
367 ..SignalFlushResult::default()
368 },
369 ),
370 (
371 true,
372 false,
373 DrainOutcome::Drained,
374 SignalFlushResult {
375 not_owned: true,
376 ..SignalFlushResult::default()
377 },
378 ),
379 (
380 true,
381 true,
382 DrainOutcome::Drained,
383 SignalFlushResult {
384 flushed: true,
385 ..SignalFlushResult::default()
386 },
387 ),
388 (
389 true,
390 true,
391 DrainOutcome::Failed,
392 SignalFlushResult {
393 failed: true,
394 ..SignalFlushResult::default()
395 },
396 ),
397 (
398 true,
399 true,
400 DrainOutcome::TimedOut,
401 SignalFlushResult {
402 timed_out: true,
403 ..SignalFlushResult::default()
404 },
405 ),
406 ];
407
408 for (installed, owned, outcome, want) in cases {
409 let got = signal_flush_result(installed, owned, outcome);
410 assert_eq!(
411 got, want,
412 "installed={installed} owned={owned} outcome={outcome:?}"
413 );
414 }
415 }
416
417 #[test]
420 fn not_installed_takes_precedence_over_not_owned() {
421 let got = signal_flush_result(false, false, DrainOutcome::TimedOut);
422 assert!(got.not_installed);
423 assert!(!got.not_owned);
424 assert!(!got.flushed);
425 assert!(!got.timed_out);
426 }
427
428 #[test]
431 fn a_missed_deadline_is_never_reported_as_flushed() {
432 let got = signal_flush_result(true, true, DrainOutcome::TimedOut);
433 assert!(!got.flushed);
434 assert!(got.timed_out);
435 assert!(!got.failed);
436 }
437
438 #[test]
442 fn an_in_deadline_rejection_is_failed_never_timed_out() {
443 let got = signal_flush_result(true, true, DrainOutcome::Failed);
444 assert!(!got.flushed);
445 assert!(!got.timed_out);
446 assert!(got.failed);
447 }
448
449 #[test]
452 fn one_rejected_signal_is_enough_to_report_a_rejected_drain() {
453 let clean = FlushResult {
454 logs: drained(),
455 traces: drained(),
456 metrics: drained(),
457 };
458 assert!(!any_owned_drain_rejected(&clean));
459
460 let rejected = SignalFlushResult {
461 failed: true,
462 ..SignalFlushResult::default()
463 };
464 for signal in ["logs", "traces", "metrics"] {
465 let mut result = clean.clone();
466 match signal {
467 "logs" => result.logs = rejected,
468 "traces" => result.traces = rejected,
469 _ => result.metrics = rejected,
470 }
471 assert!(
472 any_owned_drain_rejected(&result),
473 "a rejected {signal} drain must be reported"
474 );
475 }
476 }
477}