Add histograms for pool-ke session durations.

Author: davidv1992

nts-pool-ke/src/pool_ke.rs

@@ -5,7 +5,10 @@ use std::{
 };
 
 use notify::{RecursiveMode, Watcher};
-use opentelemetry::{KeyValue, metrics::Counter};
+use opentelemetry::{
+    KeyValue,
+    metrics::{Counter, Histogram},
+};
 use pool_nts::{
     AlgorithmDescription, BufferBorrowingReader, ClientRequest, ErrorCode, ErrorResponse,
     FixedKeyRequest, KeyExchangeResponse, MAX_MESSAGE_SIZE, NoAgreementResponse, NtsError,
@@ -27,6 +30,7 @@ use crate::{
     error::PoolError,
     haproxy::parse_haproxy_header,
     servers::{ConnectionType, Server, ServerConnection, ServerManager},
+    telemetry::TIMING_HISTOGRAM_BUCKET_BOUNDARIES,
     util::load_certificates,
 };
 
@@ -44,7 +48,9 @@ struct NtsPoolKe<S> {
     server_tls: RwLock<TlsAcceptor>,
     monitoring_keys: RwLock<Arc<HashSet<String>>>,
     session_counter: Counter<u64>,
+    session_duration: Histogram<f64>,
     upstream_session_counter: Counter<u64>,
+    upstream_session_duration: Histogram<f64>,
     server_manager: S,
 }
 
@@ -63,17 +69,33 @@ impl<S: ServerManager + 'static> NtsPoolKe<S> {
             .with_description("number of ke sessions with clients")
             .build();
 
+        let session_duration = meter
+            .f64_histogram("session_duration")
+            .with_description("Duration of the session")
+            .with_unit("s")
+            .with_boundaries(TIMING_HISTOGRAM_BUCKET_BOUNDARIES.to_vec())
+            .build();
+
         let upstream_session_counter = meter
             .u64_counter("upstream_cookie_sessions")
             .with_description("number of ke sessions with upstream for getting cookies")
             .build();
 
+        let upstream_session_duration = meter
+            .f64_histogram("upstream_get_cookie_duration")
+            .with_description("Duration to get cookies from upstream time source")
+            .with_unit("s")
+            .with_boundaries(TIMING_HISTOGRAM_BUCKET_BOUNDARIES.to_vec())
+            .build();
+
         Ok(NtsPoolKe {
             config,
             server_tls,
             monitoring_keys,
             session_counter,
+            session_duration,
             upstream_session_counter,
+            upstream_session_duration,
             server_manager,
         })
     }
@@ -105,6 +127,7 @@ impl<S: ServerManager + 'static> NtsPoolKe<S> {
                 _ = shutdown.recv() => { break; }
                 accept_result = listener.accept() => { accept_result? }
             };
+            let start_time = std::time::Instant::now();
             let self_clone = self.clone();
 
             tracker.spawn(async move {
@@ -116,6 +139,13 @@ impl<S: ServerManager + 'static> NtsPoolKe<S> {
                 .await
                 {
                     Err(_) => {
+                        self_clone.session_duration.record(
+                            start_time.elapsed().as_secs_f64(),
+                            &[
+                                KeyValue::new("outcome", "timeout"),
+                                KeyValue::new("is_monitor", is_monitor),
+                            ],
+                        );
                         ::tracing::debug!(?source_address, "NTS Pool KE timed out");
                         self_clone.session_counter.add(
                             1,
@@ -126,6 +156,13 @@ impl<S: ServerManager + 'static> NtsPoolKe<S> {
                         );
                     }
                     Ok(Err(err)) => {
+                        self_clone.session_duration.record(
+                            start_time.elapsed().as_secs_f64(),
+                            &[
+                                KeyValue::new("outcome", "error"),
+                                KeyValue::new("is_monitor", is_monitor),
+                            ],
+                        );
                         ::tracing::debug!(?err, ?source_address, "NTS Pool KE failed");
                         self_clone.session_counter.add(
                             1,
@@ -136,6 +173,13 @@ impl<S: ServerManager + 'static> NtsPoolKe<S> {
                         );
                     }
                     Ok(Ok(())) => {
+                        self_clone.session_duration.record(
+                            start_time.elapsed().as_secs_f64(),
+                            &[
+                                KeyValue::new("outcome", "success"),
+                                KeyValue::new("is_monitor", is_monitor),
+                            ],
+                        );
                         ::tracing::debug!(?source_address, "NTS Pool KE completed");
                         self_clone.session_counter.add(
                             1,
@@ -539,14 +583,29 @@ impl<S: ServerManager + 'static> NtsPoolKe<S> {
             }
         }
 
-        // TODO: Implement connection reuse
+        let start_time = std::time::Instant::now();
         match tokio::time::timeout(self.config.timesource_timeout, async {
             let server_stream = server.connect(connection_type).await?;
             workaround_lifetime_bug(buffer, request, server_stream).await
         })
         .await
         {
             Ok(v) => {
+                self.upstream_session_duration.record(
+                    start_time.elapsed().as_secs_f64(),
+                    &[
+                        KeyValue::new(
+                            "outcome",
+                            match v {
+                                Ok(_) => "success",
+                                Err(_) => "error",
+                            },
+                        ),
+                        KeyValue::new("server", server.name().clone()),
+                        KeyValue::new("uuid", server.uuid().clone()),
+                        KeyValue::new("is_monitor", for_monitor),
+                    ],
+                );
                 self.upstream_session_counter.add(
                     1,
                     &[
@@ -565,6 +624,15 @@ impl<S: ServerManager + 'static> NtsPoolKe<S> {
                 v
             }
             Err(_) => {
+                self.upstream_session_duration.record(
+                    start_time.elapsed().as_secs_f64(),
+                    &[
+                        KeyValue::new("outcome", "timeout"),
+                        KeyValue::new("server", server.name().clone()),
+                        KeyValue::new("uuid", server.uuid().clone()),
+                        KeyValue::new("is_monitor", for_monitor),
+                    ],
+                );
                 self.upstream_session_counter.add(
                     1,
                     &[

nts-pool-ke/src/telemetry.rs

@@ -1,6 +1,9 @@
 use opentelemetry_otlp::WithExportConfig;
 use opentelemetry_sdk::metrics::SdkMeterProvider;
 
+// For now, three buckets per decade, grouping everything below 1 ms and above 5s
+pub(crate) const TIMING_HISTOGRAM_BUCKET_BOUNDARIES: &[f64] = &[0.0, 1e-3, 2e-3, 5e-3, 1e-2, 2e-2, 5e-2, 1e-1, 2e-1, 5e-1, 1.0, 2.0, 5.0];
+
 fn build_otlp_exporter() -> Option<opentelemetry_otlp::MetricExporter> {
     let otlp_url = match std::env::var("OTEL_METRICS_EXPORT_DESTINATION") {
         Ok(otlp_url) => otlp_url,