diff --git a/src/config/mod.rs b/src/config/mod.rs index bbbb208b..501759b3 100644 --- a/src/config/mod.rs +++ b/src/config/mod.rs @@ -159,13 +159,6 @@ pub struct Config { /// (discarding, retrying activations, etc.) are executed. pub upkeep_task_interval_ms: u64, - /// The number of milliseconds between upkeep runs that indicates unhealthy - /// performance that should trigger a restart. - pub upkeep_unhealthy_interval_ms: u64, - - /// Whether to skip the health check if the pods are in a bad state. - pub health_check_killswitched: bool, - /// The number of seconds that deadline resets /// are skipped after startup. This delay allows workers /// time to publish results after a broker restart. @@ -303,8 +296,6 @@ impl Default for Config { store: StoreConfig::default(), runtime_config_path: None, upkeep_task_interval_ms: 1000, - upkeep_unhealthy_interval_ms: 5000, - health_check_killswitched: false, upkeep_deadline_reset_skip_after_startup_sec: 60, maintenance_task_interval_ms: 6000, runtime_metrics_interval: Duration::from_secs(1), diff --git a/src/main.rs b/src/main.rs index 30436b4c..65c92dfc 100644 --- a/src/main.rs +++ b/src/main.rs @@ -134,7 +134,6 @@ async fn main() -> Result<(), Error> { upkeep_store, startup_time, runtime_config_manager.clone(), - health_reporter.clone(), ) .await?; Ok(()) diff --git a/src/upkeep.rs b/src/upkeep.rs index d26c4606..5677f606 100644 --- a/src/upkeep.rs +++ b/src/upkeep.rs @@ -10,12 +10,9 @@ use prost_types::Timestamp; use sentry_arroyo::backends::ProducerError; use sentry_protos::taskbroker::v1::TaskActivation; use tokio::{fs, join, select, time}; -use tonic_health::ServingStatus; -use tonic_health::server::HealthReporter; use tracing::{debug, error, info, instrument, warn}; use uuid::Uuid; -use crate::SERVICE_NAME; use crate::config::Config; use crate::kafka::producer::KafkaProducer; use crate::runtime_config::RuntimeConfigManager; @@ -29,7 +26,6 @@ pub async fn upkeep( store: Arc, startup_time: DateTime, runtime_config_manager: Arc, - health_reporter: HealthReporter, ) -> Result<(), anyhow::Error> { const ASYNC_BACKTRACE_LOG_INTERVAL: Duration = Duration::from_secs(30); @@ -45,7 +41,6 @@ pub async fn upkeep( let guard = elegant_departure::get_shutdown_guard().shutdown_on_drop(); let mut timer = time::interval(Duration::from_millis(config.upkeep_task_interval_ms)); timer.set_missed_tick_behavior(time::MissedTickBehavior::Delay); - let mut last_run = Instant::now(); let mut last_vacuum = Instant::now(); let mut last_backtrace_log = Instant::now(); let mut emitted_partitions: HashSet = HashSet::new(); @@ -61,7 +56,6 @@ pub async fn upkeep( &mut last_vacuum, &mut emitted_partitions, ).await; - last_run = check_health(last_run, &config, health_reporter.clone()).await; if config.log_async_backtrace && last_backtrace_log.elapsed() >= ASYNC_BACKTRACE_LOG_INTERVAL @@ -561,40 +555,6 @@ fn create_retry_activation(mut new_activation: TaskActivation) -> TaskActivation new_activation } -/// Update health based on upkeep intervals -/// -/// Because SQLite is a shared component for upkeep, grpc, and consumer loops, when one gets slow -/// they all do. We see taskbroker getting slow in production when the underlying cloud disk -/// degrades. Restarting the application typically solves this problem. -/// Track metrics and update status so that we can measure health failures, and apply probes -/// incrementally. -pub async fn check_health( - last_run: Instant, - config: &Config, - health_reporter: HealthReporter, -) -> Instant { - let now = Instant::now(); - if config.health_check_killswitched { - metrics::counter!("upkeep.health", "status" => "killswitched").increment(1); - health_reporter - .set_service_status(SERVICE_NAME, ServingStatus::Serving) - .await; - return now; - } - if now - last_run > Duration::from_millis(config.upkeep_unhealthy_interval_ms) { - metrics::counter!("upkeep.health", "status" => "unhealthy").increment(1); - health_reporter - .set_service_status(SERVICE_NAME, ServingStatus::NotServing) - .await; - } else { - metrics::counter!("upkeep.health", "status" => "healthy").increment(1); - health_reporter - .set_service_status(SERVICE_NAME, ServingStatus::Serving) - .await; - } - now -} - #[cfg(test)] mod tests { use std::collections::HashSet;