1use chrono::{Duration, Utc};
2use ferrum_types::{
3 FerrumError, FerrumProfileEvent, MemorySnapshot, ProfileEntrypoint, ProfileEventKind,
4 ProfileStatus, ReplayReference, ResourceAction, ResourceTraceEvent, Result,
5 OBSERVABILITY_PROFILE_SCHEMA_VERSION, SYNTHETIC_RUNTIME_PRESET_HASH,
6};
7use serde::Serialize;
8use serde_json::{json, Value};
9use sha2::{Digest, Sha256};
10use std::collections::BTreeMap;
11use std::fs;
12use std::path::Path;
13use uuid::Uuid;
14
15const SYNTHETIC_MODEL: &str = "synthetic/no-weight";
16const SYNTHETIC_BACKEND: &str = "synthetic";
17
18pub fn write_observability_vertical_slice(
19 entrypoint: ProfileEntrypoint,
20 out_dir: &Path,
21) -> Result<()> {
22 let request_id = format!(
23 "obs-{}-{}",
24 entrypoint_label(entrypoint),
25 Uuid::new_v4().simple()
26 );
27 let request_dump_dir = out_dir.join("request_dump");
28 let bundle_dir = request_dump_dir.join(&request_id);
29 let replay_args = replay_command_args(entrypoint, out_dir, &request_dump_dir);
30 let replay_command = replay_command(&replay_args);
31 fs_create_dir_all(&request_dump_dir)?;
32 fs_create_dir_all(&bundle_dir)?;
33
34 let request_dump_path = request_dump_dir.join("request.json");
35 let replay_command_path = out_dir.join("replay_command.txt");
36 let request_dump_replay_command_path = request_dump_dir.join("replay_command.txt");
37 let profile_path = out_dir.join("profile.jsonl");
38 let summary_path = out_dir.join("observability_profile_summary.json");
39
40 let request = request_dump(entrypoint, &request_id, &replay_command);
41 write_json(&request_dump_path, &request)?;
42 fs_write(&replay_command_path, format!("{replay_command}\n"))?;
43 fs_write(
44 &request_dump_replay_command_path,
45 format!("{replay_command}\n"),
46 )?;
47 write_replay_bundle(
48 &bundle_dir,
49 entrypoint,
50 &request_id,
51 &replay_command,
52 &replay_args,
53 &request,
54 out_dir,
55 &request_dump_dir,
56 )?;
57
58 let events = synthetic_events(
59 entrypoint,
60 &request_id,
61 &replay_command,
62 &request_dump_dir.to_string_lossy(),
63 );
64 write_profile_jsonl(&profile_path, &events)?;
65
66 let summary = json!({
67 "schema_version": OBSERVABILITY_PROFILE_SCHEMA_VERSION,
68 "entrypoint": entrypoint_label(entrypoint),
69 "backend": SYNTHETIC_BACKEND,
70 "model": SYNTHETIC_MODEL,
71 "l0_only": true,
72 "status": "pass",
73 "request_count": 1,
74 "failed_count": 0,
75 "corrupted_count": 0,
76 "bad_text_count": 0,
77 "oom_prevented_count": 0,
78 "silent_oom_count": 0,
79 "latency_us": {
80 "p50": 320,
81 "p95": 320,
82 "p99": 320
83 },
84 "memory_high_water_bytes": 1536,
85 "resource_leak_count": 0,
86 "top_slow_phases": [
87 {"phase": "synthetic_decode", "duration_us": 200},
88 {"phase": "synthetic_prefill", "duration_us": 120}
89 ],
90 "first_failure_event": null,
91 "profile_jsonl": profile_path.to_string_lossy(),
92 "request_dump": request_dump_path.to_string_lossy(),
93 "request_dump_dir": request_dump_dir.to_string_lossy(),
94 "replay_bundle_dir": bundle_dir.to_string_lossy(),
95 "replay_command": replay_command,
96 "replay_command_path": replay_command_path.to_string_lossy()
97 });
98 write_json(&summary_path, &summary)?;
99 Ok(())
100}
101
102fn synthetic_events(
103 entrypoint: ProfileEntrypoint,
104 request_id: &str,
105 replay_command: &str,
106 bundle_dir: &str,
107) -> Vec<FerrumProfileEvent> {
108 let base = Utc::now();
109 let request_open = resource_event(
110 entrypoint,
111 request_id,
112 "request_open",
113 ResourceTraceEvent {
114 owner_kind: "request".to_string(),
115 owner_id: request_id.to_string(),
116 resource_kind: "request_slot".to_string(),
117 action: ResourceAction::RequestOpen,
118 amount: None,
119 before: None,
120 after: None,
121 capacity: Some(1),
122 underflow_amount: None,
123 reason: None,
124 error_kind: None,
125 message: None,
126 resource_error_kind: None,
127 },
128 base,
129 );
130 let reserve = resource_event(
131 entrypoint,
132 request_id,
133 "kv_reserve",
134 ResourceTraceEvent {
135 owner_kind: "request".to_string(),
136 owner_id: request_id.to_string(),
137 resource_kind: "kv_block".to_string(),
138 action: ResourceAction::Reserve,
139 amount: Some(1),
140 before: Some(0),
141 after: Some(1),
142 capacity: Some(4),
143 underflow_amount: None,
144 reason: None,
145 error_kind: None,
146 message: None,
147 resource_error_kind: None,
148 },
149 base + Duration::microseconds(10),
150 );
151 let commit = resource_event(
152 entrypoint,
153 request_id,
154 "kv_commit",
155 ResourceTraceEvent {
156 owner_kind: "request".to_string(),
157 owner_id: request_id.to_string(),
158 resource_kind: "kv_block".to_string(),
159 action: ResourceAction::Commit,
160 amount: Some(1),
161 before: Some(0),
162 after: Some(1),
163 capacity: Some(4),
164 underflow_amount: None,
165 reason: None,
166 error_kind: None,
167 message: None,
168 resource_error_kind: None,
169 },
170 base + Duration::microseconds(15),
171 );
172 let prefill = timed_event(
173 entrypoint,
174 request_id,
175 "synthetic_prefill",
176 120,
177 1024,
178 1280,
179 1280,
180 attrs([("input_tokens", json!(8)), ("output_tokens", json!(0))]),
181 base + Duration::microseconds(20),
182 );
183 let decode = timed_event(
184 entrypoint,
185 request_id,
186 "synthetic_decode",
187 200,
188 1280,
189 1536,
190 1536,
191 attrs([("input_tokens", json!(8)), ("output_tokens", json!(4))]),
192 base + Duration::microseconds(140),
193 );
194 let release = resource_event(
195 entrypoint,
196 request_id,
197 "kv_release",
198 ResourceTraceEvent {
199 owner_kind: "request".to_string(),
200 owner_id: request_id.to_string(),
201 resource_kind: "kv_block".to_string(),
202 action: ResourceAction::Release,
203 amount: Some(1),
204 before: Some(1),
205 after: Some(0),
206 capacity: Some(4),
207 underflow_amount: None,
208 reason: None,
209 error_kind: None,
210 message: None,
211 resource_error_kind: None,
212 },
213 base + Duration::microseconds(340),
214 );
215 let mut complete = base_event(
216 entrypoint,
217 request_id,
218 "request_complete",
219 ProfileEventKind::Instant,
220 base + Duration::microseconds(350),
221 );
222 complete.status = ProfileStatus::DiagnosticOnly;
223 complete.replay = Some(ReplayReference {
224 command: replay_command.to_string(),
225 bundle_dir: Some(bundle_dir.to_string()),
226 });
227 complete.attributes = attrs([
228 ("l0_only", json!(true)),
229 ("response_text", json!("synthetic ok")),
230 ]);
231 vec![
232 request_open,
233 reserve,
234 commit,
235 prefill,
236 decode,
237 release,
238 complete,
239 ]
240}
241
242fn base_event(
243 entrypoint: ProfileEntrypoint,
244 request_id: &str,
245 phase: &str,
246 event_kind: ProfileEventKind,
247 timestamp: chrono::DateTime<Utc>,
248) -> FerrumProfileEvent {
249 FerrumProfileEvent {
250 schema_version: OBSERVABILITY_PROFILE_SCHEMA_VERSION,
251 ts_unix_nanos: timestamp
252 .timestamp_nanos_opt()
253 .unwrap_or_else(|| timestamp.timestamp_micros() * 1_000),
254 event_id: format!("evt-{}-{phase}", entrypoint_label(entrypoint)),
255 request_id: request_id.to_string(),
256 correlation_id: Some(format!("corr-{}", entrypoint_label(entrypoint))),
257 entrypoint,
258 backend: SYNTHETIC_BACKEND.to_string(),
259 runtime_preset_hash: SYNTHETIC_RUNTIME_PRESET_HASH.to_string(),
260 phase: phase.to_string(),
261 event_kind,
262 timestamp,
263 status: ProfileStatus::Ok,
264 model: Some(SYNTHETIC_MODEL.to_string()),
265 duration_us: None,
266 memory: None,
267 resource: None,
268 error: None,
269 replay: None,
270 shape: BTreeMap::from([("batch_size".to_string(), json!(1))]),
271 backend_detail: None,
272 attributes: BTreeMap::new(),
273 }
274}
275
276fn resource_event(
277 entrypoint: ProfileEntrypoint,
278 request_id: &str,
279 phase: &str,
280 resource: ResourceTraceEvent,
281 timestamp: chrono::DateTime<Utc>,
282) -> FerrumProfileEvent {
283 let mut event = base_event(
284 entrypoint,
285 request_id,
286 phase,
287 ProfileEventKind::Resource,
288 timestamp,
289 );
290 event.resource = Some(resource);
291 event
292}
293
294#[allow(clippy::too_many_arguments)]
295fn timed_event(
296 entrypoint: ProfileEntrypoint,
297 request_id: &str,
298 phase: &str,
299 duration_us: u64,
300 before_bytes: u64,
301 after_bytes: u64,
302 high_water_bytes: u64,
303 attributes: BTreeMap<String, Value>,
304 timestamp: chrono::DateTime<Utc>,
305) -> FerrumProfileEvent {
306 let mut event = base_event(
307 entrypoint,
308 request_id,
309 phase,
310 ProfileEventKind::TimedSpan,
311 timestamp,
312 );
313 event.duration_us = Some(duration_us);
314 event.memory = Some(MemorySnapshot {
315 scope: "process".to_string(),
316 backend: Some(SYNTHETIC_BACKEND.to_string()),
317 before_bytes: Some(before_bytes),
318 after_bytes: Some(after_bytes),
319 current_bytes: Some(after_bytes),
320 high_water_bytes: Some(high_water_bytes),
321 available_bytes: Some(1024 * 1024),
322 });
323 event.attributes = attributes;
324 event
325}
326
327fn request_dump(
328 entrypoint: ProfileEntrypoint,
329 request_id: &str,
330 replay_command: &str,
331) -> serde_json::Value {
332 match entrypoint {
333 ProfileEntrypoint::Run => json!({
334 "schema_version": OBSERVABILITY_PROFILE_SCHEMA_VERSION,
335 "entrypoint": "run",
336 "request_id": request_id,
337 "model": SYNTHETIC_MODEL,
338 "backend": SYNTHETIC_BACKEND,
339 "l0_only": true,
340 "sanitized": true,
341 "prompt": "observability vertical slice",
342 "sampling": {"max_tokens": 4, "temperature": 0.0},
343 "replay_command": replay_command
344 }),
345 ProfileEntrypoint::Serve => json!({
346 "schema_version": OBSERVABILITY_PROFILE_SCHEMA_VERSION,
347 "entrypoint": "serve",
348 "request_id": request_id,
349 "model": SYNTHETIC_MODEL,
350 "backend": SYNTHETIC_BACKEND,
351 "l0_only": true,
352 "sanitized": true,
353 "http": {
354 "method": "POST",
355 "path": "/v1/chat/completions",
356 "body": {
357 "model": SYNTHETIC_MODEL,
358 "messages": [{"role": "user", "content": "observability vertical slice"}],
359 "max_tokens": 4,
360 "temperature": 0.0
361 }
362 },
363 "replay_command": replay_command
364 }),
365 other => json!({
366 "schema_version": OBSERVABILITY_PROFILE_SCHEMA_VERSION,
367 "entrypoint": entrypoint_label(other),
368 "request_id": request_id,
369 "model": SYNTHETIC_MODEL,
370 "backend": SYNTHETIC_BACKEND,
371 "l0_only": true,
372 "sanitized": true,
373 "replay_command": replay_command
374 }),
375 }
376}
377
378#[allow(clippy::too_many_arguments)]
379fn write_replay_bundle(
380 bundle_dir: &Path,
381 entrypoint: ProfileEntrypoint,
382 request_id: &str,
383 replay_command_text: &str,
384 replay_args: &[String],
385 request: &serde_json::Value,
386 out_dir: &Path,
387 request_dump_dir: &Path,
388) -> Result<()> {
389 let output_text = "synthetic ok";
390 let output_text_body = format!("{output_text}\n");
391 let engine_replay_args = engine_replay_command_args(bundle_dir);
392 let engine_replay_command = replay_command(&engine_replay_args);
393 let files = [
394 ("request.json", request.clone()),
395 (
396 "prompt_token_ids.json",
397 json!({
398 "schema_version": OBSERVABILITY_PROFILE_SCHEMA_VERSION,
399 "request_id": request_id,
400 "model": SYNTHETIC_MODEL,
401 "tokenizer_or_model": SYNTHETIC_MODEL,
402 "token_ids": [101, 202, 303, 404],
403 "token_count": 4,
404 "sanitized": true
405 }),
406 ),
407 (
408 "sampling_params.json",
409 json!({
410 "schema_version": OBSERVABILITY_PROFILE_SCHEMA_VERSION,
411 "request_id": request_id,
412 "sampling_params": {"max_tokens": 4, "temperature": 0.0},
413 "unavailable_reason": null
414 }),
415 ),
416 (
417 "runtime_effective_config.json",
418 json!({
419 "schema_version": OBSERVABILITY_PROFILE_SCHEMA_VERSION,
420 "request_id": request_id,
421 "entrypoint": entrypoint_label(entrypoint),
422 "profile_detail": "basic",
423 "profile_sample_rate": 1.0,
424 "profile_jsonl": out_dir.join("profile.jsonl").to_string_lossy(),
425 "memory_profile_jsonl": out_dir.join("memory_profile.jsonl").to_string_lossy(),
426 "scheduler_trace_jsonl": out_dir.join("scheduler_trace.jsonl").to_string_lossy(),
427 "request_dump_dir": request_dump_dir.to_string_lossy(),
428 "sanitized": true
429 }),
430 ),
431 (
432 "backend_selection.json",
433 json!({
434 "schema_version": OBSERVABILITY_PROFILE_SCHEMA_VERSION,
435 "request_id": request_id,
436 "backend": SYNTHETIC_BACKEND,
437 "model": SYNTHETIC_MODEL,
438 "l0_only": true
439 }),
440 ),
441 (
442 "output_token_ids.json",
443 json!({
444 "schema_version": OBSERVABILITY_PROFILE_SCHEMA_VERSION,
445 "request_id": request_id,
446 "token_ids": [909, 808],
447 "token_count": 2,
448 "finish_reason": "stop"
449 }),
450 ),
451 (
452 "bad_output_scan.json",
453 json!({
454 "schema_version": OBSERVABILITY_PROFILE_SCHEMA_VERSION,
455 "request_id": request_id,
456 "bad_output": false,
457 "bad_text_count": 0,
458 "reasons": [],
459 "first_bad_text_span": null,
460 "failure_kind": null,
461 "output_chars": output_text.chars().count(),
462 "classified_output_sha256": sha256_hex(output_text.as_bytes()),
463 "output_sha256": sha256_hex(output_text_body.as_bytes())
464 }),
465 ),
466 (
467 "replay.command.json",
468 json!({
469 "schema_version": OBSERVABILITY_PROFILE_SCHEMA_VERSION,
470 "request_id": request_id,
471 "entrypoint": entrypoint_label(entrypoint),
472 "command": replay_command_text,
473 "argv": replay_args,
474 "bundle_dir": bundle_dir.to_string_lossy(),
475 "engine_replay": {
476 "mode": "bundle_offline",
477 "requires_http_server": false,
478 "command": engine_replay_command,
479 "argv": engine_replay_args
480 },
481 "sanitized": true
482 }),
483 ),
484 ];
485 for (name, value) in files {
486 write_json(&bundle_dir.join(name), &value)?;
487 }
488 fs_write(
489 bundle_dir.join("output_text.txt").as_path(),
490 output_text_body,
491 )?;
492 Ok(())
493}
494
495fn replay_command_args(
496 entrypoint: ProfileEntrypoint,
497 out_dir: &Path,
498 request_dump_dir: &Path,
499) -> Vec<String> {
500 let subcommand = match entrypoint {
501 ProfileEntrypoint::Run => "run",
502 ProfileEntrypoint::Serve => "serve",
503 ProfileEntrypoint::BenchServe => "bench-serve",
504 ProfileEntrypoint::Synthetic => "run",
505 };
506 vec![
507 "cargo".to_string(),
508 "run".to_string(),
509 "-p".to_string(),
510 "ferrum-cli".to_string(),
511 "--".to_string(),
512 subcommand.to_string(),
513 SYNTHETIC_MODEL.to_string(),
514 "--profile-detail".to_string(),
515 "basic".to_string(),
516 "--profile-sample-rate".to_string(),
517 "1".to_string(),
518 "--profile-jsonl".to_string(),
519 out_dir.join("profile.jsonl").to_string_lossy().to_string(),
520 "--memory-profile-jsonl".to_string(),
521 out_dir
522 .join("memory_profile.jsonl")
523 .to_string_lossy()
524 .to_string(),
525 "--scheduler-trace-jsonl".to_string(),
526 out_dir
527 .join("scheduler_trace.jsonl")
528 .to_string_lossy()
529 .to_string(),
530 "--request-dump-dir".to_string(),
531 request_dump_dir.to_string_lossy().to_string(),
532 ]
533}
534
535fn engine_replay_command_args(bundle_dir: &Path) -> Vec<String> {
536 vec![
537 "cargo".to_string(),
538 "run".to_string(),
539 "-p".to_string(),
540 "ferrum-cli".to_string(),
541 "--".to_string(),
542 "replay-bundle".to_string(),
543 bundle_dir.to_string_lossy().to_string(),
544 "--out".to_string(),
545 bundle_dir
546 .join("engine_replay")
547 .to_string_lossy()
548 .to_string(),
549 "--json".to_string(),
550 ]
551}
552
553fn replay_command(args: &[String]) -> String {
554 args.iter()
555 .map(|part| shell_quote(part))
556 .collect::<Vec<_>>()
557 .join(" ")
558}
559
560fn sha256_hex(bytes: &[u8]) -> String {
561 let mut hasher = Sha256::new();
562 hasher.update(bytes);
563 format!("{:x}", hasher.finalize())
564}
565
566fn shell_quote(value: &str) -> String {
567 if value
568 .chars()
569 .all(|ch| ch.is_ascii_alphanumeric() || "-_./:".contains(ch))
570 {
571 return value.to_string();
572 }
573 format!("'{}'", value.replace('\'', "'\\''"))
574}
575
576fn entrypoint_label(entrypoint: ProfileEntrypoint) -> &'static str {
577 match entrypoint {
578 ProfileEntrypoint::Run => "run",
579 ProfileEntrypoint::Serve => "serve",
580 ProfileEntrypoint::BenchServe => "bench_serve",
581 ProfileEntrypoint::Synthetic => "synthetic",
582 }
583}
584
585fn attrs<const N: usize>(entries: [(&str, Value); N]) -> BTreeMap<String, Value> {
586 entries
587 .into_iter()
588 .map(|(key, value)| (key.to_string(), value))
589 .collect()
590}
591
592fn write_profile_jsonl(path: &Path, events: &[FerrumProfileEvent]) -> Result<()> {
593 let mut body = String::new();
594 for event in events {
595 event.validate().map_err(|err| {
596 FerrumError::internal(format!(
597 "invalid observability vertical slice event {}: {err}",
598 event.event_id
599 ))
600 })?;
601 body.push_str(&serde_json::to_string(event).map_err(|err| {
602 FerrumError::serialization(format!(
603 "failed to serialize observability event {}: {err}",
604 event.event_id
605 ))
606 })?);
607 body.push('\n');
608 }
609 fs_write(path, body)
610}
611
612fn write_json<T: Serialize>(path: &Path, value: &T) -> Result<()> {
613 let body = serde_json::to_string_pretty(value)
614 .map_err(|err| FerrumError::serialization(format!("failed to serialize JSON: {err}")))?;
615 fs_write(path, format!("{body}\n"))
616}
617
618fn fs_create_dir_all(path: &Path) -> Result<()> {
619 fs::create_dir_all(path)
620 .map_err(|err| FerrumError::io(format!("failed to create {}: {err}", path.display())))
621}
622
623fn fs_write(path: &Path, content: impl AsRef<[u8]>) -> Result<()> {
624 if let Some(parent) = path.parent() {
625 fs_create_dir_all(parent)?;
626 }
627 fs::write(path, content)
628 .map_err(|err| FerrumError::io(format!("failed to write {}: {err}", path.display())))
629}
630
631#[cfg(test)]
632mod tests {
633 use super::*;
634
635 #[test]
636 fn writes_required_run_artifacts() {
637 let root = std::env::temp_dir().join(format!(
638 "ferrum-observability-vertical-slice-{}",
639 Uuid::new_v4().simple()
640 ));
641 write_observability_vertical_slice(ProfileEntrypoint::Run, &root).unwrap();
642 assert!(root.join("profile.jsonl").is_file());
643 assert!(root.join("request_dump/request.json").is_file());
644 assert!(root.join("replay_command.txt").is_file());
645 assert!(root.join("observability_profile_summary.json").is_file());
646 let bundle_dir = fs::read_dir(root.join("request_dump"))
647 .unwrap()
648 .flatten()
649 .find_map(|entry| entry.path().is_dir().then_some(entry.path()))
650 .expect("request replay bundle directory should exist");
651 for name in [
652 "request.json",
653 "prompt_token_ids.json",
654 "sampling_params.json",
655 "runtime_effective_config.json",
656 "backend_selection.json",
657 "output_token_ids.json",
658 "output_text.txt",
659 "bad_output_scan.json",
660 "replay.command.json",
661 ] {
662 assert!(bundle_dir.join(name).is_file(), "missing {name}");
663 }
664 let replay: serde_json::Value = serde_json::from_str(
665 &fs::read_to_string(bundle_dir.join("replay.command.json")).unwrap(),
666 )
667 .unwrap();
668 let argv = replay["argv"].as_array().unwrap();
669 assert!(argv.iter().any(|part| part == "--request-dump-dir"));
670 let engine_argv = replay["engine_replay"]["argv"].as_array().unwrap();
671 assert!(engine_argv.iter().any(|part| part == "replay-bundle"));
672 assert_eq!(replay["engine_replay"]["requires_http_server"], false);
673 let output_text_bytes = fs::read(bundle_dir.join("output_text.txt")).unwrap();
674 let bad_scan: serde_json::Value = serde_json::from_str(
675 &fs::read_to_string(bundle_dir.join("bad_output_scan.json")).unwrap(),
676 )
677 .unwrap();
678 assert_eq!(
679 bad_scan["classified_output_sha256"],
680 sha256_hex(b"synthetic ok")
681 );
682 assert_eq!(bad_scan["output_sha256"], sha256_hex(&output_text_bytes));
683 let profile = fs::read_to_string(root.join("profile.jsonl")).unwrap();
684 assert!(profile.contains("\"entrypoint\":\"run\""));
685 assert!(profile.contains("\"replay\""));
686 fs::remove_dir_all(root).ok();
687 }
688}