From c123cc2f6c7aecc3ee7758d3db4aa9d27b75edb1 Mon Sep 17 00:00:00 2001 From: Malte Sander Date: Fri, 7 Aug 2026 13:14:43 +0200 Subject: [PATCH 1/3] refactor: Extract apply and update status steps Moves resource application and status computation out of the reconcile function and into dedicated controller/apply.rs and controller/update_status.rs modules, following the same shape as the airflow (#828) and hbase (#787) operators. reconcile_opa is now just validate, build, apply, update_status. KubernetesResources gains a type-state marker (Prepared, Applied), so update_status can only be called with resources that were actually applied, not merely built. Applier::apply destructures KubernetesResources without a rest pattern, so a new resource kind fails to compile instead of silently never being applied. The legacy "opacluster" field manager patch (see #444) becomes a private method on the Applier, running against the applied DaemonSets. Only the name and namespace are read from those objects, so this is equivalent to the previous behaviour. Note that delete_orphaned_resources now runs before the status patch instead of after it, matching the template operators. --- rust/operator-binary/src/controller/apply.rs | 173 ++++++++++++++++++ rust/operator-binary/src/controller/build.rs | 7 +- rust/operator-binary/src/controller/mod.rs | 17 +- .../src/controller/update_status.rs | 55 ++++++ rust/operator-binary/src/opa_controller.rs | 142 +++----------- 5 files changed, 276 insertions(+), 118 deletions(-) create mode 100644 rust/operator-binary/src/controller/apply.rs create mode 100644 rust/operator-binary/src/controller/update_status.rs diff --git a/rust/operator-binary/src/controller/apply.rs b/rust/operator-binary/src/controller/apply.rs new file mode 100644 index 00000000..b25850a4 --- /dev/null +++ b/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 = std::result::Result; + +/// 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, + ) -> Result> { + // 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( + &mut self, + resources: Vec, + ) -> Result> { + 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(()) + } +} diff --git a/rust/operator-binary/src/controller/build.rs b/rust/operator-binary/src/controller/build.rs index acc5568d..a66d777c 100644 --- a/rust/operator-binary/src/controller/build.rs +++ b/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 { +) -> Result, 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, }) } diff --git a/rust/operator-binary/src/controller/mod.rs b/rust/operator-binary/src/controller/mod.rs index f9dc9023..d9020869 100644 --- a/rust/operator-binary/src/controller/mod.rs +++ b/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 { pub daemon_sets: Vec, pub services: Vec, pub config_maps: Vec, pub service_accounts: Vec, pub role_bindings: Vec, + pub status: PhantomData, } /// Cluster-wide settings resolved once during validation, so the build steps no longer need the diff --git a/rust/operator-binary/src/controller/update_status.rs b/rust/operator-binary/src/controller/update_status.rs new file mode 100644 index 00000000..849730e4 --- /dev/null +++ b/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 = std::result::Result; + +/// Computes the cluster status from the applied resources and patches it onto the +/// [`v1alpha2::OpaCluster`]. Takes [`KubernetesResources`] 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, +) -> 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(()) +} diff --git a/rust/operator-binary/src/opa_controller.rs b/rust/operator-binary/src/opa_controller.rs index 6c4a916c..7f2f5df4 100644 --- a/rust/operator-binary/src/opa_controller.rs +++ b/rust/operator-binary/src/opa_controller.rs @@ -1,30 +1,35 @@ +//! Ensures that `Pod`s are configured and running for each [`v1alpha2::OpaCluster`]. +//! +//! This is the controller driver: it runs the `validate -> build -> apply -> update_status` +//! pipeline. The validated cluster type and the individual steps live under the +//! [`crate::controller`] module tree; this file is kept next to `main.rs` for consistency with +//! the other Stackable operators. + use std::sync::Arc; use const_format::concatcp; -use serde_json::json; use snafu::{ResultExt, Snafu}; use stackable_operator::{ cli::OperatorEnvironmentOptions, cluster_resources::ClusterResourceApplyStrategy, kube::{ - ResourceExt, core::{DeserializeGuard, error_boundary}, runtime::controller::Action, }, logging::controller::ReconcilerError, shared::time::Duration, - status::condition::{ - compute_conditions, daemonset::DaemonSetConditionBuilder, - operations::ClusterOperationsConditionBuilder, - }, utils::cluster_info::KubernetesClusterInfo, - v2::cluster_resources::cluster_resources_new, }; use strum::{EnumDiscriminants, IntoStaticStr}; use crate::{ - controller::{build, controller_name, operator_name, product_name, validate}, - crd::{OPERATOR_NAME, OpaClusterStatus, v1alpha2}, + controller::{ + apply::{self, Applier}, + build, + update_status::{self, update_status}, + validate, + }, + crd::{OPERATOR_NAME, v1alpha2}, }; pub const OPA_CONTROLLER_NAME: &str = "opacluster"; @@ -58,26 +63,11 @@ pub enum Error { #[snafu(display("failed to build the Kubernetes resources"))] BuildResources { source: build::Error }, - #[snafu(display("failed to apply Kubernetes resource"))] - ApplyResource { - source: stackable_operator::cluster_resources::Error, - }, + #[snafu(display("failed to apply the Kubernetes resources"))] + ApplyResources { source: apply::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 update status"))] - ApplyStatus { - source: stackable_operator::client::Error, - }, - - #[snafu(display("failed to delete orphaned resources"))] - DeleteOrphans { - source: stackable_operator::cluster_resources::Error, - }, + #[snafu(display("failed to update the cluster status"))] + UpdateStatus { source: update_status::Error }, } type Result = std::result::Result; @@ -103,17 +93,6 @@ pub async fn reconcile_opa( let validated_cluster = validate::validate(opa, &ctx.operator_environment).context(ValidateClusterSnafu)?; - let mut cluster_resources = cluster_resources_new( - &product_name(), - &operator_name(), - &controller_name(), - &validated_cluster.name, - &validated_cluster.namespace, - &validated_cluster.uid, - ClusterResourceApplyStrategy::from(&opa.spec.cluster_operation), - &opa.spec.object_overrides, - ); - let resources = build::build( &validated_cluster, &ctx.opa_bundle_builder_image, @@ -122,82 +101,19 @@ pub async fn reconcile_opa( ) .context(BuildResourcesSnafu)?; - let mut ds_cond_builder = DaemonSetConditionBuilder::default(); - - // Apply order: DaemonSets last, so a changed mounted ConfigMap already exists before the Pods - // (that would otherwise restart) are updated (commons-operator#111). - for service_account in resources.service_accounts { - cluster_resources - .add(client, service_account) - .await - .context(ApplyResourceSnafu)?; - } - for role_binding in resources.role_bindings { - cluster_resources - .add(client, role_binding) - .await - .context(ApplyResourceSnafu)?; - } - for service in resources.services { - cluster_resources - .add(client, service) - .await - .context(ApplyResourceSnafu)?; - } - for config_map in resources.config_maps { - cluster_resources - .add(client, config_map) - .await - .context(ApplyResourceSnafu)?; - } - for daemon_set in resources.daemon_sets { - ds_cond_builder.add( - cluster_resources - .add(client, daemon_set.clone()) - .await - .context(ApplyResourceSnafu)?, - ); - - // 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) will stay valid. - // We have to use the old field manager scope and post an empty path to get rid of it - // https://github.com/stackabletech/issues/issues/390 will implement a proper fix, e.g. also fixing Services and ConfigMaps - // For details see https://github.com/stackabletech/opa-operator/issues/444 - 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() - ); - 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(), - })?; - } - - 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)?; + let applied = Applier::new( + client, + &validated_cluster, + ClusterResourceApplyStrategy::from(&opa.spec.cluster_operation), + &opa.spec.object_overrides, + ) + .apply(resources) + .await + .context(ApplyResourcesSnafu)?; - cluster_resources - .delete_orphaned_resources(client) + update_status(client, opa, &applied) .await - .context(DeleteOrphansSnafu)?; + .context(UpdateStatusSnafu)?; Ok(Action::await_change()) } From e51056df63fd5a3558e6e34454b51519af63584c Mon Sep 17 00:00:00 2001 From: Malte Sander Date: Fri, 7 Aug 2026 13:24:34 +0200 Subject: [PATCH 2/3] chore: adapt changelog --- CHANGELOG.md | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 7cc5e959..8b2545e9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,10 +11,14 @@ 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 From 1bd520234a8e5202fe7f3613c471b072087a916b Mon Sep 17 00:00:00 2001 From: Malte Sander Date: Fri, 7 Aug 2026 13:38:48 +0200 Subject: [PATCH 3/3] docs: remove empty line in changelog --- CHANGELOG.md | 1 - 1 file changed, 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8b2545e9..f2502c17 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,7 +14,6 @@ All notable changes to this project will be documented in this file. - 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