fix(proxy): Use testodrome query id for latency measurement (#11150)

Add a new neon option "neon_query_id" to glue data with testodrome
queries. Log latency in microseconds always.

Relates to the #22486

proxy/src/compute.rs

@@ -287,9 +287,10 @@ impl ConnCfg {
         // TODO: lots of useful info but maybe we can move it elsewhere (eg traces?)
         info!(
             cold_start_info = ctx.cold_start_info().as_str(),
-            "connected to compute node at {host} ({socket_addr}) sslmode={:?}, latency={}",
+            "connected to compute node at {host} ({socket_addr}) sslmode={:?}, latency={}, query_id={}",
             self.0.get_ssl_mode(),
             ctx.get_proxy_latency(),
+            ctx.get_testodrome_id(),
         );
 
         // NB: CancelToken is supposed to hold socket_addr, but we use connect_raw.

proxy/src/context/mod.rs

@@ -63,6 +63,7 @@ struct RequestContextInner {
     success: bool,
     pub(crate) cold_start_info: ColdStartInfo,
     pg_options: Option<StartupMessageParams>,
+    testodrome_query_id: Option<String>,
 
     // extra
     // This sender is here to keep the request monitoring channel open while requests are taking place.
@@ -110,6 +111,7 @@ impl Clone for RequestContext {
             rejected: inner.rejected,
             cold_start_info: inner.cold_start_info,
             pg_options: inner.pg_options.clone(),
+            testodrome_query_id: inner.testodrome_query_id.clone(),
 
             sender: None,
             disconnect_sender: None,
@@ -160,6 +162,7 @@ impl RequestContext {
             rejected: None,
             cold_start_info: ColdStartInfo::Unknown,
             pg_options: None,
+            testodrome_query_id: None,
 
             sender: LOG_CHAN.get().and_then(|tx| tx.upgrade()),
             disconnect_sender: LOG_CHAN_DISCONNECT.get().and_then(|tx| tx.upgrade()),
@@ -210,6 +213,19 @@ impl RequestContext {
             this.set_dbname(dbname.into());
         }
 
+        // Try to get testodrome_query_id directly from parameters
+        if let Some(options_str) = options.get("options") {
+            // If not found directly, try to extract it from the options string
+            for option in options_str.split_whitespace() {
+                if option.starts_with("neon_query_id:") {
+                    if let Some(value) = option.strip_prefix("neon_query_id:") {
+                        this.set_testodrome_id(value.to_string());
+                        break;
+                    }
+                }
+            }
+        }
+
         this.pg_options = Some(options);
     }
 
@@ -355,6 +371,15 @@ impl RequestContext {
             .accumulated()
     }
 
+    pub(crate) fn get_testodrome_id(&self) -> String {
+        self.0
+            .try_lock()
+            .expect("should not deadlock")
+            .testodrome_query_id
+            .clone()
+            .unwrap_or_default()
+    }
+
     pub(crate) fn success(&self) {
         self.0
             .try_lock()
@@ -416,6 +441,10 @@ impl RequestContextInner {
         self.user = Some(user);
     }
 
+    fn set_testodrome_id(&mut self, query_id: String) {
+        self.testodrome_query_id = Some(query_id);
+    }
+
     fn has_private_peer_addr(&self) -> bool {
         match self.conn_info.addr.ip() {
             IpAddr::V4(ip) => ip.is_private(),

proxy/src/metrics.rs

@@ -406,8 +406,11 @@ impl std::fmt::Display for LatencyAccumulated {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         write!(
             f,
-            "client: {:?}, cplane: {:?}, compute: {:?}, retry: {:?}",
-            self.client, self.cplane, self.compute, self.retry
+            "client: {}, cplane: {}, compute: {}, retry: {}",
+            self.client.as_micros(),
+            self.cplane.as_micros(),
+            self.compute.as_micros(),
+            self.retry.as_micros()
         )
     }
 }