additional safeguards

crates/cli/kubectl_pod_cleanup/src/delete_pods.rs

@@ -1,6 +1,6 @@
 use std::process::Command;
 
-use log::info;
+use log::{debug, info};
 use rand::prelude::IteratorRandom;
 use rand::thread_rng;
 
@@ -42,7 +42,7 @@ pub fn delete_pod_batch(pod_names: &Vec<String>) -> AnyhowResult<()> {
 
   let stdout = String::from_utf8(output.stdout)?;
 
-  info!("Output: {}", stdout);
+  debug!("Output: {}", stdout);
 
   Ok(())
 }

crates/cli/kubectl_pod_cleanup/src/delete_pods_threaded.rs

@@ -1,4 +1,5 @@
 use std::thread;
+use std::time::{Duration, Instant};
 
 use log::error;
 
@@ -7,6 +8,14 @@ use errors::{anyhow, AnyhowResult};
 use crate::delete_pods::delete_pod_batch;
 use crate::pod_store::PodStore;
 
+/// If kubectl returns faster than this, it may be spamming the API.
+const TIMEOUT_TRIGGER_THRESHOLD : Duration = Duration::from_millis(500);
+
+const TIMEOUT_DURATION : Duration = Duration::from_secs(5);
+
+/// If we fail to delete pods this number of times, exit the thread.
+const FAILURE_TRIGGER_THRESHOLD : usize = 5;
+
 pub fn delete_pods_threaded(pod_names: Vec<String>, num_tasks: usize, batch_size: usize) -> AnyhowResult<()> {
   let pod_store = PodStore::from_names(pod_names);
 
@@ -17,6 +26,8 @@ pub fn delete_pods_threaded(pod_names: Vec<String>, num_tasks: usize, batch_size
 
     // NB: Using std::thread instead of Tokio since this API is blocking.
     let handle = thread::spawn(move || {
+      let mut failure_to_progress_count = 0;
+
       loop {
         let batch = match cloned.grab_batch(batch_size) {
           Ok(batch) => batch,
@@ -27,17 +38,36 @@ pub fn delete_pods_threaded(pod_names: Vec<String>, num_tasks: usize, batch_size
         };
 
         if batch.is_empty() {
-          break;
+          return;
+        }
+
+        if batch.len() < batch_size {
+          failure_to_progress_count += 1;
+        } else {
+          failure_to_progress_count = 0;
+        }
+
+        if failure_to_progress_count > FAILURE_TRIGGER_THRESHOLD {
+          error!("Failure to make progress.");
+          return;
         }
 
         let pod_names = batch.into_iter()
             .map(|n| n.to_string())
             .collect::<Vec<String>>();
 
+        let start = Instant::now();
+
         if let Err(err) = delete_pod_batch(&pod_names) {
           error!("threading error: {:?}", err);
           return;
         }
+
+        let duration = Instant::now().duration_since(start);
+
+        if duration.lt(&TIMEOUT_TRIGGER_THRESHOLD) {
+          thread::sleep(TIMEOUT_DURATION);
+        }
       }
     });
 

crates/cli/kubectl_pod_cleanup/src/main.rs

@@ -87,7 +87,7 @@ pub fn main() -> AnyhowResult<()> {
 
   info!("Clearing {} / {} pods...", pods_to_clear.len(), all_pods.len());
 
-  let wait_duration = Duration::from_secs(10);
+  let wait_duration = Duration::from_secs(3);
 
   info!("Clearing in {} seconds... (Last chance to cancel!)", wait_duration.as_secs());