areev_run/
otel.rs

1//! OpenTelemetry export (Wave 5): a §6.10 observer that turns run events
2//! into OTLP/HTTP **JSON** spans — hand-rolled over `std::net::TcpStream`
3//! (dependency-light policy: no HTTP client crate, no OTel SDK). One POST
4//! per run, flushed at `RunFinished`, so a collector outage costs one
5//! buffered batch, never scheduler time — the bus already guarantees the
6//! run never waits on an observer.
7//!
8//! Span model: one ROOT span per run (`areev.run`, the full wall interval),
9//! one child span per journaled effect (`node` name; task_path/attempt/
10//! effect_seq as attributes; error status on failed effects). Trace id =
11//! first 16 bytes of sha256(run_id) — stable across resumes, so a resumed
12//! run's spans land in the SAME trace.
13//!
14//! Endpoint: `http://host:port` (the standard collector `:4318`; the
15//! `/v1/traces` path is appended when absent). TLS is the collector's job
16//! in this profile — point at a local sidecar/agent, which is the normal
17//! OTel deployment shape anyway.
18
19use crate::stream::{RunEvent, RunObserver};
20use sha2::Digest;
21use std::io::{Read, Write};
22use std::sync::Mutex;
23
24pub struct OtelObserver {
25    host: String,
26    port: u16,
27    path: String,
28    state: Mutex<OtelState>,
29}
30
31#[derive(Default)]
32struct OtelState {
33    /// Per-run event buffers, attributed to the INNERMOST live run (a
34    /// subgraph child runs inline between its parent's events; a flat
35    /// buffer would let the child's RunFinished flush — and mis-trace —
36    /// the parent's spans). The observer thread stamps time — export is
37    /// observational; journaled clocks stay the replay truth.
38    stack: Vec<(String, Vec<(RunEvent, u128)>)>,
39}
40
41fn now_ns() -> u128 {
42    std::time::SystemTime::now()
43        .duration_since(std::time::UNIX_EPOCH)
44        .map(|d| d.as_nanos())
45        .unwrap_or(0)
46}
47
48fn hex(bytes: &[u8]) -> String {
49    let mut s = String::with_capacity(bytes.len() * 2);
50    for b in bytes {
51        use std::fmt::Write as _;
52        let _ = write!(s, "{b:02x}");
53    }
54    s
55}
56
57fn trace_id(run_id: &str) -> String {
58    let d = sha2::Sha256::digest(run_id.as_bytes());
59    hex(&d[..16])
60}
61
62fn span_id(seed: &str) -> String {
63    let d = sha2::Sha256::digest(seed.as_bytes());
64    hex(&d[..8])
65}
66
67impl OtelObserver {
68    /// Parse `http://host:port[/path]`. Anything else is refused loudly —
69    /// silently exporting nowhere would be worse than an error.
70    pub fn new(endpoint: &str) -> Result<Self, String> {
71        let rest = endpoint
72            .strip_prefix("http://")
73            .ok_or_else(|| format!("--otel-endpoint must be http:// (got {endpoint}); terminate TLS at a local collector"))?;
74        let (authority, path) = match rest.split_once('/') {
75            Some((a, p)) => (a, format!("/{p}")),
76            None => (rest, "/v1/traces".to_string()),
77        };
78        let (host, port) = match authority.rsplit_once(':') {
79            Some((h, p)) => (h.to_string(), p.parse::<u16>().map_err(|_| format!("bad port in {endpoint}"))?),
80            None => (authority.to_string(), 4318),
81        };
82        if host.is_empty() {
83            return Err(format!("no host in {endpoint}"));
84        }
85        Ok(OtelObserver { host, port, path, state: Mutex::new(OtelState::default()) })
86    }
87
88    fn flush(&self, run_id: &str, outcome: &str) {
89        let events = {
90            let mut st = self.state.lock().unwrap_or_else(|e| e.into_inner());
91            match st.stack.iter().rposition(|(id, _)| id == run_id) {
92                Some(pos) => st.stack.remove(pos).1,
93                None => return,
94            }
95        };
96        let payload = build_otlp(run_id, outcome, &events);
97        let body = payload.to_string();
98        // Fire-and-forget with a bounded timeout: telemetry must never wedge
99        // the observer thread (which would fill the bus and start dropping).
100        let addr = format!("{}:{}", self.host, self.port);
101        let req = format!(
102            "POST {} HTTP/1.1\r\nHost: {}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
103            self.path, self.host, body.len(), body
104        );
105        let timeout = std::time::Duration::from_secs(5);
106        // connect_timeout, never a bare connect: a blackholed collector
107        // route must cost seconds, not the OS's minutes — EventBus::drop
108        // joins this thread on drive's return.
109        let resolved = std::net::ToSocketAddrs::to_socket_addrs(&addr.as_str())
110            .ok()
111            .and_then(|mut it| it.next());
112        let Some(sockaddr) = resolved else { return };
113        if let Ok(mut stream) = std::net::TcpStream::connect_timeout(&sockaddr, timeout) {
114            let _ = stream.set_write_timeout(Some(timeout));
115            let _ = stream.set_read_timeout(Some(timeout));
116            if stream.write_all(req.as_bytes()).is_ok() {
117                let mut buf = [0u8; 512];
118                let _ = stream.read(&mut buf); // drain the status line; best-effort
119            }
120        }
121    }
122}
123
124impl RunObserver for OtelObserver {
125    fn event(&self, ev: &RunEvent) {
126        {
127            let mut st = self.state.lock().unwrap_or_else(|e| e.into_inner());
128            match ev {
129                RunEvent::RunStarted { run_id } | RunEvent::RunResumed { run_id }
130                    if !st.stack.iter().any(|(id, _)| id == run_id) =>
131                {
132                    st.stack.push((run_id.clone(), Vec::new()));
133                }
134                _ => {}
135            }
136            // Attribute to the innermost live run (last on the stack).
137            if let Some((_, events)) = st.stack.last_mut() {
138                events.push((ev.clone(), now_ns()));
139            } else {
140                st.stack.push((String::new(), vec![(ev.clone(), now_ns())]));
141            }
142        }
143        if let RunEvent::RunFinished { run_id, outcome, .. } = ev {
144            self.flush(run_id, outcome);
145        }
146    }
147}
148
149/// The OTLP/JSON payload (`resourceSpans` shape, span kind INTERNAL).
150/// Public-for-tests: the fixture asserts the exact structure without a
151/// socket.
152pub fn build_otlp(run_id: &str, outcome: &str, events: &[(RunEvent, u128)]) -> serde_json::Value {
153    use serde_json::json;
154    let tid = trace_id(run_id);
155    let root_sid = span_id(run_id);
156    let start = events.first().map(|(_, t)| *t).unwrap_or_else(now_ns);
157    let end = events.last().map(|(_, t)| *t).unwrap_or(start);
158
159    let mut spans = Vec::new();
160    // Effect spans: NodeDispatched opens, the matching EffectSettled closes.
161    for (ev, at) in events {
162        if let RunEvent::NodeDispatched { superstep, node, task_path, attempt, effect_seq } = ev {
163            let key = format!("{run_id}/{task_path}/{node}/{attempt}/{effect_seq}");
164            let settle = events.iter().find_map(|(e2, t2)| match e2 {
165                RunEvent::EffectSettled {
166                    node: n2, task_path: p2, attempt: a2, effect_seq: s2, ok, ..
167                } if n2 == node && p2 == task_path && a2 == attempt && s2 == effect_seq => {
168                    Some((*t2, *ok))
169                }
170                _ => None,
171            });
172            let (end_ns, ok) = settle.unwrap_or((*at, true));
173            spans.push(json!({
174                "traceId": tid,
175                "spanId": span_id(&key),
176                "parentSpanId": root_sid,
177                "name": node,
178                "kind": 1,
179                "startTimeUnixNano": at.to_string(),
180                "endTimeUnixNano": end_ns.to_string(),
181                "attributes": [
182                    {"key": "areev.superstep", "value": {"intValue": superstep.to_string()}},
183                    {"key": "areev.task_path", "value": {"stringValue": task_path}},
184                    {"key": "areev.attempt", "value": {"intValue": attempt.to_string()}},
185                    {"key": "areev.effect_seq", "value": {"intValue": effect_seq.to_string()}},
186                ],
187                "status": if ok { json!({"code": 1}) } else { json!({"code": 2, "message": "effect failed"}) },
188            }));
189        }
190    }
191    spans.push(json!({
192        "traceId": tid,
193        "spanId": root_sid,
194        "name": "areev.run",
195        "kind": 1,
196        "startTimeUnixNano": start.to_string(),
197        "endTimeUnixNano": end.to_string(),
198        "attributes": [
199            {"key": "areev.run_id", "value": {"stringValue": run_id}},
200            {"key": "areev.outcome", "value": {"stringValue": outcome}},
201        ],
202        "status": if outcome.contains("Completed") { json!({"code": 1}) } else { json!({"code": 2, "message": outcome}) },
203    }));
204
205    json!({
206        "resourceSpans": [{
207            "resource": {"attributes": [
208                {"key": "service.name", "value": {"stringValue": "areev-run"}},
209            ]},
210            "scopeSpans": [{
211                "scope": {"name": "areev-run", "version": env!("CARGO_PKG_VERSION")},
212                "spans": spans,
213            }],
214        }],
215    })
216}
217
218#[cfg(test)]
219mod tests {
220    use super::*;
221    use crate::stream::RunEvent;
222
223    #[test]
224    fn endpoint_parsing_is_strict() {
225        let o = OtelObserver::new("http://localhost:4318").unwrap();
226        assert_eq!((o.host.as_str(), o.port, o.path.as_str()), ("localhost", 4318, "/v1/traces"));
227        let o = OtelObserver::new("http://col.internal/custom/traces").unwrap();
228        assert_eq!((o.port, o.path.as_str()), (4318, "/custom/traces"));
229        assert!(OtelObserver::new("https://secure:4318").is_err(), "TLS is the collector's job");
230        assert!(OtelObserver::new("localhost").is_err());
231    }
232
233    #[test]
234    fn otlp_payload_carries_root_and_effect_spans() {
235        let events = vec![
236            (RunEvent::RunStarted { run_id: "r1".into() }, 1_000),
237            (
238                RunEvent::NodeDispatched {
239                    superstep: 1, node: "greet".into(), task_path: String::new(),
240                    attempt: 1, effect_seq: 0,
241                },
242                2_000,
243            ),
244            (
245                RunEvent::EffectSettled {
246                    superstep: 1, node: "greet".into(), task_path: String::new(),
247                    attempt: 1, effect_seq: 0, ok: false,
248                },
249                3_000,
250            ),
251            (
252                RunEvent::RunFinished {
253                    run_id: "r1".into(), outcome: "Failed".into(), dropped_events: 0,
254                },
255                4_000,
256            ),
257        ];
258        let p = build_otlp("r1", "Failed", &events);
259        let spans = p["resourceSpans"][0]["scopeSpans"][0]["spans"].as_array().unwrap();
260        assert_eq!(spans.len(), 2, "one effect + the root");
261        let effect = &spans[0];
262        assert_eq!(effect["name"], "greet");
263        assert_eq!(effect["status"]["code"], 2, "failed effect exports error status");
264        assert_eq!(effect["endTimeUnixNano"], "3000");
265        let root = &spans[1];
266        assert_eq!(root["name"], "areev.run");
267        assert_eq!(effect["parentSpanId"], root["spanId"]);
268        assert_eq!(effect["traceId"], root["traceId"]);
269        // Stable trace identity: a resume exports into the same trace.
270        assert_eq!(root["traceId"], build_otlp("r1", "x", &[])["resourceSpans"][0]["scopeSpans"][0]["spans"][0]["traceId"]);
271    }
272}
areev_run/otel.rs

areev_run/
otel.rs
