rust(feat): Add grpc status codes to sift-stream metrics, better logs (#530)

tsift · web-flow · commit 5acdd3a61057 · 2026-04-09T15:57:38.000-05:00
diff --git a/rust/crates/sift_rs/src/lib.rs b/rust/crates/sift_rs/src/lib.rs
@@ -55,3 +55,4 @@ pub mod retry;
 pub use retry::{DefaultGrpcRetry, RetryConfig, RetryDecider, RetryExt, Retrying};
 pub use sift_connect::{Credentials, SiftChannel, SiftChannelBuilder};
 pub use tonic::codec::CompressionEncoding;
+pub use tonic::{Code as GrpcCode, Status as GrpcStatus};
diff --git a/rust/crates/sift_stream/src/metrics/mod.rs b/rust/crates/sift_stream/src/metrics/mod.rs
@@ -122,6 +122,8 @@ pub struct SiftStreamMetricsSnapshot {
     pub old_messages_failed_adding_to_backup: u64,
     /// Current retry attempt count
     pub cur_retry_count: u64,
+    /// Count of stream completions per gRPC status code, indexed by code value (0=Ok .. 16=Unauthenticated, 17=UnknownGrpcCode)
+    pub grpc_status_counts: [u64; 18],
     /// Depth of the ingestion channel
     pub ingestion_channel_depth: u64,
     /// Depth of the backup channel
@@ -386,6 +388,7 @@ pub struct SiftStreamMetrics {
     pub(crate) old_messages_dropped_for_ingestion: U64Counter,
     pub(crate) old_messages_failed_adding_to_backup: U64Counter,
     pub(crate) cur_retry_count: U64Signal,
+    pub(crate) grpc_status_counts: [U64Counter; 18],
     pub(crate) ingestion_channel_depth: U64Signal,
     pub(crate) backup_channel_depth: U64Signal,
     pub(crate) checkpoint: CheckpointMetrics,
@@ -430,6 +433,7 @@ impl SiftStreamMetrics {
             old_messages_dropped_for_ingestion,
             old_messages_failed_adding_to_backup,
             cur_retry_count,
+            grpc_status_counts: std::array::from_fn(|i| self.grpc_status_counts[i].get()),
             ingestion_channel_depth,
             backup_channel_depth,
             checkpoint: self.checkpoint.snapshot(),
@@ -514,6 +518,114 @@ impl SiftStreamMetricsSnapshot {
                 data_type: ChannelDataType::Uint64.into(),
                 ..Default::default()
             },
+            ChannelConfig {
+                name: format!("{channel_prefix}.grpc_status_counts.ok"),
+                description: "gRPC status code Ok (0) count".into(),
+                data_type: ChannelDataType::Uint64.into(),
+                ..Default::default()
+            },
+            ChannelConfig {
+                name: format!("{channel_prefix}.grpc_status_counts.cancelled"),
+                description: "gRPC status code Cancelled (1) count".into(),
+                data_type: ChannelDataType::Uint64.into(),
+                ..Default::default()
+            },
+            ChannelConfig {
+                name: format!("{channel_prefix}.grpc_status_counts.unknown"),
+                description: "gRPC status code Unknown (2) count".into(),
+                data_type: ChannelDataType::Uint64.into(),
+                ..Default::default()
+            },
+            ChannelConfig {
+                name: format!("{channel_prefix}.grpc_status_counts.invalid_argument"),
+                description: "gRPC status code InvalidArgument (3) count".into(),
+                data_type: ChannelDataType::Uint64.into(),
+                ..Default::default()
+            },
+            ChannelConfig {
+                name: format!("{channel_prefix}.grpc_status_counts.deadline_exceeded"),
+                description: "gRPC status code DeadlineExceeded (4) count".into(),
+                data_type: ChannelDataType::Uint64.into(),
+                ..Default::default()
+            },
+            ChannelConfig {
+                name: format!("{channel_prefix}.grpc_status_counts.not_found"),
+                description: "gRPC status code NotFound (5) count".into(),
+                data_type: ChannelDataType::Uint64.into(),
+                ..Default::default()
+            },
+            ChannelConfig {
+                name: format!("{channel_prefix}.grpc_status_counts.already_exists"),
+                description: "gRPC status code AlreadyExists (6) count".into(),
+                data_type: ChannelDataType::Uint64.into(),
+                ..Default::default()
+            },
+            ChannelConfig {
+                name: format!("{channel_prefix}.grpc_status_counts.permission_denied"),
+                description: "gRPC status code PermissionDenied (7) count".into(),
+                data_type: ChannelDataType::Uint64.into(),
+                ..Default::default()
+            },
+            ChannelConfig {
+                name: format!("{channel_prefix}.grpc_status_counts.resource_exhausted"),
+                description: "gRPC status code ResourceExhausted (8) count".into(),
+                data_type: ChannelDataType::Uint64.into(),
+                ..Default::default()
+            },
+            ChannelConfig {
+                name: format!("{channel_prefix}.grpc_status_counts.failed_precondition"),
+                description: "gRPC status code FailedPrecondition (9) count".into(),
+                data_type: ChannelDataType::Uint64.into(),
+                ..Default::default()
+            },
+            ChannelConfig {
+                name: format!("{channel_prefix}.grpc_status_counts.aborted"),
+                description: "gRPC status code Aborted (10) count".into(),
+                data_type: ChannelDataType::Uint64.into(),
+                ..Default::default()
+            },
+            ChannelConfig {
+                name: format!("{channel_prefix}.grpc_status_counts.out_of_range"),
+                description: "gRPC status code OutOfRange (11) count".into(),
+                data_type: ChannelDataType::Uint64.into(),
+                ..Default::default()
+            },
+            ChannelConfig {
+                name: format!("{channel_prefix}.grpc_status_counts.unimplemented"),
+                description: "gRPC status code Unimplemented (12) count".into(),
+                data_type: ChannelDataType::Uint64.into(),
+                ..Default::default()
+            },
+            ChannelConfig {
+                name: format!("{channel_prefix}.grpc_status_counts.internal"),
+                description: "gRPC status code Internal (13) count".into(),
+                data_type: ChannelDataType::Uint64.into(),
+                ..Default::default()
+            },
+            ChannelConfig {
+                name: format!("{channel_prefix}.grpc_status_counts.unavailable"),
+                description: "gRPC status code Unavailable (14) count".into(),
+                data_type: ChannelDataType::Uint64.into(),
+                ..Default::default()
+            },
+            ChannelConfig {
+                name: format!("{channel_prefix}.grpc_status_counts.data_loss"),
+                description: "gRPC status code DataLoss (15) count".into(),
+                data_type: ChannelDataType::Uint64.into(),
+                ..Default::default()
+            },
+            ChannelConfig {
+                name: format!("{channel_prefix}.grpc_status_counts.unauthenticated"),
+                description: "gRPC status code Unauthenticated (16) count".into(),
+                data_type: ChannelDataType::Uint64.into(),
+                ..Default::default()
+            },
+            ChannelConfig {
+                name: format!("{channel_prefix}.grpc_status_counts.unknown_grpc_code"),
+                description: "Unknown gRPC status code (>16) count".into(),
+                data_type: ChannelDataType::Uint64.into(),
+                ..Default::default()
+            },
             ChannelConfig {
                 name: format!("{channel_prefix}.ingestion_channel_depth"),
                 description: "Ingestion channel depth".into(),
@@ -678,6 +790,78 @@ impl SiftStreamMetricsSnapshot {
                 &format!("{channel_prefix}.cur_retry_count"),
                 self.cur_retry_count,
             ),
+            ChannelValue::new(
+                &format!("{channel_prefix}.grpc_status_counts.ok"),
+                self.grpc_status_counts[0],
+            ),
+            ChannelValue::new(
+                &format!("{channel_prefix}.grpc_status_counts.cancelled"),
+                self.grpc_status_counts[1],
+            ),
+            ChannelValue::new(
+                &format!("{channel_prefix}.grpc_status_counts.unknown"),
+                self.grpc_status_counts[2],
+            ),
+            ChannelValue::new(
+                &format!("{channel_prefix}.grpc_status_counts.invalid_argument"),
+                self.grpc_status_counts[3],
+            ),
+            ChannelValue::new(
+                &format!("{channel_prefix}.grpc_status_counts.deadline_exceeded"),
+                self.grpc_status_counts[4],
+            ),
+            ChannelValue::new(
+                &format!("{channel_prefix}.grpc_status_counts.not_found"),
+                self.grpc_status_counts[5],
+            ),
+            ChannelValue::new(
+                &format!("{channel_prefix}.grpc_status_counts.already_exists"),
+                self.grpc_status_counts[6],
+            ),
+            ChannelValue::new(
+                &format!("{channel_prefix}.grpc_status_counts.permission_denied"),
+                self.grpc_status_counts[7],
+            ),
+            ChannelValue::new(
+                &format!("{channel_prefix}.grpc_status_counts.resource_exhausted"),
+                self.grpc_status_counts[8],
+            ),
+            ChannelValue::new(
+                &format!("{channel_prefix}.grpc_status_counts.failed_precondition"),
+                self.grpc_status_counts[9],
+            ),
+            ChannelValue::new(
+                &format!("{channel_prefix}.grpc_status_counts.aborted"),
+                self.grpc_status_counts[10],
+            ),
+            ChannelValue::new(
+                &format!("{channel_prefix}.grpc_status_counts.out_of_range"),
+                self.grpc_status_counts[11],
+            ),
+            ChannelValue::new(
+                &format!("{channel_prefix}.grpc_status_counts.unimplemented"),
+                self.grpc_status_counts[12],
+            ),
+            ChannelValue::new(
+                &format!("{channel_prefix}.grpc_status_counts.internal"),
+                self.grpc_status_counts[13],
+            ),
+            ChannelValue::new(
+                &format!("{channel_prefix}.grpc_status_counts.unavailable"),
+                self.grpc_status_counts[14],
+            ),
+            ChannelValue::new(
+                &format!("{channel_prefix}.grpc_status_counts.data_loss"),
+                self.grpc_status_counts[15],
+            ),
+            ChannelValue::new(
+                &format!("{channel_prefix}.grpc_status_counts.unauthenticated"),
+                self.grpc_status_counts[16],
+            ),
+            ChannelValue::new(
+                &format!("{channel_prefix}.grpc_status_counts.unknown_grpc_code"),
+                self.grpc_status_counts[17],
+            ),
             ChannelValue::new(
                 &format!("{channel_prefix}.ingestion_channel_depth"),
                 self.ingestion_channel_depth,
@@ -779,6 +963,7 @@ impl Default for SiftStreamMetrics {
             old_messages_dropped_for_ingestion: U64Counter::default(),
             old_messages_failed_adding_to_backup: U64Counter::default(),
             cur_retry_count: U64Signal::default(),
+            grpc_status_counts: Default::default(),
             ingestion_channel_depth: U64Signal::default(),
             backup_channel_depth: U64Signal::default(),
             checkpoint: CheckpointMetrics::default(),
diff --git a/rust/crates/sift_stream/src/stream/tasks/ingestion.rs b/rust/crates/sift_stream/src/stream/tasks/ingestion.rs
@@ -8,7 +8,10 @@ use crate::{
 };
 use sift_connect::SiftChannel;
 use sift_error::prelude::*;
-use sift_rs::{CompressionEncoding, ingest::v1::ingest_service_client::IngestServiceClient};
+use sift_rs::{
+    CompressionEncoding, GrpcCode, GrpcStatus,
+    ingest::v1::ingest_service_client::IngestServiceClient,
+};
 use std::{
     future::Future,
     pin::Pin,
@@ -138,7 +141,6 @@ impl IngestionTask {
 
                     // Currently the stream result is not used, so to simplify we return a unit value.
                     res.map(|_| ())
-                        .map_err(|e| Error::new(ErrorKind::StreamError, e))
                 }));
 
                 #[cfg(feature = "tracing")]
@@ -153,6 +155,7 @@ impl IngestionTask {
                 res = stream.as_mut().unwrap() => {
                     match res {
                         Ok(_) => {
+                            self.config.metrics.grpc_status_counts[0].increment();
                             self.config.metrics.cur_retry_count.set(0);
                             current_wait = Duration::ZERO;
                         }
@@ -192,19 +195,15 @@ impl IngestionTask {
                                 sift_stream_id = %self.config.sift_stream_id,
                                 "checkpoint succeeded - data streamed to Sift successfully"
                             );
+                            self.config.metrics.grpc_status_counts[0].increment();
                             self.config.metrics.cur_retry_count.set(0);
                         }
                         Ok(Err(e)) => {
                             current_wait = self.handle_failed_stream(&e, stream_created_at, current_wait, first_message_id.load(Ordering::Relaxed), last_message_id.load(Ordering::Relaxed))?;
                         }
-                        Err(elapsed) => {
-                            #[cfg(feature = "tracing")]
-                            tracing::error!(
-                                sift_stream_id = %self.config.sift_stream_id,
-                                error = %elapsed,
-                                "timed out waiting for checkpoint completion from Sift"
-                            );
-                            current_wait = self.handle_failed_stream(&Error::new(ErrorKind::StreamError, elapsed), stream_created_at, current_wait, first_message_id.load(Ordering::Relaxed), last_message_id.load(Ordering::Relaxed))?;
+                        Err(_elapsed) => {
+                            let timeout_status = GrpcStatus::deadline_exceeded("checkpoint timed out waiting for Sift");
+                            current_wait = self.handle_failed_stream(&timeout_status, stream_created_at, current_wait, first_message_id.load(Ordering::Relaxed), last_message_id.load(Ordering::Relaxed))?;
                         }
                     }
 
@@ -216,7 +215,7 @@ impl IngestionTask {
                     match ctrl_msg {
                         Ok(ControlMessage::BackupFull) => {
                             #[cfg(feature = "tracing")]
-                            tracing::info!(
+                            tracing::trace!(
                                 sift_stream_id = %self.config.sift_stream_id,
                                 "backup full"
                             );
@@ -243,19 +242,33 @@ impl IngestionTask {
     /// Handle a failed stream operation, sending the re-ingest signal and logging the error and incrementing metrics.
     fn handle_failed_stream(
         &mut self,
-        e: &Error,
+        status: &GrpcStatus,
         stream_created_at: Instant,
         current_wait: Duration,
         first_message_id: u64,
         last_message_id: u64,
     ) -> Result<Duration> {
+        let code = i32::from(status.code());
+        let code_idx = if (0..=16).contains(&code) {
+            code as usize
+        } else {
+            17
+        };
+        self.config.metrics.grpc_status_counts[code_idx].increment();
+
         #[cfg(feature = "tracing")]
-        tracing::error!(
-            sift_stream_id = %self.config.sift_stream_id,
-            retry_counter = self.config.metrics.cur_retry_count.get(),
-            error = %e,
-            "stream failed - failed to ingest data to Sift - if backups are enabled, backup files will be re-ingested"
-        );
+        {
+            let msg = match status.code() {
+                GrpcCode::Cancelled => "stream connection went idle",
+                _ => "stream connection is being reset",
+            };
+            tracing::warn!(
+                sift_stream_id = %self.config.sift_stream_id,
+                retry_counter = self.config.metrics.cur_retry_count.get(),
+                grpc_status = ?status.code(),
+                msg
+            );
+        }
 
         self.config
             .metrics
@@ -285,7 +298,7 @@ impl IngestionTask {
     }
 
     /// Shuts down the ingestion task by awaiting the stream one last time and sending the final checkpoint complete signal to the backup manager.
-    async fn shutdown<T: Future<Output = Result<()>> + Send + 'static>(
+    async fn shutdown<T: Future<Output = std::result::Result<(), GrpcStatus>> + Send + 'static>(
         &mut self,
         mut stream: Option<Pin<Box<T>>>,
         first_message_id: Arc<AtomicU64>,
