telemetrylib/
span_exporter_file.rs

1use chrono::{DateTime, Utc};
2use core::fmt;
3use opentelemetry_sdk::error::{OTelSdkError, OTelSdkResult};
4use opentelemetry_sdk::trace::SpanData;
5use serde_json::{Map, Value};
6use std::fs::OpenOptions;
7use std::io::Write;
8use std::path::PathBuf;
9use std::sync::atomic;
10
11use opentelemetry_sdk::Resource;
12
13/// An OpenTelemetry exporter that writes Spans to a file on export.
14pub struct SpanExporterFile {
15    resource: Resource,
16    is_shutdown: atomic::AtomicBool,
17    path: PathBuf,
18}
19
20impl fmt::Debug for SpanExporterFile {
21    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
22        f.write_str("SpanExporterFile")
23    }
24}
25
26impl SpanExporterFile {
27    /// Creates a new SpanExporterFile that writes Spans to a file on export.
28    ///
29    /// # Arguments
30    /// * `path` - The path where metrics will be written. This can be either a file or directory path.
31    ///
32    /// # Behavior
33    /// - If the path points to a directory:
34    ///   - The directory must exist
35    ///   - Metrics will be written to a file named "signals.json" within that directory
36    /// - If the path points to a file:
37    ///   - If the file exists, new metrics will be appended to it (existing data is preserved)
38    ///   - If the file doesn't exist, it will be created
39    ///   - The parent directory must exist
40    ///
41    /// # Errors
42    /// Returns an error if:
43    /// - The parent directory doesn't exist
44    /// - The path points to a directory that doesn't exist
45    /// - The user doesn't have write permissions for the target location
46    pub fn new(path: PathBuf) -> Result<Self, OTelSdkError> {
47        // TODO: Check if the file exists and has write permissions - https://github.com/valkey-io/valkey-glide/issues/3720
48        Ok(Self {
49            resource: Resource::builder_empty().build(),
50            is_shutdown: atomic::AtomicBool::new(false),
51            path,
52        })
53    }
54}
55
56impl opentelemetry_sdk::trace::SpanExporter for SpanExporterFile {
57    /// Write Spans to JSON file
58    async fn export(&self, batch: Vec<SpanData>) -> OTelSdkResult {
59        if self.is_shutdown.load(atomic::Ordering::SeqCst) {
60            return Err(OTelSdkError::AlreadyShutdown);
61        }
62
63        // TODO: Move the writes to Tokio task - https://github.com/valkey-io/valkey-glide/issues/3720
64        let Ok(mut file) = OpenOptions::new()
65            .create(true)
66            .append(true)
67            .open(&self.path)
68        else {
69            return Err(OTelSdkError::InternalFailure(format!(
70                "Unable to open exporter file: {} for append.",
71                self.path.display()
72            )));
73        };
74
75        let spans = to_jsons(batch);
76
77        for span in &spans {
78            if let Ok(s) = serde_json::to_string(&span)
79                && let Err(e) = writeln!(file, "{}", s)
80            {
81                return Err(OTelSdkError::InternalFailure(format!(
82                    "File write error. {e}",
83                )));
84            }
85        }
86        Ok(())
87    }
88
89    fn shutdown(&self) -> OTelSdkResult {
90        self.is_shutdown.store(true, atomic::Ordering::SeqCst);
91        Ok(())
92    }
93
94    fn set_resource(&mut self, res: &Resource) {
95        self.resource = res.clone();
96    }
97}
98
99fn to_jsons(batch: Vec<SpanData>) -> Vec<Value> {
100    let mut spans = Vec::<Value>::new();
101    for span in &batch {
102        let mut map = Map::new();
103        map.insert(
104            "scope".to_owned(),
105            Value::String(span.instrumentation_scope.name().to_string()),
106        );
107        if let Some(version) = &span.instrumentation_scope.version() {
108            map.insert("version".to_owned(), Value::String(version.to_string()));
109        }
110        if let Some(schema_url) = &span.instrumentation_scope.schema_url() {
111            map.insert(
112                "schema_url".to_owned(),
113                Value::String(schema_url.to_string()),
114            );
115        }
116
117        let mut scope_attributes = Vec::<Value>::new();
118        for kv in span.instrumentation_scope.attributes() {
119            let mut attr = Map::new();
120            attr.insert(kv.key.to_string(), Value::String(kv.value.to_string()));
121            scope_attributes.push(Value::Object(attr));
122        }
123        map.insert(
124            "scope_attributes".to_owned(),
125            Value::Array(scope_attributes),
126        );
127        map.insert("name".to_owned(), Value::String(span.name.to_string()));
128        map.insert(
129            "span_id".to_owned(),
130            Value::String(span.span_context.span_id().to_string()),
131        );
132        map.insert(
133            "parent_span_id".to_owned(),
134            Value::String(span.parent_span_id.to_string()),
135        );
136        map.insert(
137            "trace_id".to_owned(),
138            Value::String(span.span_context.trace_id().to_string()),
139        );
140        map.insert(
141            "kind".to_owned(),
142            Value::String(format!("{:?}", span.span_kind)),
143        );
144
145        let datetime: DateTime<Utc> = span.start_time.into();
146        map.insert(
147            "start_time".to_owned(),
148            Value::String(datetime.timestamp_micros().to_string()),
149        );
150
151        let datetime: DateTime<Utc> = span.end_time.into();
152        map.insert(
153            "end_time".to_owned(),
154            Value::String(datetime.timestamp_micros().to_string()),
155        );
156
157        map.insert(
158            "status".to_owned(),
159            Value::String(format!("{:?}", span.status)),
160        );
161
162        // Add the span attributes
163        let mut span_attributes = Vec::<Value>::new();
164        for kv in span.attributes.iter() {
165            let mut attr = Map::new();
166            attr.insert(kv.key.to_string(), Value::String(kv.value.to_string()));
167            span_attributes.push(Value::Object(attr));
168        }
169        map.insert("span_attributes".to_owned(), Value::Array(span_attributes));
170
171        // Add span events
172        let mut events = Vec::<Value>::new();
173        for event in span.events.iter() {
174            let mut evt = Map::new();
175            evt.insert("name".to_owned(), Value::String(event.name.to_string()));
176            let datetime: DateTime<Utc> = event.timestamp.into();
177            evt.insert(
178                "timestamp".to_owned(),
179                Value::String(datetime.format("%Y-%m-%d %H:%M:%S%.6f").to_string()),
180            );
181
182            let mut event_attributes = Vec::<Value>::new();
183            for kv in event.attributes.iter() {
184                let mut attr = Map::new();
185                attr.insert(kv.key.to_string(), Value::String(kv.value.to_string()));
186                event_attributes.push(Value::Object(attr));
187            }
188            evt.insert(
189                "event_attributes".to_owned(),
190                Value::Array(event_attributes),
191            );
192            events.push(Value::Object(evt));
193        }
194        map.insert("events".to_owned(), Value::Array(events));
195
196        let mut links = Vec::<Value>::new();
197        for link in span.links.iter() {
198            let mut lk = Map::new();
199            lk.insert(
200                "trace_id".to_owned(),
201                Value::String(link.span_context.trace_id().to_string()),
202            );
203            lk.insert(
204                "span_id".to_owned(),
205                Value::String(link.span_context.span_id().to_string()),
206            );
207            links.push(Value::Object(lk));
208        }
209        map.insert("links".to_owned(), Value::Array(links));
210        spans.push(Value::Object(map));
211    }
212    spans
213}
telemetrylib/span_exporter_file.rs

telemetrylib/
span_exporter_file.rs
