[codex] add connector runtime latency metrics (#31319)

## Summary

- measure successful legacy `app/list` latency with
`codex.apps.installed.duration_ms`, segmented by `path=legacy` and
`reload`
- measure successful host-owned `codex_apps` startup and explicit
refresh latency with `codex.apps.refresh.duration_ms`
- add the refresh trigger to successful
`codex.mcp.tools.fetch_uncached.duration_ms` samples for `codex_apps`
without changing other MCP-server samples

## Why

This establishes a small latency baseline for the current connector path
before `ConnectorRuntimeManager`, `app/installed`, and `app/read` land.
Error-rate and broader runtime-state instrumentation are intentionally
deferred.

This is telemetry-only and does not change connector behavior.

## Validation

- `just test -p codex-mcp` (94 passed)
- `just test -p codex-app-server list_apps` (13 passed)
- `just fix -p codex-mcp`
- `just fix -p codex-app-server`
- `just fmt`
- `git diff --check`
This commit is contained in:
Matthew Zeng authored and GitHub committed 2026-07-07 10:49:06 -07:00
1 parent 78df1237d1
commit 1345c16dd7
3 files changed
+68 -21

No files matched your search

@@ -47,6 +47,8 @@ impl AppsRequestProcessor {
request_id: &ConnectionRequestId, request_id: &ConnectionRequestId,
params: AppsListParams, params: AppsListParams,
) -> Result<Option<AppsListResponse>, JSONRPCErrorError> { ) -> Result<Option<AppsListResponse>, JSONRPCErrorError> {
let installed_start = Instant::now();
let reload = params.force_refetch;
let thread = if let Some(thread_id) = params.thread_id.as_deref() { let thread = if let Some(thread_id) = params.thread_id.as_deref() {
let (_, loaded_thread) = self.load_thread(thread_id).await?; let (_, loaded_thread) = self.load_thread(thread_id).await?;
Some(loaded_thread) Some(loaded_thread)
@@ -70,20 +72,24 @@ impl AppsRequestProcessor {
.features .features
.apps_enabled_for_auth(auth.as_ref().is_some_and(CodexAuth::uses_codex_backend)) .apps_enabled_for_auth(auth.as_ref().is_some_and(CodexAuth::uses_codex_backend))
{ {
return Ok(Some(AppsListResponse { let response = AppsListResponse {
data: Vec::new(), data: Vec::new(),
next_cursor: None, next_cursor: None,
})); };
record_legacy_apps_installed_duration(installed_start, reload);
return Ok(Some(response));
} }
if !self if !self
.workspace_codex_plugins_enabled(&config, auth.as_ref()) .workspace_codex_plugins_enabled(&config, auth.as_ref())
.await .await
{ {
return Ok(Some(AppsListResponse { let response = AppsListResponse {
data: Vec::new(), data: Vec::new(),
next_cursor: None, next_cursor: None,
})); };
record_legacy_apps_installed_duration(installed_start, reload);
return Ok(Some(response));
} }
let request = request_id.clone(); let request = request_id.clone();
@@ -103,6 +109,7 @@ impl AppsRequestProcessor {
environment_manager, environment_manager,
mcp_manager, mcp_manager,
plugins_manager, plugins_manager,
installed_start,
) => {} ) => {}
} }
}); });
@@ -113,6 +120,7 @@ impl AppsRequestProcessor {
self.shutdown_token.cancel(); self.shutdown_token.cancel();
} }
#[allow(clippy::too_many_arguments)]
async fn apps_list_task( async fn apps_list_task(
outgoing: Arc<OutgoingMessageSender>, outgoing: Arc<OutgoingMessageSender>,
request_id: ConnectionRequestId, request_id: ConnectionRequestId,
@@ -121,7 +129,9 @@ impl AppsRequestProcessor {
environment_manager: Arc<EnvironmentManager>, environment_manager: Arc<EnvironmentManager>,
mcp_manager: Arc<McpManager>, mcp_manager: Arc<McpManager>,
plugins_manager: Arc<PluginsManager>, plugins_manager: Arc<PluginsManager>,
installed_start: Instant,
) { ) {
let reload = params.force_refetch;
let retry_params = params.clone(); let retry_params = params.clone();
let retry_config = config.clone(); let retry_config = config.clone();
let retry_environment_manager = Arc::clone(&environment_manager); let retry_environment_manager = Arc::clone(&environment_manager);
@@ -136,6 +146,9 @@ impl AppsRequestProcessor {
plugins_manager, plugins_manager,
) )
.await; .await;
if result.is_ok() {
record_legacy_apps_installed_duration(installed_start, reload);
}
let should_retry = result let should_retry = result
.as_ref() .as_ref()
.is_ok_and(|(_, codex_apps_ready)| !codex_apps_ready); .is_ok_and(|(_, codex_apps_ready)| !codex_apps_ready);
@@ -363,6 +376,20 @@ impl AppsRequestProcessor {
} }
const APP_LIST_LOAD_TIMEOUT: Duration = Duration::from_secs(90); const APP_LIST_LOAD_TIMEOUT: Duration = Duration::from_secs(90);
// `app/list` is the legacy request-path baseline for the future `app/installed` endpoint;
// `path=legacy` keeps it separate from the new snapshot-backed implementation in dashboards.
const APPS_INSTALLED_DURATION_METRIC: &str = "codex.apps.installed.duration_ms";
fn record_legacy_apps_installed_duration(started_at: Instant, reload: bool) {
let reload = if reload { "true" } else { "false" };
if let Some(metrics) = codex_otel::global() {
let _ = metrics.record_duration(
APPS_INSTALLED_DURATION_METRIC,
started_at.elapsed(),
&[("path", "legacy"), ("reload", reload)],
);
}
}
enum AppListLoadResult { enum AppListLoadResult {
Accessible(Result<AccessibleConnectorsStatus, String>), Accessible(Result<AccessibleConnectorsStatus, String>),
+10 -11
View File
@@ -23,8 +23,8 @@ use crate::elicitation::ElicitationReviewerHandle;
use crate::mcp::CODEX_APPS_MCP_SERVER_NAME; use crate::mcp::CODEX_APPS_MCP_SERVER_NAME;
use crate::mcp::ToolPluginProvenance; use crate::mcp::ToolPluginProvenance;
use crate::rmcp_client::AsyncManagedClient; use crate::rmcp_client::AsyncManagedClient;
use crate::rmcp_client::CODEX_APPS_REFRESH_DURATION_METRIC;
use crate::rmcp_client::DEFAULT_STARTUP_TIMEOUT; use crate::rmcp_client::DEFAULT_STARTUP_TIMEOUT;
use crate::rmcp_client::MCP_TOOLS_FETCH_UNCACHED_DURATION_METRIC;
use crate::rmcp_client::MCP_TOOLS_LIST_DURATION_METRIC; use crate::rmcp_client::MCP_TOOLS_LIST_DURATION_METRIC;
use crate::rmcp_client::ManagedClient; use crate::rmcp_client::ManagedClient;
use crate::rmcp_client::StartupOutcomeError; use crate::rmcp_client::StartupOutcomeError;
@@ -530,6 +530,7 @@ impl McpConnectionManager {
/// cache is enabled and the latest filtered tools are returned directly to /// cache is enabled and the latest filtered tools are returned directly to
/// the caller. On failure, existing shared cache contents remain unchanged. /// the caller. On failure, existing shared cache contents remain unchanged.
pub async fn hard_refresh_codex_apps_tools_cache(&self) -> Result<Vec<ToolInfo>> { pub async fn hard_refresh_codex_apps_tools_cache(&self) -> Result<Vec<ToolInfo>> {
let refresh_start = Instant::now();
let managed_client = self let managed_client = self
.clients .clients
.get(CODEX_APPS_MCP_SERVER_NAME) .get(CODEX_APPS_MCP_SERVER_NAME)
@@ -539,7 +540,6 @@ impl McpConnectionManager {
.context("failed to get client")?; .context("failed to get client")?;
let list_start = Instant::now(); let list_start = Instant::now();
let fetch_start = Instant::now();
let fetch_ticket = managed_client let fetch_ticket = managed_client
.codex_apps_tools_cache_context .codex_apps_tools_cache_context
.as_ref() .as_ref()
@@ -547,6 +547,7 @@ impl McpConnectionManager {
let tools = list_tools_for_client_uncached( let tools = list_tools_for_client_uncached(
CODEX_APPS_MCP_SERVER_NAME, CODEX_APPS_MCP_SERVER_NAME,
/*is_codex_apps_mcp_server*/ true, /*is_codex_apps_mcp_server*/ true,
/*codex_apps_refresh_trigger*/ "explicit",
&managed_client.client, &managed_client.client,
managed_client.tool_timeout, managed_client.tool_timeout,
managed_client.server_instructions.as_deref(), managed_client.server_instructions.as_deref(),
@@ -555,11 +556,6 @@ impl McpConnectionManager {
.with_context(|| { .with_context(|| {
format!("failed to refresh tools for MCP server '{CODEX_APPS_MCP_SERVER_NAME}'") format!("failed to refresh tools for MCP server '{CODEX_APPS_MCP_SERVER_NAME}'")
})?; })?;
emit_duration(
MCP_TOOLS_FETCH_UNCACHED_DURATION_METRIC,
fetch_start.elapsed(),
&[],
);
let tools = let tools =
match ( match (
@@ -582,10 +578,13 @@ impl McpConnectionManager {
tool.tool = tool_with_model_visible_input_schema(&tool.tool); tool.tool = tool_with_model_visible_input_schema(&tool.tool);
self.with_server_metadata(tool) self.with_server_metadata(tool)
}); });
Ok(normalize_tools_for_model_with_prefix( let tools = normalize_tools_for_model_with_prefix(tools, self.prefix_mcp_tool_names);
tools, emit_duration(
self.prefix_mcp_tool_names, CODEX_APPS_REFRESH_DURATION_METRIC,
)) refresh_start.elapsed(),
&[("path", "legacy"), ("trigger", "explicit")],
);
Ok(tools)
} }
/// Returns resources from servers selected by `include_server`. Each key /// Returns resources from servers selected by `include_server`. Each key
+27 -6
View File
@@ -82,6 +82,7 @@ pub const OPENAI_FORM_CAPABILITY: &str = "openai/form";
pub(crate) const MCP_TOOLS_LIST_DURATION_METRIC: &str = "codex.mcp.tools.list.duration_ms"; pub(crate) const MCP_TOOLS_LIST_DURATION_METRIC: &str = "codex.mcp.tools.list.duration_ms";
pub(crate) const MCP_TOOLS_FETCH_UNCACHED_DURATION_METRIC: &str = pub(crate) const MCP_TOOLS_FETCH_UNCACHED_DURATION_METRIC: &str =
"codex.mcp.tools.fetch_uncached.duration_ms"; "codex.mcp.tools.fetch_uncached.duration_ms";
pub(crate) const CODEX_APPS_REFRESH_DURATION_METRIC: &str = "codex.apps.refresh.duration_ms";
pub(crate) const DEFAULT_STARTUP_TIMEOUT: Duration = Duration::from_secs(30); pub(crate) const DEFAULT_STARTUP_TIMEOUT: Duration = Duration::from_secs(30);
pub(crate) const DEFAULT_TOOL_TIMEOUT: Duration = Duration::from_secs(300); pub(crate) const DEFAULT_TOOL_TIMEOUT: Duration = Duration::from_secs(300);
@@ -304,6 +305,7 @@ impl ManagedClientStartup {
.unwrap_or_default(); .unwrap_or_default();
let cancel_token_for_fut = cancel_token; let cancel_token_for_fut = cancel_token;
async move { async move {
let refresh_start = is_codex_apps_mcp_server.then(Instant::now);
let outcome = match async { let outcome = match async {
if let Err(error) = validate_mcp_server_name(&server_name) { if let Err(error) = validate_mcp_server_name(&server_name) {
return Err(error.into()); return Err(error.into());
@@ -349,6 +351,15 @@ impl ManagedClientStartup {
Ok(result) => result, Ok(result) => result,
Err(CancelErr::Cancelled) => Err(StartupOutcomeError::Cancelled), Err(CancelErr::Cancelled) => Err(StartupOutcomeError::Cancelled),
}; };
if outcome.is_ok()
&& let Some(refresh_start) = refresh_start
{
emit_duration(
CODEX_APPS_REFRESH_DURATION_METRIC,
refresh_start.elapsed(),
&[("path", "legacy"), ("trigger", "initial")],
);
}
startup_complete.store(true, Ordering::Release); startup_complete.store(true, Ordering::Release);
outcome outcome
@@ -564,10 +575,12 @@ impl From<anyhow::Error> for StartupOutcomeError {
pub(crate) async fn list_tools_for_client_uncached( pub(crate) async fn list_tools_for_client_uncached(
server_name: &str, server_name: &str,
is_codex_apps_mcp_server: bool, is_codex_apps_mcp_server: bool,
codex_apps_refresh_trigger: &'static str,
client: &Arc<RmcpClient>, client: &Arc<RmcpClient>,
timeout: Option<Duration>, timeout: Option<Duration>,
server_instructions: Option<&str>, server_instructions: Option<&str>,
) -> Result<Vec<ToolInfo>> { ) -> Result<Vec<ToolInfo>> {
let fetch_start = Instant::now();
let resp = client let resp = client
.list_tools_with_connector_ids(/*params*/ None, timeout) .list_tools_with_connector_ids(/*params*/ None, timeout)
.await?; .await?;
@@ -583,6 +596,19 @@ pub(crate) async fn list_tools_for_client_uncached(
) )
}) })
.collect(); .collect();
if is_codex_apps_mcp_server {
emit_duration(
MCP_TOOLS_FETCH_UNCACHED_DURATION_METRIC,
fetch_start.elapsed(),
&[("trigger", codex_apps_refresh_trigger)],
);
} else {
emit_duration(
MCP_TOOLS_FETCH_UNCACHED_DURATION_METRIC,
fetch_start.elapsed(),
&[],
);
}
Ok(tools) Ok(tools)
} }
@@ -820,24 +846,19 @@ async fn start_server_task(
.and_then(|exp| exp.get(MCP_SANDBOX_STATE_META_CAPABILITY)) .and_then(|exp| exp.get(MCP_SANDBOX_STATE_META_CAPABILITY))
.is_some(); .is_some();
let list_start = Instant::now(); let list_start = Instant::now();
let fetch_start = Instant::now();
let fetch_ticket = codex_apps_tools_cache_context let fetch_ticket = codex_apps_tools_cache_context
.as_ref() .as_ref()
.map(|cache_context| cache_context.begin_fetch(CodexAppsToolsFetchSource::Startup)); .map(|cache_context| cache_context.begin_fetch(CodexAppsToolsFetchSource::Startup));
let tools = list_tools_for_client_uncached( let tools = list_tools_for_client_uncached(
&server_name, &server_name,
is_codex_apps_mcp_server, is_codex_apps_mcp_server,
/*codex_apps_refresh_trigger*/ "initial",
&client, &client,
startup_timeout, startup_timeout,
initialize_result.instructions.as_deref(), initialize_result.instructions.as_deref(),
) )
.await .await
.map_err(StartupOutcomeError::from)?; .map_err(StartupOutcomeError::from)?;
emit_duration(
MCP_TOOLS_FETCH_UNCACHED_DURATION_METRIC,
fetch_start.elapsed(),
&[],
);
let server_info = mcp_server_info_from_implementation(initialize_result.server_info); let server_info = mcp_server_info_from_implementation(initialize_result.server_info);
let tools = match (codex_apps_tools_cache_context.as_ref(), fetch_ticket) { let tools = match (codex_apps_tools_cache_context.as_ref(), fetch_ticket) {
(Some(cache_context), Some(fetch_ticket)) => { (Some(cache_context), Some(fetch_ticket)) => {