Support registering at a subpath

In addition, stop looking for active deployments by their ID; instead
use the deployment id annotation which we store against a replicaset.

Author: jackkleeman

Cargo.lock

@@ -77,6 +77,7 @@ reqwest = { version = "0.12.15", default-features = false, features = [
     "rustls-tls",
 ] }
 fnv = "1.0.7"
+url = { version = "2.5.4", features = ["serde"] }
 
 [dev-dependencies]
 assert-json-diff = "2.0.2"

Cargo.toml

@@ -363,6 +363,7 @@ This field contains Restate-specific configuration.
 | Field | Type | Description |
 |---|---|---|
 | `register` | `object` | **Required**. The location of the Restate Admin API to register this deployment against. See details below. |
+| `servicePath` | `string` | Optional path to append to the Service url when registering with Restate. |
 
 The `register` field must specify exactly one of `cluster`, `service`, or `url`.
 

README.md

@@ -57,6 +57,10 @@ class Spec {
 class Restate {
   /// The location of the Restate Admin API to register this deployment against
   register: Register
+
+  /// Optional path to append to the Service url when registering with Restate. If not provided, the
+  /// service will be registered at the root path "/".
+  servicePath: String?
 }
 
 /// The location of the Restate Admin API to register this deployment against

crd/RestateDeployment.pkl

@@ -104,6 +104,10 @@ spec:
                         description: A url of the restate admin endpoint against which to register the deployment Exactly one of `cluster`, `service` or `url` must be specified
                         type: string
                     type: object
+                  servicePath:
+                    description: Optional path to append to the Service url when registering with Restate. If not provided, the service will be registered at the root path "/".
+                    nullable: true
+                    type: string
                 required:
                 - register
                 type: object

crd/restatedeployments.yaml

@@ -3,6 +3,7 @@ use std::sync::Arc;
 use chrono::{DateTime, Utc};
 use serde::Serialize;
 use tokio::sync::RwLock;
+use url::Url;
 
 pub mod restatecluster;
 pub mod restatedeployment;
@@ -83,3 +84,20 @@ impl State {
         }
     }
 }
+
+pub fn service_url(
+    service_name: &str,
+    service_namespace: &str,
+    port: i32,
+    path: Option<&str>,
+) -> Result<Url, url::ParseError> {
+    let mut url = Url::parse(&format!(
+        "http://{service_name}.{service_namespace}.svc.cluster.local:{port}",
+    ))?;
+
+    if let Some(path) = path {
+        url.set_path(path)
+    }
+
+    Ok(url)
+}

src/controllers/mod.rs

@@ -27,8 +27,9 @@ use serde::Deserialize;
 use serde_json::json;
 use tokio::sync::RwLock;
 use tracing::*;
+use url::Url;
 
