feat: Management Access utilities

CHANGELOG.md

@@ -5,6 +5,7 @@ Please see the relevant crate changelogs:
 - [config-utils](./crates/config-utils/CHANGELOG.md)
 - [containerdebug](./crates/containerdebug/CHANGELOG.md)
 - [k8s-version](./crates/k8s-version/CHANGELOG.md)
+- [stackable-agent](./crates/stackable-agent/CHANGELOG.md)
 - [stackable-certs](./crates/stackable-certs/CHANGELOG.md)
 - [stackable-operator](./crates/stackable-operator/CHANGELOG.md)
 - [stackable-operator-derive](./crates/stackable-operator-derive/CHANGELOG.md)

Cargo.toml

@@ -120,6 +120,13 @@ zeroize = "1.9.1"
 # Enable all pedantic lints (with lower priority so individual lints can override)
 pedantic = { level = "deny", priority = -1 }
 
+# NOTE (@Techassi): Since Rust 1.99.0 this lint is triggered by fields which use `#[darling(multiple)]`.
+# This seems to be a false-positive as far as I can see. Allowing this lint at the container or
+# field level did not work. That's why it is now allowed globally.
+# TODO (@Techassi): Remove this once the fix (see https://github.com/TedDriggs/darling/pull/441) is
+# released.
+redundant_field_names = "allow"
+
 # Pedantic lints we don't enforce (yet)
 doc_markdown = "allow"
 missing_errors_doc = "allow"

crates/stackable-agent/CHANGELOG.md

