Skip to content

Force existing non-latest deployments back to latest #160

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
titanous wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Force existing non-latest deployments back to latest #160

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
119 changes: 100 additions & 19 deletions src/controllers/restatedeployment/controller.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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}"
Expand Down Expand Up @@ -899,6 +927,7 @@ impl RestateDeployment {
ctx: &Context,
service_endpoint: &Url,
use_http11: Option<bool>,
force: bool,
) -> Result<String> {
debug!(
"Registering endpoint '{service_endpoint}' to Restate at '{}'",
Expand All @@ -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
Expand All @@ -940,22 +974,27 @@ impl RestateDeployment {
Ok(resp.id)
}

pub(super) async fn list_deployments(&self, ctx: &Context) -> Result<HashMap<String, bool>> {
pub(super) async fn list_deployments(
&self,
ctx: &Context,
) -> Result<HashMap<String, DeploymentState>> {
// 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)]
Expand All @@ -967,6 +1006,7 @@ impl RestateDeployment {
struct DeploymentQueryResultRow {
deployment_id: String,
active: bool,
latest: bool,
}

let resp = ctx
Expand All @@ -984,19 +1024,23 @@ impl RestateDeployment {
.await
.map_err(Error::AdminCallFailed)?;

let mut endpoints = HashMap::with_capacity(response.rows.len());
let mut endpoints: HashMap<String, DeploymentState> =
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);
}
}
}
Expand Down Expand Up @@ -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| {
Expand Down
38 changes: 26 additions & 12 deletions src/controllers/restatedeployment/reconcilers/knative.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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::{
Expand Down Expand Up @@ -640,14 +641,19 @@ async fn register_or_lookup_deployment(
config: &Configuration,
route: &Route,
) -> Result<String> {
// 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());
}

Expand All @@ -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)
Expand Down Expand Up @@ -743,7 +757,7 @@ pub async fn cleanup_old_configurations(
ctx: &Context,
rsd_uid: &str,
rsd: &RestateDeployment,
deployments: &std::collections::HashMap<String, bool>,
deployments: &std::collections::HashMap<String, DeploymentState>,
active_tag: Option<&str>,
) -> Result<(i32, Option<chrono::DateTime<chrono::Utc>>)> {
// Use reflector cache instead of API list() call
Expand Down Expand Up @@ -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;
Expand Down
7 changes: 4 additions & 3 deletions src/controllers/restatedeployment/reconcilers/replicaset.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -274,7 +275,7 @@ pub async fn cleanup_old_replicasets(
rs_api: &Api<ReplicaSet>,
rsd_uid: &str,
rsd: &RestateDeployment,
deployments: &HashMap<String, bool>,
deployments: &HashMap<String, DeploymentState>,
except_rs: Option<&str>,
) -> Result<(i32, Option<chrono::DateTime<chrono::Utc>>)> {
let replicasets_cell = std::cell::Cell::new(Vec::new());
Expand Down Expand Up @@ -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;
Expand Down
Loading