Add exponential backoff and log suppression for scheduler disconnection

When the scheduler disconnects or becomes unavailable, the executor poll loop previously logged a WARN message every ~100ms, spamming logs during graceful shutdown scenarios.

Changes:
- Add exponential backoff (100ms to 30s) when scheduler connection fails
- Reduce log level from WARN to DEBUG after 5 consecutive failures
- Log restoration message when connection is re-established
- Prevents log spam when scheduler shuts down before executor

Author: phillipleblanc

ballista/executor/src/execution_loop.rs

@@ -43,6 +43,13 @@ use std::ops::Deref;
 use std::sync::mpsc::{Receiver, Sender, TryRecvError};
 use std::time::{SystemTime, UNIX_EPOCH};
 use std::{sync::Arc, time::Duration};
+
+/// Initial backoff delay when scheduler connection fails (100ms).
+const INITIAL_BACKOFF_MS: u64 = 100;
+/// Maximum backoff delay when scheduler connection fails (30s).
+const MAX_BACKOFF_MS: u64 = 30_000;
+/// Number of consecutive failures before reducing log level from WARN to DEBUG.
+const QUIET_AFTER_FAILURES: u32 = 5;
 use tokio::sync::oneshot::Sender as OneShotSender;
 use tokio::sync::{OwnedSemaphorePermit, Semaphore};
 use tonic::codegen::{Body, Bytes, StdError};
@@ -83,16 +90,16 @@ where
         }
     });
 
+    // Track consecutive scheduler connection failures for backoff and log suppression
+    let mut consecutive_failures: u32 = 0;
+    let mut current_backoff_ms: u64 = INITIAL_BACKOFF_MS;
+
     loop {
         // Wait for task slots to be available before asking for new work
         let permit = available_task_slots.acquire().await.unwrap();
         // Make the slot available again
         drop(permit);
 
-        // Keeps track of whether we received task in last iteration
-        // to avoid going in sleep mode between polling
-        let mut active_job = false;
-
         let task_status: Vec<TaskStatus> =
             sample_tasks_status(&mut task_status_receiver).await;
 
@@ -107,8 +114,21 @@ where
 
         *report_ready;
 
+        // Keeps track of whether we received task in last iteration
+        // to avoid going in sleep mode between polling
+        let active_job;
+
         match poll_work_result {
             Ok(result) => {
+                // Reset backoff state on successful connection
+                if consecutive_failures > 0 {
+                    info!(
+                        "Scheduler connection restored after {consecutive_failures} failed attempts"
+                    );
+                }
+                consecutive_failures = 0;
+                current_backoff_ms = INITIAL_BACKOFF_MS;
+
                 let PollWorkResult {
                     tasks,
                     jobs_to_clean,
@@ -197,7 +217,23 @@ where
                 }
             }
             Err(error) => {
-                warn!("Executor poll work loop failed. If this continues to happen the Scheduler might be marked as dead. Error: {error}");
+                consecutive_failures = consecutive_failures.saturating_add(1);
+
+                // Log at WARN level for first few failures, then reduce to DEBUG to avoid log spam
+                if consecutive_failures <= QUIET_AFTER_FAILURES {
+                    warn!(
+                        "Executor poll work loop failed (attempt {consecutive_failures}). If this continues, the Scheduler might be unavailable. Error: {error}"
+                    );
+                } else {
+                    debug!(
+                        "Executor poll work loop failed (attempt {consecutive_failures}). Error: {error}"
+                    );
+                }
+
+                // Apply exponential backoff before retrying
+                tokio::time::sleep(Duration::from_millis(current_backoff_ms)).await;
+                current_backoff_ms = (current_backoff_ms * 2).min(MAX_BACKOFF_MS);
+                continue;
             }
         }