fix: Dont fail when attempting to remove non-existant up metric

near · Jun 27, 2024 · a185c02 · a185c02
1 parent 0728655
commit a185c02
Show file tree

Hide file tree

Showing 2 changed files with 14 additions and 7 deletions.
diff --git a/block-streamer/src/block_stream.rs b/block-streamer/src/block_stream.rs
@@ -148,8 +148,9 @@ impl BlockStream {
             task.cancellation_token.cancel();
             let _ = task.handle.await?;
 
-            metrics::BLOCK_STREAM_UP
-                .remove_label_values(&[&self.indexer_config.get_full_name()])?;
+            // Fails if metric doesn't exist, i.e. task was never polled
+            let _ = metrics::BLOCK_STREAM_UP
+                .remove_label_values(&[&self.indexer_config.get_full_name()]);
 
             return Ok(());
         }

diff --git a/block-streamer/src/server/block_streamer_service.rs b/block-streamer/src/server/block_streamer_service.rs
@@ -58,6 +58,7 @@ impl BlockStreamerService {
 
 #[tonic::async_trait]
 impl blockstreamer::block_streamer_server::BlockStreamer for BlockStreamerService {
+    #[tracing::instrument(skip(self))]
     async fn start_stream(
         &self,
         request: Request<blockstreamer::StartStreamRequest>,
@@ -117,7 +118,11 @@ impl blockstreamer::block_streamer_server::BlockStreamer for BlockStreamerServic
                 self.delta_lake_client.clone(),
                 self.lake_s3_client.clone(),
             )
-            .map_err(|_| Status::internal("Failed to start block stream"))?;
+            .map_err(|err| {
+                tracing::error!(?err, "Failed to start block stream");
+
+                Status::internal("Failed to start block stream")
+            })?;
 
         let mut lock = self.get_block_streams_lock()?;
         lock.insert(indexer_config.get_hash_id(), block_stream);
@@ -127,6 +132,7 @@ impl blockstreamer::block_streamer_server::BlockStreamer for BlockStreamerServic
         }))
     }
 
+    #[tracing::instrument(skip(self))]
     async fn stop_stream(
         &self,
         request: Request<blockstreamer::StopStreamRequest>,
@@ -148,10 +154,10 @@ impl blockstreamer::block_streamer_server::BlockStreamer for BlockStreamerServic
                 )))
             }
             Some(mut block_stream) => {
-                block_stream
-                    .cancel()
-                    .await
-                    .map_err(|_| Status::internal("Failed to cancel block stream"))?;
+                block_stream.cancel().await.map_err(|err| {
+                    tracing::error!(?err, "Failed to cancel block stream");
+                    Status::internal("Failed to cancel block stream")
+                })?;
             }
         }