From bfc8171fa77e12be92d3a73890bbb9af2dfe72cb Mon Sep 17 00:00:00 2001 From: Brandon Thomas Date: Wed, 11 Oct 2023 00:02:36 -0400 Subject: [PATCH] additional safeguards --- .../kubectl_pod_cleanup/src/delete_pods.rs | 4 +-- .../src/delete_pods_threaded.rs | 32 ++++++++++++++++++- crates/cli/kubectl_pod_cleanup/src/main.rs | 2 +- 3 files changed, 34 insertions(+), 4 deletions(-) diff --git a/crates/cli/kubectl_pod_cleanup/src/delete_pods.rs b/crates/cli/kubectl_pod_cleanup/src/delete_pods.rs index 6022d726e7..6c446ec6e5 100644 --- a/crates/cli/kubectl_pod_cleanup/src/delete_pods.rs +++ b/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) -> AnyhowResult<()> { let stdout = String::from_utf8(output.stdout)?; - info!("Output: {}", stdout); + debug!("Output: {}", stdout); Ok(()) } diff --git a/crates/cli/kubectl_pod_cleanup/src/delete_pods_threaded.rs b/crates/cli/kubectl_pod_cleanup/src/delete_pods_threaded.rs index a7198cbd07..c84d82026e 100644 --- a/crates/cli/kubectl_pod_cleanup/src/delete_pods_threaded.rs +++ b/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, 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, 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, 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::>(); + 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); + } } }); diff --git a/crates/cli/kubectl_pod_cleanup/src/main.rs b/crates/cli/kubectl_pod_cleanup/src/main.rs index 0b6a421890..f2a653fc73 100644 --- a/crates/cli/kubectl_pod_cleanup/src/main.rs +++ b/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());