kaizen/telemetry/
mod.rs

1// SPDX-License-Identifier: AGPL-3.0-or-later
2//! Optional pluggable sinks that receive the same redacted [`IngestExportBatch`] as Kaizen sync.
3//! Fan-out runs in parallel with the primary `POST` (see `sync::engine`); outbox is committed only
4//! when the primary succeeds (and, when `fail_open` is `false`, when the fan-out completes `Ok`).
5
6mod batch_metadata;
7mod file;
8mod resolve;
9
10#[cfg(feature = "telemetry-datadog")]
11mod datadog;
12#[cfg(feature = "telemetry-dev")]
13mod dev;
14#[cfg(feature = "telemetry-otlp")]
15mod otlp;
16#[cfg(feature = "telemetry-posthog")]
17mod posthog;
18
19use crate::core::config::{ExporterConfig, TelemetryConfig};
20use crate::sync::IngestExportBatch;
21use anyhow::Result;
22use std::path::Path;
23use std::sync::Arc;
24
25pub use batch_metadata::telemetry_file_line;
26pub use file::{FileExporter, default_ndjson_path, resolve_file_exporter_path};
27
28pub use resolve::DatadogResolved;
29pub use resolve::OtlpResolved;
30pub use resolve::PostHogResolved;
31
32/// Third-party and OTel sinks use the same batch types as the HTTP ingest.
33pub trait TelemetryExporter: Send + Sync {
34    fn name(&self) -> &str;
35    fn export(&self, batch: &IngestExportBatch) -> Result<()>;
36}
37
38/// Built from `TelemetryConfig` via [`load_exporters`]. Empty is a no-op.
39pub struct ExporterRegistry {
40    exporters: Vec<Arc<dyn TelemetryExporter>>,
41}
42
43impl ExporterRegistry {
44    pub fn empty() -> Self {
45        Self {
46            exporters: Vec::new(),
47        }
48    }
49
50    pub fn is_empty(&self) -> bool {
51        self.exporters.is_empty()
52    }
53
54    pub fn from_vec(exporters: Vec<Arc<dyn TelemetryExporter>>) -> Self {
55        Self { exporters }
56    }
57
58    /// When `fail_open` is `true`, log each exporter error and continue. If `false`, return the first error.
59    pub fn fan_out(&self, fail_open: bool, batch: &IngestExportBatch) -> Result<()> {
60        for e in &self.exporters {
61            let r = e.export(batch);
62            if let Err(err) = r {
63                tracing::warn!(exporter = e.name(), error = %err, "telemetry exporter");
64                if !fail_open {
65                    return Err(err);
66                }
67            }
68        }
69        Ok(())
70    }
71}
72
73/// Build exporters from TOML + environment. Missing creds for a sink log a warning and skip it.
74/// `workspace` resolves relative `file` paths (see [`resolve_file_exporter_path`]).
75pub fn load_exporters(cfg: &TelemetryConfig, workspace: &Path) -> ExporterRegistry {
76    let mut v: Vec<Arc<dyn TelemetryExporter>> = Vec::new();
77    for entry in &cfg.exporters {
78        if let Some(exp) = build_exporter(entry, workspace) {
79            v.push(exp);
80        }
81    }
82    ExporterRegistry::from_vec(v)
83}
84
85fn build_exporter(c: &ExporterConfig, workspace: &Path) -> Option<Arc<dyn TelemetryExporter>> {
86    if !c.is_enabled() {
87        return None;
88    }
89    match c {
90        ExporterConfig::None => None,
91        ExporterConfig::File { path, .. } => {
92            let p = file::resolve_file_exporter_path(path.as_deref(), workspace);
93            Some(Arc::new(file::FileExporter::new(p)) as _)
94        }
95        ExporterConfig::Dev { .. } => {
96            #[cfg(feature = "telemetry-dev")]
97            {
98                Some(Arc::new(dev::DevExporter) as _)
99            }
100            #[cfg(not(feature = "telemetry-dev"))]
101            {
102                tracing::warn!(
103                    "telemetry `dev` exporter configured but `telemetry-dev` is not enabled"
104                );
105                None
106            }
107        }
108        ExporterConfig::PostHog { .. } => {
109            let r = PostHogResolved::from_config(c)?;
110            #[cfg(feature = "telemetry-posthog")]
111            {
112                Some(Arc::new(posthog::PostHogExporter::new(&r.host, &r.project_api_key)) as _)
113            }
114            #[cfg(not(feature = "telemetry-posthog"))]
115            {
116                let _ = &r;
117                tracing::warn!(
118                    "PostHog configured but the `telemetry-posthog` feature is not enabled"
119                );
120                None
121            }
122        }
123        ExporterConfig::Datadog { .. } => {
124            let r = DatadogResolved::from_config(c)?;
125            #[cfg(feature = "telemetry-datadog")]
126            {
127                Some(Arc::new(datadog::DatadogExporter::new(&r.site, &r.api_key)) as _)
128            }
129            #[cfg(not(feature = "telemetry-datadog"))]
130            {
131                let _ = &r;
132                tracing::warn!(
133                    "Datadog configured but the `telemetry-datadog` feature is not enabled"
134                );
135                None
136            }
137        }
138        ExporterConfig::Otlp { .. } => {
139            let r = OtlpResolved::from_config(c)?;
140            #[cfg(feature = "telemetry-otlp")]
141            {
142                Some(Arc::new(otlp::OtlpExporter::new(&r.endpoint)) as _)
143            }
144            #[cfg(not(feature = "telemetry-otlp"))]
145            {
146                let _ = &r;
147                tracing::warn!("OTLP configured but the `telemetry-otlp` feature is not enabled");
148                None
149            }
150        }
151    }
152}
kaizen/telemetry/mod.rs

kaizen/telemetry/
mod.rs
