telemetrylib/
metrics_exporter_file.rs

1use chrono::{DateTime, Utc};
2use opentelemetry_sdk::error::{OTelSdkError, OTelSdkResult};
3use opentelemetry_sdk::metrics::Temporality;
4use opentelemetry_sdk::metrics::data::{
5    AggregatedMetrics, Gauge, Histogram, Metric, MetricData, ResourceMetrics, Sum,
6};
7use opentelemetry_sdk::metrics::exporter::PushMetricExporter;
8use serde_json::{Map, Value};
9use std::fs::OpenOptions;
10use std::io::Write;
11use std::path::PathBuf;
12use std::result::Result;
13use std::sync::atomic::{AtomicBool, Ordering};
14use std::time::Duration;
15
16/// An OpenTelemetry exporter that writes Metrics to a file on export.
17pub struct FileMetricExporter {
18    is_shutdown: AtomicBool,
19    path: PathBuf,
20}
21
22impl FileMetricExporter {
23    /// Creates a new FileMetricExporter that writes metrics to the specified path.
24    ///
25    /// # Arguments
26    /// * `path` - The path where metrics will be written. This can be either a file or directory path.
27    ///
28    /// # Behavior
29    /// - If the path points to a directory:
30    ///   - The directory must exist
31    ///   - Metrics will be written to a file named "signals.json" within that directory
32    /// - If the path points to a file:
33    ///   - If the file exists, new metrics will be appended to it (existing data is preserved)
34    ///   - If the file doesn't exist, it will be created
35    ///   - The parent directory must exist
36    ///
37    /// # Errors
38    /// Returns an error if:
39    /// - The parent directory doesn't exist
40    /// - The path points to a directory that doesn't exist
41    /// - The user doesn't have write permissions for the target location
42    pub fn new(path: PathBuf) -> Result<Self, OTelSdkError> {
43        // TODO: Check if the file exists and has write permissions - https://github.com/valkey-io/valkey-glide/issues/3720
44        Ok(Self {
45            is_shutdown: AtomicBool::new(false),
46            path,
47        })
48    }
49}
50
51impl PushMetricExporter for FileMetricExporter {
52    fn temporality(&self) -> Temporality {
53        Temporality::Cumulative
54    }
55
56    async fn export(&self, metrics: &ResourceMetrics) -> OTelSdkResult {
57        if self.is_shutdown.load(Ordering::SeqCst) {
58            return Err(OTelSdkError::AlreadyShutdown);
59        }
60
61        // TODO: Move the writes to Tokio task - https://github.com/valkey-io/valkey-glide/issues/3720
62        let mut file = OpenOptions::new()
63            .create(true)
64            .append(true)
65            .open(&self.path)
66            .map_err(|err| {
67                OTelSdkError::InternalFailure(format!("Unable to open exporter file: {err}"))
68            })?;
69
70        let metrics_json = to_json(metrics).map_err(|e| {
71            OTelSdkError::InternalFailure(format!("Failed to serialize metrics to JSON: {e}"))
72        })?;
73        let json_string = serde_json::to_string(&metrics_json).map_err(|e| {
74            OTelSdkError::InternalFailure(format!("Failed to serialize metrics to JSON: {e}"))
75        })?;
76
77        writeln!(file, "{}", json_string)
78            .map_err(|e| OTelSdkError::InternalFailure(format!("File write error: {e}")))?;
79
80        Ok(())
81    }
82
83    /// No-op implementation since metrics are written immediately in export()
84    fn force_flush(&self) -> OTelSdkResult {
85        Ok(())
86    }
87
88    fn shutdown_with_timeout(&self, _timeout: Duration) -> OTelSdkResult {
89        self.is_shutdown.store(true, Ordering::SeqCst);
90        Ok(())
91    }
92}
93
94fn to_json(metrics: &ResourceMetrics) -> Result<Value, OTelSdkError> {
95    let mut root = Map::new();
96
97    // Add resource attributes
98    let mut resource_attrs = Map::new();
99    for (key, value) in metrics.resource().iter() {
100        resource_attrs.insert(key.to_string(), Value::String(value.to_string()));
101    }
102    root.insert("resource".to_owned(), Value::Object(resource_attrs));
103
104    // Add scope metrics
105    let mut scope_metrics = Vec::new();
106    for scope_metric in metrics.scope_metrics() {
107        let mut scope = Map::new();
108        let mut scope_info = Map::new();
109        scope_info.insert(
110            "name".to_owned(),
111            Value::String(scope_metric.scope().name().to_string()),
112        );
113        if let Some(version) = scope_metric.scope().version() {
114            scope_info.insert("version".to_owned(), Value::String(version.to_string()));
115        }
116        if let Some(schema_url) = scope_metric.scope().schema_url() {
117            scope_info.insert(
118                "schema_url".to_owned(),
119                Value::String(schema_url.to_string()),
120            );
121        }
122        scope.insert("scope".to_owned(), Value::Object(scope_info));
123
124        // Add metrics
125        let mut metrics_array = Vec::new();
126        for metric in scope_metric.metrics() {
127            let mut metric_obj = Map::new();
128            metric_obj.insert("name".to_owned(), Value::String(metric.name().to_string()));
129            metric_obj.insert(
130                "description".to_owned(),
131                Value::String(metric.description().to_string()),
132            );
133            metric_obj.insert("unit".to_owned(), Value::String(metric.unit().to_string()));
134
135            // Serialize data points using helper functions
136            let data_points = serialize_metric_data(metric)?;
137            metric_obj.insert("data_points".to_owned(), Value::Array(data_points));
138            metrics_array.push(Value::Object(metric_obj));
139        }
140        scope.insert("metrics".to_owned(), Value::Array(metrics_array));
141        scope_metrics.push(Value::Object(scope));
142    }
143    root.insert("scope_metrics".to_owned(), Value::Array(scope_metrics));
144
145    Ok(Value::Object(root))
146}
147
148/// Serialize metric data points based on the aggregation type.
149///
150/// In opentelemetry_sdk 0.32 the metric data model is an explicit
151/// `AggregatedMetrics` enum keyed on the numeric type (`u64`, `i64`, `f64`),
152/// each wrapping a `MetricData<T>` describing the aggregation kind. This
153/// replaces the previous `Box<dyn Any>` downcasting approach while preserving
154/// the same JSON output shape.
155fn serialize_metric_data(metric: &Metric) -> Result<Vec<Value>, OTelSdkError> {
156    match metric.data() {
157        AggregatedMetrics::U64(data) => {
158            serialize_metric_data_typed(metric, data, |v: u64| Value::Number(v.into()))
159        }
160        AggregatedMetrics::I64(data) => {
161            serialize_metric_data_typed(metric, data, |v: i64| Value::Number(v.into()))
162        }
163        AggregatedMetrics::F64(data) => {
164            serialize_metric_data_typed(metric, data, |v: f64| Value::String(v.to_string()))
165        }
166    }
167}
168
169/// Serialize the aggregation for a single numeric type `T`.
170fn serialize_metric_data_typed<T>(
171    metric: &Metric,
172    data: &MetricData<T>,
173    value_serializer: impl Fn(T) -> Value,
174) -> Result<Vec<Value>, OTelSdkError>
175where
176    T: Copy,
177{
178    match data {
179        MetricData::Sum(sum) => serialize_sum_data_points(sum, value_serializer),
180        MetricData::Gauge(gauge) => serialize_gauge_data_points(gauge, value_serializer),
181        MetricData::Histogram(histogram) => {
182            serialize_histogram_data_points(histogram, value_serializer)
183        }
184        MetricData::ExponentialHistogram(_) => Err(OTelSdkError::InternalFailure(format!(
185            "Unsupported metric type (ExponentialHistogram) for metric '{}'",
186            metric.name()
187        ))),
188    }
189}
190
191/// Generic helper to serialize Sum data points
192fn serialize_sum_data_points<T>(
193    sum: &Sum<T>,
194    value_serializer: impl Fn(T) -> Value,
195) -> Result<Vec<Value>, OTelSdkError>
196where
197    T: Copy,
198{
199    let start_time = sum.start_time();
200    let time = sum.time();
201    let mut data_points = Vec::new();
202    for point in sum.data_points() {
203        let mut dp = Map::new();
204        dp.insert("value".to_owned(), value_serializer(point.value()));
205
206        let start_time: DateTime<Utc> = start_time.into();
207        dp.insert(
208            "start_time".to_owned(),
209            Value::String(start_time.timestamp_micros().to_string()),
210        );
211
212        let time: DateTime<Utc> = time.into();
213        dp.insert(
214            "time".to_owned(),
215            Value::String(time.timestamp_micros().to_string()),
216        );
217
218        dp.insert(
219            "attributes".to_owned(),
220            attributes_to_json(point.attributes()),
221        );
222        data_points.push(Value::Object(dp));
223    }
224    Ok(data_points)
225}
226
227/// Generic helper to serialize Gauge data points
228fn serialize_gauge_data_points<T>(
229    gauge: &Gauge<T>,
230    value_serializer: impl Fn(T) -> Value,
231) -> Result<Vec<Value>, OTelSdkError>
232where
233    T: Copy,
234{
235    let time = gauge.time();
236    let mut data_points = Vec::new();
237    for point in gauge.data_points() {
238        let mut dp = Map::new();
239        dp.insert("value".to_owned(), value_serializer(point.value()));
240
241        let time: DateTime<Utc> = time.into();
242        dp.insert(
243            "time".to_owned(),
244            Value::String(time.timestamp_micros().to_string()),
245        );
246
247        dp.insert(
248            "attributes".to_owned(),
249            attributes_to_json(point.attributes()),
250        );
251        data_points.push(Value::Object(dp));
252    }
253    Ok(data_points)
254}
255
256/// Generic helper to serialize Histogram data points
257fn serialize_histogram_data_points<T>(
258    histogram: &Histogram<T>,
259    value_serializer: impl Fn(T) -> Value,
260) -> Result<Vec<Value>, OTelSdkError>
261where
262    T: Copy,
263{
264    let start_time = histogram.start_time();
265    let time = histogram.time();
266    let mut data_points = Vec::new();
267    for point in histogram.data_points() {
268        let mut dp = Map::new();
269        dp.insert("count".to_owned(), Value::Number(point.count().into()));
270        dp.insert("sum".to_owned(), value_serializer(point.sum()));
271
272        // Bucket counts are always u64 in the OTel SDK, independent of T
273        let bucket_counts: Vec<Value> = point
274            .bucket_counts()
275            .map(|count| Value::Number(count.into()))
276            .collect();
277        dp.insert("bucket_counts".to_owned(), Value::Array(bucket_counts));
278
279        // Bounds are always f64 in the OTel SDK, independent of T — serialized as strings for precision
280        let bounds: Vec<Value> = point
281            .bounds()
282            .map(|bound| Value::String(bound.to_string()))
283            .collect();
284        dp.insert("bounds".to_owned(), Value::Array(bounds));
285
286        let start_time: DateTime<Utc> = start_time.into();
287        dp.insert(
288            "start_time".to_owned(),
289            Value::String(start_time.timestamp_micros().to_string()),
290        );
291        let time: DateTime<Utc> = time.into();
292        dp.insert(
293            "time".to_owned(),
294            Value::String(time.timestamp_micros().to_string()),
295        );
296
297        dp.insert(
298            "attributes".to_owned(),
299            attributes_to_json(point.attributes()),
300        );
301        data_points.push(Value::Object(dp));
302    }
303    Ok(data_points)
304}
305
306// Helper function to convert attributes to JSON
307fn attributes_to_json<'a>(attributes: impl Iterator<Item = &'a opentelemetry::KeyValue>) -> Value {
308    let mut json_attributes = Map::new();
309    for kv in attributes {
310        json_attributes.insert(kv.key.to_string(), Value::String(kv.value.to_string()));
311    }
312    Value::Object(json_attributes)
313}
telemetrylib/metrics_exporter_file.rs

telemetrylib/
metrics_exporter_file.rs
