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            let id = unit_span_id(&state.id, key, u.dispatch_count.max(1));
162            Some(
163                Span {
164                    trace: &ctx.trace_id,
165                    id: &id,
166                    parent: Some(&root),
167                    name: format!("step {key}"),
168                    start_ms: start,
169                    end_ms: h.timestamp_ms,
170                    attrs,
171                    error,
172                }
173                .encode(),
174            )
175        };
176        let s = match &h.event {
177            Event::StepSucceeded { step, .. } => ended(step, None, "succeeded"),
178            Event::TimerFired { step } => ended(step, None, "timer fired"),
179            Event::StepAttemptFailed { step, error, .. } => ended(
180                step,
181                Some(format!("{}: {}", error.code, error.message)),
182                "failed",
183            ),
184            Event::StepAttemptUncertain { step, reason, .. } => ended(
185                step,
186                Some(format!("outcome unknown: {reason}")),
187                "uncertain",
188            ),
189            Event::WorkflowCompleted { .. }
190            | Event::WorkflowFailed { .. }
191            | Event::WorkflowCancelled { .. }
192            | Event::WorkflowTimedOut { .. } => {
193                let mut attrs = vec![
194                    attr("orbit.execution_id", state.id.clone()),
195                    attr("orbit.workflow", state.workflow.clone()),
196                    attr("orbit.workflow_version", state.version.clone()),
197                    attr("orbit.status", state.status.as_str().to_string()),
198                ];
199                if let Some(c) = state.compensation {
200                    attrs.push(attr("orbit.compensation", format!("{c:?}").to_lowercase()));
201                }
202                let error = match state.status {
203                    WorkflowStatus::Completed => None,
204                    _ => Some(
205                        state
206                            .error
207                            .as_ref()
208                            .map(|e| format!("{}: {}", e.code, e.message))
209                            .unwrap_or_else(|| state.status.as_str().to_string()),
210                    ),
211                };
212                Some(
213                    Span {
214                        trace: &ctx.trace_id,
215                        id: &root,
216                        parent: ctx.parent_span_id.as_deref(),
217                        name: format!("workflow {}", state.workflow),
218                        start_ms: state.created_at_ms,
219                        end_ms: h.timestamp_ms,
220                        attrs,
221                        error,
222                    }
223                    .encode(),
224                )
225            }
226            _ => None,
227        };
228        out.extend(s);
229    }
230    // Attempts still running when the workflow ends are not exported: they did
231    // not finish, so there is no honest end time for them.
232    out
233}
234
235/// Background OTLP exporter.
236pub struct Exporter {
237    tx: mpsc::SyncSender<Vec<Value>>,
238}
239
240impl Exporter {
241    pub fn start(config: TelemetryConfig) -> Exporter {
242        let (tx, rx) = mpsc::sync_channel::<Vec<Value>>(1024);
243        std::thread::Builder::new()
244            .name("orbit-otlp".into())
245            .spawn(move || {
246                let agent: ureq::Agent = ureq::Agent::config_builder()
247                    .http_status_as_error(false)
248                    .timeout_global(Some(Duration::from_secs(10)))
249                    .build()
250                    .into();
251                let url = format!("{}/v1/traces", config.endpoint);
252                while let Ok(first) = rx.recv() {
253                    // Batch whatever else is already queued.
254                    let mut spans = first;
255                    while let Ok(more) = rx.try_recv() {
256                        spans.extend(more);
257                        if spans.len() >= 512 {
258                            break;
259                        }
260                    }
261                    let body = json!({
262                        "resourceSpans": [{
263                            "resource": { "attributes": [
264                                attr("service.name", config.service_name.clone()),
265                                attr("service.version", env!("CARGO_PKG_VERSION")),
266                            ]},
267                            "scopeSpans": [{ "scope": { "name": "orbit-runtime", "version": env!("CARGO_PKG_VERSION") }, "spans": spans }],
268                        }]
269                    });
270                    let mut req = agent.post(&url).header("content-type", "application/json");
271                    for (k, v) in &config.headers {
272                        req = req.header(k, v);
273                    }
274                    match req.send(body.to_string()) {
275                        Ok(r) if r.status().is_success() => {}
276                        Ok(r) => tracing::warn!(status = r.status().as_u16(), "OTLP collector rejected spans"),
277                        Err(e) => tracing::warn!(error = %e, "cannot export spans to the OTLP collector"),
278                    }
279                }
280            })
281            .expect("start OTLP exporter thread");
282        Exporter { tx }
283    }
284
285    pub fn export(&self, spans: Vec<Value>) {
286        if spans.is_empty() {
287            return;
288        }
289        if self.tx.try_send(spans).is_err() {
290            tracing::warn!("OTLP export queue is full; dropping spans");
291        }
292    }
293}
orbit_runtime/telemetry.rs

orbit_runtime/
telemetry.rs
