openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-09-13 11:47:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add monitoring metrics to analytics events

Browse Source
This commit is contained in:
alexsong-oai
2026-03-25 13:29:21 -07:00
parent 621862a7d1
commit bc9ad48874
2 changed files with 295 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

132
codex-rs/core/src/analytics_client.rs
View File

@@ -15,6 +15,7 @@ use std::sync::Arc;
use std::sync::Mutex;
use std::time::Duration;
use tokio::sync::mpsc;
use tokio::sync::mpsc::error::TrySendError;
#[derive(Clone)]
pub(crate) struct TrackEventsContext {
@@ -50,6 +51,15 @@ pub(crate) enum InvocationType {
Implicit,
}
impl InvocationType {
fn tag_value(self) -> &'static str {
match self {
Self::Explicit => "explicit",
Self::Implicit => "implicit",
}
}
}
pub(crate) struct AppInvocation {
pub(crate) connector_id: Option<String>,
pub(crate) app_name: Option<String>,
@@ -110,9 +120,14 @@ impl AnalyticsEventsQueue {
}
fn try_send(&self, job: TrackEventsJob) {
if self.sender.try_send(job).is_err() {
//TODO: add a metric for this
tracing::warn!("dropping analytics events: queue is full");
if let Err(err) = self.sender.try_send(job) {
let (reason, job) = match &err {
TrySendError::Full(job) => ("queue_full", job),
TrySendError::Closed(job) => ("queue_closed", job),
};
let job_type = job.job_type();
emit_analytics_events_failure_metric(reason, job_type, job.invoke_type_tag(), &[]);
tracing::warn!("dropping analytics events job: reason={reason} job_type={job_type}");
}
}
@@ -245,6 +260,56 @@ enum TrackEventsJob {
PluginDisabled(TrackPluginManagementJob),
}
impl TrackEventsJob {
fn job_type(&self) -> &'static str {
match self {
Self::SkillInvocations(_) => "skill_invocations",
Self::AppMentioned(_) => "app_mentioned",
Self::AppUsed(_) => "app_used",
Self::PluginUsed(_) => "plugin_used",
Self::PluginInstalled(_) => "plugin_installed",
Self::PluginUninstalled(_) => "plugin_uninstalled",
Self::PluginEnabled(_) => "plugin_enabled",
Self::PluginDisabled(_) => "plugin_disabled",
}
}
fn invoke_type_tag(&self) -> Option<&'static str> {
match self {
Self::SkillInvocations(job) => job
.invocations
.first()
.map(|invocation| invocation.invocation_type.tag_value()),
Self::AppMentioned(job) => job
.mentions
.first()
.and_then(|mention| mention.invocation_type.map(InvocationType::tag_value)),
Self::AppUsed(job) => job.app.invocation_type.map(InvocationType::tag_value),
Self::PluginUsed(_)
| Self::PluginInstalled(_)
| Self::PluginUninstalled(_)
| Self::PluginEnabled(_)
| Self::PluginDisabled(_) => None,
}
}
}
fn emit_analytics_events_failure_metric<'a>(
reason: &'static str,
job_type: &'static str,
invoke_type: Option<&'static str>,
extra_tags: &[(&'a str, &'a str)],
) {
if let Some(metrics) = codex_otel::metrics::global() {
let mut tags = vec![("reason", reason), ("job_type", job_type)];
if let Some(invoke_type) = invoke_type {
tags.push(("invoke_type", invoke_type));
}
tags.extend(extra_tags.iter().copied());
let _ = metrics.counter("codex.analytics_events.emit.failure", /*inc*/ 1, &tags);
}
}
struct TrackSkillInvocationsJob {
config: Arc<Config>,
tracking: TrackEventsContext,
@@ -678,17 +743,52 @@ async fn send_track_events(
if events.is_empty() {
return;
}
let (job_type, invoke_type) = match events.first() {
Some(TrackEventRequest::SkillInvocation(event)) => (
"skill_invocations",
event
.event_params
.invoke_type
.map(InvocationType::tag_value),
),
Some(TrackEventRequest::AppMentioned(event)) => (
"app_mentioned",
event
.event_params
.invoke_type
.map(InvocationType::tag_value),
),
Some(TrackEventRequest::AppUsed(event)) => (
"app_used",
event
.event_params
.invoke_type
.map(InvocationType::tag_value),
),
Some(TrackEventRequest::PluginUsed(_)) => ("plugin_used", None),
Some(TrackEventRequest::PluginInstalled(_)) => ("plugin_installed", None),
Some(TrackEventRequest::PluginUninstalled(_)) => ("plugin_uninstalled", None),
Some(TrackEventRequest::PluginEnabled(_)) => ("plugin_enabled", None),
Some(TrackEventRequest::PluginDisabled(_)) => ("plugin_disabled", None),
None => unreachable!("events should be non-empty"),
};
let Some(auth) = auth_manager.auth().await else {
emit_analytics_events_failure_metric("auth_missing", job_type, invoke_type, &[]);
return;
};
if !auth.is_chatgpt_auth() {
emit_analytics_events_failure_metric("non_chatgpt_auth", job_type, invoke_type, &[]);
return;
}
let access_token = match auth.get_token() {
Ok(token) => token,
Err(_) => return,
Err(_) => {
emit_analytics_events_failure_metric("token_error", job_type, invoke_type, &[]);
return;
}
};
let Some(account_id) = auth.get_account_id() else {
emit_analytics_events_failure_metric("account_id_missing", job_type, invoke_type, &[]);
return;
};
@@ -710,10 +810,17 @@ async fn send_track_events(
Ok(response) if response.status().is_success() => {}
Ok(response) => {
let status = response.status();
emit_analytics_events_failure_metric(
"http_status",
job_type,
invoke_type,
&[("status_code", status.as_str())],
);
let body = response.text().await.unwrap_or_default();
tracing::warn!("events failed with status {status}: {body}");
}
Err(err) => {
emit_analytics_events_failure_metric("request_error", job_type, invoke_type, &[]);
tracing::warn!("failed to send events request: {err}");
}
}
@@ -725,6 +832,13 @@ pub(crate) fn skill_id_for_local_skill(
skill_path: &Path,
skill_name: &str,
) -> String {
tracing::info!(
?repo_url,
?repo_root,
skill_path = %skill_path.display(),
skill_name,
"building analytics skill id for local skill"
);
let path = normalize_path_for_skill_id(repo_url, repo_root, skill_path);
let prefix = if let Some(url) = repo_url {
format!("repo_{url}")
@@ -734,7 +848,15 @@ pub(crate) fn skill_id_for_local_skill(
let raw_id = format!("{prefix}_{path}_{skill_name}");
let mut hasher = Sha1::new();
hasher.update(raw_id.as_bytes());
format!("{:x}", hasher.finalize())
let skill_id = format!("{:x}", hasher.finalize());
tracing::info!(
normalized_path = path,
prefix,
raw_id,
skill_id,
"built analytics skill id for local skill"
);
skill_id
}
/// Returns a normalized path for skill ID construction.

168
codex-rs/core/src/analytics_client_tests.rs
View File

@@ -5,12 +5,19 @@ use super::CodexAppUsedEventRequest;
use super::CodexPluginEventRequest;
use super::CodexPluginUsedEventRequest;
use super::InvocationType;
use super::TrackAppMentionedJob;
use super::TrackAppUsedJob;
use super::TrackEventRequest;
use super::TrackEventsContext;
use super::TrackEventsJob;
use super::TrackPluginManagementJob;
use super::TrackPluginUsedJob;
use super::TrackSkillInvocationsJob;
use super::codex_app_metadata;
use super::codex_plugin_metadata;
use super::codex_plugin_used_metadata;
use super::normalize_path_for_skill_id;
use crate::config::ConfigBuilder;
use crate::plugins::AppConnectorId;
use crate::plugins::PluginCapabilitySummary;
use crate::plugins::PluginId;
@@ -21,6 +28,7 @@ use std::collections::HashSet;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::Mutex;
use tempfile::TempDir;
use tokio::sync::mpsc;
fn expected_absolute_path(path: &PathBuf) -> String {
@@ -271,6 +279,152 @@ fn plugin_used_dedupe_is_keyed_by_turn_and_plugin() {
assert_eq!(queue.should_enqueue_plugin_used(&turn_2, &plugin), true);
}
#[test]
fn track_events_job_type_uses_expected_tag_values() {
let codex_home = TempDir::new().expect("tempdir should create");
let config = Arc::new(load_test_config(codex_home.path()));
let tracking = TrackEventsContext {
model_slug: "gpt-5".to_string(),
thread_id: "thread-1".to_string(),
turn_id: "turn-1".to_string(),
};
let cases = vec![
(
TrackEventsJob::SkillInvocations(TrackSkillInvocationsJob {
config: Arc::clone(&config),
tracking: tracking.clone(),
invocations: Vec::new(),
}),
"skill_invocations",
),
(
TrackEventsJob::AppMentioned(TrackAppMentionedJob {
config: Arc::clone(&config),
tracking: tracking.clone(),
mentions: Vec::new(),
}),
"app_mentioned",
),
(
TrackEventsJob::AppUsed(TrackAppUsedJob {
config: Arc::clone(&config),
tracking: tracking.clone(),
app: AppInvocation {
connector_id: None,
app_name: None,
invocation_type: None,
},
}),
"app_used",
),
(
TrackEventsJob::PluginUsed(TrackPluginUsedJob {
config: Arc::clone(&config),
tracking: tracking.clone(),
plugin: sample_plugin_metadata(),
}),
"plugin_used",
),
(
TrackEventsJob::PluginInstalled(TrackPluginManagementJob {
config: Arc::clone(&config),
plugin: sample_plugin_metadata(),
}),
"plugin_installed",
),
(
TrackEventsJob::PluginUninstalled(TrackPluginManagementJob {
config: Arc::clone(&config),
plugin: sample_plugin_metadata(),
}),
"plugin_uninstalled",
),
(
TrackEventsJob::PluginEnabled(TrackPluginManagementJob {
config: Arc::clone(&config),
plugin: sample_plugin_metadata(),
}),
"plugin_enabled",
),
(
TrackEventsJob::PluginDisabled(TrackPluginManagementJob {
config,
plugin: sample_plugin_metadata(),
}),
"plugin_disabled",
),
];
for (job, expected) in cases {
assert_eq!(job.job_type(), expected);
}
}
#[test]
fn track_events_job_invoke_type_tag_uses_first_available_value() {
let codex_home = TempDir::new().expect("tempdir should create");
let config = Arc::new(load_test_config(codex_home.path()));
let tracking = TrackEventsContext {
model_slug: "gpt-5".to_string(),
thread_id: "thread-1".to_string(),
turn_id: "turn-1".to_string(),
};
let explicit_skill_job = TrackEventsJob::SkillInvocations(TrackSkillInvocationsJob {
config: Arc::clone(&config),
tracking: tracking.clone(),
invocations: vec![super::SkillInvocation {
skill_name: "doc".to_string(),
skill_scope: codex_protocol::protocol::SkillScope::Repo,
skill_path: codex_home.path().join("SKILL.md"),
invocation_type: InvocationType::Explicit,
}],
});
let implicit_app_job = TrackEventsJob::AppUsed(TrackAppUsedJob {
config: Arc::clone(&config),
tracking: tracking.clone(),
app: AppInvocation {
connector_id: Some("drive".to_string()),
app_name: Some("Google Drive".to_string()),
invocation_type: Some(InvocationType::Implicit),
},
});
let first_value_app_mentioned_job = TrackEventsJob::AppMentioned(TrackAppMentionedJob {
config: Arc::clone(&config),
tracking: tracking.clone(),
mentions: vec![
AppInvocation {
connector_id: Some("drive".to_string()),
app_name: Some("Google Drive".to_string()),
invocation_type: Some(InvocationType::Explicit),
},
AppInvocation {
connector_id: Some("calendar".to_string()),
app_name: Some("Calendar".to_string()),
invocation_type: Some(InvocationType::Implicit),
},
],
});
let empty_skill_job = TrackEventsJob::SkillInvocations(TrackSkillInvocationsJob {
config: Arc::clone(&config),
tracking: tracking.clone(),
invocations: Vec::new(),
});
let no_invoke_type_job = TrackEventsJob::PluginUsed(TrackPluginUsedJob {
config,
tracking,
plugin: sample_plugin_metadata(),
});
assert_eq!(explicit_skill_job.invoke_type_tag(), Some("explicit"));
assert_eq!(implicit_app_job.invoke_type_tag(), Some("implicit"));
assert_eq!(
first_value_app_mentioned_job.invoke_type_tag(),
Some("explicit")
);
assert_eq!(empty_skill_job.invoke_type_tag(), None);
assert_eq!(no_invoke_type_job.invoke_type_tag(), None);
}
fn sample_plugin_metadata() -> PluginTelemetryMetadata {
PluginTelemetryMetadata {
plugin_id: PluginId::parse("sample@test").expect("valid plugin id"),
@@ -287,3 +441,17 @@ fn sample_plugin_metadata() -> PluginTelemetryMetadata {
}),
}
}
fn load_test_config(codex_home: &std::path::Path) -> crate::config::Config {
tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.expect("tokio runtime should build")
.block_on(
ConfigBuilder::default()
.codex_home(codex_home.to_path_buf())
.fallback_cwd(Some(codex_home.to_path_buf()))
.build(),
)
.expect("config should load")
}