refactor: Extract apply and update status steps

CHANGELOG.md

@@ -11,10 +11,13 @@ All notable changes to this project will be documented in this file.
 - Bump `stackable-operator` to 0.114.0 ([#867]).
 - The RBAC ServiceAccount and RoleBinding are now built with the operator-rs `v2::rbac`
   functions and carry the full set of recommended labels ([#861]).
+- The reconciler now applies resources and derives the cluster status in discrete
+  apply and update_status steps for the `opa_controller` ([#872]).
 
 [#852]: https://github.com/stackabletech/opa-operator/pull/852
 [#861]: https://github.com/stackabletech/opa-operator/pull/861
 [#867]: https://github.com/stackabletech/opa-operator/pull/867
+[#872]: https://github.com/stackabletech/opa-operator/pull/872
 
 ## [26.7.0] - 2026-07-21
 

rust/operator-binary/src/controller/apply.rs

@@ -0,0 +1,173 @@
+//! The apply step in the OpaCluster controller.
+
+use std::marker::PhantomData;
+
+use serde_json::json;
+use snafu::{ResultExt, Snafu};
+use stackable_operator::{
+    client::Client,
+    cluster_resources::{ClusterResource, ClusterResourceApplyStrategy, ClusterResources},
+    deep_merger::ObjectOverrides,
+    k8s_openapi::api::apps::v1::DaemonSet,
+    kube::ResourceExt,
+    v2::cluster_resources::cluster_resources_new,
+};
+use strum::{EnumDiscriminants, IntoStaticStr};
+
+use crate::controller::{
+    Applied, KubernetesResources, Prepared, ValidatedCluster, controller_name, operator_name,
+    product_name,
+};
+
+#[derive(Snafu, Debug, EnumDiscriminants)]
+#[strum_discriminants(derive(IntoStaticStr))]
+pub enum Error {
+    #[snafu(display("failed to apply Kubernetes resource"))]
+    ApplyResource {
+        source: stackable_operator::cluster_resources::Error,
+    },
+
+    #[snafu(display("failed to apply legacy field-manager patch for DaemonSet {name}"))]
+    ApplyPatchDaemonSet {
+        source: stackable_operator::client::Error,
+        name: String,
+    },
+
+    #[snafu(display("failed to delete orphaned resources"))]
+    DeleteOrphanedResources {
+        source: stackable_operator::cluster_resources::Error,
+    },
+}
+
+type Result<T, E = Error> = std::result::Result<T, E>;
+
+/// Applier for the Kubernetes resource specifications produced by this controller.
+///
+/// The implementation is not tied to this controller and could theoretically be moved to
+/// stackable_operator if [`KubernetesResources`] would contain all possible resource types.
+pub struct Applier<'a> {
+    client: &'a Client,
+    cluster_resources: ClusterResources<'a>,
+}
+
+impl<'a> Applier<'a> {
+    pub fn new(
+        client: &'a Client,
+        cluster: &ValidatedCluster,
+        apply_strategy: ClusterResourceApplyStrategy,
+        object_overrides: &'a ObjectOverrides,
+    ) -> Applier<'a> {
+        let cluster_resources = cluster_resources_new(
+            &product_name(),
+            &operator_name(),
+            &controller_name(),
+            &cluster.name,
+            &cluster.namespace,
+            &cluster.uid,
+            apply_strategy,
+            object_overrides,
+        );
+
+        Applier {
+            client,
+            cluster_resources,
+        }
+    }
+
+    /// Applies the given Kubernetes resources and marks them as applied.
+    pub async fn apply(
+        mut self,
+        resources: KubernetesResources<Prepared>,
+    ) -> Result<KubernetesResources<Applied>> {
+        // Destructured without `..`, so adding a field to [`KubernetesResources`] fails to
+        // compile here instead of silently never being applied.
+        let KubernetesResources {
+            daemon_sets,
+            services,
+            config_maps,
+            service_accounts,
+            role_bindings,
+            status: _,
+        } = resources;
+
+        // Apply order is: DaemonSets last (a changed mounted ConfigMap must exist first, else the
+        // Pods restart a second time -- commons-operator#111). The ServiceAccount comes first
+        // because the Pods reference it at creation time.
+        let service_accounts = self.add_resources(service_accounts).await?;
+        let role_bindings = self.add_resources(role_bindings).await?;
+        let services = self.add_resources(services).await?;
+        let config_maps = self.add_resources(config_maps).await?;
+        let daemon_sets = self.add_resources(daemon_sets).await?;
+
+        self.remove_legacy_field_manager_scope(&daemon_sets).await?;
+
+        self.cluster_resources
+            .delete_orphaned_resources(self.client)
+            .await
+            .context(DeleteOrphanedResourcesSnafu)?;
+
+        Ok(KubernetesResources {
+            daemon_sets,
+            services,
+            config_maps,
+            service_accounts,
+            role_bindings,
+            status: PhantomData,
+        })
+    }
+
+    async fn add_resources<T: ClusterResource + Sync>(
+        &mut self,
+        resources: Vec<T>,
+    ) -> Result<Vec<T>> {
+        let mut applied_resources = vec![];
+
+        for resource in resources {
+            let applied_resource = self
+                .cluster_resources
+                .add(self.client, resource)
+                .await
+                .context(ApplyResourceSnafu)?;
+            applied_resources.push(applied_resource);
+        }
+
+        Ok(applied_resources)
+    }
+
+    /// Relinquishes the fields still owned by the historical field manager scope "opacluster".
+    ///
+    /// A previous version of opa-operator used the field manager scope "opacluster" to write out a
+    /// DaemonSet with the bundle-builder container called "opa-bundle-builder". During
+    /// https://github.com/stackabletech/opa-operator/pull/420 it was renamed to "bundle-builder".
+    /// As we are now using the field manager scope "opa.stackable.tech_opacluster", our old changes
+    /// (with the old container) would stay valid. We have to use the old field manager scope and
+    /// post an empty patch to get rid of it.
+    /// https://github.com/stackabletech/issues/issues/390 will implement a proper fix, which also
+    /// covers Services and ConfigMaps. For details see
+    /// https://github.com/stackabletech/opa-operator/issues/444.
+    async fn remove_legacy_field_manager_scope(&self, daemon_sets: &[DaemonSet]) -> Result<()> {
+        for daemon_set in daemon_sets {
+            tracing::trace!(
+                "Removing old field manager scope \"opacluster\" of DaemonSet {daemonset_name} to remove the \"opa-bundle-builder\" container. \
+                See https://github.com/stackabletech/opa-operator/issues/444 and https://github.com/stackabletech/issues/issues/390 for details.",
+                daemonset_name = daemon_set.name_any()
+            );
+
+            self.client
+                .apply_patch(
+                    "opacluster",
+                    daemon_set,
+                    // We can hardcode this here, as https://github.com/stackabletech/issues/issues/390
+                    // will solve the general problem and we always have created DaemonSets using
+                    // the "apps/v1" version.
+                    json!({"apiVersion": "apps/v1", "kind": "DaemonSet"}),
+                )
+                .await
+                .context(ApplyPatchDaemonSetSnafu {
+                    name: daemon_set.name_any(),
+                })?;
+        }
+
+        Ok(())
+    }
+}

rust/operator-binary/src/controller/build.rs

@@ -1,7 +1,7 @@
 //! Build steps that turn the [`ValidatedCluster`](super::ValidatedCluster) into
 //! Kubernetes resource specifications.
 
-use std::str::FromStr;
+use std::{marker::PhantomData, str::FromStr};
 
 use snafu::{ResultExt, Snafu};
 use stackable_operator::{
@@ -11,7 +11,7 @@ use stackable_operator::{
 };
 
 use crate::controller::{
-    KubernetesResources, RoleGroupName, ValidatedCluster,
+    KubernetesResources, Prepared, RoleGroupName, ValidatedCluster,
     build::resource::{
         config_map::build_rolegroup_config_map,
         daemonset::build_server_rolegroup_daemonset,
@@ -59,7 +59,7 @@ pub fn build(
     opa_bundle_builder_image: &str,
     user_info_fetcher_image: &str,
     cluster_info: &KubernetesClusterInfo,
-) -> Result<KubernetesResources, Error> {
+) -> Result<KubernetesResources<Prepared>, Error> {
     let mut daemon_sets = vec![];
     let mut services = vec![];
     let mut config_maps = vec![];
@@ -103,6 +103,7 @@ pub fn build(
         config_maps,
         service_accounts: vec![build_service_account(cluster)],
         role_bindings: vec![build_role_binding(cluster)],
+        status: PhantomData,
     })
 }
 

rust/operator-binary/src/controller/mod.rs

@@ -1,7 +1,7 @@
 //! Controller-level vocabulary: the [`ValidatedCluster`] type and the `build` / `validate`
 //! sub-modules.
 
-use std::{collections::BTreeMap, str::FromStr};
+use std::{collections::BTreeMap, marker::PhantomData, str::FromStr};
 
 // Re-exported so the rest of the controller refers to `crate::controller::RoleGroupName`.
 pub use stackable_operator::v2::types::operator::RoleGroupName;
@@ -41,7 +41,9 @@ use crate::{
     opa_controller::OPA_CONTROLLER_NAME,
 };
 
+pub mod apply;
 pub mod build;
+pub mod update_status;
 pub mod validate;
 
 /// The validated [`v1alpha2::OpaCluster`].
@@ -235,18 +237,29 @@ impl KubeResource for ValidatedCluster {
     }
 }
 
+/// Marker for prepared Kubernetes resources which are not applied yet.
+pub struct Prepared;
+
+/// Marker for applied Kubernetes resources.
+pub struct Applied;
+
 /// Every Kubernetes resource produced by the [`build`](build::build) step.
 ///
 /// OPA runs as a `DaemonSet` (one Pod per node), so there are no `StatefulSet`s, PDBs or
 /// `Listener`s. `services` holds the role-level `Service` and the per-role-group headless and
 /// metrics `Service`s; `config_maps` holds the per-role-group `ConfigMap`s and the cluster-level
 /// discovery `ConfigMap`.
-pub struct KubernetesResources {
+///
+/// `T` is a marker that indicates whether these resources are only [`Prepared`] or already
+/// [`Applied`]. It lets the type system prove that e.g. the cluster status is derived from
+/// applied resources rather than merely built ones.
+pub struct KubernetesResources<T> {
     pub daemon_sets: Vec<DaemonSet>,
     pub services: Vec<Service>,
     pub config_maps: Vec<ConfigMap>,
     pub service_accounts: Vec<ServiceAccount>,
     pub role_bindings: Vec<RoleBinding>,
+    pub status: PhantomData<T>,
 }
 
 /// Cluster-wide settings resolved once during validation, so the build steps no longer need the

rust/operator-binary/src/controller/update_status.rs

@@ -0,0 +1,55 @@
+//! The update_status step in the OpaCluster controller.
+
+use snafu::{ResultExt, Snafu};
+use stackable_operator::{
+    client::Client,
+    status::condition::{
+        compute_conditions, daemonset::DaemonSetConditionBuilder,
+        operations::ClusterOperationsConditionBuilder,
+    },
+};
+use strum::{EnumDiscriminants, IntoStaticStr};
+
+use crate::{
+    controller::{Applied, KubernetesResources},
+    crd::{OPERATOR_NAME, OpaClusterStatus, v1alpha2},
+};
+
+#[derive(Snafu, Debug, EnumDiscriminants)]
+#[strum_discriminants(derive(IntoStaticStr))]
+pub enum Error {
+    #[snafu(display("failed to update status"))]
+    ApplyStatus {
+        source: stackable_operator::client::Error,
+    },
+}
+
+type Result<T, E = Error> = std::result::Result<T, E>;
+
+/// Computes the cluster status from the applied resources and patches it onto the
+/// [`v1alpha2::OpaCluster`]. Takes [`KubernetesResources<Applied>`] so the type system proves the
+/// status derives from applied resources, not merely built ones.
+pub async fn update_status(
+    client: &Client,
+    opa: &v1alpha2::OpaCluster,
+    applied: &KubernetesResources<Applied>,
+) -> Result<()> {
+    let mut ds_cond_builder = DaemonSetConditionBuilder::default();
+    for daemon_set in &applied.daemon_sets {
+        ds_cond_builder.add(daemon_set.clone());
+    }
+
+    let cluster_operation_cond_builder =
+        ClusterOperationsConditionBuilder::new(&opa.spec.cluster_operation);
+
+    let status = OpaClusterStatus {
+        conditions: compute_conditions(opa, &[&ds_cond_builder, &cluster_operation_cond_builder]),
+    };
+
+    client
+        .apply_patch_status(OPERATOR_NAME, opa, &status)
+        .await
+        .context(ApplyStatusSnafu)?;
+
+    Ok(())
+}