chore: infinite retry while connecting to grpc server (#2432)

Signed-off-by: Yashash H L <[email protected]>
numaproj · Feb 27, 2025 · 3fb5b5d · 3fb5b5d
1 parent 4f31aad
commit 3fb5b5d
Show file tree

Hide file tree

Showing 4 changed files with 35 additions and 8 deletions.
diff --git a/rust/Cargo.lock b/rust/Cargo.lock
diff --git a/rust/monovertex/src/sink.rs b/rust/monovertex/src/sink.rs
@@ -6,13 +6,14 @@ use backoff::retry::Retry;
 use backoff::strategy::fixed;
 use tonic::transport::Channel;
 use tonic::Request;
+use tracing::warn;
 
 pub mod proto {
     tonic::include_proto!("sink.v1");
 }
 
 const RECONNECT_INTERVAL: u64 = 1000;
-const MAX_RECONNECT_ATTEMPTS: usize = 5;
+const MAX_RECONNECT_ATTEMPTS: usize = usize::MAX;
 const SINK_SOCKET: &str = "/var/run/numaflow/sink.sock";
 const FB_SINK_SOCKET: &str = "/var/run/numaflow/fb-sink.sock";
 
@@ -61,7 +62,15 @@ impl SinkClient {
 
         let channel = Retry::retry(
             interval,
-            || async { connect_with_uds(config.socket_path.clone().into()).await },
+            || async {
+                match connect_with_uds(config.socket_path.clone().into()).await {
+                    Ok(channel) => Ok(channel),
+                    Err(e) => {
+                        warn!(?e, "Failed to connect to sink UDS socket");
+                        Err(e)
+                    }
+                }
+            },
             |_: &Error| true,
         )
         .await?;

diff --git a/rust/monovertex/src/source.rs b/rust/monovertex/src/source.rs
@@ -8,12 +8,13 @@ use base64::Engine;
 use tokio_stream::StreamExt;
 use tonic::transport::Channel;
 use tonic::Request;
+use tracing::warn;
 
 pub mod proto {
     tonic::include_proto!("source.v1");
 }
 const RECONNECT_INTERVAL: u64 = 1000;
-const MAX_RECONNECT_ATTEMPTS: usize = 5;
+const MAX_RECONNECT_ATTEMPTS: usize = usize::MAX;
 const SOURCE_SOCKET: &str = "/var/run/numaflow/source.sock";
 const SOURCE_SERVER_INFO_FILE: &str = "/var/run/numaflow/sourcer-server-info";
 
@@ -48,7 +49,15 @@ impl SourceClient {
 
         let channel = Retry::retry(
             interval,
-            || async { connect_with_uds(config.socket_path.clone().into()).await },
+            || async {
+                match connect_with_uds(config.socket_path.clone().into()).await {
+                    Ok(channel) => Ok(channel),
+                    Err(e) => {
+                        warn!(?e, "Failed to connect to source UDS socket");
+                        Err(e)
+                    }
+                }
+            },
             |_: &Error| true,
         )
         .await?;

diff --git a/rust/monovertex/src/transformer.rs b/rust/monovertex/src/transformer.rs
@@ -6,14 +6,15 @@ use backoff::retry::Retry;
 use backoff::strategy::fixed;
 use tonic::transport::Channel;
 use tonic::Request;
+use tracing::warn;
 
 pub mod proto {
     tonic::include_proto!("sourcetransformer.v1");
 }
 
 const DROP: &str = "U+005C__DROP__";
 const RECONNECT_INTERVAL: u64 = 1000;
-const MAX_RECONNECT_ATTEMPTS: usize = 5;
+const MAX_RECONNECT_ATTEMPTS: usize = usize::MAX;
 const TRANSFORMER_SOCKET: &str = "/var/run/numaflow/sourcetransform.sock";
 const TRANSFORMER_SERVER_INFO_FILE: &str = "/var/run/numaflow/sourcetransformer-server-info";
 
@@ -48,7 +49,15 @@ impl TransformerClient {
 
         let channel = Retry::retry(
             interval,
-            || async { connect_with_uds(config.socket_path.clone().into()).await },
+            || async {
+                match connect_with_uds(config.socket_path.clone().into()).await {
+                    Ok(channel) => Ok(channel),
+                    Err(e) => {
+                        warn!(?e, "Failed to connect to transformer UDS socket");
+                        Err(e)
+                    }
+                }
+            },
             |_: &Error| true,
         )
         .await?;