diff --git a/CHANGELOG.md b/CHANGELOG.md index 7cc5e959..f2502c17 100644 --- a/CHANGELOG.md +++ b/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 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()) }