1use std::fmt::Write as _;
2
3use serde_json::{json, Value};
4
5use crate::api::EventMsg;
6use crate::wal::EventBody;
7
8pub const GENAI_SEMCONV_VERSION: &str = "1.36.0";
9
10#[derive(Debug, Clone)]
11pub struct ExportOptions {
12 pub include_content: bool,
13 pub max_field_bytes: usize,
14 pub model: String,
15 pub store_ns: u64,
16}
17
18impl Default for ExportOptions {
19 fn default() -> Self {
20 ExportOptions {
21 include_content: false,
22 max_field_bytes: 64 * 1024,
23 model: String::new(),
24 store_ns: 0,
25 }
26 }
27}
28
29pub fn redact(text: &str, max: usize) -> String {
30 let mut out = String::with_capacity(text.len().min(max));
31 for (i, word) in text.split(' ').enumerate() {
32 if i > 0 {
33 out.push(' ');
34 }
35 let w = word.trim_matches(|c: char| !c.is_ascii_alphanumeric() && c != '-' && c != '_');
36 let secret = (w.starts_with("sk-") && w.len() >= 24)
37 || (w.starts_with("AKIA") && w.len() == 20 && w.chars().all(|c| c.is_ascii_uppercase() || c.is_ascii_digit()))
38 || (w.starts_with("ghp_") && w.len() >= 36)
39 || (w.starts_with("xox") && w.len() >= 24 && w.as_bytes().get(4) == Some(&b'-'))
40 || (w.starts_with("AIza") && w.len() == 39);
41 if secret {
42 out.push_str("[REDACTED]");
43 } else {
44 out.push_str(word);
45 }
46 }
47 let mut out = out.replace("PRIVATE KEY-----", "PRIVATE KEY-----[REDACTED]");
48 if let Some(i) = out.find("-----BEGIN") {
49 if let Some(j) = out[i..].find("-----END") {
50 out.replace_range(i..i + j, "-----BEGIN [REDACTED] ");
51 }
52 }
53 if out.len() > max {
54 let mut cut = max;
55 while !out.is_char_boundary(cut) {
56 cut -= 1;
57 }
58 out.truncate(cut);
59 out.push_str("…[truncated]");
60 }
61 out
62}
63
64fn content(text: &str, o: &ExportOptions) -> Value {
65 if o.include_content {
66 json!(redact(text, o.max_field_bytes))
67 } else {
68 json!({ "bytes": text.len() })
69 }
70}
71
72fn body_json(body: &EventBody, o: &ExportOptions) -> Value {
73 match body {
74 EventBody::Created { parent, params } => json!({"type":"created","parent":parent,"params":params_json(params)}),
75 EventBody::Appended { text, span } => json!({"type":"appended","text": text.as_deref().map(|t| content(t, o)),"tokens":span.len()}),
76 EventBody::Message { role, text, span } => json!({"type":"message","role":role,"text":content(text,o),"tokens":span.len()}),
77 EventBody::GenerationPrompt { span } => json!({"type":"generation_prompt","tokens":span.len()}),
78 EventBody::Generated { span, text, channel, finish } => json!({"type":"generated","tokens":span.len(),"text":content(text,o),"channel":channel,"finish":finish}),
79 EventBody::ToolUse { name, arguments } => json!({"type":"tool_use","name":name,"arguments":content(arguments,o)}),
80 EventBody::ToolResult { call_id, content: c, span } => json!({"type":"tool_result","call_id":call_id,"content":content(c,o),"tokens":span.len()}),
81 EventBody::EpochBump => json!({"type":"epoch_bump"}),
82 EventBody::GenerationFingerprint { digest } => json!({"type":"generation_fingerprint","digest":format!("{digest:016x}")}),
83 EventBody::ToolParseFailure { raw } => json!({"type":"tool_parse_failure","raw":content(raw,o)}),
84 EventBody::Forked { parent, fork_at, params } => json!({"type":"forked","parent":parent,"fork_at":fork_at,"params":params_json(params)}),
85 EventBody::MetaUpdated { version, title, archived } => json!({"type":"meta_updated","version":version,"title":title.as_deref().map(|t| content(t,o)),"archived":archived}),
86 EventBody::Rebased { parent, fork_at, params, edits } => json!({"type":"rebased","parent":parent,"fork_at":fork_at,"params":params_json(params),"edits":edits.len()}),
87 EventBody::ToolExpired { call_id, reason } => json!({"type":"tool_expired","call_id":call_id,"reason":reason}),
88 EventBody::Purged { generation, descendants } => json!({"type":"purged","generation":generation,"descendants":format!("{descendants:?}")}),
89 EventBody::Rerooted { purged_parent, prefix } => json!({"type":"rerooted","purged_parent":purged_parent,"prefix_events":prefix.len()}),
90 EventBody::QosSet { class, batch_invariant } => json!({"type":"qos_set","class":class,"batch_invariant":batch_invariant}),
91 EventBody::Block { role, kind, payload, span, visibility_version } => json!({"type":"block","role":role,"kind":kind,"payload":content(payload,o),"tokens":span.len(),"visibility_version":visibility_version}),
92 EventBody::PermissionRequest { call_id, text } => json!({"type":"permission_request","call_id":call_id,"text":content(text,o)}),
93 EventBody::PermissionResponse { request_id, granted } => json!({"type":"permission_response","request_id":request_id,"granted":granted}),
94 EventBody::ToolCancelRequested { call_id } => json!({"type":"tool_cancel_requested","call_id":call_id}),
95 EventBody::ToolOutcome { call_id, outcome, note } => json!({"type":"tool_outcome","call_id":call_id,"outcome":outcome,"note":content(note,o)}),
96 EventBody::ToolReconciliation { call_id, note } => json!({"type":"tool_reconciliation","call_id":call_id,"note":content(note,o)}),
97 EventBody::ToolLease { call_id, deadline_unix_ms } => json!({"type":"tool_lease","call_id":call_id,"deadline_unix_ms":deadline_unix_ms}),
98 EventBody::Pinned { deadline_unix_ms } => json!({"type":"pinned","deadline_unix_ms":deadline_unix_ms}),
99 }
100}
101
102fn params_json(p: &crate::GenParams) -> Value {
103 json!({"temperature":p.temperature,"top_p":p.top_p,"min_p":p.min_p,"top_k":p.top_k,"seed":p.seed})
104}
105
106pub fn jsonl(session: u64, events: &[EventMsg], o: &ExportOptions) -> String {
107 let mut out = String::new();
108 for e in events {
109 let line = json!({
110 "session": session,
111 "event_id": e.event_id,
112 "epoch": e.epoch,
113 "ts_unix_ms": e.ts_unix_ms,
114 "event": body_json(&e.body, o),
115 });
116 let _ = writeln!(out, "{line}");
117 }
118 out
119}
120
121pub(crate) fn hex(bytes: &[u8]) -> String {
122 bytes.iter().map(|b| format!("{b:02x}")).collect()
123}
124
125pub fn hex_id(bytes: &[u8]) -> String {
126 hex(bytes)
127}
128
129pub fn trace_id_bytes(session: u64) -> [u8; 16] {
130 let h1 = xxhash_rust::xxh3::xxh3_64(&session.to_le_bytes());
131 let h2 = xxhash_rust::xxh3::xxh3_64_with_seed(&session.to_le_bytes(), 0x5e55);
132 let mut out = [0u8; 16];
133 out[..8].copy_from_slice(&h1.to_be_bytes());
134 out[8..].copy_from_slice(&h2.to_be_bytes());
135 out
136}
137
138pub fn store_trace_ns(id: &crate::store::StoreId) -> u64 {
139 (xxhash_rust::xxh3::xxh3_64_with_seed(&id.0, 0x5354) >> 1).max(1)
140}
141
142pub fn trace_id_bytes_ns(ns: u64, session: u64) -> [u8; 16] {
143 if ns == 0 {
144 return trace_id_bytes(session);
145 }
146 let mut b = [0u8; 16];
147 b[..8].copy_from_slice(&ns.to_le_bytes());
148 b[8..].copy_from_slice(&session.to_le_bytes());
149 let h1 = xxhash_rust::xxh3::xxh3_64(&b);
150 let h2 = xxhash_rust::xxh3::xxh3_64_with_seed(&b, 0x5e55);
151 let mut out = [0u8; 16];
152 out[..8].copy_from_slice(&h1.to_be_bytes());
153 out[8..].copy_from_slice(&h2.to_be_bytes());
154 out
155}
156
157pub fn span_id_bytes_ns(ns: u64, session: u64, event_id: u64) -> [u8; 8] {
158 if ns == 0 {
159 return span_id_bytes(session, event_id);
160 }
161 let mut b = [0u8; 24];
162 b[..8].copy_from_slice(&ns.to_le_bytes());
163 b[8..16].copy_from_slice(&session.to_le_bytes());
164 b[16..].copy_from_slice(&event_id.to_le_bytes());
165 xxhash_rust::xxh3::xxh3_64(&b).to_be_bytes()
166}
167
168pub fn trace_id(session: u64) -> String {
169 hex(&trace_id_bytes(session))
170}
171
172pub fn span_id_bytes(session: u64, event_id: u64) -> [u8; 8] {
173 let mut b = [0u8; 16];
174 b[..8].copy_from_slice(&session.to_le_bytes());
175 b[8..].copy_from_slice(&event_id.to_le_bytes());
176 xxhash_rust::xxh3::xxh3_64(&b).to_be_bytes()
177}
178
179pub fn span_id(session: u64, event_id: u64) -> String {
180 hex(&span_id_bytes(session, event_id))
181}
182
183pub const ROOT: u64 = u64::MAX;
184
185fn ns(ms: u64) -> String {
186 (ms.saturating_mul(1_000_000)).to_string()
187}
188
189pub(crate) fn attr(key: &str, v: Value) -> Value {
190 let value = match v {
191 Value::String(s) => json!({"stringValue": s}),
192 Value::Bool(b) => json!({"boolValue": b}),
193 Value::Number(n) if n.is_i64() || n.is_u64() => json!({"intValue": n.to_string()}),
194 Value::Number(n) => json!({"doubleValue": n}),
195 other => json!({"stringValue": other.to_string()}),
196 };
197 json!({"key": key, "value": value})
198}
199
200pub fn otlp_jsonl(session: u64, events: &[EventMsg], o: &ExportOptions) -> String {
201 let store_ns = o.store_ns;
202 let tid = hex(&trace_id_bytes_ns(store_ns, session));
203 let span_id =
204 |session: u64, event_id: u64| hex(&span_id_bytes_ns(store_ns, session, event_id));
205 let first_ts = events.iter().map(|e| e.ts_unix_ms).find(|&t| t > 0).unwrap_or(0);
206 let last_ts = events.iter().map(|e| e.ts_unix_ms).max().unwrap_or(0);
207 let timing = if first_ts == 0 { "unavailable" } else { "wall_clock_ms" };
208 let mut spans: Vec<Value> = Vec::new();
209 spans.push(json!({
210 "traceId": tid, "spanId": span_id(session, ROOT),
211 "name": "session", "kind": 1,
212 "startTimeUnixNano": ns(first_ts), "endTimeUnixNano": ns(last_ts),
213 "attributes": [
214 attr("superfluid.session_id", json!(session)),
215 attr("superfluid.event_id", json!(ROOT)),
216 attr("superfluid.timing", json!(timing)),
217 ],
218 }));
219 let mut open_gen: Option<(u64, u64, u64, u32)> = None;
220 let mut open_tools: std::collections::HashMap<u64, (u64, u64, String, u64)> = Default::default();
221 for e in events {
222 match &e.body {
223 EventBody::GenerationPrompt { .. } => {
224 open_gen = Some((e.event_id, e.ts_unix_ms, 0, e.epoch as u32));
225 }
226 EventBody::Generated { span, finish, .. } => {
227 let g = open_gen.get_or_insert((e.event_id, e.ts_unix_ms, 0, e.epoch as u32));
228 g.2 += span.len() as u64;
229 if *finish != superfluid_abi::finish::NONE {
230 let (start_id, start_ts, tokens, epoch) = *g;
231 spans.push(json!({
232 "traceId": tid, "spanId": span_id(session, start_id),
233 "parentSpanId": span_id(session, ROOT),
234 "name": "generate", "kind": 1,
235 "startTimeUnixNano": ns(start_ts), "endTimeUnixNano": ns(e.ts_unix_ms),
236 "attributes": [
237 attr("superfluid.session_id", json!(session)),
238 attr("superfluid.event_id", json!(start_id)),
239 attr("superfluid.parent_event_id", json!(ROOT)),
240 attr("superfluid.epoch", json!(epoch)),
241 attr("superfluid.finish", json!(finish)),
242 attr("gen_ai.system", json!("superfluid")),
243 attr("gen_ai.request.model", json!(o.model.clone())),
244 attr("gen_ai.usage.output_tokens", json!(tokens)),
245 ],
246 }));
247 open_gen = None;
248 }
249 }
250 EventBody::ToolUse { name, .. } => {
251 open_tools.insert(e.event_id, (e.event_id, e.ts_unix_ms, name.clone(), e.epoch));
252 }
253 EventBody::ToolResult { call_id, .. }
254 | EventBody::ToolExpired { call_id, .. }
255 | EventBody::ToolOutcome { call_id, .. } => {
256 if let Some((start_id, start_ts, name, epoch)) = open_tools.remove(call_id) {
257 let parent = open_gen.map(|g| g.0).unwrap_or(ROOT);
258 let expired = matches!(e.body, EventBody::ToolExpired { .. });
259 let outcome = match &e.body {
260 EventBody::ToolOutcome { outcome, .. } => {
261 if *outcome == crate::wal::tool_outcome::CANCELLED {
262 "cancelled"
263 } else {
264 "failed"
265 }
266 }
267 _ if expired => "expired",
268 _ => "result",
269 };
270 spans.push(json!({
271 "traceId": tid, "spanId": span_id(session, start_id),
272 "parentSpanId": span_id(session, parent),
273 "name": "tool_use", "kind": 3,
274 "startTimeUnixNano": ns(start_ts), "endTimeUnixNano": ns(e.ts_unix_ms),
275 "attributes": [
276 attr("superfluid.session_id", json!(session)),
277 attr("superfluid.event_id", json!(start_id)),
278 attr("superfluid.parent_event_id", json!(parent)),
279 attr("superfluid.epoch", json!(epoch)),
280 attr("gen_ai.tool.name", json!(name)),
281 attr("gen_ai.tool.call.id", json!(start_id.to_string())),
282 attr("superfluid.tool.outcome", json!(outcome)),
283 ],
284 }));
285 }
286 }
287 _ => {}
288 }
289 }
290 for (_, (start_id, start_ts, name, epoch)) in open_tools {
291 spans.push(json!({
292 "traceId": tid, "spanId": span_id(session, start_id),
293 "parentSpanId": span_id(session, open_gen.map(|g| g.0).unwrap_or(ROOT)),
294 "name": "tool_use", "kind": 3,
295 "startTimeUnixNano": ns(start_ts), "endTimeUnixNano": ns(start_ts),
296 "attributes": [
297 attr("superfluid.session_id", json!(session)),
298 attr("superfluid.event_id", json!(start_id)),
299 attr("superfluid.epoch", json!(epoch)),
300 attr("gen_ai.tool.name", json!(name)),
301 attr("gen_ai.tool.call.id", json!(start_id.to_string())),
302 attr("superfluid.tool.outcome", json!("open")),
303 ],
304 }));
305 }
306 let req = trace_request(
307 vec![
308 attr("service.name", json!("superfluid")),
309 attr("superfluid.semconv_version", json!(GENAI_SEMCONV_VERSION)),
310 attr("superfluid.content_included", json!(o.include_content)),
311 ],
312 "superfluid.session",
313 spans,
314 );
315 format!("{req}\n")
316}
317
318pub(crate) fn trace_request(resource: Vec<Value>, scope: &str, spans: Vec<Value>) -> Value {
319 json!({
320 "resourceSpans": [{
321 "resource": {"attributes": resource},
322 "scopeSpans": [{
323 "scope": {"name": scope, "version": env!("CARGO_PKG_VERSION")},
324 "spans": spans,
325 }],
326 }]
327 })
328}
329
330pub fn push_otlp_traces(endpoint: &str, session: u64, events: &[EventMsg], o: &ExportOptions) -> Result<u16, String> {
331 let body = otlp_jsonl(session, events, o);
332 let url = format!("{}/v1/traces", endpoint.trim_end_matches('/'));
333 let agent = ureq::Agent::config_builder()
334 .timeout_global(Some(std::time::Duration::from_secs(10)))
335 .build()
336 .new_agent();
337 let resp = agent
338 .post(&url)
339 .header("content-type", "application/json")
340 .send(body.as_bytes())
341 .map_err(|e| format!("otlp push to {url}: {e}"))?;
342 Ok(resp.status().as_u16())
343}
344
345use std::sync::atomic::{AtomicBool, Ordering};
346use std::sync::Arc;
347use std::time::Duration;
348
349use crate::scheduler::{SchedStats, TTFT_BUCKETS_MS};
350
351pub fn otlp_metrics_json(s: &SchedStats, now_nanos: u64, start_nanos: u64) -> String {
352 let g = |a: &std::sync::atomic::AtomicU64| a.load(Ordering::Relaxed);
353 let now = now_nanos.to_string();
354 let start = start_nanos.to_string();
355
356 let sum = |name: &str, value: u64| {
357 json!({
358 "name": name, "unit": "1",
359 "sum": {
360 "aggregationTemporality": 2, "isMonotonic": true,
361 "dataPoints": [{
362 "asInt": value.to_string(),
363 "startTimeUnixNano": start, "timeUnixNano": now
364 }]
365 }
366 })
367 };
368 let gauge_pt = |name: &str, value: u64, attrs: Value| {
369 json!({
370 "name": name, "unit": "1",
371 "gauge": { "dataPoints": [{
372 "asInt": value.to_string(), "timeUnixNano": now, "attributes": attrs
373 }]}
374 })
375 };
376 let attr = |k: &str, v: &str| json!([{ "key": k, "value": { "stringValue": v } }]);
377
378 let mut metrics = vec![
379 sum("superfluid_ticks_total", g(&s.ticks)),
380 sum("superfluid_prefill_tokens_total", g(&s.prefill_tokens)),
381 sum("superfluid_decode_tokens_total", g(&s.decode_tokens)),
382 sum("superfluid_warm_prefix_tokens_total", g(&s.warm_prefix_tokens)),
383 sum("superfluid_cold_admissions_total", g(&s.cold_admissions)),
384 sum("superfluid_unservable_cold_admissions_total", g(&s.unservable_cold_admissions)),
385 sum("superfluid_unservable_park_refusals_total", g(&s.unservable_park_refusals)),
386 sum("superfluid_preemptions_total", g(&s.preemptions)),
387 sum("superfluid_starvation_grants_total", g(&s.starvation_grants)),
388 sum("superfluid_os_pressure_events_total", g(&s.os_pressure_events)),
389 sum("superfluid_pressure_evictions_total", g(&s.pressure_evictions)),
390 sum("superfluid_pressure_bytes_evicted_total", g(&s.pressure_bytes_evicted)),
391 sum("superfluid_worker_respawns_total", g(&s.worker_respawns)),
392 sum("superfluid_spec_proposed_total", g(&s.spec_proposed)),
393 sum("superfluid_spec_accepted_total", g(&s.spec_accepted)),
394 sum("superfluid_resumes_total", g(&s.resumes)),
395 sum("superfluid_pins_yielded_total", g(&s.pins_yielded)),
396 sum("superfluid_pins_expired_total", g(&s.pins_expired)),
397 ];
398
399 let park_points: Vec<Value> = [("lossless", g(&s.parks_lossless)), ("lossy", g(&s.parks_lossy))]
400 .into_iter()
401 .map(|(enc, v)| {
402 json!({
403 "asInt": v.to_string(), "startTimeUnixNano": start, "timeUnixNano": now,
404 "attributes": attr("encoding", enc)
405 })
406 })
407 .collect();
408 metrics.push(json!({
409 "name": "superfluid_parks_total", "unit": "1",
410 "sum": { "aggregationTemporality": 2, "isMonotonic": true, "dataPoints": park_points }
411 }));
412
413 metrics.push(gauge_pt("superfluid_lanes_active", g(&s.lanes_active), json!([])));
414 metrics.push(gauge_pt("superfluid_pins_held", g(&s.pins_held), json!([])));
415 metrics.push(gauge_pt("superfluid_pinned_bytes", g(&s.pinned_bytes), json!([])));
416 for (c, name) in ["interactive_chat", "inline_completion", "foreground_agent", "background_agent"]
417 .iter()
418 .enumerate()
419 {
420 metrics.push(gauge_pt("superfluid_queue_depth", g(&s.queue_depth[c]), attr("class", name)));
421 }
422
423 let bounds: Vec<f64> = TTFT_BUCKETS_MS.iter().map(|b| *b as f64 / 1000.0).collect();
424 let bucket_counts: Vec<String> = (0..=TTFT_BUCKETS_MS.len())
425 .map(|i| g(&s.ttft_buckets[i]).to_string())
426 .collect();
427 metrics.push(json!({
428 "name": "superfluid_ttft_seconds", "unit": "s",
429 "histogram": { "aggregationTemporality": 2, "dataPoints": [{
430 "startTimeUnixNano": start, "timeUnixNano": now,
431 "count": g(&s.ttft_count).to_string(),
432 "sum": g(&s.ttft_sum_ms) as f64 / 1000.0,
433 "bucketCounts": bucket_counts,
434 "explicitBounds": bounds
435 }]}
436 }));
437
438 serde_json::to_string(&json!({
439 "resourceMetrics": [{
440 "resource": { "attributes": attr("service.name", "superfluid") },
441 "scopeMetrics": [{
442 "scope": { "name": "superfluid", "version": env!("CARGO_PKG_VERSION") },
443 "metrics": metrics
444 }]
445 }]
446 }))
447 .expect("otlp metrics json")
448}
449
450pub fn push_otlp_metrics(endpoint: &str, s: &SchedStats) -> Result<u16, String> {
451 let now = crate::wal::now_unix_ms() * 1_000_000;
452 let body = otlp_metrics_json(s, now, now);
453 let url = format!("{}/v1/metrics", endpoint.trim_end_matches('/'));
454 let agent = ureq::Agent::config_builder()
455 .timeout_global(Some(Duration::from_secs(10)))
456 .build()
457 .new_agent();
458 let resp = agent
459 .post(&url)
460 .header("content-type", "application/json")
461 .send(body.as_bytes())
462 .map_err(|e| format!("otlp metrics push to {url}: {e}"))?;
463 Ok(resp.status().as_u16())
464}
465
466pub struct MetricsPusher {
467 stop: Arc<AtomicBool>,
468 handle: Option<std::thread::JoinHandle<()>>,
469}
470
471impl MetricsPusher {
472 pub const MIN_INTERVAL: Duration = Duration::from_millis(10);
473
474 pub fn start(endpoint: String, interval: Duration, stats: Arc<SchedStats>) -> MetricsPusher {
475 let interval = interval.max(Self::MIN_INTERVAL);
476 let stop = Arc::new(AtomicBool::new(false));
477 let s = Arc::clone(&stop);
478 let slice = Duration::from_millis(100).min(interval.max(Duration::from_millis(1)));
479 let handle = std::thread::spawn(move || {
480 while !s.load(Ordering::Relaxed) {
481 let mut waited = Duration::ZERO;
482 while waited < interval && !s.load(Ordering::Relaxed) {
483 std::thread::sleep(slice);
484 waited += slice;
485 }
486 if s.load(Ordering::Relaxed) {
487 break;
488 }
489 if let Err(e) = push_otlp_metrics(&endpoint, &stats) {
490 tracing::debug!("otlp metrics push failed (retry next interval): {e}");
491 }
492 }
493 });
494 MetricsPusher {
495 stop,
496 handle: Some(handle),
497 }
498 }
499}
500
501impl Drop for MetricsPusher {
502 fn drop(&mut self) {
503 self.stop.store(true, Ordering::Relaxed);
504 if let Some(h) = self.handle.take() {
505 let _ = h.join();
506 }
507 }
508}