Add NATS event callback metrics

lib/naxum/src/middleware/ack/on_failure.rs

@@ -25,7 +25,7 @@ impl OnFailure for DefaultOnFailure {
             trace!("nacking message");
             if let Err(err) = acker.ack_with(jetstream::AckKind::Nak(None)).await {
                 warn!(
-                    error = ?err,
+                    si.error.message = ?err,
                     subject = head.subject.as_str(),
                     "failed to nack the message",
                 );

lib/naxum/src/middleware/ack/on_success.rs

@@ -25,7 +25,7 @@ impl OnSuccess for DefaultOnSuccess {
             trace!("double acking message");
             if let Err(err) = acker.double_ack().await {
                 warn!(
-                    error = ?err,
+                    si.error.message = ?err,
                     subject = head.subject.as_str(),
                     "failed to double ack the message",
                 );

lib/naxum/src/serve.rs

@@ -271,7 +271,7 @@ where
         Err(err) => {
             if failed_count > MAX_FAILED_MESSAGES {
                 warn!(
-                    error = ?err,
+                    si.error.message = ?err,
                     "failed to read message in after {} consecutive failures; closing stream",
                     failed_count,
                 );

lib/si-data-nats/BUCK

@@ -4,6 +4,7 @@ rust_library(
     name = "si-data-nats",
     deps = [
         "//lib/telemetry-rs:telemetry",
+        "//lib/telemetry-utils-rs:telemetry-utils",
         "//third-party/rust:async-nats",
         "//third-party/rust:bytes",
         "//third-party/rust:futures",

lib/si-data-nats/Cargo.toml

@@ -10,6 +10,7 @@ publish.workspace = true
 
 [dependencies]
 telemetry = { path = "../../lib/telemetry-rs" }
+telemetry-utils = { path = "../../lib/telemetry-utils-rs" }
 
 async-nats = { workspace = true }
 bytes = { workspace = true }

lib/si-data-nats/src/lib.rs

@@ -13,6 +13,7 @@ use async_nats::{subject::ToSubject, ToServerAddrs};
 use bytes::Bytes;
 use serde::{Deserialize, Serialize};
 use telemetry::prelude::*;
+use telemetry_utils::metric;
 use thiserror::Error;
 use tokio::sync::Mutex;
 
@@ -121,6 +122,29 @@ impl Client {
         if let Some(connection_name) = &config.connection_name {
             options = options.name(connection_name);
         }
+
+        options = options.event_callback(|event| async move {
+            match event {
+                Event::Connected => metric!(counter.nats.event_callback.connected = 1),
+                Event::Disconnected => metric!(counter.nats.event_callback.disconnected = 1),
+                Event::Closed => metric!(counter.nats.event_callback.closed = 1),
+                Event::LameDuckMode => metric!(counter.nats.event_callback.lame_duck_mode = 1),
+                Event::Draining => metric!(counter.nats.event_callback.draining = 1),
+                Event::SlowConsumer(sid) => {
+                    metric!(counter.nats.event_callback.slow_consumer = 1);
+                    warn!(%sid, "nats event callback: slow consumers for subscription");
+                }
+                Event::ServerError(server_error) => {
+                    metric!(counter.nats.event_callback.server_error = 1);
+                    error!(si.error.message = ?server_error, "nats event callback: server error")
+                }
+                Event::ClientError(client_error) => {
+                    metric!(counter.nats.event_callback.client_error = 1);
+                    error!(si.error.message = ?client_error, "nats event callback: client error")
+                }
+            }
+        });
+
         Self::connect_with_options(&config.url, config.subject_prefix.clone(), options).await
     }
 

lib/veritech-server/src/handlers.rs

@@ -309,7 +309,7 @@ where
             };
             request.dec_run_metric();
             if let Err(err) = publisher.publish_result(&func_result_error).await {
-                error!(error = ?err, "failed to publish errored result");
+                error!(si.error.message = ?err, "failed to publish errored result");
             }
         }
     }

lib/veritech-server/src/server.rs

@@ -165,7 +165,7 @@ impl Server {
     #[inline]
     pub async fn run(self) {
         if let Err(err) = self.try_run().await {
-            error!(error = ?err, "error while running veritech main loop");
+            error!(si.error.message = ?err, "error while running veritech main loop");
         }
     }