-use crate::controllers::{Diagnostics, State};
+use crate::controllers::{service_url, Diagnostics, State};
 use crate::metrics::Metrics;
 use crate::resources::restateclusters::RestateCluster;
 use crate::resources::restatedeployments::{
@@ -151,7 +152,7 @@ impl RestateDeployment {
         let rs_api = Api::<ReplicaSet>::namespaced(ctx.client.clone(), namespace);
         let svc_api = Api::<Service>::namespaced(ctx.client.clone(), namespace);
 
-        let admin_endpoint = self.spec.restate.register.url()?;
+        let admin_endpoint = self.spec.restate.register.admin_url()?;
 
         let pod_template_annotation = reconcilers::replicaset::pod_template_annotation(self);
 
@@ -277,15 +278,28 @@ impl RestateDeployment {
         )
         .await?;
 
-        let service_endpoint =
-            format!("http://{versioned_name}.{namespace}.svc.cluster.local:9080/");
+        let service_endpoint = service_url(
+            &versioned_name,
+            namespace,
+            9080,
+            self.spec.restate.service_path.as_deref(),
+        )?;
 
-        let mut endpoints = self
-            .list_endpoints(&ctx.http_client, &admin_endpoint)
+        let mut deployments = self
+            .list_deployments(&ctx.http_client, &admin_endpoint)
             .await?;
 
-        // if the endpoint is not active, register it
-        if !endpoints.get(&service_endpoint).cloned().unwrap_or(false) {
+        let existing_deployment_id = replicaset
+            .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()
+        }) {
             if let Some(cluster_name) = &self.spec.restate.register.cluster {
                 // wait for the cluster to be ready before registering to it
                 validate_cluster_status(rsc_api, cluster_name).await?;
@@ -302,7 +316,8 @@ impl RestateDeployment {
             )
             .await?;
             // if registration succeeded, treat this as an active endpoint
-            endpoints.insert(service_endpoint.clone(), true);
+            // if we fail after this point we will re-register and should get the same deployment id
+            deployments.insert(deployment_id.clone(), true);
 
             debug!("Updating deployment-id annotation of ReplicaSet/Service {versioned_name} in namespace {namespace}");
 
@@ -338,7 +353,7 @@ impl RestateDeployment {
             &ctx.http_client,
             &admin_endpoint,
             self,
-            &endpoints,
+            &deployments,
         )
         .await?;
 
@@ -475,8 +490,8 @@ impl RestateDeployment {
     /// Register a service version with the Restate cluster
     async fn register_service_with_restate(
         client: &reqwest::Client,
-        admin_endpoint: &str,
-        service_endpoint: &str,
+        admin_endpoint: &Url,
+        service_endpoint: &Url,
     ) -> Result<String> {
         debug!("Registering endpoint '{service_endpoint}' to Restate at '{admin_endpoint}'",);
 
@@ -486,7 +501,7 @@ impl RestateDeployment {
         }
 
         let resp: DeploymentResponse = client
-            .post(format!("{}/deployments", admin_endpoint))
+            .post(admin_endpoint.join("/deployments")?)
             .json(&serde_json::json!({
                 "uri": service_endpoint,
             }))
@@ -502,14 +517,14 @@ impl RestateDeployment {
         Ok(resp.id)
     }
 
-    async fn list_endpoints(
+    pub async fn list_deployments(
         &self,
         http_client: &reqwest::Client,
-        admin_endpoint: &str,
+        admin_endpoint: &Url,
     ) -> Result<HashMap<String, bool>> {
-        // This query finds endpoint urls,  noting those that are the latest for a particular service, or have an active invocation
+        // This query finds deployments, noting those that are the latest for a particular service, or have an active invocation
         let sql_query = r#"
-            SELECT d.endpoint, (s.name IS NOT NULL OR i.id IS NOT NULL) as active
+            SELECT d.id as deployment_id, (s.name IS NOT NULL OR i.id IS NOT NULL) as active
             FROM sys_deployment d
             LEFT JOIN sys_service s ON (d.id = s.deployment_id)
             LEFT JOIN sys_invocation_status i ON (d.id = i.pinned_deployment_id);
@@ -522,12 +537,12 @@ impl RestateDeployment {
 
         #[derive(Deserialize)]
         struct DeploymentQueryResultRow {
-            endpoint: String,
+            deployment_id: String,
             active: bool,
         }
 
         let response: DeploymentQueryResult = http_client
-            .post(format!("{}/query", admin_endpoint))
+            .post(admin_endpoint.join("/query")?)
             .header(reqwest::header::ACCEPT, "application/json")
             .json(&serde_json::json!({
                 "query": sql_query
@@ -544,10 +559,10 @@ impl RestateDeployment {
         let mut endpoints = HashMap::with_capacity(response.rows.len());
 
         for row in response.rows {
-            match endpoints.entry(row.endpoint) {
+            match endpoints.entry(row.deployment_id) {
                 std::collections::hash_map::Entry::Occupied(mut entry) => {
-                    // technically there can be multiple deployments with the same endpoint url (via different headers, etc)
-                    // we treat the endpoint as active if any such deployment is active
+                    // 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);
                     }
@@ -595,10 +610,10 @@ impl RestateDeployment {
             };
         }
 
-        let admin_endpoint = self.spec.restate.register.url()?;
+        let admin_endpoint = self.spec.restate.register.admin_url()?;
 
-        let endpoints = self
-            .list_endpoints(&ctx.http_client, &admin_endpoint)
+        let deployments = self
+            .list_deployments(&ctx.http_client, &admin_endpoint)
             .await?;
 
         let (active_count, next_removal) = reconcilers::replicaset::cleanup_old_replicasets(
@@ -608,7 +623,7 @@ impl RestateDeployment {
             &ctx.http_client,
             &admin_endpoint,
             self,
-            &endpoints,
+            &deployments,
         )
         .await?;
 

src/controllers/restatedeployment/controller.rs

@@ -11,6 +11,7 @@ use kube::runtime::reflector::Store;
 use kube::{Resource, ResourceExt};
 use serde_json::json;
 use tracing::*;
+use url::Url;
 
 use crate::controllers::restatedeployment::controller::{
     APP_MANAGED_BY_LABEL, OWNED_BY_LABEL, RESTATE_DEPLOYMENT_ID_ANNOTATION,
@@ -114,6 +115,11 @@ pub fn generate_pod_template_hash(rsd: &RestateDeployment, pod_template: &str) -
         hasher.write(cluster.as_bytes());
     }
 
+    // if you change the path, it creates a new deployment, which means we want a new replicaset too to keep things 1:1
+    if let Some(service_path) = rsd.spec.restate.service_path.as_deref() {
+        hasher.write(service_path.as_bytes());
+    }
+
     if let Some(collision_count) = rsd.status.as_ref().and_then(|s| s.collision_count) {
         hasher.write(&collision_count.to_be_bytes());
     }
@@ -150,9 +156,9 @@ pub async fn cleanup_old_replicasets(
     rs_api: &Api<ReplicaSet>,
     replicasets_store: &Store<ReplicaSet>,
     http_client: &reqwest::Client,
-    admin_endpoint: &str,
+    admin_endpoint: &Url,
     rsd: &RestateDeployment,
-    endpoints: &HashMap<String, bool>,
+    deployments: &HashMap<String, bool>,
 ) -> Result<(i32, Option<chrono::DateTime<chrono::Utc>>)> {
     let owner_name = rsd.name_any();
 
@@ -191,14 +197,16 @@ pub async fn cleanup_old_replicasets(
 
     for rs in replicasets {
         let rs_name = rs.name_any();
-        let service_endpoint = format!("http://{}.{}.svc.cluster.local:9080/", rs_name, namespace);
 
-        // Skip active versions
-        let endpoint = endpoints.get(&service_endpoint).cloned();
-        let endpoint_exists = endpoint.is_some();
-        let endpoint_active = endpoint.unwrap_or(false);
+        let rs_deployment_id = rs.annotations().get(RESTATE_DEPLOYMENT_ID_ANNOTATION);
 
-        if endpoint_active {
+        // Skip active deployments
+        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);
+
+        if deployment_active {
             active_count += 1;
 
             if rs
@@ -251,7 +259,7 @@ pub async fn cleanup_old_replicasets(
         match (
             current_remove_at,
             current_remove_at_in_past,
-            endpoint_exists,
+            deployment_exists,
         ) {
             (_, true, _) | (_, _, false) => {
                 // we are past the remove at time, or the endpoint was removed by other means; can now scale it down
@@ -290,24 +298,22 @@ pub async fn cleanup_old_replicasets(
                     continue;
                 }
 
-                if endpoint_exists {
-                    if let Some(deployment_id) =
-                        rs.annotations().get(RESTATE_DEPLOYMENT_ID_ANNOTATION)
-                    {
-                        debug!("Force-deleting Restate deployment {deployment_id} as its associated with old ReplicaSet {rs_name} in namespace {namespace}");
-
-                        let resp = http_client
-                            .delete(format!(
-                                "{admin_endpoint}/deployments/{deployment_id}?force=true"
-                            ))
-                            .send()
-                            .await
-                            .map_err(Error::AdminCallFailed)?;
-
-                        // for idempotency we have to allow 404
-                        if resp.status() != reqwest::StatusCode::NOT_FOUND {
-                            let _ = resp.error_for_status().map_err(Error::AdminCallFailed)?;
-                        }
+                if deployment_exists {
+                    let rs_deployment_id = rs_deployment_id.unwrap();
+
+                    debug!("Force-deleting Restate deployment {rs_deployment_id} as its associated with old ReplicaSet {rs_name} in namespace {namespace}");
+
+                    let resp = http_client
+                        .delete(format!(
+                            "{admin_endpoint}/deployments/{rs_deployment_id}?force=true"
+                        ))
+                        .send()
+                        .await
+                        .map_err(Error::AdminCallFailed)?;
+
+                    // for idempotency we have to allow 404
+                    if resp.status() != reqwest::StatusCode::NOT_FOUND {
+                        let _ = resp.error_for_status().map_err(Error::AdminCallFailed)?;
                     }
                 }
 

src/controllers/restatedeployment/reconcilers/replicaset.rs

@@ -51,6 +51,9 @@ pub enum Error {
 
     #[error("This RestateDeployment is backing recently-active versions in Restate. It will be removed after the drain delay period.")]
     DeploymentDraining { requeue_after: Option<Duration> },
+
+    #[error(transparent)]
+    InvalidUrl(#[from] url::ParseError),
 }
 
 pub type Result<T, E = Error> = std::result::Result<T, E>;
@@ -70,6 +73,7 @@ impl Error {
             Error::HashCollision => "HashCollision",
             Error::DeploymentInUse => "DeploymentInUse",
             Error::DeploymentDraining { .. } => "DeploymentDraining",
+            Error::InvalidUrl { .. } => "InvalidUrl",
         }
     }
 }