Add login path observability: login_duration histogram, in-flight gauge, abandoned counter

pgdog/src/frontend/client/mod.rs

@@ -30,6 +30,7 @@ use crate::net::messages::{
 };
 use crate::net::{MessageBuffer, ProtocolMessage, Stream, parameter::Parameters};
 use crate::state::State;
+use crate::stats::logins::LoginTimer;
 use crate::stats::memory::MemoryUsage;
 use crate::util::{safe_timeout, user_database_from_params};
 
@@ -138,6 +139,7 @@ impl Client {
         addr: SocketAddr,
         config: Arc<ConfigAndUsers>,
         protocol_version: ProtocolVersion,
+        mut login_timer: LoginTimer,
     ) -> Result<(), Error> {
         let login_timeout = Duration::from_millis(config.config.general.client_login_timeout);
 
@@ -148,6 +150,7 @@ impl Client {
         .await
         {
             Ok(Ok(Some(mut client))) => {
+                login_timer.success();
                 if client.admin {
                     // Admin clients are not waited on during shutdown.
                     spawn(async move {
@@ -160,10 +163,16 @@ impl Client {
                 Ok(())
             }
             Err(_) => {
+                // Login timeouts are logged; the drop still counts the
+                // connection as an abandoned login.
                 error!("client login timeout [{}]", addr);
                 Ok(())
             }
-            Ok(Ok(None)) => Ok(()),
+            Ok(Ok(None)) => {
+                // Login was rejected with an explicit response.
+                login_timer.disarm();
+                Ok(())
+            }
             Ok(Err(err)) => Err(err),
         }
     }

pgdog/src/frontend/client/test/mod.rs

@@ -487,6 +487,7 @@ async fn test_client_login_timeout() {
             addr,
             crate::config::config(),
             ProtocolVersion::V3_0,
+            crate::stats::logins::LoginTimer::new(),
         )
         .await
     });

pgdog/src/frontend/client/test/test_client.rs

@@ -341,9 +341,16 @@ impl SpawnedClient {
         let handle = tokio::spawn(async move {
             let (stream, addr) = listener.accept().await.unwrap();
             let stream = Stream::plain(stream, 4096);
-            Client::spawn(stream, params, addr, config(), ProtocolVersion::V3_0)
-                .await
-                .unwrap();
+            Client::spawn(
+                stream,
+                params,
+                addr,
+                config(),
+                ProtocolVersion::V3_0,
+                crate::stats::logins::LoginTimer::new(),
+            )
+            .await
+            .unwrap();
         });
 
         let conn = TcpStream::connect(format!("127.0.0.1:{}", port))

pgdog/src/frontend/listener.rs

@@ -10,6 +10,7 @@ use crate::net::messages::{FrontendPid, NegotiateProtocolVersion, Startup, hello
 use crate::net::tls::{acceptor, peer_certificate_present, peer_identity};
 use crate::net::{self, Stream, tweak};
 use crate::sighup::Sighup;
+use crate::stats::logins::LoginTimer;
 use tokio::net::{TcpListener, TcpSocket, TcpStream, lookup_host};
 use tokio::signal::ctrl_c;
 use tokio::{select, spawn};
@@ -204,6 +205,7 @@ impl Listener {
         let mut stream = Stream::plain(stream, config.config.memory.net_buffer);
 
         let tls = acceptor();
+        let mut login_timer = LoginTimer::new();
 
         loop {
             let startup = match Startup::from_stream(&mut stream).await {
@@ -222,6 +224,7 @@ impl Listener {
 
             match startup {
                 Startup::Ssl => {
+                    login_timer.engage();
                     if let Some(tls) = tls.as_ref() {
                         stream.send_flush(&SslReply::Yes).await?;
                         let plain = stream.take()?;
@@ -248,6 +251,7 @@ impl Listener {
                 }
 
                 Startup::GssEnc => {
+                    login_timer.engage();
                     // GSS encryption is not yet supported; reject and wait for a normal startup.
                     stream.send_flush(&SslReply::No).await?;
                 }
@@ -257,6 +261,7 @@ impl Listener {
                     params,
                     unrecognized_options,
                 } => {
+                    login_timer.engage();
                     let negotiated = version
                         .negotiated()
                         .ok_or_else(|| net::Error::UnsupportedStartup(version.as_i32()))?;
@@ -273,11 +278,20 @@ impl Listener {
                             .await?;
                     }
 
-                    Box::pin(Client::spawn(stream, params, addr, config, negotiated)).await?;
+                    Box::pin(Client::spawn(
+                        stream,
+                        params,
+                        addr,
+                        config,
+                        negotiated,
+                        login_timer,
+                    ))
+                    .await?;
                     break;
                 }
 
                 Startup::Cancel { ref id } => {
+                    login_timer.disarm();
                     if comms().verify_cancel(id) {
                         let _ = databases().cancel(FrontendPid::from(id)).await;
                     }

pgdog/src/stats/http_server.rs

@@ -11,7 +11,8 @@ use tokio::select;
 use tracing::{info, warn};
 
 use super::{
-    Clients, ClientsLocked, Listeners, LookupMetrics, MirrorStatsMetrics, Pools, QueryCache, TwoPc,
+    Clients, ClientsLocked, Listeners, Logins, LookupMetrics, MirrorStatsMetrics, Pools,
+    QueryCache, TwoPc,
 };
 use crate::tasks;
 
@@ -41,6 +42,12 @@ async fn metrics(_: Request<hyper::body::Incoming>) -> Result<Response<Full<Byte
         .collect();
     let query_cache = query_cache.join("\n");
     let two_pc = TwoPc::load();
+    let logins: Vec<_> = Logins::load()
+        .into_iter()
+        .chain(std::iter::once(Logins::histogram()))
+        .map(|m| m.to_string())
+        .collect();
+    let logins = logins.join("\n");
     let metrics_data = clients.to_string()
         + "\n"
         + &clients_locked.to_string()
@@ -55,6 +62,8 @@ async fn metrics(_: Request<hyper::body::Incoming>) -> Result<Response<Full<Byte
         + "\n"
         + &query_cache
         + "\n"
+        + &logins
+        + "\n"
         + &two_pc.to_string();
     let response = Response::builder()
         .header(