diff --git a/src/controllers/restatedeployment/controller.rs b/src/controllers/restatedeployment/controller.rs index cebee63..bc7b958 100644 --- a/src/controllers/restatedeployment/controller.rs +++ b/src/controllers/restatedeployment/controller.rs @@ -56,6 +56,29 @@ pub(super) const RESTATE_DEPLOYMENT_ID_ANNOTATION: &str = "restate.dev/deploymen pub(super) const OWNED_BY_LABEL: &str = "restate.dev/owned-by"; pub(super) const APP_MANAGED_BY_LABEL: &str = "app.kubernetes.io/managed-by"; +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub(super) struct DeploymentState { + pub(super) active: bool, + pub(super) latest: bool, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(super) enum RegistrationAction { + Register { force: bool }, + Reuse, +} + +pub(super) fn registration_action( + existing_deployment_id: Option<&str>, + state: Option<&DeploymentState>, +) -> RegistrationAction { + match (existing_deployment_id, state) { + (Some(_), Some(state)) if state.latest => RegistrationAction::Reuse, + (Some(_), Some(_)) => RegistrationAction::Register { force: true }, + _ => RegistrationAction::Register { force: false }, + } +} + pub(super) struct Context { /// Kubernetes client pub client: Client, @@ -438,13 +461,11 @@ impl RestateDeployment { .annotations() .get(RESTATE_DEPLOYMENT_ID_ANNOTATION); - // if the repliceset doesn't have a deployment id, or its deployment id is not active, register it - if existing_deployment_id.is_none_or(|existing_deployment_id| { - !deployments - .get(existing_deployment_id) - .cloned() - .unwrap_or_default() - }) { + let registration_action = registration_action( + existing_deployment_id.map(String::as_str), + existing_deployment_id.and_then(|id| deployments.get(id)), + ); + if let RegistrationAction::Register { force } = registration_action { let valid = async { if let Some(cluster_name) = &self.spec.restate.register.cluster { // wait for the cluster to be ready before registering to it @@ -492,11 +513,18 @@ impl RestateDeployment { &ctx, &service_endpoint, self.spec.restate.use_http11.as_ref().cloned(), + force, ) .await?; // if registration succeeded, treat this as an active endpoint // if we fail after this point we will re-register and should get the same deployment id - deployments.insert(deployment_id.clone(), true); + deployments.insert( + deployment_id.clone(), + DeploymentState { + active: true, + latest: true, + }, + ); debug!( "Updating deployment-id annotation of ReplicaSet/Service {versioned_name} in namespace {namespace}" @@ -899,6 +927,7 @@ impl RestateDeployment { ctx: &Context, service_endpoint: &Url, use_http11: Option, + force: bool, ) -> Result { debug!( "Registering endpoint '{service_endpoint}' to Restate at '{}'", @@ -918,8 +947,13 @@ impl RestateDeployment { payload["use_http_11"] = serde_json::Value::Bool(use_http11); } + let path = if force { + "/deployments?force=true" + } else { + "/deployments" + }; let resp = ctx - .request(Method::POST, &self.spec.restate.register, "/deployments")? + .request(Method::POST, &self.spec.restate.register, path)? .json(&payload) .send() .await @@ -940,22 +974,27 @@ impl RestateDeployment { Ok(resp.id) } - pub(super) async fn list_deployments(&self, ctx: &Context) -> Result> { + pub(super) async fn list_deployments( + &self, + ctx: &Context, + ) -> Result> { // This query finds deployments, noting those that are the latest for a particular service, or have an active invocation let sql_query = r#" - WITH active_deployments AS ( + WITH latest_deployments AS ( SELECT DISTINCT deployment_id as id FROM sys_service WHERE deployment_id IS NOT NULL - UNION + ), active_invocations AS ( SELECT DISTINCT pinned_deployment_id as id FROM sys_invocation_status WHERE pinned_deployment_id IS NOT NULL AND status != 'completed' ) SELECT d.id as deployment_id, - a.id IS NOT NULL as active + l.id IS NOT NULL OR a.id IS NOT NULL as active, + l.id IS NOT NULL as latest FROM sys_deployment d - LEFT JOIN active_deployments a ON d.id = a.id; + LEFT JOIN latest_deployments l ON d.id = l.id + LEFT JOIN active_invocations a ON d.id = a.id; "#; #[derive(Deserialize)] @@ -967,6 +1006,7 @@ impl RestateDeployment { struct DeploymentQueryResultRow { deployment_id: String, active: bool, + latest: bool, } let resp = ctx @@ -984,19 +1024,23 @@ impl RestateDeployment { .await .map_err(Error::AdminCallFailed)?; - let mut endpoints = HashMap::with_capacity(response.rows.len()); + let mut endpoints: HashMap = + HashMap::with_capacity(response.rows.len()); for row in response.rows { + let state = DeploymentState { + active: row.active, + latest: row.latest, + }; match endpoints.entry(row.deployment_id) { std::collections::hash_map::Entry::Occupied(mut entry) => { // two rows with same deployment id shouldnt happen... // we treat the deployment as active if any row is active - if !entry.get() { - entry.insert(row.active); - } + entry.get_mut().active |= state.active; + entry.get_mut().latest |= state.latest; } std::collections::hash_map::Entry::Vacant(entry) => { - entry.insert(row.active); + entry.insert(state); } } } @@ -1411,6 +1455,43 @@ mod tests { use super::*; use serde_json::json; + #[test] + fn registration_converges_an_old_version_to_latest() { + let active_latest = DeploymentState { + active: true, + latest: true, + }; + let active_old = DeploymentState { + active: true, + latest: false, + }; + let inactive_old = DeploymentState { + active: false, + latest: false, + }; + + assert_eq!( + registration_action(Some("dp_latest"), Some(&active_latest)), + RegistrationAction::Reuse + ); + assert_eq!( + registration_action(Some("dp_old"), Some(&active_old)), + RegistrationAction::Register { force: true } + ); + assert_eq!( + registration_action(Some("dp_inactive_old"), Some(&inactive_old)), + RegistrationAction::Register { force: true } + ); + assert_eq!( + registration_action(Some("dp_gone"), None), + RegistrationAction::Register { force: false } + ); + assert_eq!( + registration_action(None, None), + RegistrationAction::Register { force: false } + ); + } + /// Build a minimal ReplicaSet-mode RestateDeployment for selector tests. fn make_rsd(match_labels: Option<&[(&str, &str)]>, image: &str) -> RestateDeployment { let selector = match_labels.map(|labels| { diff --git a/src/controllers/restatedeployment/reconcilers/knative.rs b/src/controllers/restatedeployment/reconcilers/knative.rs index 4ea5993..db078a5 100644 --- a/src/controllers/restatedeployment/reconcilers/knative.rs +++ b/src/controllers/restatedeployment/reconcilers/knative.rs @@ -9,7 +9,8 @@ use tracing::*; use url::Url; use crate::controllers::restatedeployment::controller::{ - Context, RESTATE_DEPLOYMENT_ID_ANNOTATION, + Context, DeploymentState, RESTATE_DEPLOYMENT_ID_ANNOTATION, RegistrationAction, + registration_action, }; use crate::controllers::restatedeployment::reconcilers::replicaset::generate_pod_template_hash; use crate::resources::knative::{ @@ -640,14 +641,19 @@ async fn register_or_lookup_deployment( config: &Configuration, route: &Route, ) -> Result { - // Check if Configuration already has deployment-id annotation - if let Some(annotations) = &config.metadata.annotations - && let Some(deployment_id) = annotations.get(RESTATE_DEPLOYMENT_ID_ANNOTATION) - { - trace!( - deployment_id = %deployment_id, - "Found existing deployment ID in Configuration annotation" - ); + let existing_deployment_id = config + .metadata + .annotations + .as_ref() + .and_then(|annotations| annotations.get(RESTATE_DEPLOYMENT_ID_ANNOTATION)); + let deployments = rsd.list_deployments(ctx).await?; + let action = registration_action( + existing_deployment_id.map(String::as_str), + existing_deployment_id.and_then(|id| deployments.get(id)), + ); + if action == RegistrationAction::Reuse { + let deployment_id = existing_deployment_id.expect("reuse requires deployment ID"); + trace!(deployment_id = %deployment_id, "Reusing latest deployment ID"); return Ok(deployment_id.clone()); } @@ -669,8 +675,16 @@ async fn register_or_lookup_deployment( .register .maybe_tunnel_url(&ctx.rce_store, url)?; + let RegistrationAction::Register { force } = action else { + unreachable!("reuse returned above") + }; let deployment_id = rsd - .register_service_with_restate(ctx, &url, rsd.spec.restate.use_http11.as_ref().cloned()) + .register_service_with_restate( + ctx, + &url, + rsd.spec.restate.use_http11.as_ref().cloned(), + force, + ) .await?; Ok(deployment_id) @@ -743,7 +757,7 @@ pub async fn cleanup_old_configurations( ctx: &Context, rsd_uid: &str, rsd: &RestateDeployment, - deployments: &std::collections::HashMap, + deployments: &std::collections::HashMap, active_tag: Option<&str>, ) -> Result<(i32, Option>)> { // Use reflector cache instead of API list() call @@ -821,7 +835,7 @@ pub async fn cleanup_old_configurations( let deployment = config_deployment_id .and_then(|config_deployment_id| deployments.get(config_deployment_id).cloned()); let deployment_exists = deployment.is_some(); - let deployment_active = deployment.unwrap_or(false); + let deployment_active = deployment.is_some_and(|state| state.active); if deployment_active { active_count += 1; diff --git a/src/controllers/restatedeployment/reconcilers/replicaset.rs b/src/controllers/restatedeployment/reconcilers/replicaset.rs index 23266d4..3838329 100644 --- a/src/controllers/restatedeployment/reconcilers/replicaset.rs +++ b/src/controllers/restatedeployment/reconcilers/replicaset.rs @@ -16,7 +16,8 @@ use serde_json::json; use tracing::*; use crate::controllers::restatedeployment::controller::{ - APP_MANAGED_BY_LABEL, Context, OWNED_BY_LABEL, RESTATE_DEPLOYMENT_ID_ANNOTATION, + APP_MANAGED_BY_LABEL, Context, DeploymentState, OWNED_BY_LABEL, + RESTATE_DEPLOYMENT_ID_ANNOTATION, }; use crate::resources::restatecloudenvironments::InProcessTunnelParams; use crate::resources::restatedeployments::RestateDeployment; @@ -274,7 +275,7 @@ pub async fn cleanup_old_replicasets( rs_api: &Api, rsd_uid: &str, rsd: &RestateDeployment, - deployments: &HashMap, + deployments: &HashMap, except_rs: Option<&str>, ) -> Result<(i32, Option>)> { let replicasets_cell = std::cell::Cell::new(Vec::new()); @@ -339,7 +340,7 @@ pub async fn cleanup_old_replicasets( let deployment = rs_deployment_id .and_then(|rs_deployment_id| deployments.get(rs_deployment_id).cloned()); let deployment_exists = deployment.is_some(); - let deployment_active = deployment.unwrap_or(false); + let deployment_active = deployment.is_some_and(|state| state.active); if deployment_active { active_count += 1;