Browse Source

feat: report rpm limiter mode details

main
orangels 1 month ago
parent
commit
2b9fae7ebd
2 changed files with 343 additions and 4 deletions
  1. +236
    -4
      src/cli.rs
  2. +107
    -0
      src/report.rs

+ 236
- 4
src/cli.rs View File

@@ -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<ScheduledProbe>,
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<u64, PhaseAccumulator>,
sliding_seconds: BTreeMap<u64, PhaseAccumulator>,
}

impl RpmModeSummaryBuilder {
fn record(&mut self, phase: ProbePhase, second: Option<u64>, 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<RpmModeDetailReport> {
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<u64, PhaseAccumulator>) -> Vec<ProbeSecondReport> {
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<f64> {
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<u32>) -> Result<u32> {
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: {}",


+ 107
- 0
src/report.rs View File

@@ -66,6 +66,7 @@ pub struct RpmRunReport {
pub started_at: DateTime<Utc>,
pub duration_ms: u128,
pub target_rpm: u32,
pub actual_rpm: Option<f64>,
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<RpmModeDetailReport>,
pub errors: Vec<ErrorCount>,
}

#[derive(Debug, Clone, Serialize)]
pub struct RpmModeDetailReport {
pub burst: Option<PhaseSummaryReport>,
pub refill_probe: Vec<ProbeSecondReport>,
pub sliding_probe: Vec<ProbeSecondReport>,
pub window_boundary: Option<WindowBoundaryReport>,
pub inference: Option<LimiterInferenceReport>,
}

#[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<String>,
}

#[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<PathBuf> {
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\""));
}
}

Loading…
Cancel
Save