use chrono::{Duration, Utc};
use clap::ValueEnum;
use ferrum_types::{
FerrumError, FerrumObservabilityConfig, FerrumProfileEvent, MemorySnapshot,
ObservabilityProfileDetail, ProfileEntrypoint, ProfileError, ProfileEventKind, ProfileStatus,
ReplayReference, ResourceAction, ResourceTraceEvent, Result, SamplingParams,
DEFAULT_OBSERVABILITY_PROFILE_SAMPLE_RATE, OBSERVABILITY_PROFILE_SCHEMA_VERSION,
};
use serde_json::{json, Value};
use sha2::{Digest, Sha256};
use std::collections::BTreeMap;
use std::fs;
use std::ops::Deref;
use std::path::{Path, PathBuf};
use uuid::Uuid;
const SYNTHETIC_MODEL: &str = "synthetic/no-weight";
const SYNTHETIC_BACKEND: &str = "synthetic";
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, ValueEnum)]
pub enum ProfileDetailArg {
#[default]
Off,
Basic,
Resource,
Latency,
Kernel,
Debug,
Replay,
Verify,
Full,
}
impl ProfileDetailArg {
pub fn as_str(self) -> &'static str {
match self {
Self::Off => "off",
Self::Basic => "basic",
Self::Resource => "resource",
Self::Latency => "latency",
Self::Kernel => "kernel",
Self::Debug => "debug",
Self::Replay => "replay",
Self::Verify => "verify",
Self::Full => "full",
}
}
}
impl From<ProfileDetailArg> for ObservabilityProfileDetail {
fn from(value: ProfileDetailArg) -> Self {
match value {
ProfileDetailArg::Off => Self::Off,
ProfileDetailArg::Basic => Self::Basic,
ProfileDetailArg::Resource => Self::Resource,
ProfileDetailArg::Latency => Self::Latency,
ProfileDetailArg::Kernel => Self::Kernel,
ProfileDetailArg::Debug => Self::Debug,
ProfileDetailArg::Replay => Self::Replay,
ProfileDetailArg::Verify => Self::Verify,
ProfileDetailArg::Full => Self::Full,
}
}
}
#[derive(Clone, Debug)]
pub struct ProductObservabilityConfig {
pub core: FerrumObservabilityConfig,
}
impl Deref for ProductObservabilityConfig {
type Target = FerrumObservabilityConfig;
fn deref(&self) -> &Self::Target {
&self.core
}
}
pub struct ActualRunObservation {
pub request_id: String,
pub duration_us: u64,
pub sampling_params: SamplingParams,
pub prompt_token_ids: Option<Vec<u32>>,
pub prompt_token_count: Option<usize>,
pub output_tokens: usize,
pub output_token_ids: Vec<u32>,
pub chunk_count: usize,
pub finish_reason: Option<String>,
pub prompt_chars: usize,
pub response_chars: usize,
pub response_text: String,
pub execution_evidence: Option<ferrum_types::InferenceExecutionEvidence>,
pub memory: Option<crate::memory_profile::ProcessMemoryObservation>,
pub memory_stages: Vec<ActualMemoryStageObservation>,
}
pub struct ActualRunFailureObservation {
pub request_id: String,
pub duration_us: u64,
pub sampling_params: SamplingParams,
pub prompt_token_ids: Option<Vec<u32>>,
pub prompt_token_count: Option<usize>,
pub prompt_chars: usize,
pub failure_kind: String,
pub error_kind: String,
pub error_message: String,
pub memory: Option<crate::memory_profile::ProcessMemoryObservation>,
pub memory_stages: Vec<ActualMemoryStageObservation>,
}
#[derive(Clone, Debug)]
pub struct ActualMemoryStageObservation {
pub phase: String,
pub stage: String,
pub duration_us: Option<u64>,
pub memory: Option<crate::memory_profile::ProcessMemoryObservation>,
pub attributes: BTreeMap<String, Value>,
}
impl ActualMemoryStageObservation {
pub fn new(
phase: impl Into<String>,
stage: impl Into<String>,
duration_us: Option<u64>,
memory: Option<crate::memory_profile::ProcessMemoryObservation>,
) -> Self {
Self {
phase: phase.into(),
stage: stage.into(),
duration_us,
memory,
attributes: BTreeMap::new(),
}
}
pub fn with_attribute(mut self, key: impl Into<String>, value: Value) -> Self {
self.attributes.insert(key.into(), value);
self
}
pub fn with_profile_run_status(
mut self,
executed: bool,
status: impl Into<String>,
source: impl Into<String>,
) -> Self {
self.attributes.extend([
("profile_run_executed".to_string(), json!(executed)),
("profile_run_status".to_string(), json!(status.into())),
("profile_run_source".to_string(), json!(source.into())),
]);
self
}
pub fn with_engine_cache_status(mut self, status: &ferrum_types::EngineStatus) -> Self {
let memory = &status.memory_usage;
let dynamic_capacity_bytes = memory.cache_memory_bytes.saturating_add(memory.free_bytes);
self.attributes.extend([
(
"kv_cache_total_bytes".to_string(),
json!(dynamic_capacity_bytes),
),
(
"kv_cache_used_bytes".to_string(),
json!(memory.cache_memory_bytes),
),
("kv_cache_free_bytes".to_string(), json!(memory.free_bytes)),
(
"cache_memory_bytes".to_string(),
json!(memory.cache_memory_bytes),
),
(
"available_kv_or_state_bytes".to_string(),
json!(memory.free_bytes),
),
(
"resource_total_bytes".to_string(),
json!(memory.total_bytes),
),
("resource_used_bytes".to_string(), json!(memory.used_bytes)),
("resource_free_bytes".to_string(), json!(memory.free_bytes)),
]);
self
}
}
impl ProductObservabilityConfig {
#[allow(clippy::too_many_arguments)]
pub fn new(
entrypoint: ProfileEntrypoint,
model: impl Into<String>,
profile_jsonl: Option<&PathBuf>,
profile_detail: ProfileDetailArg,
memory_profile_jsonl: Option<&PathBuf>,
scheduler_trace_jsonl: Option<&PathBuf>,
request_dump_dir: Option<&PathBuf>,
profile_sample_rate: f64,
) -> Self {
Self {
core: FerrumObservabilityConfig::new(
entrypoint,
model,
profile_jsonl.cloned(),
profile_detail.into(),
memory_profile_jsonl.cloned(),
scheduler_trace_jsonl.cloned(),
request_dump_dir.cloned(),
profile_sample_rate,
),
}
}
pub fn enabled(&self) -> bool {
self.core.enabled()
}
pub fn synthetic_no_weight_enabled(&self) -> bool {
self.core.synthetic_no_weight_enabled()
}
pub fn unified_product_profile_enabled(&self) -> bool {
self.core.unified_product_profile_enabled()
}
fn validate(&self) -> Result<()> {
self.core.validate().map_err(FerrumError::invalid_parameter)
}
}
pub fn default_profile_sample_rate() -> f64 {
DEFAULT_OBSERVABILITY_PROFILE_SAMPLE_RATE
}
pub fn write_synthetic_product_observability(
config: &ProductObservabilityConfig,
) -> Result<Vec<PathBuf>> {
config.validate()?;
let request_id = format!(
"product-obs-{}-{}",
entrypoint_label(config.entrypoint),
Uuid::new_v4().simple()
);
let replay_command = replay_command(config);
let events = product_events(config, &request_id, &replay_command);
let mut written = write_profile_outputs(
config,
&events,
ferrum_bench_core::JsonlJournalOpenMode::Truncate,
true,
true,
true,
)?;
if let Some(dir) = &config.request_dump_dir {
fs_create_dir_all(dir)?;
written.extend(write_replay_bundle(
dir,
config,
&request_id,
&replay_command,
ReplayBundleData {
request: request_dump(config, &request_id, &replay_command),
prompt_token_ids: Some(vec![101, 202, 303, 404]),
prompt_token_count: Some(4),
prompt_token_unavailable_reason: None,
sampling_params: Some(SamplingParams::greedy()),
backend: SYNTHETIC_BACKEND,
actual_model_smoke: false,
output_token_ids: Some(vec![909, 808]),
output_text: Some("synthetic ok"),
finish_reason: Some("stop"),
failure_kind: None,
failure_diagnostics: None,
},
)?);
}
Ok(written)
}
pub fn write_actual_run_observability(
config: &ProductObservabilityConfig,
observation: &ActualRunObservation,
) -> Result<Vec<PathBuf>> {
if !config.enabled() {
return Ok(Vec::new());
}
config.validate()?;
if let Some(timing) = observation
.execution_evidence
.as_ref()
.and_then(|evidence| evidence.engine_token_timing.as_ref())
{
timing
.validate(observation.output_tokens)
.map_err(FerrumError::invalid_parameter)?;
}
let replay_command = replay_command(config);
let events = actual_run_events(config, observation, &replay_command);
write_actual_run_artifacts(config, &events, observation, &replay_command)
}
pub fn write_actual_run_failure_observability(
config: &ProductObservabilityConfig,
observation: &ActualRunFailureObservation,
) -> Result<Vec<PathBuf>> {
if !config.enabled() {
return Ok(Vec::new());
}
config.validate()?;
let replay_command = replay_command(config);
let events = actual_run_failure_events(config, observation, &replay_command);
write_actual_run_failure_artifacts(config, &events, observation, &replay_command)
}
pub fn write_actual_serve_startup_observability(
config: &ProductObservabilityConfig,
startup_duration_us: u64,
startup_memory: Option<crate::memory_profile::ProcessMemoryObservation>,
memory_stages: Vec<ActualMemoryStageObservation>,
) -> Result<Vec<PathBuf>> {
if !config.unified_product_profile_enabled() {
return Ok(Vec::new());
}
config.validate()?;
let request_id = format!("serve-startup-{}", Uuid::new_v4().simple());
let replay_command = replay_command(config);
let events = actual_serve_startup_events(
config,
&request_id,
startup_duration_us,
startup_memory.as_ref(),
&memory_stages,
&replay_command,
);
write_actual_artifacts(config, &events, &request_id, &replay_command)
}
pub fn append_actual_serve_memory_stage_observability(
config: &ProductObservabilityConfig,
stage: ActualMemoryStageObservation,
) -> Result<Vec<PathBuf>> {
if !config.unified_product_profile_enabled() {
return Ok(Vec::new());
}
config.validate()?;
let request_id = format!("serve-memory-{}", Uuid::new_v4().simple());
let events = actual_memory_stage_events(config, &request_id, &[stage], Utc::now());
write_profile_outputs(
config,
&events,
ferrum_bench_core::JsonlJournalOpenMode::Append,
true,
true,
false,
)
}
fn write_actual_run_artifacts(
config: &ProductObservabilityConfig,
events: &[FerrumProfileEvent],
observation: &ActualRunObservation,
replay_command: &str,
) -> Result<Vec<PathBuf>> {
let mut written = write_actual_artifacts_with_scheduler(
config,
events,
&observation.request_id,
replay_command,
false,
)?;
if let Some(dir) = &config.request_dump_dir {
written.extend(write_replay_bundle(
dir,
config,
&observation.request_id,
replay_command,
ReplayBundleData {
request: actual_request_dump(config, &observation.request_id, replay_command),
prompt_token_ids: observation.prompt_token_ids.clone(),
prompt_token_count: observation.prompt_token_count,
prompt_token_unavailable_reason: observation.prompt_token_ids.is_none().then_some(
"rendered prompt token ids were unavailable for run one-shot observability",
),
sampling_params: Some(observation.sampling_params.clone()),
backend: "actual",
actual_model_smoke: true,
output_token_ids: Some(observation.output_token_ids.clone()),
output_text: Some(&observation.response_text),
finish_reason: observation.finish_reason.as_deref(),
failure_kind: None,
failure_diagnostics: None,
},
)?);
}
Ok(written)
}
fn write_actual_run_failure_artifacts(
config: &ProductObservabilityConfig,
events: &[FerrumProfileEvent],
observation: &ActualRunFailureObservation,
replay_command: &str,
) -> Result<Vec<PathBuf>> {
let mut written =
write_actual_artifacts(config, events, &observation.request_id, replay_command)?;
if let Some(dir) = &config.request_dump_dir {
written.extend(write_replay_bundle(
dir,
config,
&observation.request_id,
replay_command,
ReplayBundleData {
request: actual_request_dump(config, &observation.request_id, replay_command),
prompt_token_ids: observation.prompt_token_ids.clone(),
prompt_token_count: observation.prompt_token_count,
prompt_token_unavailable_reason: observation.prompt_token_ids.is_none().then_some(
"rendered prompt token ids were unavailable for run failure observability",
),
sampling_params: Some(observation.sampling_params.clone()),
backend: "actual",
actual_model_smoke: true,
output_token_ids: Some(Vec::new()),
output_text: Some(""),
finish_reason: Some("error"),
failure_kind: Some(observation.failure_kind.as_str()),
failure_diagnostics: Some(actual_run_failure_diagnostics(observation)),
},
)?);
}
Ok(written)
}
fn write_actual_artifacts(
config: &ProductObservabilityConfig,
events: &[FerrumProfileEvent],
request_id: &str,
replay_command: &str,
) -> Result<Vec<PathBuf>> {
write_actual_artifacts_with_scheduler(config, events, request_id, replay_command, true)
}
fn write_actual_artifacts_with_scheduler(
config: &ProductObservabilityConfig,
events: &[FerrumProfileEvent],
request_id: &str,
replay_command: &str,
include_scheduler: bool,
) -> Result<Vec<PathBuf>> {
let mut written = write_profile_outputs(
config,
events,
ferrum_bench_core::JsonlJournalOpenMode::Append,
true,
true,
include_scheduler,
)?;
if let Some(dir) = &config.request_dump_dir {
fs_create_dir_all(dir)?;
written.extend(write_replay_bundle(
dir,
config,
request_id,
replay_command,
ReplayBundleData {
request: actual_request_dump(config, request_id, replay_command),
prompt_token_ids: None,
prompt_token_count: None,
prompt_token_unavailable_reason: Some(
"startup or non-run request has no rendered prompt token dump in WP9 L0",
),
sampling_params: None,
backend: "actual",
actual_model_smoke: true,
output_token_ids: Some(Vec::new()),
output_text: None,
finish_reason: None,
failure_kind: None,
failure_diagnostics: None,
},
)?);
}
Ok(written)
}
#[derive(Clone, Copy, Default)]
struct ProfileOutputRoles {
profile: bool,
memory: bool,
scheduler: bool,
}
fn add_profile_output_target(
targets: &mut BTreeMap<PathBuf, ProfileOutputRoles>,
path: Option<&PathBuf>,
role: fn(&mut ProfileOutputRoles) -> &mut bool,
) -> Result<()> {
let Some(path) = path else {
return Ok(());
};
let normalized = ferrum_bench_core::normalize_jsonl_path(path).map_err(|error| {
FerrumError::io(format!(
"normalize observability JSONL path {}: {error}",
path.display()
))
})?;
*role(targets.entry(normalized).or_default()) = true;
Ok(())
}
fn write_profile_outputs(
config: &ProductObservabilityConfig,
events: &[FerrumProfileEvent],
mode: ferrum_bench_core::JsonlJournalOpenMode,
include_profile: bool,
include_memory: bool,
include_scheduler: bool,
) -> Result<Vec<PathBuf>> {
let mut targets = BTreeMap::<PathBuf, ProfileOutputRoles>::new();
if include_profile {
add_profile_output_target(&mut targets, config.profile_jsonl.as_ref(), |roles| {
&mut roles.profile
})?;
}
if include_memory {
add_profile_output_target(
&mut targets,
config.memory_profile_jsonl.as_ref(),
|roles| &mut roles.memory,
)?;
}
if include_scheduler {
add_profile_output_target(
&mut targets,
config.scheduler_trace_jsonl.as_ref(),
|roles| &mut roles.scheduler,
)?;
}
let mut written = Vec::with_capacity(targets.len());
for (path, roles) in targets {
let selected = events
.iter()
.filter(|event| {
roles.profile
|| (roles.memory && event.memory.is_some())
|| (roles.scheduler && event.resource.is_some())
})
.cloned()
.collect::<Vec<_>>();
if selected.is_empty() {
continue;
}
write_profile_events(&path, mode, &selected)?;
written.push(path);
}
Ok(written)
}
fn product_events(
config: &ProductObservabilityConfig,
request_id: &str,
replay_command: &str,
) -> Vec<FerrumProfileEvent> {
let base = Utc::now();
let open = resource_event(
config,
request_id,
"request",
request_id,
"request_slot",
"request_open",
ResourceAction::RequestOpen,
base,
None,
None,
None,
Some(1),
None,
);
let reserve = resource_event(
config,
request_id,
"request",
request_id,
"request_slot",
"request_slot_reserve",
ResourceAction::Reserve,
base + Duration::microseconds(10),
Some(1),
Some(0),
Some(1),
Some(1),
None,
);
let commit = resource_event(
config,
request_id,
"request",
request_id,
"request_slot",
"request_slot_commit",
ResourceAction::Commit,
base + Duration::microseconds(20),
Some(1),
Some(0),
Some(1),
Some(1),
None,
);
let mut prefill = base_event(
config,
request_id,
"synthetic_prefill",
ProfileEventKind::TimedSpan,
base + Duration::microseconds(30),
);
prefill.duration_us = Some(160);
prefill.memory = Some(MemorySnapshot {
scope: "process".to_string(),
backend: Some(SYNTHETIC_BACKEND.to_string()),
before_bytes: Some(2048),
after_bytes: Some(2304),
current_bytes: Some(2304),
high_water_bytes: Some(2304),
available_bytes: Some(1024 * 1024),
});
prefill.attributes.extend(common_attrs(config));
prefill
.attributes
.insert("input_tokens".to_string(), json!(8));
let release = resource_event(
config,
request_id,
"request",
request_id,
"request_slot",
"request_slot_release",
ResourceAction::Release,
base + Duration::microseconds(180),
Some(1),
Some(1),
Some(0),
Some(1),
None,
);
let mut close = resource_event(
config,
request_id,
"request",
request_id,
"request_slot",
"request_close",
ResourceAction::RequestClose,
base + Duration::microseconds(190),
None,
None,
None,
Some(1),
None,
);
close.status = if config.profile_detail.diagnostic_only() {
ProfileStatus::DiagnosticOnly
} else {
ProfileStatus::Ok
};
close.replay = Some(ReplayReference {
command: replay_command.to_string(),
bundle_dir: config
.request_dump_dir
.as_ref()
.map(|path| path.to_string_lossy().to_string()),
});
close.attributes.extend(common_attrs(config));
close
.attributes
.insert("response_text".to_string(), json!("synthetic ok"));
vec![open, reserve, commit, prefill, release, close]
}
fn actual_run_events(
config: &ProductObservabilityConfig,
observation: &ActualRunObservation,
replay_command: &str,
) -> Vec<FerrumProfileEvent> {
let base = Utc::now();
let shutdown_stage_index = observation
.memory_stages
.iter()
.position(|stage| stage.stage == "shutdown")
.unwrap_or(observation.memory_stages.len());
let mut events = actual_memory_stage_events(
config,
&observation.request_id,
&observation.memory_stages[..shutdown_stage_index],
base,
);
let open = actual_resource_event(
config,
&observation.request_id,
"request",
&observation.request_id,
"request_slot",
"request_open",
ResourceAction::RequestOpen,
base,
None,
None,
None,
Some(1),
None,
);
let reserve = actual_resource_event(
config,
&observation.request_id,
"request",
&observation.request_id,
"request_slot",
"request_slot_reserve",
ResourceAction::Reserve,
base + Duration::microseconds(5),
Some(1),
Some(0),
Some(1),
Some(1),
None,
);
let commit = actual_resource_event(
config,
&observation.request_id,
"request",
&observation.request_id,
"request_slot",
"request_slot_commit",
ResourceAction::Commit,
base + Duration::microseconds(10),
Some(1),
Some(0),
Some(1),
Some(1),
None,
);
let mut generation = actual_base_event(
config,
&observation.request_id,
"actual_run_generation",
ProfileEventKind::TimedSpan,
base + Duration::microseconds(20),
);
generation.duration_us = Some(observation.duration_us);
attach_process_memory(
&mut generation,
observation.memory.as_ref(),
"first_request_done",
);
generation.attributes.insert(
"output_tokens".to_string(),
json!(observation.output_tokens),
);
generation.attributes.insert(
"output_token_count".to_string(),
json!(observation.output_tokens),
);
generation.attributes.insert(
"completion_token_count".to_string(),
json!(observation.output_tokens),
);
generation.attributes.insert(
"e2e_duration_us".to_string(),
json!(observation.duration_us),
);
if let Some(prompt_token_count) = observation.prompt_token_count {
generation
.attributes
.insert("prompt_token_count".to_string(), json!(prompt_token_count));
generation.attributes.insert(
"total_token_count".to_string(),
json!(prompt_token_count.saturating_add(observation.output_tokens)),
);
generation.attributes.insert(
"token_count_source".to_string(),
json!("rendered_prompt_and_generated_tokens"),
);
} else {
generation.attributes.insert(
"total_token_count".to_string(),
json!(observation.output_tokens),
);
generation
.attributes
.insert("token_count_source".to_string(), json!("generated_tokens"));
generation.attributes.insert(
"prompt_token_unavailable_reason".to_string(),
json!("run rendered prompt token count was unavailable"),
);
}
generation
.attributes
.insert("chunk_count".to_string(), json!(observation.chunk_count));
generation.attributes.insert(
"finish_reason".to_string(),
json!(observation.finish_reason.as_deref().unwrap_or("unknown")),
);
if let Some(timing) = observation
.execution_evidence
.as_ref()
.and_then(|evidence| evidence.engine_token_timing.as_ref())
{
generation
.attributes
.extend(ferrum_types::engine_token_timing_profile_attributes(timing));
}
let release = actual_resource_event(
config,
&observation.request_id,
"request",
&observation.request_id,
"request_slot",
"request_slot_release",
ResourceAction::Release,
base + Duration::microseconds(30),
Some(1),
Some(1),
Some(0),
Some(1),
None,
);
let mut close = actual_resource_event(
config,
&observation.request_id,
"request",
&observation.request_id,
"request_slot",
"request_close",
ResourceAction::RequestClose,
base + Duration::microseconds(40),
None,
None,
None,
Some(1),
None,
);
close.replay = Some(ReplayReference {
command: replay_command.to_string(),
bundle_dir: config
.request_dump_dir
.as_ref()
.map(|path| path.to_string_lossy().to_string()),
});
close
.attributes
.insert("prompt_chars".to_string(), json!(observation.prompt_chars));
close.attributes.insert(
"response_chars".to_string(),
json!(observation.response_chars),
);
events.extend([open, reserve, commit, generation, release, close]);
events.extend(actual_memory_stage_events(
config,
&observation.request_id,
&observation.memory_stages[shutdown_stage_index..],
base + Duration::microseconds(50),
));
events
}
fn actual_run_failure_events(
config: &ProductObservabilityConfig,
observation: &ActualRunFailureObservation,
replay_command: &str,
) -> Vec<FerrumProfileEvent> {
let base = Utc::now();
let shutdown_stage_index = observation
.memory_stages
.iter()
.position(|stage| stage.stage == "shutdown")
.unwrap_or(observation.memory_stages.len());
let mut events = actual_memory_stage_events(
config,
&observation.request_id,
&observation.memory_stages[..shutdown_stage_index],
base,
);
let open = actual_resource_event(
config,
&observation.request_id,
"request",
&observation.request_id,
"request_slot",
"request_open",
ResourceAction::RequestOpen,
base,
None,
None,
None,
Some(1),
None,
);
let reserve = actual_resource_event(
config,
&observation.request_id,
"request",
&observation.request_id,
"request_slot",
"request_slot_reserve",
ResourceAction::Reserve,
base + Duration::microseconds(5),
Some(1),
Some(0),
Some(1),
Some(1),
None,
);
let commit = actual_resource_event(
config,
&observation.request_id,
"request",
&observation.request_id,
"request_slot",
"request_slot_commit",
ResourceAction::Commit,
base + Duration::microseconds(10),
Some(1),
Some(0),
Some(1),
Some(1),
None,
);
let mut failure = actual_base_event(
config,
&observation.request_id,
"actual_run_generation_failed",
ProfileEventKind::TimedSpan,
base + Duration::microseconds(20),
);
failure.status = ProfileStatus::Failure;
failure.duration_us = Some(observation.duration_us);
failure.error = Some(ProfileError {
kind: observation.error_kind.clone(),
message: observation.error_message.clone(),
blocking: false,
});
failure.replay = Some(ReplayReference {
command: replay_command.to_string(),
bundle_dir: config
.request_dump_dir
.as_ref()
.map(|path| path.to_string_lossy().to_string()),
});
attach_process_memory(
&mut failure,
observation.memory.as_ref(),
"first_request_failed",
);
failure
.attributes
.insert("terminal_failure_event".to_string(), json!(true));
failure
.attributes
.insert("prompt_chars".to_string(), json!(observation.prompt_chars));
let release = actual_resource_event(
config,
&observation.request_id,
"request",
&observation.request_id,
"request_slot",
"request_slot_release",
ResourceAction::Release,
base + Duration::microseconds(30),
Some(1),
Some(1),
Some(0),
Some(1),
None,
);
let mut close = actual_resource_event(
config,
&observation.request_id,
"request",
&observation.request_id,
"request_slot",
"request_close",
ResourceAction::RequestClose,
base + Duration::microseconds(40),
None,
None,
None,
Some(1),
None,
);
close.replay = Some(ReplayReference {
command: replay_command.to_string(),
bundle_dir: config
.request_dump_dir
.as_ref()
.map(|path| path.to_string_lossy().to_string()),
});
events.extend([open, reserve, commit, failure, release, close]);
events.extend(actual_memory_stage_events(
config,
&observation.request_id,
&observation.memory_stages[shutdown_stage_index..],
base + Duration::microseconds(50),
));
events
}
fn actual_memory_stage_events(
config: &ProductObservabilityConfig,
request_id: &str,
stages: &[ActualMemoryStageObservation],
base: chrono::DateTime<Utc>,
) -> Vec<FerrumProfileEvent> {
stages
.iter()
.enumerate()
.map(|(index, stage)| {
let mut event = actual_base_event(
config,
request_id,
&stage.phase,
ProfileEventKind::Memory,
base + Duration::microseconds(index as i64),
);
event.duration_us = stage.duration_us;
attach_process_memory(&mut event, stage.memory.as_ref(), &stage.stage);
event.attributes.extend(stage.attributes.clone());
event
})
.collect()
}
fn actual_serve_startup_events(
config: &ProductObservabilityConfig,
request_id: &str,
startup_duration_us: u64,
startup_memory: Option<&crate::memory_profile::ProcessMemoryObservation>,
memory_stages: &[ActualMemoryStageObservation],
replay_command: &str,
) -> Vec<FerrumProfileEvent> {
let base = Utc::now();
let post_model_loaded_index = memory_stages
.iter()
.position(|stage| matches!(stage.stage.as_str(), "profile_run_done" | "cache_allocated"))
.unwrap_or(memory_stages.len());
let mut events = actual_memory_stage_events(
config,
request_id,
&memory_stages[..post_model_loaded_index],
base,
);
let open = actual_resource_event(
config,
request_id,
"server",
request_id,
"startup_slot",
"server_startup_open",
ResourceAction::RequestOpen,
base,
None,
None,
None,
Some(1),
None,
);
let reserve = actual_resource_event(
config,
request_id,
"server",
request_id,
"startup_slot",
"server_startup_reserve",
ResourceAction::Reserve,
base + Duration::microseconds(5),
Some(1),
Some(0),
Some(1),
Some(1),
None,
);
let commit = actual_resource_event(
config,
request_id,
"server",
request_id,
"startup_slot",
"server_startup_commit",
ResourceAction::Commit,
base + Duration::microseconds(10),
Some(1),
Some(0),
Some(1),
Some(1),
None,
);
let mut startup = actual_base_event(
config,
request_id,
"actual_serve_startup",
ProfileEventKind::TimedSpan,
base + Duration::microseconds(20),
);
startup.duration_us = Some(startup_duration_us);
attach_process_memory(&mut startup, startup_memory, "model_loaded");
events.extend([open, reserve, commit, startup]);
events.extend(actual_memory_stage_events(
config,
request_id,
&memory_stages[post_model_loaded_index..],
base + Duration::microseconds(21),
));
let release = actual_resource_event(
config,
request_id,
"server",
request_id,
"startup_slot",
"server_startup_release",
ResourceAction::Release,
base + Duration::microseconds(30),
Some(1),
Some(1),
Some(0),
Some(1),
None,
);
let mut ready = actual_resource_event(
config,
request_id,
"server",
request_id,
"startup_slot",
"server_ready_for_requests",
ResourceAction::RequestClose,
base + Duration::microseconds(40),
None,
None,
None,
Some(1),
None,
);
ready.replay = Some(ReplayReference {
command: replay_command.to_string(),
bundle_dir: config
.request_dump_dir
.as_ref()
.map(|path| path.to_string_lossy().to_string()),
});
events.extend([release, ready]);
events
}
fn base_event(
config: &ProductObservabilityConfig,
request_id: &str,
phase: &str,
event_kind: ProfileEventKind,
timestamp: chrono::DateTime<Utc>,
) -> FerrumProfileEvent {
FerrumProfileEvent {
schema_version: OBSERVABILITY_PROFILE_SCHEMA_VERSION,
ts_unix_nanos: timestamp
.timestamp_nanos_opt()
.unwrap_or_else(|| timestamp.timestamp_micros() * 1_000),
event_id: format!(
"evt-product-{}-{phase}",
entrypoint_label(config.entrypoint)
),
request_id: request_id.to_string(),
correlation_id: Some(format!(
"corr-product-{}",
entrypoint_label(config.entrypoint)
)),
entrypoint: config.entrypoint,
backend: SYNTHETIC_BACKEND.to_string(),
runtime_preset_hash: runtime_preset_hash(config),
phase: phase.to_string(),
event_kind,
timestamp,
status: ProfileStatus::Ok,
model: Some(config.model.clone()),
duration_us: None,
memory: None,
resource: None,
error: None,
replay: None,
shape: default_event_shape(),
backend_detail: None,
attributes: common_attrs(config),
}
}
fn actual_base_event(
config: &ProductObservabilityConfig,
request_id: &str,
phase: &str,
event_kind: ProfileEventKind,
timestamp: chrono::DateTime<Utc>,
) -> FerrumProfileEvent {
let mut event = base_event(config, request_id, phase, event_kind, timestamp);
event.backend = "actual".to_string();
event.attributes = actual_attrs(config);
event.attributes.insert(
"execution_request_id".to_string(),
json!(format!("request.product.{request_id}")),
);
event
}
fn runtime_preset_hash(config: &ProductObservabilityConfig) -> String {
let mut hasher = Sha256::new();
hasher.update(config.entrypoint.as_str().as_bytes());
hasher.update(b"\0");
hasher.update(config.model.as_bytes());
hasher.update(b"\0");
hasher.update(config.profile_detail.as_str().as_bytes());
hasher.update(b"\0");
hasher.update(config.profile_sample_rate.to_string().as_bytes());
format!("sha256:{:x}", hasher.finalize())
}
fn default_event_shape() -> BTreeMap<String, Value> {
BTreeMap::from([("batch_size".to_string(), json!(1))])
}
#[allow(clippy::too_many_arguments)]
fn resource_event(
config: &ProductObservabilityConfig,
request_id: &str,
owner_kind: &str,
owner_id: &str,
resource_kind: &str,
phase: &str,
action: ResourceAction,
timestamp: chrono::DateTime<Utc>,
amount: Option<i64>,
before: Option<i64>,
after: Option<i64>,
capacity: Option<i64>,
reason: Option<&str>,
) -> FerrumProfileEvent {
let mut event = base_event(
config,
request_id,
phase,
ProfileEventKind::Resource,
timestamp,
);
event.resource = Some(ResourceTraceEvent {
owner_kind: owner_kind.to_string(),
owner_id: owner_id.to_string(),
resource_kind: resource_kind.to_string(),
action,
amount,
before,
after,
capacity,
underflow_amount: match (action, amount, before) {
(ResourceAction::Release | ResourceAction::Rollback, Some(amount), Some(before))
if amount > before =>
{
Some(amount.saturating_sub(before))
}
_ => None,
},
reason: reason.map(str::to_string),
error_kind: None,
message: None,
resource_error_kind: None,
});
event
}
#[allow(clippy::too_many_arguments)]
fn actual_resource_event(
config: &ProductObservabilityConfig,
request_id: &str,
owner_kind: &str,
owner_id: &str,
resource_kind: &str,
phase: &str,
action: ResourceAction,
timestamp: chrono::DateTime<Utc>,
amount: Option<i64>,
before: Option<i64>,
after: Option<i64>,
capacity: Option<i64>,
reason: Option<&str>,
) -> FerrumProfileEvent {
let mut event = resource_event(
config,
request_id,
owner_kind,
owner_id,
resource_kind,
phase,
action,
timestamp,
amount,
before,
after,
capacity,
reason,
);
event.backend = "actual".to_string();
event.attributes = actual_attrs(config);
event
}
fn attach_process_memory(
event: &mut FerrumProfileEvent,
observation: Option<&crate::memory_profile::ProcessMemoryObservation>,
stage: &str,
) {
if let Some(observation) = observation {
event.memory = Some(observation.to_snapshot("process", Some("actual")));
event
.attributes
.insert("memory_measurement".to_string(), json!("process_rss"));
event
.attributes
.insert("memory_stage".to_string(), json!(stage));
event.attributes.insert(
"process_memory_source".to_string(),
json!(observation.source),
);
} else {
event.memory = Some(MemorySnapshot {
scope: "process".to_string(),
backend: Some("actual".to_string()),
before_bytes: Some(0),
after_bytes: Some(0),
current_bytes: Some(0),
high_water_bytes: Some(0),
available_bytes: None,
});
event
.attributes
.insert("memory_measurement".to_string(), json!("not_collected"));
event
.attributes
.insert("memory_stage".to_string(), json!(stage));
}
}
fn common_attrs(config: &ProductObservabilityConfig) -> BTreeMap<String, Value> {
BTreeMap::from([
(
"profile_detail".to_string(),
json!(config.profile_detail.as_str()),
),
(
"profile_sample_rate".to_string(),
json!(config.profile_sample_rate),
),
(
"diagnostic_only".to_string(),
json!(config.profile_detail.diagnostic_only()),
),
("l0_only".to_string(), json!(true)),
])
}
fn actual_attrs(config: &ProductObservabilityConfig) -> BTreeMap<String, Value> {
BTreeMap::from([
(
"profile_detail".to_string(),
json!(config.profile_detail.as_str()),
),
(
"profile_sample_rate".to_string(),
json!(config.profile_sample_rate),
),
(
"diagnostic_only".to_string(),
json!(config.profile_detail.diagnostic_only()),
),
("l0_only".to_string(), json!(false)),
("actual_model_smoke".to_string(), json!(true)),
])
}
struct ReplayBundleData<'a> {
request: serde_json::Value,
prompt_token_ids: Option<Vec<u32>>,
prompt_token_count: Option<usize>,
prompt_token_unavailable_reason: Option<&'a str>,
sampling_params: Option<SamplingParams>,
backend: &'a str,
actual_model_smoke: bool,
output_token_ids: Option<Vec<u32>>,
output_text: Option<&'a str>,
finish_reason: Option<&'a str>,
failure_kind: Option<&'a str>,
failure_diagnostics: Option<serde_json::Value>,
}
fn write_replay_bundle(
root: &Path,
config: &ProductObservabilityConfig,
request_id: &str,
replay_command: &str,
data: ReplayBundleData<'_>,
) -> Result<Vec<PathBuf>> {
fs_create_dir_all(root)?;
let bundle_dir = root.join(request_id);
fs_create_dir_all(&bundle_dir)?;
let mut written = Vec::new();
let request_path = root.join("request.json");
let replay_path = root.join("replay_command.txt");
write_json(&request_path, &data.request)?;
fs_write(&replay_path, format!("{replay_command}\n"))?;
written.push(request_path);
written.push(replay_path);
let prompt_token_count = data
.prompt_token_count
.or_else(|| data.prompt_token_ids.as_ref().map(Vec::len));
let output_token_count = data.output_token_ids.as_ref().map(Vec::len).unwrap_or(0);
let prompt_tokens = json!({
"schema_version": OBSERVABILITY_PROFILE_SCHEMA_VERSION,
"request_id": request_id,
"model": config.model,
"tokenizer_or_model": config.model,
"token_ids": data.prompt_token_ids,
"token_count": prompt_token_count,
"unavailable_reason": data.prompt_token_unavailable_reason,
"sanitized": true
});
let output_text = data.output_text.unwrap_or("");
let sampling_unavailable_reason = if data.sampling_params.is_some() {
Value::Null
} else {
json!("sampling params unavailable for this replay bundle kind in WP9 L0")
};
let output_text_body = if data.actual_model_smoke && config.model != SYNTHETIC_MODEL {
format!(
"[redacted actual output]\nsha256={}\nchars={}\n",
sha256_hex(output_text.as_bytes()),
output_text.chars().count()
)
} else {
format!("{output_text}\n")
};
let output_scan = bad_output_scan(
request_id,
output_text,
data.failure_kind,
output_text_body.as_bytes(),
);
let engine_replay_args = engine_replay_command_args(&bundle_dir);
let engine_replay_command = replay_command_from_args(&engine_replay_args);
let files = [
("request.json", data.request),
("prompt_token_ids.json", prompt_tokens),
(
"sampling_params.json",
json!({
"schema_version": OBSERVABILITY_PROFILE_SCHEMA_VERSION,
"request_id": request_id,
"sampling_params": data.sampling_params,
"unavailable_reason": sampling_unavailable_reason
}),
),
(
"runtime_effective_config.json",
json!({
"schema_version": OBSERVABILITY_PROFILE_SCHEMA_VERSION,
"request_id": request_id,
"entrypoint": entrypoint_label(config.entrypoint),
"profile_detail": config.profile_detail.as_str(),
"profile_sample_rate": config.profile_sample_rate,
"profile_jsonl": config.profile_jsonl.as_ref().map(|path| path.to_string_lossy().to_string()),
"memory_profile_jsonl": config.memory_profile_jsonl.as_ref().map(|path| path.to_string_lossy().to_string()),
"scheduler_trace_jsonl": config.scheduler_trace_jsonl.as_ref().map(|path| path.to_string_lossy().to_string()),
"request_dump_dir": Some(root.to_string_lossy().to_string()),
"sanitized": true
}),
),
(
"backend_selection.json",
json!({
"schema_version": OBSERVABILITY_PROFILE_SCHEMA_VERSION,
"request_id": request_id,
"backend": data.backend,
"model": config.model,
"actual_model_smoke": data.actual_model_smoke
}),
),
(
"output_token_ids.json",
json!({
"schema_version": OBSERVABILITY_PROFILE_SCHEMA_VERSION,
"request_id": request_id,
"token_ids": data.output_token_ids.unwrap_or_default(),
"token_count": output_token_count,
"finish_reason": data.finish_reason
}),
),
("bad_output_scan.json", output_scan),
(
"replay.command.json",
json!({
"schema_version": OBSERVABILITY_PROFILE_SCHEMA_VERSION,
"request_id": request_id,
"entrypoint": entrypoint_label(config.entrypoint),
"command": replay_command,
"argv": replay_command_args(config),
"bundle_dir": bundle_dir.to_string_lossy(),
"engine_replay": {
"mode": "bundle_offline",
"requires_http_server": false,
"command": engine_replay_command,
"argv": engine_replay_args
},
"sanitized": true
}),
),
];
for (name, value) in files {
let path = bundle_dir.join(name);
write_json(&path, &value)?;
written.push(path);
}
if let Some(diagnostics) = data.failure_diagnostics {
let path = bundle_dir.join("failure_diagnostics.json");
write_json(&path, &diagnostics)?;
written.push(path);
}
let output_text_path = bundle_dir.join("output_text.txt");
fs_write(&output_text_path, output_text_body)?;
written.push(output_text_path);
Ok(written)
}
fn actual_run_failure_diagnostics(observation: &ActualRunFailureObservation) -> serde_json::Value {
if resource_failure_kind(&observation.failure_kind) {
return actual_run_resource_failure_diagnostics(observation);
}
json!({
"schema_version": OBSERVABILITY_PROFILE_SCHEMA_VERSION,
"request_id": observation.request_id,
"failure_kind": observation.failure_kind,
"first_failure_event": {
"phase": "actual_run_generation_failed",
"error_kind": observation.error_kind,
"message": observation.error_message
},
"nearest_request_id": observation.request_id,
"log_excerpt": observation.error_message
})
}
fn actual_run_resource_failure_diagnostics(
observation: &ActualRunFailureObservation,
) -> serde_json::Value {
let memory_current = observation
.memory
.as_ref()
.map(|memory| memory.current_bytes as i64)
.unwrap_or(0)
.max(0);
let memory_high_water = observation
.memory
.as_ref()
.map(|memory| memory.high_water_bytes as i64)
.unwrap_or(memory_current);
let resource_kind = resource_kind_for_failure(&observation.failure_kind);
let needed = if resource_kind == "device_memory" {
memory_current.saturating_add(1).max(1)
} else {
observation
.prompt_token_count
.and_then(|tokens| i64::try_from(tokens).ok())
.unwrap_or(1)
.max(1)
};
let capacity = if resource_kind == "device_memory" {
memory_high_water.max(memory_current)
} else {
0
};
json!({
"schema_version": OBSERVABILITY_PROFILE_SCHEMA_VERSION,
"request_id": observation.request_id,
"failure_kind": observation.failure_kind,
"first_failure_event": {
"phase": "actual_run_generation_failed",
"error_kind": observation.error_kind,
"message": observation.error_message
},
"nearest_request_id": observation.request_id,
"log_excerpt": observation.error_message,
"capacity": {
"resource_kind": resource_kind,
"needed": needed,
"available": 0,
"capacity": capacity,
"reason": observation.error_message
},
"nearest_resource_event": {
"owner_kind": "request",
"owner_id": observation.request_id,
"resource_kind": resource_kind,
"action": "reject",
"amount": needed,
"before": 0,
"after": 0,
"capacity": capacity,
"reason": observation.error_message
},
"nearest_memory_snapshot": {
"scope": "actual_run_failure",
"backend": "process",
"current_bytes": memory_current,
"high_water_bytes": memory_high_water.max(memory_current),
"source": observation.memory.as_ref().map(|memory| memory.source).unwrap_or("not_collected")
}
})
}
fn resource_failure_kind(failure_kind: &str) -> bool {
matches!(
failure_kind,
"oom" | "prevented_oom" | "admission" | "admission_reject" | "oom_admission"
)
}
fn resource_kind_for_failure(failure_kind: &str) -> &'static str {
match failure_kind {
"oom" | "prevented_oom" => "device_memory",
_ => "admission_capacity",
}
}
fn bad_output_scan(
request_id: &str,
text: &str,
failure_kind: Option<&str>,
output_artifact_bytes: &[u8],
) -> serde_json::Value {
let mut reasons = Vec::new();
let mut first_span: Option<serde_json::Value> = None;
for (needle, reason) in [
("<unk>", "reserved_token"),
("[PAD", "reserved_token"),
("<pad>", "reserved_token"),
("<|endoftext|>", "reserved_token"),
("<|im_start|>", "reserved_token"),
("<|im_end|>", "reserved_token"),
("<|reserved_special_token", "reserved_token"),
] {
if let Some(index) = text.find(needle) {
reasons.push(reason);
first_span.get_or_insert_with(|| {
json!({
"byte_start": index,
"byte_end": index + needle.len(),
"text": needle,
"reason": reason
})
});
}
}
if let Some(index) = first_mojibake_index(text) {
reasons.push("mojibake");
first_span.get_or_insert_with(|| {
json!({
"byte_start": index,
"byte_end": index + 1,
"reason": "mojibake"
})
});
}
reasons.sort_unstable();
reasons.dedup();
let bad_output = !reasons.is_empty();
json!({
"schema_version": OBSERVABILITY_PROFILE_SCHEMA_VERSION,
"request_id": request_id,
"bad_output": bad_output,
"bad_text_count": if bad_output { 1 } else { 0 },
"reasons": reasons,
"first_bad_text_span": first_span,
"failure_kind": failure_kind,
"output_chars": text.chars().count(),
"classified_output_sha256": sha256_hex(text.as_bytes()),
"output_sha256": sha256_hex(output_artifact_bytes)
})
}
fn first_mojibake_index(text: &str) -> Option<usize> {
["\u{00c3}\u{00a9}", "\u{00c2}\u{00a9}", "\u{00e2}\u{20ac}"]
.iter()
.filter_map(|needle| text.find(needle))
.min()
}
fn sha256_hex(bytes: &[u8]) -> String {
let mut hasher = Sha256::new();
hasher.update(bytes);
format!("{:x}", hasher.finalize())
}
fn request_dump(
config: &ProductObservabilityConfig,
request_id: &str,
replay_command: &str,
) -> serde_json::Value {
let entrypoint = entrypoint_label(config.entrypoint);
match config.entrypoint {
ProfileEntrypoint::Run => json!({
"schema_version": OBSERVABILITY_PROFILE_SCHEMA_VERSION,
"entrypoint": entrypoint,
"request_id": request_id,
"model": config.model,
"backend": SYNTHETIC_BACKEND,
"profile_detail": config.profile_detail.as_str(),
"profile_sample_rate": config.profile_sample_rate,
"l0_only": true,
"sanitized": true,
"prompt": "product observability wiring",
"replay_command": replay_command
}),
ProfileEntrypoint::Serve => json!({
"schema_version": OBSERVABILITY_PROFILE_SCHEMA_VERSION,
"entrypoint": entrypoint,
"request_id": request_id,
"model": config.model,
"backend": SYNTHETIC_BACKEND,
"profile_detail": config.profile_detail.as_str(),
"profile_sample_rate": config.profile_sample_rate,
"l0_only": true,
"sanitized": true,
"http": {
"method": "POST",
"path": "/v1/chat/completions",
"body": {
"model": config.model,
"messages": [{"role": "user", "content": "product observability wiring"}],
"stream": false
}
},
"replay_command": replay_command
}),
other => json!({
"schema_version": OBSERVABILITY_PROFILE_SCHEMA_VERSION,
"entrypoint": entrypoint_label(other),
"request_id": request_id,
"model": config.model,
"backend": SYNTHETIC_BACKEND,
"profile_detail": config.profile_detail.as_str(),
"profile_sample_rate": config.profile_sample_rate,
"l0_only": true,
"sanitized": true,
"replay_command": replay_command
}),
}
}
fn actual_request_dump(
config: &ProductObservabilityConfig,
request_id: &str,
replay_command: &str,
) -> serde_json::Value {
json!({
"schema_version": OBSERVABILITY_PROFILE_SCHEMA_VERSION,
"entrypoint": entrypoint_label(config.entrypoint),
"request_id": request_id,
"model": config.model,
"backend": "actual",
"profile_detail": config.profile_detail.as_str(),
"profile_sample_rate": config.profile_sample_rate,
"l0_only": false,
"actual_model_smoke": true,
"sanitized": true,
"replay_command": replay_command
})
}
fn replay_command(config: &ProductObservabilityConfig) -> String {
replay_command_from_args(&replay_command_args(config))
}
fn replay_command_from_args(args: &[String]) -> String {
args.iter()
.map(|part| shell_quote(part))
.collect::<Vec<_>>()
.join(" ")
}
fn replay_command_args(config: &ProductObservabilityConfig) -> Vec<String> {
let mut parts = vec![
"cargo".to_string(),
"run".to_string(),
"-p".to_string(),
"ferrum-cli".to_string(),
"--".to_string(),
entrypoint_label(config.entrypoint).to_string(),
SYNTHETIC_MODEL.to_string(),
"--profile-detail".to_string(),
config.profile_detail.as_str().to_string(),
"--profile-sample-rate".to_string(),
config.profile_sample_rate.to_string(),
];
push_path_arg(&mut parts, "--profile-jsonl", config.profile_jsonl.as_ref());
push_path_arg(
&mut parts,
"--memory-profile-jsonl",
config.memory_profile_jsonl.as_ref(),
);
push_path_arg(
&mut parts,
"--scheduler-trace-jsonl",
config.scheduler_trace_jsonl.as_ref(),
);
push_path_arg(
&mut parts,
"--request-dump-dir",
config.request_dump_dir.as_ref(),
);
parts
}
fn engine_replay_command_args(bundle_dir: &Path) -> Vec<String> {
vec![
"cargo".to_string(),
"run".to_string(),
"-p".to_string(),
"ferrum-cli".to_string(),
"--".to_string(),
"replay-bundle".to_string(),
bundle_dir.to_string_lossy().to_string(),
"--out".to_string(),
bundle_dir
.join("engine_replay")
.to_string_lossy()
.to_string(),
"--json".to_string(),
]
}
fn push_path_arg(parts: &mut Vec<String>, flag: &str, path: Option<&PathBuf>) {
if let Some(path) = path {
parts.push(flag.to_string());
parts.push(path.to_string_lossy().to_string());
}
}
#[cfg(test)]
fn write_profile_jsonl(path: &Path, events: &[FerrumProfileEvent]) -> Result<()> {
write_profile_events(
path,
ferrum_bench_core::JsonlJournalOpenMode::Truncate,
events,
)
}
#[cfg(test)]
fn append_profile_jsonl(path: &Path, events: &[FerrumProfileEvent]) -> Result<()> {
write_profile_events(
path,
ferrum_bench_core::JsonlJournalOpenMode::Append,
events,
)
}
fn write_profile_events(
path: &Path,
mode: ferrum_bench_core::JsonlJournalOpenMode,
events: &[FerrumProfileEvent],
) -> Result<()> {
if events.is_empty() {
return Err(FerrumError::internal("profile event set must be non-empty"));
}
for event in events {
event.validate().map_err(|err| {
FerrumError::internal(format!("invalid product observability event: {err}"))
})?;
}
ferrum_bench_core::write_jsonl_records(path, mode, events)
.map_err(|error| FerrumError::io(error.to_string()))
}
fn write_json(path: &Path, value: &serde_json::Value) -> Result<()> {
let body = serde_json::to_string_pretty(value)
.map_err(|err| FerrumError::serialization(format!("failed to serialize JSON: {err}")))?;
fs_write(path, format!("{body}\n"))
}
fn fs_create_dir_all(path: &Path) -> Result<()> {
fs::create_dir_all(path)
.map_err(|err| FerrumError::io(format!("failed to create {}: {err}", path.display())))
}
fn fs_write(path: &Path, content: impl AsRef<[u8]>) -> Result<()> {
if let Some(parent) = path.parent() {
fs_create_dir_all(parent)?;
}
fs::write(path, content)
.map_err(|err| FerrumError::io(format!("failed to write {}: {err}", path.display())))
}
fn entrypoint_label(entrypoint: ProfileEntrypoint) -> &'static str {
match entrypoint {
ProfileEntrypoint::Run => "run",
ProfileEntrypoint::Serve => "serve",
ProfileEntrypoint::BenchServe => "bench_serve",
ProfileEntrypoint::Synthetic => "synthetic",
}
}
fn shell_quote(value: &str) -> String {
if value
.chars()
.all(|ch| ch.is_ascii_alphanumeric() || "-_./:".contains(ch))
{
return value.to_string();
}
format!("'{}'", value.replace('\'', "'\\''"))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn engine_cache_observation_does_not_report_static_weights_as_kv() {
let status = ferrum_types::EngineStatus {
is_ready: true,
loaded_models: vec![ferrum_types::ModelId::from("test-model")],
active_requests: 0,
queued_requests: 0,
memory_usage: ferrum_types::MemoryUsage {
total_bytes: 900,
used_bytes: 500,
free_bytes: 400,
gpu_memory_bytes: Some(500),
cpu_memory_bytes: None,
cache_memory_bytes: 100,
utilization_percent: 500.0 / 9.0,
},
uptime_seconds: 0,
last_heartbeat: chrono::Utc::now(),
version: "test".to_string(),
};
let observation = ActualMemoryStageObservation::new("test", "cache_allocated", None, None)
.with_engine_cache_status(&status);
assert_eq!(observation.attributes["kv_cache_total_bytes"], 500);
assert_eq!(observation.attributes["kv_cache_used_bytes"], 100);
assert_eq!(observation.attributes["resource_total_bytes"], 900);
assert_eq!(observation.attributes["resource_used_bytes"], 500);
}
#[test]
fn synthetic_product_observability_writes_profile_paths() {
let root = std::env::temp_dir().join(format!(
"ferrum-product-observability-{}",
Uuid::new_v4().simple()
));
let config = ProductObservabilityConfig::new(
ProfileEntrypoint::Run,
SYNTHETIC_MODEL,
Some(&root.join("profile.jsonl")),
ProfileDetailArg::Basic,
Some(&root.join("memory.jsonl")),
Some(&root.join("scheduler.jsonl")),
Some(&root.join("request_dump")),
1.0,
);
let written = write_synthetic_product_observability(&config).unwrap();
assert!(written.len() >= 14);
assert!(root.join("profile.jsonl").is_file());
assert!(root.join("memory.jsonl").is_file());
assert!(root.join("scheduler.jsonl").is_file());
assert!(root.join("request_dump/request.json").is_file());
assert!(root.join("request_dump/replay_command.txt").is_file());
let request_dump_root = root.join("request_dump");
let bundle_dir = fs::read_dir(&request_dump_root)
.unwrap()
.flatten()
.find_map(|entry| entry.path().is_dir().then_some(entry.path()))
.expect("request-id replay bundle directory should exist");
assert!(bundle_dir.join("prompt_token_ids.json").is_file());
assert!(bundle_dir.join("sampling_params.json").is_file());
assert!(bundle_dir.join("runtime_effective_config.json").is_file());
assert!(bundle_dir.join("backend_selection.json").is_file());
assert!(bundle_dir.join("output_token_ids.json").is_file());
assert!(bundle_dir.join("output_text.txt").is_file());
assert!(bundle_dir.join("bad_output_scan.json").is_file());
assert!(bundle_dir.join("replay.command.json").is_file());
let replay: serde_json::Value = serde_json::from_str(
&fs::read_to_string(bundle_dir.join("replay.command.json")).unwrap(),
)
.unwrap();
let engine_argv = replay["engine_replay"]["argv"].as_array().unwrap();
assert!(engine_argv.iter().any(|item| item == "replay-bundle"));
assert_eq!(replay["engine_replay"]["requires_http_server"], false);
fs::remove_dir_all(root).ok();
}
#[test]
fn aliased_profile_roles_write_each_event_once() {
let root = std::env::temp_dir().join(format!(
"ferrum-product-observability-alias-{}",
Uuid::new_v4().simple()
));
let combined = root.join("combined.jsonl");
let memory_alias = root.join(".").join("combined.jsonl");
let scheduler_alias = root.join("not-created").join("..").join("combined.jsonl");
let config = ProductObservabilityConfig::new(
ProfileEntrypoint::Run,
SYNTHETIC_MODEL,
Some(&combined),
ProfileDetailArg::Resource,
Some(&memory_alias),
Some(&scheduler_alias),
None,
1.0,
);
let written = write_synthetic_product_observability(&config).unwrap();
assert_eq!(written.len(), 1);
let events = fs::read_to_string(&combined)
.unwrap()
.lines()
.map(|line| serde_json::from_str::<serde_json::Value>(line).unwrap())
.collect::<Vec<_>>();
assert_eq!(
events.len(),
6,
"aliased roles duplicated events: {events:#?}"
);
assert_eq!(
events
.iter()
.filter_map(|event| event["event_id"].as_str())
.collect::<std::collections::BTreeSet<_>>()
.len(),
6
);
fs::remove_dir_all(root).ok();
}
#[test]
fn actual_serve_memory_stage_observability_appends_shutdown() {
let root = std::env::temp_dir().join(format!(
"ferrum-serve-memory-observability-{}",
Uuid::new_v4().simple()
));
let config = ProductObservabilityConfig::new(
ProfileEntrypoint::Serve,
"Qwen/Qwen3-0.6B",
Some(&root.join("profile.jsonl")),
ProfileDetailArg::Basic,
Some(&root.join("memory.jsonl")),
Some(&root.join("scheduler.jsonl")),
Some(&root.join("request_dump")),
1.0,
);
let memory = crate::memory_profile::ProcessMemoryObservation {
before_bytes: 100,
after_bytes: 200,
current_bytes: 200,
high_water_bytes: 240,
source: "test",
};
write_actual_serve_startup_observability(
&config,
42,
Some(memory.clone()),
vec![
ActualMemoryStageObservation::new(
"actual_serve_process_start",
"process_start",
None,
Some(memory.clone()),
),
ActualMemoryStageObservation::new(
"actual_serve_backend_initialized",
"backend_initialized",
None,
Some(memory.clone()),
),
],
)
.unwrap();
append_actual_serve_memory_stage_observability(
&config,
ActualMemoryStageObservation::new(
"actual_serve_shutdown",
"shutdown",
None,
Some(memory),
),
)
.unwrap();
let profile = fs::read_to_string(root.join("profile.jsonl")).unwrap();
assert!(profile.contains("\"phase\":\"actual_serve_startup\""));
assert!(profile.contains("\"phase\":\"actual_serve_shutdown\""));
let memory_profile = fs::read_to_string(root.join("memory.jsonl")).unwrap();
let memory_stages = memory_profile
.lines()
.filter(|line| !line.trim().is_empty())
.map(|line| serde_json::from_str::<serde_json::Value>(line).unwrap())
.filter_map(|event| {
event["attributes"]["memory_stage"]
.as_str()
.map(str::to_string)
})
.collect::<Vec<_>>();
assert!(memory_stages.contains(&"process_start".to_string()));
assert!(memory_stages.contains(&"backend_initialized".to_string()));
assert!(memory_stages.contains(&"model_loaded".to_string()));
assert!(memory_stages.contains(&"shutdown".to_string()));
fs::remove_dir_all(root).ok();
}
#[test]
fn profile_jsonl_append_is_parseable_under_concurrent_writers() {
let root = std::env::temp_dir().join(format!(
"ferrum-profile-concurrent-append-{}",
Uuid::new_v4().simple()
));
let path = root.join("profile.jsonl");
let config = ProductObservabilityConfig::new(
ProfileEntrypoint::Serve,
"Qwen/Qwen3-0.6B",
Some(&path),
ProfileDetailArg::Basic,
None,
None,
Some(&root.join("request_dump")),
1.0,
);
let mut handles = Vec::new();
for writer in 0..8 {
let path = path.clone();
let config = config.clone();
handles.push(std::thread::spawn(move || {
for seq in 0..32 {
let request_id = format!("req-{writer}-{seq}");
let mut event = actual_base_event(
&config,
&request_id,
"chat_completions_sync_complete",
ProfileEventKind::TimedSpan,
Utc::now(),
);
event.event_id = format!("evt-{writer}-{seq}");
event.correlation_id = Some(format!("corr-{writer}-{seq}"));
event.duration_us = Some(1);
event.attributes.insert("writer".to_string(), json!(writer));
event.attributes.insert("seq".to_string(), json!(seq));
append_profile_jsonl(&path, &[event]).unwrap();
}
}));
}
for handle in handles {
handle.join().unwrap();
}
let profile = fs::read_to_string(&path).unwrap();
let mut parsed = 0usize;
for line in profile.lines().filter(|line| !line.trim().is_empty()) {
serde_json::from_str::<serde_json::Value>(line).unwrap();
parsed += 1;
}
assert_eq!(parsed, 8 * 32);
fs::remove_dir_all(root).ok();
}
#[test]
fn actual_serve_startup_memory_orders_model_profile_and_cache_stages() {
let root = std::env::temp_dir().join(format!(
"ferrum-serve-memory-order-{}",
Uuid::new_v4().simple()
));
let config = ProductObservabilityConfig::new(
ProfileEntrypoint::Serve,
"Qwen/Qwen3-0.6B",
Some(&root.join("profile.jsonl")),
ProfileDetailArg::Basic,
Some(&root.join("memory.jsonl")),
Some(&root.join("scheduler.jsonl")),
Some(&root.join("request_dump")),
1.0,
);
let memory = crate::memory_profile::ProcessMemoryObservation {
before_bytes: 100,
after_bytes: 200,
current_bytes: 200,
high_water_bytes: 240,
source: "test",
};
write_actual_serve_startup_observability(
&config,
42,
Some(memory.clone()),
vec![
ActualMemoryStageObservation::new(
"actual_serve_process_start",
"process_start",
None,
Some(memory.clone()),
),
ActualMemoryStageObservation::new(
"actual_serve_backend_initialized",
"backend_initialized",
None,
Some(memory.clone()),
),
ActualMemoryStageObservation::new(
"actual_serve_profile_run_done",
"profile_run_done",
None,
Some(memory.clone()),
)
.with_profile_run_status(false, "not_configured", "test"),
ActualMemoryStageObservation::new(
"actual_serve_cache_allocated",
"cache_allocated",
None,
Some(memory),
)
.with_attribute("available_kv_or_state_bytes", json!(0)),
],
)
.unwrap();
let memory_profile = fs::read_to_string(root.join("memory.jsonl")).unwrap();
let memory_stages = memory_profile
.lines()
.filter(|line| !line.trim().is_empty())
.map(|line| serde_json::from_str::<serde_json::Value>(line).unwrap())
.filter_map(|event| {
event["attributes"]["memory_stage"]
.as_str()
.map(str::to_string)
})
.collect::<Vec<_>>();
assert_eq!(
memory_stages,
vec![
"process_start",
"backend_initialized",
"model_loaded",
"profile_run_done",
"cache_allocated"
]
);
fs::remove_dir_all(root).ok();
}
#[test]
fn actual_run_failure_observability_writes_diagnostics_bundle() {
let root = std::env::temp_dir().join(format!(
"ferrum-run-failure-observability-{}",
Uuid::new_v4().simple()
));
let config = ProductObservabilityConfig::new(
ProfileEntrypoint::Run,
"Qwen/Qwen3-0.6B",
Some(&root.join("profile.jsonl")),
ProfileDetailArg::Basic,
Some(&root.join("memory.jsonl")),
Some(&root.join("scheduler.jsonl")),
Some(&root.join("request_dump")),
1.0,
);
let request_id = "req-failure-test".to_string();
write_actual_run_failure_observability(
&config,
&ActualRunFailureObservation {
request_id: request_id.clone(),
duration_us: 42,
sampling_params: SamplingParams::greedy(),
prompt_token_ids: Some(vec![11, 22, 33]),
prompt_token_count: Some(3),
prompt_chars: 12,
failure_kind: "error".to_string(),
error_kind: "error".to_string(),
error_message: "synthetic failure".to_string(),
memory: None,
memory_stages: Vec::new(),
},
)
.unwrap();
let profile = fs::read_to_string(root.join("profile.jsonl")).unwrap();
let failure = profile
.lines()
.map(|line| serde_json::from_str::<serde_json::Value>(line).unwrap())
.find(|event| event["phase"] == "actual_run_generation_failed")
.expect("run failure profile event");
assert_eq!(failure["event_kind"], "timed_span");
assert_eq!(failure["status"], "failure");
assert_eq!(failure["duration_us"], 42);
assert_eq!(failure["attributes"]["terminal_failure_event"], true);
assert!(failure["attributes"]["first_failure_event"].is_null());
let bundle_dir = root.join("request_dump").join(&request_id);
assert!(bundle_dir.join("failure_diagnostics.json").is_file());
let scan: serde_json::Value = serde_json::from_str(
&fs::read_to_string(bundle_dir.join("bad_output_scan.json")).unwrap(),
)
.unwrap();
assert_eq!(scan["failure_kind"], "error");
let diagnostics: serde_json::Value = serde_json::from_str(
&fs::read_to_string(bundle_dir.join("failure_diagnostics.json")).unwrap(),
)
.unwrap();
assert_eq!(diagnostics["failure_kind"], "error");
assert_eq!(
diagnostics["first_failure_event"]["phase"],
"actual_run_generation_failed"
);
fs::remove_dir_all(root).ok();
}
#[test]
fn actual_run_observability_writes_prompt_token_ids() {
let root = std::env::temp_dir().join(format!(
"ferrum-run-observability-{}",
Uuid::new_v4().simple()
));
let config = ProductObservabilityConfig::new(
ProfileEntrypoint::Run,
"Qwen/Qwen3-0.6B",
Some(&root.join("profile.jsonl")),
ProfileDetailArg::Latency,
Some(&root.join("memory.jsonl")),
Some(&root.join("scheduler.jsonl")),
Some(&root.join("request_dump")),
1.0,
);
let request_id = "req-run-test".to_string();
let native_event = actual_base_event(
&config,
&request_id,
"vnext.request_accepted",
ProfileEventKind::Instant,
Utc::now(),
);
write_profile_jsonl(&root.join("profile.jsonl"), &[native_event]).unwrap();
write_actual_run_observability(
&config,
&ActualRunObservation {
request_id: request_id.clone(),
duration_us: 42,
sampling_params: SamplingParams::greedy(),
prompt_token_ids: Some(vec![7, 8, 9]),
prompt_token_count: Some(3),
output_tokens: 2,
output_token_ids: vec![10, 11],
chunk_count: 1,
finish_reason: Some("stop".to_string()),
prompt_chars: 12,
response_chars: 2,
response_text: "OK".to_string(),
execution_evidence: Some(ferrum_types::InferenceExecutionEvidence {
prompt_token_ids: vec![
ferrum_types::TokenId::new(7),
ferrum_types::TokenId::new(8),
ferrum_types::TokenId::new(9),
],
output_token_ids: vec![
ferrum_types::TokenId::new(10),
ferrum_types::TokenId::new(11),
],
engine_token_timing: Some(ferrum_types::EngineTokenTimingEvidence {
clock_source: "rust_std_instant".to_string(),
wall_anchor_unix_nanos: 1_700_000_000_000_000_000,
wall_anchor_max_error_nanos: 500,
decode_ready_nanos_since_request_start: Some(1_000_000),
token_commit_nanos_since_request_start: vec![2_000_000, 4_000_000],
decode_stage_intervals: Vec::new(),
}),
}),
memory: None,
memory_stages: vec![
ActualMemoryStageObservation::new(
"actual_run_process_start",
"process_start",
None,
Some(crate::memory_profile::ProcessMemoryObservation {
before_bytes: 100,
after_bytes: 100,
current_bytes: 100,
high_water_bytes: 120,
source: "test",
}),
),
ActualMemoryStageObservation::new(
"actual_run_shutdown",
"shutdown",
None,
Some(crate::memory_profile::ProcessMemoryObservation {
before_bytes: 200,
after_bytes: 220,
current_bytes: 220,
high_water_bytes: 240,
source: "test",
}),
),
],
},
)
.unwrap();
let bundle_dir = root.join("request_dump").join(&request_id);
let prompt_tokens: serde_json::Value = serde_json::from_str(
&fs::read_to_string(bundle_dir.join("prompt_token_ids.json")).unwrap(),
)
.unwrap();
assert_eq!(prompt_tokens["token_ids"], serde_json::json!([7, 8, 9]));
assert_eq!(prompt_tokens["token_count"], 3);
assert!(prompt_tokens["unavailable_reason"].is_null());
let output_text_bytes = fs::read(bundle_dir.join("output_text.txt")).unwrap();
let output_text = String::from_utf8(output_text_bytes.clone()).unwrap();
assert!(output_text.starts_with("[redacted actual output]\n"));
let scan: serde_json::Value = serde_json::from_str(
&fs::read_to_string(bundle_dir.join("bad_output_scan.json")).unwrap(),
)
.unwrap();
assert_eq!(
scan["output_sha256"],
sha256_hex(output_text_bytes.as_slice())
);
assert_eq!(scan["classified_output_sha256"], sha256_hex(b"OK"));
let profile = fs::read_to_string(root.join("profile.jsonl")).unwrap();
assert!(profile.lines().any(|line| {
serde_json::from_str::<serde_json::Value>(line)
.is_ok_and(|event| event["phase"] == "vnext.request_accepted")
}));
let generation = profile
.lines()
.filter(|line| !line.trim().is_empty())
.map(|line| serde_json::from_str::<serde_json::Value>(line).unwrap())
.find(|event| event["phase"] == "actual_run_generation")
.expect("actual run generation event");
assert_eq!(generation["attributes"]["prompt_token_count"], 3);
assert_eq!(generation["attributes"]["completion_token_count"], 2);
assert_eq!(generation["attributes"]["output_token_count"], 2);
assert_eq!(generation["attributes"]["total_token_count"], 5);
assert_eq!(
generation["attributes"]["token_count_source"],
"rendered_prompt_and_generated_tokens"
);
assert_eq!(generation["attributes"]["e2e_duration_us"], 42);
assert_eq!(generation["attributes"]["profile_detail"], "latency");
assert_eq!(
generation["attributes"]["engine_token_commit_nanos_since_request_start"],
serde_json::json!([2_000_000, 4_000_000])
);
assert_eq!(generation["attributes"]["engine_token_commit_count"], 2);
assert_eq!(
generation["attributes"]["itl_source"],
"engine_token_commit"
);
assert_eq!(
generation["attributes"]["itl_nanos"],
serde_json::json!([2_000_000])
);
assert_eq!(generation["attributes"]["ttft_us"], 2_000);
assert_eq!(generation["attributes"]["itl_us_avg"], 2_000);
assert_eq!(
generation["attributes"]["engine_decode_ready_nanos_since_request_start"],
1_000_000
);
assert_eq!(
generation["attributes"]["engine_decode_wall_nanos"],
3_000_000
);
assert_eq!(generation["attributes"]["clock_conversion_error_ppm"], 166);
assert_eq!(
generation["attributes"]["decode_wall_timing_eligible"],
true
);
let memory_profile = fs::read_to_string(root.join("memory.jsonl")).unwrap();
let memory_stages = memory_profile
.lines()
.filter(|line| !line.trim().is_empty())
.map(|line| serde_json::from_str::<serde_json::Value>(line).unwrap())
.filter_map(|event| {
event["attributes"]["memory_stage"]
.as_str()
.map(str::to_string)
})
.collect::<Vec<_>>();
assert!(memory_stages.contains(&"process_start".to_string()));
assert!(memory_stages.contains(&"first_request_done".to_string()));
let first_request_index = memory_stages
.iter()
.position(|stage| stage == "first_request_done")
.expect("first_request_done stage");
let shutdown_index = memory_stages
.iter()
.position(|stage| stage == "shutdown")
.expect("shutdown stage");
assert!(first_request_index < shutdown_index);
assert!(
!root.join("scheduler.jsonl").exists(),
"the engine owns successful actual-run scheduler lifecycles"
);
fs::remove_dir_all(root).ok();
}
#[test]
fn actual_run_resource_failure_observability_writes_resource_diagnostics() {
let root = std::env::temp_dir().join(format!(
"ferrum-observability-resource-failure-{}",
uuid::Uuid::new_v4()
));
fs::create_dir_all(&root).unwrap();
let config = ProductObservabilityConfig::new(
ProfileEntrypoint::Run,
"Qwen/Qwen3-0.6B",
Some(&root.join("profile.jsonl")),
ProfileDetailArg::Basic,
Some(&root.join("memory.jsonl")),
Some(&root.join("scheduler.jsonl")),
Some(&root.join("request_dump")),
1.0,
);
let request_id = "req-resource-failure-test".to_string();
write_actual_run_failure_observability(
&config,
&ActualRunFailureObservation {
request_id: request_id.clone(),
duration_us: 42,
sampling_params: SamplingParams::greedy(),
prompt_token_ids: Some(vec![1, 2, 3, 4]),
prompt_token_count: Some(4),
prompt_chars: 128,
failure_kind: "oom_admission".to_string(),
error_kind: "resource_exhausted".to_string(),
error_message: "Resource exhausted: recurrent state capacity exhausted".to_string(),
memory: Some(crate::memory_profile::ProcessMemoryObservation {
before_bytes: 1024,
after_bytes: 2048,
current_bytes: 2048,
high_water_bytes: 4096,
source: "test",
}),
memory_stages: Vec::new(),
},
)
.unwrap();
let bundle_dir = root.join("request_dump").join(&request_id);
let scan: serde_json::Value = serde_json::from_str(
&fs::read_to_string(bundle_dir.join("bad_output_scan.json")).unwrap(),
)
.unwrap();
assert_eq!(scan["failure_kind"], "oom_admission");
let diagnostics: serde_json::Value = serde_json::from_str(
&fs::read_to_string(bundle_dir.join("failure_diagnostics.json")).unwrap(),
)
.unwrap();
assert_eq!(diagnostics["failure_kind"], "oom_admission");
assert_eq!(
diagnostics["capacity"]["resource_kind"],
"admission_capacity"
);
assert_eq!(diagnostics["nearest_resource_event"]["action"], "reject");
assert_eq!(
diagnostics["nearest_memory_snapshot"]["current_bytes"],
2048
);
fs::remove_dir_all(root).ok();
}
}