orbit_runtime/
telemetry.rs

1//! OpenTelemetry tracing: OTLP/HTTP (JSON encoding) export of execution
2//! traces, accepted by the OpenTelemetry Collector, Jaeger, Grafana Tempo,
3//! Honeycomb and other OTLP backends.
4//!
5//! * One trace per workflow execution; child workflows join their parent's
6//!   trace, nested under the step that started them.
7//! * A root span per execution, and one span per task attempt (dispatch to
8//!   result), timer and compensation, with worker, attempt and error attributes.
9//! * Trace and span ids are derived deterministically (SHA-256) from
10//!   execution, unit and dispatch identifiers, so a runtime restart never
11//!   produces duplicate or orphaned spans.
12//! * Export never blocks the engine: spans are queued to a background thread
13//!   and dropped (with a warning) if the collector cannot keep up.
14//!
15//! Configured with `orbit server --otlp-endpoint`, or the standard
16//! `OTEL_EXPORTER_OTLP_ENDPOINT`, `OTEL_EXPORTER_OTLP_HEADERS` and
17//! `OTEL_SERVICE_NAME` environment variables.
18
19use crate::model::{Event, ExecutionState, HistoryEvent, WorkflowStatus};
20use serde_json::{json, Value};
21use sha2::{Digest, Sha256};
22use std::sync::mpsc;
23use std::time::Duration;
24
25#[derive(Debug, Clone)]
26pub struct TelemetryConfig {
27    /// Base OTLP/HTTP endpoint, e.g. `http://localhost:4318`; spans are posted
28    /// to `<endpoint>/v1/traces`.
29    pub endpoint: String,
30    /// Extra request headers, e.g. an API key.
31    pub headers: Vec<(String, String)>,
32    pub service_name: String,
33}
34
35impl TelemetryConfig {
36    /// From `OTEL_EXPORTER_OTLP_ENDPOINT` / `_HEADERS` and `OTEL_SERVICE_NAME`.
37    pub fn from_env() -> Option<TelemetryConfig> {
38        let endpoint = std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT")
39            .ok()
40            .filter(|e| !e.is_empty())?;
41        Some(TelemetryConfig::new(&endpoint))
42    }
43
44    pub fn new(endpoint: &str) -> TelemetryConfig {
45        let headers = std::env::var("OTEL_EXPORTER_OTLP_HEADERS")
46            .unwrap_or_default()
47            .split(',')
48            .filter_map(|kv| kv.split_once('='))
49            .map(|(k, v)| (k.trim().to_string(), v.trim().to_string()))
50            .collect();
51        TelemetryConfig {
52            endpoint: endpoint.trim_end_matches('/').to_string(),
53            headers,
54            service_name: std::env::var("OTEL_SERVICE_NAME")
55                .unwrap_or_else(|_| "orbit-runtime".into()),
56        }
57    }
58}
59
60/// Where a trace and its root span hang: the trace id and the parent span id
61/// (the parent workflow's step span, for child workflows).
62#[derive(Debug, Clone)]
63pub struct TraceContext {
64    pub trace_id: String,
65    pub parent_span_id: Option<String>,
66}
67
68fn hex(bytes: &[u8]) -> String {
69    bytes.iter().map(|b| format!("{b:02x}")).collect()
70}
71
72fn digest(s: &str) -> [u8; 32] {
73    Sha256::digest(s.as_bytes()).into()
74}
75
76/// Trace id of a top-level execution.
77pub fn trace_id(root_execution: &str) -> String {
78    hex(&digest(&format!("orbit-trace:{root_execution}"))[..16])
79}
80
81pub fn root_span_id(execution: &str) -> String {
82    hex(&digest(&format!("orbit-span:{execution}"))[..8])
83}
84
85/// Span id of the `dispatch`-th dispatch of `unit` in `execution`.
86pub fn unit_span_id(execution: &str, unit: &str, dispatch: u32) -> String {
87    hex(&digest(&format!("orbit-span:{execution}/{unit}#{dispatch}"))[..8])
88}
89
90fn attr(key: &str, value: impl Into<Value>) -> Value {
91    let v: Value = value.into();
92    let typed = match &v {
93        Value::Bool(b) => json!({ "boolValue": b }),
94        Value::Number(n) if n.is_i64() || n.is_u64() => json!({ "intValue": n.to_string() }),
95        Value::Number(n) => json!({ "doubleValue": n }),
96        Value::String(s) => json!({ "stringValue": s }),
97        other => json!({ "stringValue": other.to_string() }),
98    };
99    json!({ "key": key, "value": typed })
100}
101
102fn nanos(ms: u64) -> String {
103    (ms as u128 * 1_000_000).to_string()
104}
105
106/// One finished span, before OTLP encoding.
107struct Span<'a> {
108    trace: &'a str,
109    id: &'a str,
110    parent: Option<&'a str>,
111    name: String,
112    start_ms: u64,
113    end_ms: u64,
114    attrs: Vec<Value>,
115    error: Option<String>,
116}
117
118impl Span<'_> {
119    fn encode(self) -> Value {
120        let mut s = json!({
121            "traceId": self.trace,
122            "spanId": self.id,
123            "name": self.name,
124            "kind": 1,
125            "startTimeUnixNano": nanos(self.start_ms),
126            "endTimeUnixNano": nanos(self.end_ms.max(self.start_ms)),
127            "attributes": self.attrs,
128            "status": match &self.error {
129                Some(m) => json!({ "code": 2, "message": m }),
130                None => json!({ "code": 1 }),
131            },
132        });
133        if let Some(p) = self.parent {
134            s["parentSpanId"] = json!(p);
135        }
136        s
137    }
138}
139
140/// Spans finished by one committed batch of events.
141pub fn spans_for(state: &ExecutionState, ctx: &TraceContext, batch: &[HistoryEvent]) -> Vec<Value> {
142    let root = root_span_id(&state.id);
143    let mut out = vec![];
144    for h in batch {
145        let ended = |key: &str, error: Option<String>, outcome: &str| -> Option<Value> {
146            let u = state.units.get(key)?;
147            let start = u.dispatched_at_ms.or(u.scheduled_at_ms)?;
148            let mut attrs = vec![
149                attr("orbit.execution_id", state.id.clone()),
150                attr("orbit.workflow", state.workflow.clone()),
151                attr("orbit.unit", key.to_string()),
152                attr("orbit.attempt", u.attempt),
153                attr("orbit.outcome", outcome.to_string()),
154            ];
155            if let Some(w) = &u.worker_id {
156                attrs.push(attr("orbit.worker_id", w.clone()));
157            }
158            if u.redelivery {
159                attrs.push(attr("orbit.redelivery", true));
160            }
161            // Later runs (continue-as-new) reuse unit keys: keep their spans apart.
162            let id = if state.run > 1 {
163                unit_span_id(
164                    &state.id,
165                    &format!("run{}/{key}", state.run),
166                    u.dispatch_count.max(1),
167                )
168            } else {
169                unit_span_id(&state.id, key, u.dispatch_count.max(1))
170            };
171            Some(
172                Span {
173                    trace: &ctx.trace_id,
174                    id: &id,
175                    parent: Some(&root),
176                    name: format!("step {key}"),
177                    start_ms: start,
178                    end_ms: h.timestamp_ms,
179                    attrs,
180                    error,
181                }
182                .encode(),
183            )
184        };
185        let s = match &h.event {
186            Event::StepSucceeded { step, .. } => ended(step, None, "succeeded"),
187            Event::TimerFired { step } => ended(step, None, "timer fired"),
188            Event::StepAttemptFailed { step, error, .. } => ended(
189                step,
190                Some(format!("{}: {}", error.code, error.message)),
191                "failed",
192            ),
193            Event::StepAttemptUncertain { step, reason, .. } => ended(
194                step,
195                Some(format!("outcome unknown: {reason}")),
196                "uncertain",
197            ),
198            Event::WorkflowCompleted { .. }
199            | Event::WorkflowFailed { .. }
200            | Event::WorkflowCancelled { .. }
201            | Event::WorkflowTimedOut { .. } => {
202                let mut attrs = vec![
203                    attr("orbit.execution_id", state.id.clone()),
204                    attr("orbit.workflow", state.workflow.clone()),
205                    attr("orbit.workflow_version", state.version.clone()),
206                    attr("orbit.status", state.status.as_str().to_string()),
207                ];
208                if let Some(c) = state.compensation {
209                    attrs.push(attr("orbit.compensation", format!("{c:?}").to_lowercase()));
210                }
211                let error = match state.status {
212                    WorkflowStatus::Completed => None,
213                    _ => Some(
214                        state
215                            .error
216                            .as_ref()
217                            .map(|e| format!("{}: {}", e.code, e.message))
218                            .unwrap_or_else(|| state.status.as_str().to_string()),
219                    ),
220                };
221                Some(
222                    Span {
223                        trace: &ctx.trace_id,
224                        id: &root,
225                        parent: ctx.parent_span_id.as_deref(),
226                        name: format!("workflow {}", state.workflow),
227                        start_ms: state.created_at_ms,
228                        end_ms: h.timestamp_ms,
229                        attrs,
230                        error,
231                    }
232                    .encode(),
233                )
234            }
235            _ => None,
236        };
237        out.extend(s);
238    }
239    // Attempts still running when the workflow ends are not exported: they did
240    // not finish, so there is no honest end time for them.
241    out
242}
243
244/// Background OTLP exporter.
245pub struct Exporter {
246    tx: mpsc::SyncSender<Vec<Value>>,
247}
248
249impl Exporter {
250    pub fn start(config: TelemetryConfig) -> Exporter {
251        let (tx, rx) = mpsc::sync_channel::<Vec<Value>>(1024);
252        std::thread::Builder::new()
253            .name("orbit-otlp".into())
254            .spawn(move || {
255                let agent: ureq::Agent = ureq::Agent::config_builder()
256                    .http_status_as_error(false)
257                    .timeout_global(Some(Duration::from_secs(10)))
258                    .build()
259                    .into();
260                let url = format!("{}/v1/traces", config.endpoint);
261                while let Ok(first) = rx.recv() {
262                    // Batch whatever else is already queued.
263                    let mut spans = first;
264                    while let Ok(more) = rx.try_recv() {
265                        spans.extend(more);
266                        if spans.len() >= 512 {
267                            break;
268                        }
269                    }
270                    let body = json!({
271                        "resourceSpans": [{
272                            "resource": { "attributes": [
273                                attr("service.name", config.service_name.clone()),
274                                attr("service.version", env!("CARGO_PKG_VERSION")),
275                            ]},
276                            "scopeSpans": [{ "scope": { "name": "orbit-runtime", "version": env!("CARGO_PKG_VERSION") }, "spans": spans }],
277                        }]
278                    });
279                    let mut req = agent.post(&url).header("content-type", "application/json");
280                    for (k, v) in &config.headers {
281                        req = req.header(k, v);
282                    }
283                    match req.send(body.to_string()) {
284                        Ok(r) if r.status().is_success() => {}
285                        Ok(r) => tracing::warn!(status = r.status().as_u16(), "OTLP collector rejected spans"),
286                        Err(e) => tracing::warn!(error = %e, "cannot export spans to the OTLP collector"),
287                    }
288                }
289            })
290            .expect("start OTLP exporter thread");
291        Exporter { tx }
292    }
293
294    pub fn export(&self, spans: Vec<Value>) {
295        if spans.is_empty() {
296            return;
297        }
298        if self.tx.try_send(spans).is_err() {
299            tracing::warn!("OTLP export queue is full; dropping spans");
300        }
301    }
302}
orbit_runtime/telemetry.rs

orbit_runtime/
telemetry.rs
