github/codex
4
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-19 16:17:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

Measure rollout persistence size reductions (#50454)

Browse Source 
## What changed

- Record original and persisted item sizes in `codex.rollout.persistence.item_bytes_v2`, distinguished by stage, and record positive size reductions in `codex.rollout.persistence.bytes_removed`. Count the full original size as removed for dropped items.
- Use explicit histogram boundaries from 1 KiB to 64 MiB for item sizes, bytes removed, and `codex.rollout.persistence.turn_bytes_v2`. Larger values remain represented in the overflow bucket and histogram sum.
- Align item and turn size stages and report a measurement error when a kept item's persisted size cannot be measured.

GitOrigin-RevId: 6487acf1c6f2c041ab9680ab398491d3c1541388
This commit is contained in:
Owen Lin
2026-10-02 23:06:56 +00:00
committed by copyberry
parent fd75aa116c
commit 3c3a990da0
1 changed files with 82 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/rollout/src/persistence_metrics.rs
+82 -15
View File
@@ -14,13 +14,31 @@ use crate::ResponseItemEnvelope;
use crate::RolloutItem;
use crate::policy::persisted_rollout_item;
const ITEM_BYTES_METRIC: &str = "codex.rollout.persistence.item_bytes";
const ITEM_BYTES_METRIC: &str = "codex.rollout.persistence.item_bytes_v2";
const BYTES_REMOVED_METRIC: &str = "codex.rollout.persistence.bytes_removed";
const APPEND_METRIC: &str = "codex.rollout.persistence.append";
const TURN_BYTES_METRIC: &str = "codex.rollout.persistence.turn_bytes";
const TURN_BYTES_METRIC: &str = "codex.rollout.persistence.turn_bytes_v2";
const MEASUREMENT_ERROR_METRIC: &str = "codex.rollout.persistence.measurement_error";
const SAMPLE_DENOMINATOR: u64 = 100;
const SAMPLE_RATE_LABEL: &str = "0.01";
const KIB: f64 = 1_024.0;
const MIB: f64 = 1_024.0 * KIB;
// Cover the persistence cap while retaining enough range for large items and turns. Values above
// 64 MiB remain visible in the overflow bucket and histogram sum.
const ROLLOUT_BYTES_BOUNDARIES: &[f64] = &[
KIB,
4.0 * KIB,
16.0 * KIB,
64.0 * KIB,
256.0 * KIB,
MIB,
4.0 * MIB,
16.0 * MIB,
64.0 * MIB,
];
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PersistenceDecision {
Kept,
@@ -315,6 +333,26 @@ fn response_item_type(item: &ResponseItemEnvelope) -> &'static str {
}
}
fn record_item_bytes(
metrics: &MetricsClient,
stage: &'static str,
item: &RolloutItemMeasurement,
payload_bytes: u64,
) {
let _ = metrics.histogram_with_boundaries(
ITEM_BYTES_METRIC,
saturating_i64(payload_bytes),
ROLLOUT_BYTES_BOUNDARIES,
&[
("stage", stage),
("decision", item.decision.as_str()),
("rollout_item_type", item.rollout_item_type.as_str()),
("encoding", "rollout_item_json_v1"),
("sample_rate", SAMPLE_RATE_LABEL),
],
);
}
#[derive(Clone)]
pub struct RolloutPersistenceTelemetry {
metrics: Option<MetricsClient>,
@@ -348,16 +386,44 @@ impl RolloutPersistenceTelemetry {
for item in &measurement.items {
if let Some(payload_bytes) = item.payload_bytes {
let _ = metrics.histogram(
ITEM_BYTES_METRIC,
saturating_i64(payload_bytes),
&[
("decision", item.decision.as_str()),
("rollout_item_type", item.rollout_item_type.as_str()),
("encoding", "rollout_item_json_v1"),
("sample_rate", SAMPLE_RATE_LABEL),
],
);
record_item_bytes(metrics, "before_projection", item, payload_bytes);
let persisted_payload_bytes = match item.persisted_payload_bytes {
Some(persisted_payload_bytes) => {
record_item_bytes(
metrics,
"after_projection",
item,
persisted_payload_bytes,
);
persisted_payload_bytes
}
None if item.decision == PersistenceDecision::Dropped => 0,
None => {
let _ = metrics.counter(
MEASUREMENT_ERROR_METRIC,
/*inc*/ 1,
&[
("rollout_item_type", item.rollout_item_type.as_str()),
("phase", "serialize_projection"),
],
);
continue;
}
};
let bytes_removed = payload_bytes.saturating_sub(persisted_payload_bytes);
if bytes_removed > 0 {
let _ = metrics.histogram_with_boundaries(
BYTES_REMOVED_METRIC,
saturating_i64(bytes_removed),
ROLLOUT_BYTES_BOUNDARIES,
&[
("decision", item.decision.as_str()),
("rollout_item_type", item.rollout_item_type.as_str()),
("encoding", "rollout_item_json_v1"),
("sample_rate", SAMPLE_RATE_LABEL),
],
);
}
} else {
let _ = metrics.counter(
MEASUREMENT_ERROR_METRIC,
@@ -406,12 +472,13 @@ impl RolloutPersistenceTelemetry {
}
for turn in turn_update.completed {
for (stage, totals) in [
("pre_filter", turn.totals.pre_filter),
("post_filter", turn.totals.post_filter),
("before_projection", turn.totals.pre_filter),
("after_projection", turn.totals.post_filter),
] {
let _ = metrics.histogram(
let _ = metrics.histogram_with_boundaries(
TURN_BYTES_METRIC,
saturating_i64(totals.payload_bytes),
ROLLOUT_BYTES_BOUNDARIES,
&[
("stage", stage),
("outcome", turn.outcome.as_str()),