Add client-side exec-server RPC attempt metrics (#42883)

## What changed

- Record `exec_server_client_requests_total` for every client RPC call attempt,
  labeled by protocol method.
- Count attempts before local admission so rejected, timed-out, cancelled, and
  transport-failed calls are included, while notifications and responses are not.
- Export the counter to configured OpenTelemetry collectors while excluding it
  from the built-in Statsig metrics set.

## Testing

- Cover every RPC call entry point, successful responses, local and transport
  failures, timeouts, cancellation, notifications, and disabled metrics.
- Verify that the OTLP HTTP exporter retains the counter.

GitOrigin-RevId: d266e840d5871aa2c34bed8ce44f3345e2b4650f

codex-rs/exec-server/src/rpc_client_metrics_tests.rs

@@ -0,0 +1,252 @@
+//! Exercise caller-side counters through the RPC transport, including failures and cancellation.
+
+use std::collections::BTreeMap;
+use std::time::Duration;
+
+use codex_otel::MetricsClient;
+use codex_otel::MetricsConfig;
+use opentelemetry_sdk::metrics::InMemoryMetricExporter;
+use opentelemetry_sdk::metrics::data::AggregatedMetrics;
+use opentelemetry_sdk::metrics::data::MetricData;
+use pretty_assertions::assert_eq;
+use serde_json::Value;
+use tokio::sync::mpsc;
+use tokio::sync::watch;
+
+use super::RpcCallError;
+use super::RpcClient;
+use super::RpcClientEvent;
+use crate::connection::JsonRpcConnection;
+use crate::connection::JsonRpcConnectionEvent;
+use crate::connection::JsonRpcTransport;
+use crate::protocol::FS_READ_FILE_METHOD;
+use crate::protocol::INITIALIZED_METHOD;
+use crate::protocol::JSONRPCMessage;
+use crate::protocol::JSONRPCResponse;
+use crate::protocol::RequestId;
+
+struct Harness {
+    client: RpcClient,
+    metrics: MetricsClient,
+    outgoing: mpsc::Receiver<JSONRPCMessage>,
+    incoming: mpsc::Sender<JsonRpcConnectionEvent>,
+    _events: mpsc::Receiver<RpcClientEvent>,
+    _disconnected: watch::Sender<bool>,
+}
+
+impl Harness {
+    fn new() -> Self {
+        let metrics = MetricsClient::new(
+            MetricsConfig::in_memory(
+                "test",
+                "exec-server-client-test",
+                env!("CARGO_PKG_VERSION"),
+                InMemoryMetricExporter::default(),
+            )
+            .with_runtime_reader(),
+        )
+        .expect("metrics client");
+        let (outgoing_tx, outgoing) = mpsc::channel(/*buffer*/ 8);
+        let (incoming, incoming_rx) = mpsc::channel(/*buffer*/ 8);
+        let (disconnected, disconnected_rx) = watch::channel(/*init*/ false);
+        let (mut client, events) = RpcClient::new(JsonRpcConnection {
+            outgoing_tx,
+            incoming_rx,
+            disconnected_rx,
+            task_handles: Vec::new(),
+            transport: JsonRpcTransport::Plain,
+        });
+        client.metrics = Some(metrics.clone());
+        Self {
+            client,
+            metrics,
+            outgoing,
+            incoming,
+            _events: events,
+            _disconnected: disconnected,
+        }
+    }
+
+    fn counts(&self) -> BTreeMap<String, u64> {
+        let snapshot = self.metrics.snapshot().expect("metrics snapshot");
+        let mut counts = BTreeMap::new();
+        for metric in snapshot
+            .scope_metrics()
+            .flat_map(opentelemetry_sdk::metrics::data::ScopeMetrics::metrics)
+            .filter(|metric| metric.name() == "exec_server_client_requests_total")
+        {
+            let AggregatedMetrics::U64(MetricData::Sum(sum)) = metric.data() else {
+                panic!("client request count should be a u64 sum");
+            };
+            for point in sum.data_points() {
+                let attributes = point
+                    .attributes()
+                    .map(|attribute| {
+                        (
+                            attribute.key.as_str(),
+                            attribute.value.as_str().into_owned(),
+                        )
+                    })
+                    .collect::<Vec<_>>();
+                assert_eq!(attributes.len(), 1, "only the method is labeled");
+                assert_eq!(attributes[0].0, "method");
+                *counts.entry(attributes[0].1.clone()).or_default() += point.value();
+            }
+        }
+        counts
+    }
+}
+
+#[derive(Clone, Copy)]
+enum CallKind {
+    Regular,
+    Untraced,
+    WithTimeout,
+    Cleanup,
+}
+
+#[tokio::test]
+async fn each_request_entry_point_counts_once_and_preserves_response() {
+    for kind in [
+        CallKind::Regular,
+        CallKind::Untraced,
+        CallKind::WithTimeout,
+        CallKind::Cleanup,
+    ] {
+        let mut harness = Harness::new();
+        let params = serde_json::json!({"path": "/sensitive-test-path"});
+        let request = async {
+            match kind {
+                CallKind::Regular => {
+                    harness
+                        .client
+                        .call::<_, Value>(FS_READ_FILE_METHOD, &params)
+                        .await
+                }
+                CallKind::Untraced => {
+                    harness
+                        .client
+                        .call_untraced::<_, Value>(FS_READ_FILE_METHOD, &params)
+                        .await
+                }
+                CallKind::WithTimeout => {
+                    harness
+                        .client
+                        .call_with_timeout::<_, Value>(
+                            FS_READ_FILE_METHOD,
+                            &params,
+                            Duration::from_secs(1),
+                        )
+                        .await
+                }
+                CallKind::Cleanup => {
+                    harness
+                        .client
+                        .call_for_cleanup::<_, Value>(FS_READ_FILE_METHOD, &params)
+                        .await
+                }
+            }
+        };
+        let server = async {
+            let Some(JSONRPCMessage::Request(request)) = harness.outgoing.recv().await else {
+                panic!("expected request");
+            };
+            assert_eq!(request.method, FS_READ_FILE_METHOD);
+            assert_eq!(request.params, Some(params.clone()));
+            harness
+                .incoming
+                .send(JsonRpcConnectionEvent::Message(JSONRPCMessage::Response(
+                    JSONRPCResponse {
+                        id: request.id,
+                        result: serde_json::json!({"ok": true}),
+                    },
+                )))
+                .await
+                .expect("response accepted");
+        };
+        let (response, ()) = tokio::join!(request, server);
+        assert_eq!(
+            response.expect("RPC response"),
+            serde_json::json!({"ok": true})
+        );
+        assert_eq!(
+            harness.counts(),
+            BTreeMap::from([(FS_READ_FILE_METHOD.to_string(), 1)])
+        );
+    }
+}
+
+#[tokio::test]
+async fn local_rejections_and_closed_transport_count_as_attempts() {
+    let harness = Harness::new();
+    let slots = harness
+        .client
+        .shared_call_slots
+        .acquire_many(super::MAX_IN_FLIGHT_REGULAR_CALLS as u32)
+        .await
+        .expect("occupy slots");
+    let rejected = harness
+        .client
+        .call::<_, Value>(FS_READ_FILE_METHOD, &())
+        .await;
+    assert!(matches!(
+        rejected,
+        Err(RpcCallError::PendingRequestLimitExceeded { .. })
+    ));
+    drop(slots);
+    harness.client.close_transport().await;
+    let closed = harness
+        .client
+        .call::<_, Value>(FS_READ_FILE_METHOD, &())
+        .await;
+    assert!(matches!(closed, Err(RpcCallError::Closed)));
+    assert_eq!(
+        harness.counts(),
+        BTreeMap::from([(FS_READ_FILE_METHOD.to_string(), 2)])
+    );
+}
+
+#[tokio::test(start_paused = true)]
+async fn timeout_and_cancellation_count_as_attempts() {
+    let mut harness = Harness::new();
+    let timed_out = harness
+        .client
+        .call_with_timeout::<_, Value>(FS_READ_FILE_METHOD, &(), Duration::from_secs(1))
+        .await;
+    assert!(matches!(timed_out, Err(RpcCallError::TimedOut { .. })));
+    harness.outgoing.recv().await.expect("timed out request");
+    let mut cancelled = Box::pin(harness.client.call::<_, Value>(FS_READ_FILE_METHOD, &()));
+    assert!(futures::poll!(cancelled.as_mut()).is_pending());
+    harness.outgoing.recv().await.expect("cancelled request");
+    drop(cancelled);
+    assert_eq!(
+        harness.counts(),
+        BTreeMap::from([(FS_READ_FILE_METHOD.to_string(), 2)])
+    );
+}
+
+#[tokio::test]
+async fn notifications_responses_and_disabled_metrics_do_not_record_attempts() {
+    let mut harness = Harness::new();
+    harness
+        .client
+        .notify(INITIALIZED_METHOD, &())
+        .await
+        .expect("notification");
+    harness
+        .client
+        .respond(RequestId::Integer(1), &())
+        .await
+        .expect("response");
+    assert_eq!(harness.counts(), BTreeMap::new());
+    harness.client.metrics = None;
+    harness.client.close_transport().await;
+    assert!(matches!(
+        harness
+            .client
+            .call::<_, Value>(FS_READ_FILE_METHOD, &())
+            .await,
+        Err(RpcCallError::Closed)
+    ));
+    assert_eq!(harness.counts(), BTreeMap::new());
+}