From 2b9fae7ebdbc25fac1ae9524b774ea35664e8704 Mon Sep 17 00:00:00 2001 From: orangels Date: Wed, 6 May 2026 21:25:54 +0800 Subject: [PATCH] feat: report rpm limiter mode details --- src/cli.rs | 240 +++++++++++++++++++++++++++++++++++++++++++++++++- src/report.rs | 107 ++++++++++++++++++++++ 2 files changed, 343 insertions(+), 4 deletions(-) diff --git a/src/cli.rs b/src/cli.rs index 93dbeb0..85e6524 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -3,8 +3,10 @@ use crate::benchmarks::judge; use crate::config::AppConfig; use crate::metrics::{LatencySummary, Metrics, MetricsSummary}; use crate::report::{ - BenchmarkReport, BenchmarkSummaryReport, DatasetReport, LatencyReport, RpmReport, RpmRunReport, - RpmSummaryReport, RunReport, WrongCaseReport, write_benchmark_report, write_rpm_report, + BenchmarkReport, BenchmarkSummaryReport, DatasetReport, LatencyReport, LimiterInferenceKind, + LimiterInferenceReport, PhaseSummaryReport, ProbeSecondReport, RpmModeDetailReport, RpmReport, + RpmRunReport, RpmSummaryReport, RunReport, WindowBoundaryReport, WrongCaseReport, + write_benchmark_report, write_rpm_report, }; use crate::rpm_modes::{ ProbePhase, RpmMode, ScheduledProbe, burst_schedule, sliding_window_schedule, @@ -16,6 +18,7 @@ use chrono::Utc; use clap::{Parser, Subcommand}; use futures::{StreamExt, stream}; use regex::Regex; +use std::collections::BTreeMap; use std::path::{Path, PathBuf}; use std::time::{Duration, Instant}; use tokio::time::{Instant as TokioInstant, sleep_until}; @@ -396,6 +399,7 @@ async fn run_rpm(config_path: PathBuf, options: RpmCommandOptions) -> Result<()> let started_at = Utc::now(); let started = Instant::now(); let mut metrics = Metrics::new(); + let mut mode_summary = RpmModeSummaryBuilder::default(); let results = run_scheduled_requests( provider_config.protocol, @@ -406,8 +410,8 @@ async fn run_rpm(config_path: PathBuf, options: RpmCommandOptions) -> Result<()> .await; for result in results { - let _phase = result.phase; - let _second = result.second; + let success = result.result.is_ok(); + mode_summary.record(result.phase, result.second, success); match result.result { Ok(response) => metrics.record_success(response.status, response.elapsed_ms as u64), Err(error) => metrics.record_failure(error_code(&error)), @@ -423,6 +427,7 @@ async fn run_rpm(config_path: PathBuf, options: RpmCommandOptions) -> Result<()> started_at, duration_ms: started.elapsed().as_millis(), target_rpm: mode_plan.target_rpm, + actual_rpm: actual_rpm(summary.total, started.elapsed()), temperature: 0.0, max_tokens: 1024, }, @@ -432,6 +437,8 @@ async fn run_rpm(config_path: PathBuf, options: RpmCommandOptions) -> Result<()> failure: summary.failed, latency_ms: latency_report(&summary.latency_ms), }, + mode: mode_plan.mode_name.to_string(), + mode_detail: mode_summary.into_report(options.mode), errors: summary.errors, }; let report_path = write_rpm_report(Path::new("."), &report)?; @@ -441,6 +448,7 @@ async fn run_rpm(config_path: PathBuf, options: RpmCommandOptions) -> Result<()> #[derive(Debug)] struct RpmModePlan { + mode_name: &'static str, target_rpm: u32, probes: Vec, default_concurrency: usize, @@ -511,12 +519,227 @@ fn build_rpm_mode_plan( let default_concurrency = probes.len().max(1); Ok(RpmModePlan { + mode_name: mode_name(mode), target_rpm, default_concurrency, probes, }) } +fn mode_name(mode: RpmMode) -> &'static str { + match mode { + RpmMode::Sustained => "sustained", + RpmMode::Burst => "burst", + RpmMode::TokenBucket => "token-bucket", + RpmMode::SlidingWindow => "sliding-window", + RpmMode::WindowBoundary => "window-boundary", + RpmMode::Diagnose => "diagnose", + } +} + +#[derive(Default)] +struct RpmModeSummaryBuilder { + phases: BTreeMap<&'static str, PhaseAccumulator>, + refill_seconds: BTreeMap, + sliding_seconds: BTreeMap, +} + +impl RpmModeSummaryBuilder { + fn record(&mut self, phase: ProbePhase, second: Option, success: bool) { + match phase { + ProbePhase::Burst => self.phase("burst").record(success), + ProbePhase::RefillProbe => { + self.phase("refill_probe").record(success); + self.refill_seconds + .entry(second.unwrap_or(0)) + .or_default() + .record(success); + } + ProbePhase::SlidingProbe => { + self.phase("sliding_probe").record(success); + self.sliding_seconds + .entry(second.unwrap_or(0)) + .or_default() + .record(success); + } + ProbePhase::BeforeBoundary => self.phase("before_boundary").record(success), + ProbePhase::AfterBoundary => self.phase("after_boundary").record(success), + } + } + + fn into_report(self, mode: RpmMode) -> Option { + if mode == RpmMode::Sustained { + return None; + } + + let burst = self.phases.get("burst").map(PhaseAccumulator::to_report); + let refill_probe = probe_seconds_report(self.refill_seconds); + let sliding_probe = probe_seconds_report(self.sliding_seconds); + let window_boundary = match ( + self.phases.get("before_boundary"), + self.phases.get("after_boundary"), + ) { + (Some(before), Some(after)) => Some(WindowBoundaryReport { + before: before.to_report(), + after: after.to_report(), + }), + _ => None, + }; + let inference = if mode == RpmMode::Diagnose { + Some(infer_limiter( + burst.as_ref(), + &refill_probe, + &sliding_probe, + window_boundary.as_ref(), + )) + } else { + None + }; + + Some(RpmModeDetailReport { + burst, + refill_probe, + sliding_probe, + window_boundary, + inference, + }) + } + + fn phase(&mut self, name: &'static str) -> &mut PhaseAccumulator { + self.phases.entry(name).or_default() + } +} + +#[derive(Default)] +struct PhaseAccumulator { + sent: u64, + success: u64, + failure: u64, +} + +impl PhaseAccumulator { + fn record(&mut self, success: bool) { + self.sent += 1; + if success { + self.success += 1; + } else { + self.failure += 1; + } + } + + fn to_report(&self) -> PhaseSummaryReport { + PhaseSummaryReport { + sent: self.sent, + success: self.success, + failure: self.failure, + } + } + + fn success_rate(&self) -> f64 { + if self.sent == 0 { + 0.0 + } else { + self.success as f64 / self.sent as f64 + } + } +} + +fn probe_seconds_report(seconds: BTreeMap) -> Vec { + seconds + .into_iter() + .map(|(second, accumulator)| ProbeSecondReport { + second, + sent: accumulator.sent, + success: accumulator.success, + failure: accumulator.failure, + }) + .collect() +} + +fn infer_limiter( + _burst: Option<&PhaseSummaryReport>, + refill_probe: &[ProbeSecondReport], + sliding_probe: &[ProbeSecondReport], + window_boundary: Option<&WindowBoundaryReport>, +) -> LimiterInferenceReport { + let mut signals = Vec::new(); + + if let Some(boundary) = window_boundary { + let before_rate = phase_success_rate(&boundary.before); + let after_rate = phase_success_rate(&boundary.after); + if after_rate > before_rate + 0.3 { + signals.push("after-boundary success rate was much higher than before-boundary".into()); + return LimiterInferenceReport { + likely_limiter: LimiterInferenceKind::FixedWindow, + confidence: "medium".to_string(), + signals, + }; + } + } + + let refill_sent: u64 = refill_probe.iter().map(|probe| probe.sent).sum(); + let refill_success: u64 = refill_probe.iter().map(|probe| probe.success).sum(); + if refill_sent > 0 && refill_success as f64 / refill_sent as f64 >= 0.5 { + signals.push("refill probes recovered at a steady rate".into()); + return LimiterInferenceReport { + likely_limiter: LimiterInferenceKind::TokenBucket, + confidence: "medium".to_string(), + signals, + }; + } + + let early = sliding_probe + .iter() + .filter(|probe| probe.second <= 30) + .fold(PhaseAccumulator::default(), |mut acc, probe| { + acc.sent += probe.sent; + acc.success += probe.success; + acc.failure += probe.failure; + acc + }); + let late = sliding_probe + .iter() + .filter(|probe| probe.second >= 60) + .fold(PhaseAccumulator::default(), |mut acc, probe| { + acc.sent += probe.sent; + acc.success += probe.success; + acc.failure += probe.failure; + acc + }); + if late.sent > 0 && late.success_rate() > early.success_rate() + 0.3 { + signals.push("probe recovery improved near the 60 second rolling window".into()); + return LimiterInferenceReport { + likely_limiter: LimiterInferenceKind::SlidingWindow, + confidence: "medium".to_string(), + signals, + }; + } + + signals.push("signals did not clearly match a limiter model".into()); + LimiterInferenceReport { + likely_limiter: LimiterInferenceKind::Unknown, + confidence: "low".to_string(), + signals, + } +} + +fn phase_success_rate(phase: &PhaseSummaryReport) -> f64 { + if phase.sent == 0 { + 0.0 + } else { + phase.success as f64 / phase.sent as f64 + } +} + +fn actual_rpm(total_requests: u64, elapsed: Duration) -> Option { + let elapsed_seconds = elapsed.as_secs_f64(); + if elapsed_seconds == 0.0 { + None + } else { + Some(total_requests as f64 / elapsed_seconds * 60.0) + } +} + fn require_positive(name: &str, value: Option) -> Result { let value = value.with_context(|| format!("{name} is required for this rpm mode"))?; require_positive_value(name, value) @@ -687,7 +910,16 @@ fn print_benchmark_report(report: &BenchmarkReport, report_path: &Path) { } fn print_rpm_report(report: &RpmReport, report_path: &Path) { + println!("mode: {}", report.mode); println!("target_rpm: {}", report.run.target_rpm); + println!( + "actual_rpm: {}", + report + .run + .actual_rpm + .map(|rpm| format!("{rpm:.2}")) + .unwrap_or_else(|| "n/a".to_string()) + ); println!("actual_requests: {}", report.summary.actual_requests); println!( "success: {} failed: {}", diff --git a/src/report.rs b/src/report.rs index 12a64c7..dafc3d6 100644 --- a/src/report.rs +++ b/src/report.rs @@ -66,6 +66,7 @@ pub struct RpmRunReport { pub started_at: DateTime, pub duration_ms: u128, pub target_rpm: u32, + pub actual_rpm: Option, pub temperature: f32, pub max_tokens: u32, } @@ -81,13 +82,61 @@ pub struct RpmSummaryReport { #[derive(Debug, Clone, Serialize)] pub struct RpmReport { pub benchmark: String, + pub mode: String, pub provider: String, pub model: String, pub run: RpmRunReport, pub summary: RpmSummaryReport, + pub mode_detail: Option, pub errors: Vec, } +#[derive(Debug, Clone, Serialize)] +pub struct RpmModeDetailReport { + pub burst: Option, + pub refill_probe: Vec, + pub sliding_probe: Vec, + pub window_boundary: Option, + pub inference: Option, +} + +#[derive(Debug, Clone, Serialize)] +pub struct PhaseSummaryReport { + pub sent: u64, + pub success: u64, + pub failure: u64, +} + +#[derive(Debug, Clone, Serialize)] +pub struct ProbeSecondReport { + pub second: u64, + pub sent: u64, + pub success: u64, + pub failure: u64, +} + +#[derive(Debug, Clone, Serialize)] +pub struct WindowBoundaryReport { + pub before: PhaseSummaryReport, + pub after: PhaseSummaryReport, +} + +#[derive(Debug, Clone, Serialize)] +pub struct LimiterInferenceReport { + pub likely_limiter: LimiterInferenceKind, + pub confidence: String, + pub signals: Vec, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum LimiterInferenceKind { + TokenBucket, + FixedWindow, + SlidingWindow, + Unknown, +} + pub fn write_benchmark_report(root: &Path, report: &BenchmarkReport) -> Result { write_report( root, @@ -231,6 +280,7 @@ mod tests { started_at: Utc.with_ymd_and_hms(2026, 5, 6, 1, 2, 3).unwrap(), duration_ms: 1000, target_rpm: 60, + actual_rpm: Some(60.0), temperature: 0.0, max_tokens: 1024, }, @@ -244,6 +294,8 @@ mod tests { p99: Some(10), }, }, + mode: "sustained".to_string(), + mode_detail: None, errors: vec![], }; @@ -252,4 +304,59 @@ mod tests { assert!(!json.contains("sensitive prompt")); assert!(!json.contains("\"prompt\"")); } + + #[test] + fn rpm_report_serializes_mode_details_and_inference() { + let report = RpmReport { + benchmark: "rpm".to_string(), + mode: "token-bucket".to_string(), + provider: "anthropic".to_string(), + model: "claude/test".to_string(), + run: RpmRunReport { + started_at: Utc.with_ymd_and_hms(2026, 5, 6, 1, 2, 3).unwrap(), + duration_ms: 2000, + target_rpm: 120, + actual_rpm: Some(120.0), + temperature: 0.0, + max_tokens: 1024, + }, + summary: RpmSummaryReport { + actual_requests: 4, + success: 4, + failure: 0, + latency_ms: LatencyReport { + p50: Some(10), + p95: Some(20), + p99: Some(30), + }, + }, + mode_detail: Some(RpmModeDetailReport { + burst: Some(PhaseSummaryReport { + sent: 2, + success: 2, + failure: 0, + }), + refill_probe: vec![ProbeSecondReport { + second: 1, + sent: 2, + success: 2, + failure: 0, + }], + sliding_probe: vec![], + window_boundary: None, + inference: Some(LimiterInferenceReport { + likely_limiter: LimiterInferenceKind::TokenBucket, + confidence: "medium".to_string(), + signals: vec!["probe success approximated refill".to_string()], + }), + }), + errors: vec![], + }; + + let json = serde_json::to_string(&report).expect("serialize report"); + + assert!(json.contains("\"mode\":\"token-bucket\"")); + assert!(json.contains("\"refill_probe\"")); + assert!(json.contains("\"likely_limiter\":\"token_bucket\"")); + } }