feat: Platform Access utilities

crates/stackable-operator/CHANGELOG.md

@@ -4,6 +4,18 @@ All notable changes to this project will be documented in this file.
 
 ## [Unreleased]
 
+### Added
+
+- Add utility for product agents ([#1290]):
+  - `TlsClientCredential`, enum for TLS-based platform access used for adding volumes and mounts.
+  - `Anonymous`, for platform access to products that don't authenticate the agent.
+  - `ManagementPolicy`, whether an agent adopts existing resources and deletes resources in the product.
+  - `Ready`/`Synced` conditions of resources managed by agents.
+  - Annotation marking resources owned by an agent.
+  - Labels and selectors for agents.
+
+[#1290]: https://github.com/stackabletech/operator-rs/pull/1290
+
 ## [0.119.0] - 2026-09-23
 
 ### Removed

crates/stackable-operator/src/commons/mod.rs

@@ -6,6 +6,7 @@ pub mod cluster_operation;
 pub mod networking;
 pub mod opa;
 pub mod pdb;
+pub mod platform_access;
 pub mod product_image_selection;
 pub mod random_secret_creation;
 pub mod rbac;

crates/stackable-operator/src/commons/platform_access/condition.rs

@@ -0,0 +1,144 @@
+use k8s_openapi::apimachinery::pkg::apis::meta::v1::{Condition, Time};
+use strum::Display;
+
+/// Condition types of resources an agent manages inside a product.
+#[derive(Clone, Copy, Debug, Display, Eq, PartialEq)]
+pub enum ManagedConditionType {
+    /// The resource exists in the product and is usable.
+    Ready,
+
+    /// The last reconciliation applied the spec to the resource in the product.
+    Synced,
+}
+
+/// The desired state of one condition, without the bookkeeping fields.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct ManagedCondition {
+    pub type_: ManagedConditionType,
+
+    pub status: bool,
+
+    /// A CamelCase reason like `Created` or `UnsupportedChange`.
+    pub reason: String,
+
+    pub message: String,
+}
+
+/// Merges the given conditions into the current ones.
+///
+/// `lastTransitionTime` only changes if the status of a condition changes. Conditions of other
+/// types are kept.
+pub fn update_conditions(
+    current: &[Condition],
+    updates: Vec<ManagedCondition>,
+    now: &Time,
+) -> Vec<Condition> {
+    let mut conditions = current.to_vec();
+
+    for update in updates {
+        let type_ = update.type_.to_string();
+        let status = if update.status { "True" } else { "False" }.to_owned();
+        let existing = conditions
+            .iter()
+            .position(|condition| condition.type_ == type_);
+
+        let last_transition_time = match existing {
+            Some(index) if conditions[index].status == status => {
+                conditions[index].last_transition_time.clone()
+            }
+            _ => now.clone(),
+        };
+
+        let condition = Condition {
+            last_transition_time,
+            message: update.message,
+            observed_generation: None,
+            reason: update.reason,
+            status,
+            type_,
+        };
+
+        match existing {
+            Some(index) => conditions[index] = condition,
+            None => conditions.push(condition),
+        }
+    }
+
+    conditions
+}
+
+#[cfg(test)]
+mod tests {
+    use k8s_openapi::jiff::Timestamp;
+
+    use super::*;
+
+    fn time(seconds: i64) -> Time {
+        Time(Timestamp::from_second(seconds).expect("the timestamp should be valid"))
+    }
+
+    fn ready(status: bool, reason: &str) -> ManagedCondition {
+        ManagedCondition {
+            type_: ManagedConditionType::Ready,
+            status,
+            reason: reason.to_owned(),
+            message: String::new(),
+        }
+    }
+
+    #[test]
+    fn add_missing_condition() {
+        let conditions = update_conditions(&[], vec![ready(true, "Created")], &time(10));
+
+        assert_eq!(1, conditions.len());
+        assert_eq!("Ready", conditions[0].type_);
+        assert_eq!("True", conditions[0].status);
+        assert_eq!("Created", conditions[0].reason);
+        assert_eq!(time(10), conditions[0].last_transition_time);
+    }
+
+    #[test]
+    fn keep_transition_time_if_status_is_unchanged() {
+        let current = update_conditions(&[], vec![ready(true, "Created")], &time(10));
+
+        let conditions = update_conditions(&current, vec![ready(true, "Updated")], &time(20));
+
+        assert_eq!("Updated", conditions[0].reason);
+        assert_eq!(time(10), conditions[0].last_transition_time);
+    }
+
+    #[test]
+    fn update_transition_time_if_status_changes() {
+        let current = update_conditions(&[], vec![ready(true, "Created")], &time(10));
+
+        let conditions =
+            update_conditions(&current, vec![ready(false, "KafkaRejected")], &time(20));
+
+        assert_eq!("False", conditions[0].status);
+        assert_eq!(time(20), conditions[0].last_transition_time);
+    }
+
+    #[test]
+    fn keep_conditions_of_other_types() {
+        let other = Condition {
+            last_transition_time: time(5),
+            message: String::new(),
+            observed_generation: None,
+            reason: "Other".to_owned(),
+            status: "True".to_owned(),
+            type_: "Other".to_owned(),
+        };
+
+        let conditions = update_conditions(
+            std::slice::from_ref(&other),
+            vec![ready(true, "Created")],
+            &time(10),
+        );
+
+        let types: Vec<_> = conditions
+            .iter()
+            .map(|condition| condition.type_.as_str())
+            .collect();
+        assert_eq!(vec!["Other", "Ready"], types);
+    }
+}

crates/stackable-operator/src/commons/platform_access/mod.rs

@@ -0,0 +1,5 @@
+//! Helpers for agents managing resources inside a product. The CRD fragments live in
+//! [`crate::crd::platform_access`].
+
+pub mod condition;
+pub mod ownership;

crates/stackable-operator/src/commons/platform_access/ownership.rs

@@ -0,0 +1,37 @@
+use kube::api::ObjectMeta;
+
+/// Annotation an agent sets on an object once it created or adopted the resource in the product.
+///
+/// The value is the RFC 3339 timestamp of that moment. Agents only delete resources they own.
+pub const OWNED_SINCE_ANNOTATION: &str = "stackable.tech/owned-since";
+
+/// Whether the agent owns the resource in the product that the object with this metadata manages.
+pub fn is_owned(metadata: &ObjectMeta) -> bool {
+    metadata
+        .annotations
+        .as_ref()
+        .is_some_and(|annotations| annotations.contains_key(OWNED_SINCE_ANNOTATION))
+}
+
+#[cfg(test)]
+mod tests {
+    use std::collections::BTreeMap;
+
+    use rstest::rstest;
+
+    use super::*;
+
+    #[rstest]
+    #[case::no_annotations(None, false)]
+    #[case::other_annotation(Some(("stackable.tech/other", "x")), false)]
+    #[case::owned(Some((OWNED_SINCE_ANNOTATION, "2026-10-06T09:12:44Z")), true)]
+    fn ownership(#[case] annotation: Option<(&str, &str)>, #[case] expected: bool) {
+        let metadata = ObjectMeta {
+            annotations: annotation
+                .map(|(key, value)| BTreeMap::from([(key.to_owned(), value.to_owned())])),
+            ..ObjectMeta::default()
+        };
+
+        assert_eq!(expected, is_owned(&metadata));
+    }
+}

crates/stackable-operator/src/crd/mod.rs

@@ -7,6 +7,7 @@ use serde::{Deserialize, Serialize};
 pub mod authentication;
 pub mod git_sync;
 pub mod listener;
+pub mod platform_access;
 pub mod s3;
 pub mod scaler;
 

crates/stackable-operator/src/crd/platform_access/anonymous/mod.rs

@@ -0,0 +1,32 @@
+use schemars::JsonSchema;
+use serde::{Deserialize, Serialize};
+
+use crate::versioned::versioned;
+
+#[versioned(version(name = "v1alpha1"))]
+pub mod versioned {
+    /// No credential: the product does not authenticate the agent.
+    #[derive(Clone, Debug, Default, Deserialize, Eq, JsonSchema, PartialEq, Serialize)]
+    pub struct Anonymous {}
+}
+
+#[cfg(test)]
+mod tests {
+    use super::v1alpha1::Anonymous;
+
+    #[test]
+    fn deserialize_empty_object() {
+        let anonymous: Anonymous =
+            serde_json::from_str("{}").expect("an empty object should be valid");
+
+        assert_eq!(Anonymous {}, anonymous);
+    }
+
+    #[test]
+    fn schema_is_an_object() {
+        let schema = serde_json::to_value(schemars::schema_for!(Anonymous))
+            .expect("the schema should be serializable");
+
+        assert_eq!("object", schema["type"]);
+    }
+}

crates/stackable-operator/src/crd/platform_access/management/mod.rs

@@ -0,0 +1,44 @@
+use schemars::JsonSchema;
+use serde::{Deserialize, Serialize};
+
+use crate::versioned::versioned;
+
+mod v1alpha1_impl;
+
+#[versioned(version(name = "v1alpha1"))]
+pub mod versioned {
+    /// What an agent may do with a resource inside the product beyond applying its spec.
+    #[derive(Clone, Debug, Default, Deserialize, Eq, JsonSchema, PartialEq, Serialize)]
+    #[serde(default, rename_all = "camelCase")]
+    pub struct ManagementPolicy {
+        /// Whether a resource that already exists in the product, but was not created by the
+        /// agent, is taken over. `Refuse` by default.
+        pub adoption: AdoptionPolicy,
+
+        /// What happens to the resource in the product when this object is deleted. `Retain` by
+        /// default.
+        pub deletion: DeletionPolicy,
+    }
+
+    /// Whether an agent takes over a resource that already exists in the product.
+    #[derive(Clone, Copy, Debug, Default, Deserialize, Eq, JsonSchema, PartialEq, Serialize)]
+    pub enum AdoptionPolicy {
+        /// Report the existing resource as a conflict and leave it untouched.
+        #[default]
+        Refuse,
+
+        /// Take over the existing resource and apply the spec to it.
+        Adopt,
+    }
+
+    /// What happens to a resource in the product when the object managing it is deleted.
+    #[derive(Clone, Copy, Debug, Default, Deserialize, Eq, JsonSchema, PartialEq, Serialize)]
+    pub enum DeletionPolicy {
+        /// Keep the resource in the product.
+        #[default]
+        Retain,
+
+        /// Delete the resource in the product, if the agent created or adopted it.
+        Delete,
+    }
+}

crates/stackable-operator/src/crd/platform_access/management/v1alpha1_impl.rs

@@ -0,0 +1,44 @@
+use crate::crd::platform_access::management::v1alpha1::{
+    AdoptionPolicy, DeletionPolicy, ManagementPolicy,
+};
+
+impl ManagementPolicy {
+    /// Whether the agent takes over a resource that already exists in the product.
+    pub fn adopts_existing(&self) -> bool {
+        self.adoption == AdoptionPolicy::Adopt
+    }
+
+    /// Whether the agent deletes the resource in the product when this object is deleted.
+    pub fn deletes_on_cleanup(&self) -> bool {
+        self.deletion == DeletionPolicy::Delete
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use rstest::rstest;
+
+    use super::*;
+
+    #[rstest]
+    #[case::defaults("{}", false, false)]
+    #[case::adopt(r#"{"adoption": "Adopt"}"#, true, false)]
+    #[case::delete(r#"{"deletion": "Delete"}"#, false, true)]
+    #[case::explicit_defaults(r#"{"adoption": "Refuse", "deletion": "Retain"}"#, false, false)]
+    fn deserialize_policy(
+        #[case] input: &str,
+        #[case] adopts_existing: bool,
+        #[case] deletes_on_cleanup: bool,
+    ) {
+        let policy: ManagementPolicy =
+            serde_json::from_str(input).expect("the policy should be valid");
+
+        assert_eq!(adopts_existing, policy.adopts_existing());
+        assert_eq!(deletes_on_cleanup, policy.deletes_on_cleanup());
+    }
+
+    #[test]
+    fn reject_unknown_policy() {
+        assert!(serde_json::from_str::<ManagementPolicy>(r#"{"deletion": "Orphan"}"#).is_err());
+    }
+}

crates/stackable-operator/src/crd/platform_access/mod.rs

@@ -0,0 +1,9 @@
+//! CRD fragments for resources managed by product agents, and for the credentials the agents use.
+
+pub mod anonymous;
+pub mod management;
+pub mod tls;
+
+pub mod v1alpha1 {
+    pub use super::{anonymous::v1alpha1::*, management::v1alpha1::*, tls::v1alpha1::*};
+}