diff --git a/release-notes/unreleased/185-rate-limit-restatedeployment-queries.md b/release-notes/unreleased/185-rate-limit-restatedeployment-queries.md new file mode 100644 index 0000000..0bda440 --- /dev/null +++ b/release-notes/unreleased/185-rate-limit-restatedeployment-queries.md @@ -0,0 +1,25 @@ +# Release Notes for Issue #185: Rate limit RestateDeployment cleanup queries + +## Behavioral Change + +### What Changed + +The operator now rate-limits expensive RestateDeployment deployment-usage queries per +Restate Admin API endpoint. Only one such query may run for an endpoint at a time, and +retries are spaced with exponential backoff and jitter. + +### Why This Matters + +When a deployment remains in use, repeated reconciles can otherwise repeatedly issue a +costly usage query. Coordinating these retries protects the Restate environment while +preserving normal cleanup once the deployment can be removed. + +### Impact on Users + +- Existing and new deployments may take longer to retry cleanup after a query-related + failure or while another deployment targeting the same endpoint is being checked. +- No manifest, Helm value, or migration change is required. + +### Related Issues + +- Issue #185: Use vqueues for pinned RestateDeployment accounting when capability is proven diff --git a/src/controllers/restatedeployment/controller.rs b/src/controllers/restatedeployment/controller.rs index 62c413c..955b3e8 100644 --- a/src/controllers/restatedeployment/controller.rs +++ b/src/controllers/restatedeployment/controller.rs @@ -1,4 +1,5 @@ use std::collections::BTreeMap; +use std::hash::{Hash, Hasher}; use std::sync::Arc; use std::time::Duration; @@ -50,6 +51,7 @@ use crate::controllers::restatedeployment::cleanup::{ }; use crate::controllers::restatedeployment::reconcilers; use crate::controllers::restatedeployment::registration::{self, RegistrationAction}; +use crate::controllers::restatedeployment::retry::ExpensiveOperationRetries; use super::reconcilers::replicaset::{ POD_TEMPLATE_HASH_LABEL, RESTATE_POD_TEMPLATE_ANNOTATION, RESTATE_TUNNEL_NAME_ANNOTATION, @@ -86,6 +88,8 @@ pub(super) struct Context { pub metrics: Metrics, /// HTTP client pub http_client: reqwest::Client, + /// Process-local, endpoint-scoped retry coordination for expensive admin work. + pub expensive_operation_retries: ExpensiveOperationRetries, } impl Context { @@ -115,6 +119,7 @@ impl Context { metrics, diagnostics: state.diagnostics.clone(), http_client: reqwest::Client::new(), + expensive_operation_retries: ExpensiveOperationRetries::new(), }) } @@ -139,6 +144,65 @@ impl Context { Ok(request_builder) } + + /// A stable cache key for capabilities of the downstream Restate server. Authentication is + /// intentionally excluded: token rotation does not change which system tables exist. + pub fn admin_endpoint_key(&self, admin_endpoint: &RestateAdminEndpoint) -> Result { + Ok(admin_endpoint + .admin_url(&self.rce_store, &self.cluster_dns)? + .to_string()) + } + + fn expensive_operation_key(&self, rsd: &RestateDeployment) -> (String, String) { + let endpoint = self + .admin_endpoint_key(&rsd.spec.restate.register) + // Invalid endpoint configuration cannot be globally coordinated. Preserve the + // floor rather than hiding the configuration error behind a retry storm. + .unwrap_or_else(|_| "".into()); + let resource = rsd.uid().unwrap_or_else(|| { + format!("{}/{}", rsd.namespace().unwrap_or_default(), rsd.name_any()) + }); + (endpoint, resource) + } + + fn admit_expensive_operation(&self, rsd: &RestateDeployment) -> Result<()> { + let (endpoint, resource) = self.expensive_operation_key(rsd); + self.expensive_operation_retries + .admit(endpoint, resource) + .map_err(|requeue_after| Error::ExpensiveOperationDeferred { requeue_after }) + } + + fn expensive_retry_after(&self, rsd: &RestateDeployment) -> Duration { + let (endpoint, resource) = self.expensive_operation_key(rsd); + self.expensive_operation_retries.failure(endpoint, resource) + } + + fn finish_expensive_operation(&self, rsd: &RestateDeployment) { + let (endpoint, _) = self.expensive_operation_key(rsd); + self.expensive_operation_retries.finish(&endpoint); + } + + fn reset_expensive_retries(&self, rsd: &RestateDeployment) { + let (endpoint, resource) = self.expensive_operation_key(rsd); + self.expensive_operation_retries + .reset_resource(&endpoint, &resource); + } +} + +impl RestateDeployment { + /// Spread otherwise-healthy periodic reconciliations over a minute, using stable resource + /// identity so an operator restart does not align every deployment again. This is a poll, + /// not an error retry, so a small positive jitter is preferable to an exact global cadence. + fn healthy_requeue_after(&self) -> Duration { + const BASE: Duration = Duration::from_secs(5 * 60); + const JITTER: u64 = 60; + + let mut hasher = fnv::FnvHasher::default(); + self.uid() + .unwrap_or_else(|| self.name_any()) + .hash(&mut hasher); + BASE + Duration::from_secs(hasher.finish() % (JITTER + 1)) + } } /// Check an admin API response, returning the response if successful or an error @@ -226,8 +290,10 @@ fn root_cause(err: &Error) -> &Error { } } +#[cfg(test)] fn error_policy(_rs: Arc, err: &Error, _ctx: C) -> Action { match root_cause(err) { + Error::ExpensiveOperationDeferred { requeue_after } => Action::requeue(*requeue_after), // A drain knows its own deadline; the blanket interval would make a short // drainDelaySeconds cost up to 30s per version anyway. A deletion blocked on // in-flight invocations sets its own interval too, backing off as the wait grows @@ -243,6 +309,30 @@ fn error_policy(_rs: Arc, err: &Error, _ctx: C) -> Action { } } +/// Controller-framework errors do not pass through `reconcile_status`, including finalizer +/// failures during deletion. Apply the same endpoint-scoped protection there while preserving a +/// real drain deadline when one exists. +fn restate_deployment_error_policy( + rsd: Arc, + err: &Error, + ctx: Arc, +) -> Action { + match root_cause(err) { + Error::ExpensiveOperationDeferred { requeue_after } => Action::requeue(*requeue_after), + Error::DeploymentDraining { + requeue_after: Some(requeue_after), + } => Action::requeue(*requeue_after), + Error::DeploymentInUse { + requeue_after: Some(requeue_after), + .. + } => Action::requeue((*requeue_after).max(ctx.expensive_retry_after(&rsd))), + Error::AdminCallFailed(_) | Error::AdminCallRejected { .. } => { + Action::requeue(ctx.expensive_retry_after(&rsd)) + } + _ => Action::requeue(Duration::from_secs(30)), + } +} + impl RestateDeployment { /// Resolve the RestateCloudEnvironment values a `tunnelMode: in-process` /// deployment derives its identity from (None for every other mode). They feed @@ -678,7 +768,7 @@ impl RestateDeployment { .and_then(|s| s.conditions.as_ref()) .and_then(|c| c.iter().find(|cond| cond.r#type == "Ready")); - let (result, message, reason, status) = if is_knative { + let (mut result, message, reason, status) = if is_knative { // Delegate to Knative reconciler let knative_result = if self.spec.restate.is_in_process_tunnel() { // An in-process tunnel carries no traffic the Knative autoscaler can @@ -701,10 +791,10 @@ impl RestateDeployment { if secs < 5 * 60 { Action::requeue(Duration::from_secs(secs)) } else { - Action::requeue(Duration::from_secs(5 * 60)) + Action::requeue(self.healthy_requeue_after()) } } - None => Action::requeue(Duration::from_secs(5 * 60)), + None => Action::requeue(self.healthy_requeue_after()), }; ( @@ -846,10 +936,10 @@ impl RestateDeployment { if secs < 5 * 60 { Action::requeue(Duration::from_secs(secs)) } else { - Action::requeue(Duration::from_secs(5 * 60)) + Action::requeue(self.healthy_requeue_after()) } } - None => Action::requeue(Duration::from_secs(5 * 60)), + None => Action::requeue(self.healthy_requeue_after()), }; status_from_replica_set( @@ -979,6 +1069,26 @@ impl RestateDeployment { } }; + // These states all ran, or are about to immediately re-run, deployment usage + // accounting against Restate. Coordinate their retries per endpoint so one stuck + // resource cannot keep a full scan at a fixed cadence, and several resources sharing + // an environment cannot line up their retries. Route/Configuration readiness is + // deliberately not included: those paths fail before the Restate query and retain + // their short Kubernetes readiness polling. + if status == "True" { + ctx.reset_expensive_retries(self); + } else if result.is_ok() && requeues_after_expensive_admin_work(&reason) { + let requeue_after = ctx.expensive_retry_after(self); + debug!( + name = %self.name_any(), + namespace = %namespace, + reason = %reason, + requeue_after_secs = %requeue_after.as_secs(), + "Backing off an expensive RestateDeployment retry" + ); + result = Ok(Action::requeue(requeue_after)); + } + // Emit a K8s Warning event for admin API failures so they're visible // via `kubectl describe` and `kubectl get events` if reason == "AdminCallFailed" || reason == "AdminCallRejected" { @@ -1051,8 +1161,21 @@ impl RestateDeployment { ctx: &Context, mode: CleanupMode, ) -> Result { - let sql_query = deployment_usage_query(mode); + ctx.admit_expensive_operation(self)?; + let response = self.query_deployment_usage(ctx, mode).await; + ctx.finish_expensive_operation(self); + match response { + Ok(response) => Ok(response.into_map()), + Err(err) => Err(err), + } + } + async fn query_deployment_usage( + &self, + ctx: &Context, + mode: CleanupMode, + ) -> Result { + let sql_query = deployment_usage_query(mode); let resp = ctx .request(Method::POST, &self.spec.restate.register, "/query")? .header(reqwest::header::ACCEPT, "application/json") @@ -1062,13 +1185,11 @@ impl RestateDeployment { .send() .await .map_err(Error::AdminCallFailed)?; - let response: DeploymentUsageRows = check_admin_response(resp) + check_admin_response(resp) .await? .json() .await - .map_err(Error::AdminCallFailed)?; - - Ok(response.into_map()) + .map_err(Error::AdminCallFailed) } /// How long ago deletion was requested, for pacing the retries of a blocked one. @@ -1190,6 +1311,24 @@ impl RestateDeployment { } } +/// Readiness conditions that occur after the normal ReplicaSet-mode usage query. The controller +/// reports them as `Ready=False` rather than reconciliation errors, so they must opt into the +/// shared retry coordinator here instead of relying on the controller framework's error policy. +fn requeues_after_expensive_admin_work(reason: &str) -> bool { + matches!( + reason, + "AdminCallFailed" + | "AdminCallRejected" + | "ForeignDeployment" + | "NotLatest" + | "ClusterNotReady" + | "ReplicaSetNoStatus" + | "ReplicaSetScaling" + | "ReplicaSetPodNotReady" + | "ReplicaSetPodNotAvailable" + ) +} + /// Build the `.status.labelSelector` for a ReplicaSet-mode RestateDeployment, /// scoped to the latest version's pods by appending the pod-template-hash. /// @@ -1248,8 +1387,8 @@ pub fn validate_replica_set_status( status } else { return Err(Error::DeploymentNotReady { - message: "ReplicaSetNoStatus".into(), - reason: "ReplicaSet has no status set; it may have just been created".into(), + message: "ReplicaSet has no status set; it may have just been created".into(), + reason: "ReplicaSetNoStatus".into(), requeue_after: None, replica_set_status: status.cloned().map(Box::new), }); @@ -1494,7 +1633,7 @@ pub async fn run(client: Client, metrics: Metrics, state: State) { .owns_stream(hpa_reflector) .run( reconcile, - error_policy, + restate_deployment_error_policy, Context::new( client, replicasets_store, diff --git a/src/controllers/restatedeployment/mod.rs b/src/controllers/restatedeployment/mod.rs index 33532f2..4358a84 100644 --- a/src/controllers/restatedeployment/mod.rs +++ b/src/controllers/restatedeployment/mod.rs @@ -1,6 +1,7 @@ pub(crate) mod cleanup; pub mod controller; pub(crate) mod registration; +pub(crate) mod retry; mod reconcilers; diff --git a/src/controllers/restatedeployment/retry.rs b/src/controllers/restatedeployment/retry.rs new file mode 100644 index 0000000..f645e59 --- /dev/null +++ b/src/controllers/restatedeployment/retry.rs @@ -0,0 +1,170 @@ +//! Process-local admission control and retry coordination for expensive Restate admin queries. + +use std::collections::HashMap; +use std::hash::{BuildHasher, Hash, RandomState}; +use std::sync::Mutex; +use std::time::{Duration, Instant}; + +const FLOOR: Duration = Duration::from_secs(30); +const CEILING: Duration = Duration::from_secs(5 * 60); +const MAX_JITTER: Duration = Duration::from_secs(60); +const ENDPOINT_SPACING: Duration = Duration::from_secs(30); + +#[derive(Debug, Clone, Hash, PartialEq, Eq)] +struct RetryKey { + endpoint: String, + resource: String, +} + +#[derive(Debug, Default)] +struct RetryState { + failures: u32, + not_before: Option, +} + +#[derive(Debug, Default)] +struct EndpointState { + in_flight: bool, + next_admission: Option, +} + +#[derive(Default)] +struct Inner { + retries: HashMap, + endpoints: HashMap, +} + +/// Coordinates expensive deployment-usage queries for one operator process. +/// +/// A permit is held until the HTTP operation completes, rather than for a fixed time, so a slow +/// legacy query cannot overlap a later reconcile. The endpoint spacing is enforced when a caller +/// actually obtains that permit. Delayed callers do not reserve queue positions: a watch event +/// may cause an earlier healthy caller to run first, but cannot starve a caller on a stale slot. +pub(super) struct ExpensiveOperationRetries { + inner: Mutex, +} + +impl ExpensiveOperationRetries { + pub(super) fn new() -> Self { + Self { + inner: Mutex::new(Inner::default()), + } + } + + /// Acquires the endpoint permit or returns a controller requeue delay. + pub(super) fn admit(&self, endpoint: String, resource: String) -> Result<(), Duration> { + let now = Instant::now(); + let key = RetryKey { + endpoint: endpoint.clone(), + resource, + }; + let mut inner = self.inner.lock().expect("retry coordinator lock poisoned"); + let not_before = inner.retries.entry(key.clone()).or_default().not_before; + if let Some(not_before) = not_before.filter(|time| *time > now) { + return Err(not_before.saturating_duration_since(now)); + } + + { + let endpoint_state = inner.endpoints.entry(endpoint).or_default(); + if endpoint_state.in_flight { + return Err(ENDPOINT_SPACING); + } + if let Some(next_admission) = endpoint_state.next_admission.filter(|time| *time > now) { + return Err(next_admission.saturating_duration_since(now)); + } + endpoint_state.in_flight = true; + endpoint_state.next_admission = Some(now + ENDPOINT_SPACING); + } + + inner.retries.entry(key).or_default().not_before = None; + Ok(()) + } + + /// Releases the endpoint permit after the HTTP request completes. It intentionally keeps the + /// resource's retry history; the whole reconcile must succeed before that state is reset. + pub(super) fn finish(&self, endpoint: &str) { + if let Some(state) = self + .inner + .lock() + .expect("retry coordinator lock poisoned") + .endpoints + .get_mut(endpoint) + { + state.in_flight = false; + } + } + + /// Records one failed query-bearing reconcile and schedules a fresh exponential retry. + pub(super) fn failure(&self, endpoint: String, resource: String) -> Duration { + let now = Instant::now(); + let key = RetryKey { endpoint, resource }; + let mut inner = self.inner.lock().expect("retry coordinator lock poisoned"); + let retry = inner.retries.entry(key.clone()).or_default(); + retry.failures = retry.failures.saturating_add(1); + let delay = backoff_with_positive_jitter(&key, retry.failures); + retry.not_before = Some(now + delay); + delay + } + + /// A fully successful reconcile clears only its own retry history. + pub(super) fn reset_resource(&self, endpoint: &str, resource: &str) { + self.inner + .lock() + .expect("retry coordinator lock poisoned") + .retries + .remove(&RetryKey { + endpoint: endpoint.into(), + resource: resource.into(), + }); + } +} + +fn backoff_with_positive_jitter(key: &RetryKey, failures: u32) -> Duration { + let multiplier = 1_u32 << failures.saturating_sub(1).min(4); + let base = FLOOR.saturating_mul(multiplier).min(CEILING); + let jitter_cap = (base / 5).min(MAX_JITTER); + let jitter = RandomState::new().hash_one(key) % (jitter_cap.as_nanos() as u64 + 1); + base + Duration::from_nanos(jitter) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn backoff_grows_and_keeps_jitter_at_the_cap() { + let key = RetryKey { + endpoint: "https://admin.example.test".into(), + resource: "uid-a".into(), + }; + for (failures, floor) in [(1, 30), (2, 60), (3, 120), (4, 240), (5, 300), (6, 300)] { + let delay = backoff_with_positive_jitter(&key, failures); + assert!(delay >= Duration::from_secs(floor)); + assert!(delay <= CEILING + MAX_JITTER); + } + } + + #[test] + fn endpoint_permit_remains_held_until_the_query_finishes() { + let retries = ExpensiveOperationRetries::new(); + let endpoint = "https://admin.example.test"; + assert!(retries.admit(endpoint.into(), "uid-a".into()).is_ok()); + // This is still deferred even after the normal spacing period would have elapsed, + // because the only release path is `finish`. + assert!(retries.admit(endpoint.into(), "uid-b".into()).is_err()); + retries.finish(endpoint); + // Spacing is now the only remaining limiter. + assert!(retries.admit(endpoint.into(), "uid-b".into()).is_err()); + } + + #[test] + fn failure_history_is_keyed_by_the_expensive_operation_not_the_reason() { + let retries = ExpensiveOperationRetries::new(); + let endpoint = "https://admin.example.test"; + let first = retries.failure(endpoint.into(), "uid-a".into()); + retries.reset_resource(endpoint, "uid-a"); + let reset = retries.failure(endpoint.into(), "uid-a".into()); + assert!(first >= FLOOR); + assert!(reset >= FLOOR); + } +} diff --git a/src/lib.rs b/src/lib.rs index f49f833..b1a28ed 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -64,6 +64,9 @@ pub enum Error { body: String, }, + #[error("expensive Restate admin operation deferred for {requeue_after:?}")] + ExpensiveOperationDeferred { requeue_after: Duration }, + #[error("Encountered a hash collision, will retry with a new template hash")] HashCollision, @@ -126,6 +129,7 @@ impl Error { Error::InvalidSigningKeyError(_) => "InvalidSigningKeyError", Error::AdminCallFailed(_) => "AdminCallFailed", Error::AdminCallRejected { .. } => "AdminCallRejected", + Error::ExpensiveOperationDeferred { .. } => "ExpensiveOperationDeferred", Error::HashCollision => "HashCollision", Error::DeploymentNotLatest { .. } => "DeploymentNotLatest", Error::DeploymentInUse { .. } => "DeploymentInUse",