blotter/commands/
export.rs

1//! OTLP 1.11.0 JSON mapping for the OpenTelemetry file-exporter JSONL format.
2//!
3//! This module owns the outward bridge only. Blotter's event schema remains
4//! internal, and no OpenTelemetry crate is required for this compatibility
5//! snapshot.
6
7use crate::cli::{ExportArgs, ExportFormat};
8use crate::error::{AppError, AppResult};
9use crate::output;
10use crate::store;
11use crate::{Impact, ListItem, parse_since};
12use jiff::Timestamp;
13use serde::Serialize;
14use std::io::Write;
15use std::path::{Path, PathBuf};
16
17const EVENT_NAME: &str = "blotter.friction.reported";
18
19/// Argument validation that must run before the clock is resolved, so a missing
20/// `--format` is reported ahead of any environment error.
21pub fn validate(args: &ExportArgs) -> AppResult<()> {
22    let Some(ExportFormat::OtlpJson) = args.format else {
23        return Err(AppError::invalid_argument(
24            "export requires --format otlp-json",
25            "Run `blotter export --format otlp-json`.",
26        ));
27    };
28    Ok(())
29}
30
31pub fn run(
32    args: ExportArgs,
33    file: Option<PathBuf>,
34    _pretty: bool,
35    now: Timestamp,
36) -> AppResult<i32> {
37    validate(&args)?;
38    let since = args
39        .since
40        .as_deref()
41        .map(|value| parse_since(value, now))
42        .transpose()?;
43
44    let resolved = store::discover(file)?;
45    let store::LoadedFold { items, .. } = store::load_folded(&resolved)?;
46
47    let data = LogsData::from_items(items, since)?;
48    write_otlp_json(&data)?;
49    Ok(0)
50}
51
52#[derive(Serialize)]
53#[serde(rename_all = "camelCase")]
54struct LogsData {
55    resource_logs: Vec<ResourceLogs>,
56}
57
58impl LogsData {
59    fn from_items(items: Vec<ListItem>, since: Option<Timestamp>) -> AppResult<Self> {
60        let mut cuts: Vec<_> = items
61            .into_iter()
62            .filter(|item| {
63                item.kind == "cut"
64                    && since.is_none_or(|threshold| {
65                        item.ts
66                            .parse::<Timestamp>()
67                            .is_ok_and(|timestamp| timestamp >= threshold)
68                    })
69            })
70            .collect();
71        cuts.sort_by(|left, right| {
72            left.ts
73                .parse::<Timestamp>()
74                .expect("folded items have valid RFC3339 timestamps")
75                .cmp(
76                    &right
77                        .ts
78                        .parse::<Timestamp>()
79                        .expect("folded items have valid RFC3339 timestamps"),
80                )
81                .then_with(|| left.id.cmp(&right.id))
82        });
83
84        let log_records = cuts
85            .iter()
86            .map(LogRecord::from_item)
87            .collect::<AppResult<Vec<_>>>()?;
88
89        Ok(Self {
90            resource_logs: vec![ResourceLogs {
91                resource: Resource {},
92                scope_logs: vec![ScopeLogs {
93                    scope: InstrumentationScope {
94                        name: "blotter",
95                        version: env!("CARGO_PKG_VERSION"),
96                    },
97                    log_records,
98                }],
99            }],
100        })
101    }
102}
103
104#[derive(Serialize)]
105struct Resource {}
106
107#[derive(Serialize)]
108#[serde(rename_all = "camelCase")]
109struct ResourceLogs {
110    resource: Resource,
111    scope_logs: Vec<ScopeLogs>,
112}
113
114#[derive(Serialize)]
115#[serde(rename_all = "camelCase")]
116struct ScopeLogs {
117    scope: InstrumentationScope,
118    log_records: Vec<LogRecord>,
119}
120
121#[derive(Serialize)]
122struct InstrumentationScope {
123    name: &'static str,
124    version: &'static str,
125}
126
127#[derive(Serialize)]
128#[serde(rename_all = "camelCase")]
129struct LogRecord {
130    event_name: &'static str,
131    time_unix_nano: String,
132    severity_number: u8,
133    severity_text: &'static str,
134    body: AnyValue,
135    attributes: Vec<KeyValue>,
136}
137
138impl LogRecord {
139    fn from_item(item: &ListItem) -> AppResult<Self> {
140        let timestamp = item
141            .ts
142            .parse::<Timestamp>()
143            .expect("folded items have valid RFC3339 timestamps");
144        // OTLP timeUnixNano is an unsigned fixed64: a record outside that range
145        // rejects the whole export rather than emitting a partial or placeholder.
146        let time_unix_nano = u64::try_from(timestamp.as_nanosecond()).map_err(|_| {
147            AppError::invalid_input(
148                format!(
149                    "record {} has timestamp {} outside the OTLP unsigned 64-bit nanosecond range",
150                    item.id, item.ts
151                ),
152                "Correct that record's timestamp, or exclude it with --since, then export again.",
153            )
154        })?;
155        let impact = item.impact.expect("cut items have impact");
156        let (severity_number, severity_text) = severity_fields(impact);
157        let status = export_status(item);
158        let mut attributes = vec![
159            string_attribute("blotter.friction.id", &item.id),
160            string_attribute("blotter.friction.impact", impact.as_str()),
161            string_attribute("blotter.friction.status", status),
162            string_attribute("blotter.friction.agent", &item.agent),
163            tags_attribute(&item.tags),
164            string_attribute("blotter.friction.cwd", &item.cwd),
165        ];
166        if status == "resolved" {
167            let resolution = item
168                .resolution
169                .as_ref()
170                .expect("resolved export status has a resolution");
171            attributes.push(string_attribute(
172                "blotter.friction.resolved_ts",
173                &resolution.ts,
174            ));
175        }
176        Ok(Self {
177            event_name: EVENT_NAME,
178            time_unix_nano: time_unix_nano.to_string(),
179            severity_number,
180            severity_text,
181            body: AnyValue::StringValue(item.text.clone()),
182            attributes,
183        })
184    }
185}
186
187fn severity_fields(impact: Impact) -> (u8, &'static str) {
188    match impact {
189        Impact::Low => (9, "INFO"),
190        Impact::Material => (13, "WARN"),
191        Impact::Blocking => (17, "ERROR"),
192    }
193}
194
195fn export_status(item: &ListItem) -> &'static str {
196    match item.resolution.as_ref() {
197        Some(resolution) if resolution.dropped => "dropped",
198        Some(_) => "resolved",
199        None => "open",
200    }
201}
202
203#[derive(Serialize)]
204struct KeyValue {
205    key: &'static str,
206    value: AnyValue,
207}
208
209fn string_attribute(key: &'static str, value: &str) -> KeyValue {
210    KeyValue {
211        key,
212        value: AnyValue::StringValue(value.to_owned()),
213    }
214}
215
216fn tags_attribute(tags: &[String]) -> KeyValue {
217    KeyValue {
218        key: "blotter.friction.tags",
219        value: AnyValue::ArrayValue(ArrayValue {
220            values: tags.iter().cloned().map(AnyValue::StringValue).collect(),
221        }),
222    }
223}
224
225#[derive(Serialize)]
226#[serde(rename_all = "camelCase")]
227enum AnyValue {
228    StringValue(String),
229    ArrayValue(ArrayValue),
230}
231
232#[derive(Serialize)]
233struct ArrayValue {
234    values: Vec<AnyValue>,
235}
236
237fn write_otlp_json(data: &LogsData) -> AppResult<()> {
238    let mut output =
239        output::stdout_writer().map_err(|error| AppError::from_io(error, Path::new("stdout")))?;
240    serde_json::to_writer(&mut output, data)
241        .map_err(|error| AppError::from_io(std::io::Error::other(error), Path::new("stdout")))?;
242    writeln!(output).map_err(|error| AppError::from_io(error, Path::new("stdout")))?;
243    output
244        .flush()
245        .map_err(|error| AppError::from_io(error, Path::new("stdout")))
246}
blotter/commands/export.rs

blotter/commands/
export.rs