@@ -0,0 +1,21 @@
+# Changelog
+
+All notable changes to this project will be documented in this file.
+
+## [Unreleased]
+
+### Added
+
+- `DeletionPolicy`, whether an agent deletes a resource in the product ([#1290]).
+- `ResourceOperation` with `reconciliationPaused`, which stops an agent from changing a resource
+  in the product ([#1290]).
+- `Ready`/`Synced` conditions of resources managed by agents (`condition`), as `metav1.Condition`
+  with `True`/`False`/`Unknown` and `observedGeneration` ([#1290]).
+
+## [0.0.1]
+
+### Added
+
+- Add the `stackable-agent` crate as a workspace member ([#1290]).
+
+[#1290]: https://github.com/stackabletech/operator-rs/pull/1290

crates/stackable-agent/Cargo.toml

@@ -0,0 +1,23 @@
+[package]
+name = "stackable-agent"
+description = "Building blocks for agents managing resources inside products of the Stackable Data Platform"
+version = "0.0.1"
+authors.workspace = true
+license.workspace = true
+edition.workspace = true
+repository.workspace = true
+
+[dependencies]
+stackable-versioned = { path = "../stackable-versioned" }
+
+k8s-openapi.workspace = true
+schemars.workspace = true
+serde.workspace = true
+strum.workspace = true
+
+[dev-dependencies]
+rstest.workspace = true
+serde_json.workspace = true
+
+[lints]
+workspace = true

crates/stackable-agent/src/condition.rs

@@ -0,0 +1,186 @@
+//! Conditions of resources an agent manages inside a product, as `metav1.Condition`.
+
+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 status of a condition.
+#[derive(Clone, Copy, Debug, Display, Eq, PartialEq)]
+pub enum ConditionStatus {
+    True,
+    False,
+    Unknown,
+}
+
+/// The desired state of one condition, without the bookkeeping fields.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct ManagedCondition {
+    pub type_: ManagedConditionType,
+
+    /// `Unknown` if the agent could not determine the state, e.g. because the product is not
+    /// reachable.
+    pub status: ConditionStatus,
+
+    /// A CamelCase reason like `Available` 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. `observed_generation` is the `metadata.generation` of the object the conditions
+/// were determined for.
+pub fn update_conditions(
+    current: &[Condition],
+    updates: Vec<ManagedCondition>,
+    now: &Time,
+    observed_generation: Option<i64>,
+) -> Vec<Condition> {
+    let mut conditions = current.to_vec();
+
+    for update in updates {
+        let type_ = update.type_.to_string();
+        let status = update.status.to_string();
+        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,
+            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: ConditionStatus, 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(ConditionStatus::True, "Created")],
+            &time(10),
+            Some(3),
+        );
+
+        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);
+        assert_eq!(Some(3), conditions[0].observed_generation);
+    }
+
+    #[test]
+    fn keep_transition_time_if_status_is_unchanged() {
+        let current = update_conditions(
+            &[],
+            vec![ready(ConditionStatus::True, "Created")],
+            &time(10),
+            Some(1),
+        );
+
+        let conditions = update_conditions(
+            &current,
+            vec![ready(ConditionStatus::True, "Updated")],
+            &time(20),
+            Some(2),
+        );
+
+        assert_eq!("Updated", conditions[0].reason);
+        assert_eq!(time(10), conditions[0].last_transition_time);
+        assert_eq!(Some(2), conditions[0].observed_generation);
+    }
+
+    #[test]
+    fn update_transition_time_if_status_changes() {
+        let current = update_conditions(
+            &[],
+            vec![ready(ConditionStatus::True, "Created")],
+            &time(10),
+            Some(1),
+        );
+
+        let conditions = update_conditions(
+            &current,
+            vec![ready(ConditionStatus::Unknown, "ConnectionFailed")],
+            &time(20),
+            Some(1),
+        );
+
+        assert_eq!("Unknown", 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(ConditionStatus::True, "Created")],
+            &time(10),
+            Some(1),
+        );
+
+        let types: Vec<_> = conditions
+            .iter()
+            .map(|condition| condition.type_.as_str())
+            .collect();
+        assert_eq!(vec!["Other", "Ready"], types);
+        assert_eq!(None, conditions[0].observed_generation);
+    }
+}

crates/stackable-agent/src/deletion.rs

@@ -0,0 +1,44 @@
+use schemars::JsonSchema;
+use serde::{Deserialize, Serialize};
+use stackable_versioned::versioned;
+
+#[versioned(version(name = "v1alpha1"))]
+pub mod versioned {
+    /// 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.
+        Delete,
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use rstest::rstest;
+
+    use super::v1alpha1::DeletionPolicy;
+
+    #[rstest]
+    #[case::retain(r#""Retain""#, DeletionPolicy::Retain)]
+    #[case::delete(r#""Delete""#, DeletionPolicy::Delete)]
+    fn deserialize_policy(#[case] input: &str, #[case] expected: DeletionPolicy) {
+        let policy: DeletionPolicy =
+            serde_json::from_str(input).expect("the policy should be valid");
+
+        assert_eq!(expected, policy);
+    }
+
+    #[test]
+    fn retain_by_default() {
+        assert_eq!(DeletionPolicy::Retain, DeletionPolicy::default());
+    }
+
+    #[test]
+    fn reject_unknown_policy() {
+        assert!(serde_json::from_str::<DeletionPolicy>(r#""Orphan""#).is_err());
+    }
+}

crates/stackable-agent/src/lib.rs

@@ -0,0 +1,5 @@
+//! Building blocks for agents managing resources inside products, like topics in Apache Kafka.
+
+pub mod condition;
+pub mod deletion;
+pub mod operation;

crates/stackable-agent/src/operation.rs

@@ -0,0 +1,38 @@
+use schemars::JsonSchema;
+use serde::{Deserialize, Serialize};
+use stackable_versioned::versioned;
+
+#[versioned(version(name = "v1alpha1"))]
+pub mod versioned {
+    /// Operations on this resource in the product, independent of the `clusterOperation` of the
+    /// product cluster.
+    #[derive(Clone, Debug, Default, Deserialize, Eq, JsonSchema, PartialEq, Serialize)]
+    #[serde(rename_all = "camelCase")]
+    pub struct ResourceOperation {
+        /// Stops the agent from changing the resource in the product. The status is still updated.
+        #[serde(default)]
+        pub reconciliation_paused: bool,
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::v1alpha1::ResourceOperation;
+
+    #[test]
+    fn not_paused_by_default() {
+        let operation: ResourceOperation =
+            serde_json::from_str("{}").expect("the operation should be valid");
+
+        assert!(!operation.reconciliation_paused);
+    }
+
+    #[test]
+    fn deserialize_paused() {
+        let operation: ResourceOperation =
+            serde_json::from_str(r#"{"reconciliationPaused": true}"#)
+                .expect("the operation should be valid");
+
+        assert!(operation.reconciliation_paused);
+    }
+}