diff --git a/CHANGELOG.md b/CHANGELOG.md index 67a2863a..3af1a759 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,10 +11,19 @@ functions and carry the full set of recommended labels ([#761]). - BREAKING: The `nodes` role is now required by the CRD; a SupersetCluster without it was previously accepted by the API server but reconciled to no `nodes` resources ([#761]). +- The reconciler now applies resources and derives the cluster status in discrete + apply and update_status steps for the `controller` reconcile ([#772]). + +### Removed + +- The migration that copied the Flask `SECRET_KEY` from the credentials Secret (where SDP 26.3 + kept it, under `connections.secretKey`) into the operator-owned Secret. It was only needed for + the upgrade from SDP 26.3 to 26.7, which has been released ([#772]). [#756]: https://github.com/stackabletech/superset-operator/pull/756 [#761]: https://github.com/stackabletech/superset-operator/pull/761 [#765]: https://github.com/stackabletech/superset-operator/pull/765 +[#772]: https://github.com/stackabletech/superset-operator/pull/772 ## [26.7.0] - 2026-07-21 diff --git a/rust/operator-binary/src/controller.rs b/rust/operator-binary/src/controller.rs index 17011681..165794ca 100644 --- a/rust/operator-binary/src/controller.rs +++ b/rust/operator-binary/src/controller.rs @@ -1,26 +1,25 @@ //! Ensures that `Pod`s are configured and running for each [`SupersetCluster`] +pub mod apply; pub(crate) mod build; pub mod dereference; +pub mod update_status; pub mod validate; -use std::{collections::BTreeMap, str::FromStr, sync::Arc}; +use std::{collections::BTreeMap, marker::PhantomData, str::FromStr, sync::Arc}; use const_format::concatcp; use snafu::{ResultExt, Snafu}; use stackable_operator::{ - builder::meta::ObjectMetaBuilder, cli::OperatorEnvironmentOptions, - client::Client, cluster_resources::ClusterResourceApplyStrategy, commons::{ affinity::StackableAffinity, product_image_selection::ResolvedProductImage, - random_secret_creation::{self, create_random_secret_if_not_exists}, resources::{NoRuntimeLimits, Resources}, }, crd::listener, k8s_openapi::api::{ apps::v1::{Deployment, StatefulSet}, - core::v1::{ConfigMap, Secret, Service, ServiceAccount}, + core::v1::{ConfigMap, Service, ServiceAccount}, policy::v1::PodDisruptionBudget, rbac::v1::RoleBinding, }, @@ -33,13 +32,8 @@ use stackable_operator::{ kvp::Labels, logging::controller::ReconcilerError, shared::time::Duration, - status::condition::{ - compute_conditions, deployment::DeploymentConditionBuilder, - operations::ClusterOperationsConditionBuilder, statefulset::StatefulSetConditionBuilder, - }, v2::{ HasName, HasUid, NameIsValidLabelValue, - cluster_resources::cluster_resources_new, kvp::label::{recommended_labels, role_group_selector}, product_logging::framework::{ValidatedContainerLogConfigChoice, VectorContainerLogConfig}, role_group_utils::ResourceNames, @@ -54,20 +48,22 @@ use stackable_operator::{ }, }; use strum::{EnumDiscriminants, IntoStaticStr}; -use tracing::instrument; use crate::{ OPERATOR_NAME, + controller::{ + apply::{Applier, ensure_secrets}, + update_status::update_status, + }, crd::{ - APP_NAME, INTERNAL_SECRET_SECRET_KEY, SupersetRole, + APP_NAME, SupersetRole, authentication::SupersetClientAuthenticationDetailsResolved, authorization::SupersetOpaConfigResolved, databases::{ CeleryBrokerConnection, CeleryResultsBackendConnection, MetadataDatabaseConnection, }, v1alpha1::{ - SupersetCluster, SupersetClusterStatus, SupersetConfig, SupersetConfigOverrides, - SupersetStorageConfig, + SupersetCluster, SupersetConfig, SupersetConfigOverrides, SupersetStorageConfig, }, }, }; @@ -82,11 +78,21 @@ pub struct Ctx { pub operator_environment: OperatorEnvironmentOptions, } +/// Marker for prepared Kubernetes resources which are not applied yet. +pub struct Prepared; + +/// Marker for Kubernetes resources which are already applied. +pub struct Applied; + /// Every Kubernetes resource produced by the build step. /// /// The `Node` role is provisioned via a `StatefulSet` (it serves the Superset web UI), while the /// `Worker`/`Beat` Celery roles are provisioned via `Deployment`s; the build step collects both. -pub struct KubernetesResources { +/// +/// `T` marks whether these resources are merely [`Prepared`] or already [`Applied`]. The marker +/// lets the compiler prove that the cluster status is derived from the applied resources (which +/// carry the state the API server returned) rather than from the built ones. +pub struct KubernetesResources { pub stateful_sets: Vec, pub deployments: Vec, pub services: Vec, @@ -95,6 +101,7 @@ pub struct KubernetesResources { pub pod_disruption_budgets: Vec, pub service_accounts: Vec, pub role_bindings: Vec, + pub status: PhantomData, } /// Per-role configuration extracted during validation. @@ -380,49 +387,22 @@ pub enum Error { #[snafu(display("failed to validate cluster"))] Validate { source: validate::Error }, - #[snafu(display("failed to delete orphaned resources"))] - DeleteOrphanedResources { - source: stackable_operator::cluster_resources::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 ensure the SECRET_KEY Secret exists"))] + EnsureSecrets { source: apply::Error }, - #[snafu(display("failed to update status"))] - ApplyStatus { - source: stackable_operator::client::Error, - }, + #[snafu(display("failed to apply the Kubernetes resources"))] + ApplyResources { source: apply::Error }, + + #[snafu(display("failed to update the cluster status"))] + UpdateStatus { source: update_status::Error }, #[snafu(display("SupersetCluster object is invalid"))] InvalidSupersetCluster { source: error_boundary::InvalidObject, }, - - #[snafu(display("failed to create SECRET_KEY secret"))] - CreateSecretKeySecret { - source: random_secret_creation::Error, - }, - - #[snafu(display("failed to retrieve credentials secret {secret_name:?}"))] - RetrieveCredentialsSecret { - source: stackable_operator::client::Error, - secret_name: String, - }, - - #[snafu(display("object is missing metadata to build owner reference"))] - ObjectMissingMetadataForOwnerRef { - source: stackable_operator::builder::meta::Error, - }, - - #[snafu(display("failed to create SECRET_KEY secret from migrated value"))] - CreateRandomSecret { - source: stackable_operator::client::Error, - }, } type Result = std::result::Result; @@ -433,6 +413,18 @@ impl ReconcilerError for Error { } } +/// Reconcile function of the SupersetCluster controller. +/// +/// The reconcile function performs the following steps: +/// 1. Dereference the objects the SupersetCluster refers to (client required). +/// 2. Validate the cluster specification together with the dereferenced objects, yielding a +/// [`ValidatedCluster`] (no client required). +/// 3. Build the Kubernetes resource specifications from the validated cluster (no client +/// required). +/// 4. Ensure the Secrets exist that the resources mount but that the operator cannot build, +/// because their value has to be generated once and then kept (client required). +/// 5. Apply the resource specifications and delete the orphaned ones (client required). +/// 6. Update the cluster status from the applied resources (client required). pub async fn reconcile_superset( superset: Arc>, ctx: Arc, @@ -447,9 +439,6 @@ pub async fn reconcile_superset( let client = &ctx.client; - let cluster_operation_cond_builder = - ClusterOperationsConditionBuilder::new(&superset.spec.cluster_config.cluster_operation); - let dereferenced = dereference::dereference(client, superset) .await .context(DereferenceSnafu)?; @@ -461,181 +450,29 @@ pub async fn reconcile_superset( ) .context(ValidateSnafu)?; - let mut cluster_resources = cluster_resources_new( - &product_name(), - &operator_name(), - &controller_name(), - &validated.name, - &validated.namespace, - &validated.uid, + let resources = build::build(&validated).context(BuildResourcesSnafu)?; + + ensure_secrets(client, &validated) + .await + .context(EnsureSecretsSnafu)?; + + let applied = Applier::new( + client, + &validated, ClusterResourceApplyStrategy::from(&superset.spec.cluster_config.cluster_operation), &superset.spec.object_overrides, - ); - - // TODO: Can be removed after SDP 26.7 is released (it's only a migration from 26.3 - 26.7) - // (don't forget about the snafu Error variants). - // Removal is tracked in https://github.com/stackabletech/superset-operator/issues/755 - migrate_legacy_secret_key_secret_from_26_3(superset, &validated, client).await?; - create_random_secret_if_not_exists( - &validated.cluster_config.secret_key_secret_name, - INTERNAL_SECRET_SECRET_KEY, - 256, - &validated, - client, ) + .apply(resources) .await - .context(CreateSecretKeySecretSnafu)?; - - let resources = build::build(&validated).context(BuildResourcesSnafu)?; - - let mut statefulset_cond_builder = StatefulSetConditionBuilder::default(); - let mut deployment_cond_builder = DeploymentConditionBuilder::default(); - - // The StatefulSets/Deployments are applied last, so every ConfigMap and Secret they mount - // already exists — otherwise a changed mount would restart the Pods. - // See https://github.com/stackabletech/commons-operator/issues/111 for details. - 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 listener in resources.listeners { - cluster_resources - .add(client, listener) - .await - .context(ApplyResourceSnafu)?; - } - for pdb in resources.pod_disruption_budgets { - cluster_resources - .add(client, pdb) - .await - .context(ApplyResourceSnafu)?; - } - for statefulset in resources.stateful_sets { - statefulset_cond_builder.add( - cluster_resources - .add(client, statefulset) - .await - .context(ApplyResourceSnafu)?, - ); - } - for deployment in resources.deployments { - deployment_cond_builder.add( - cluster_resources - .add(client, deployment) - .await - .context(ApplyResourceSnafu)?, - ); - } + .context(ApplyResourcesSnafu)?; - cluster_resources - .delete_orphaned_resources(client) - .await - .context(DeleteOrphanedResourcesSnafu)?; - - let status = SupersetClusterStatus { - conditions: compute_conditions( - superset, - &[ - &statefulset_cond_builder, - &deployment_cond_builder, - &cluster_operation_cond_builder, - ], - ), - }; - client - .apply_patch_status(OPERATOR_NAME, superset, &status) + update_status(client, superset, &applied) .await - .context(ApplyStatusSnafu)?; + .context(UpdateStatusSnafu)?; Ok(Action::await_change()) } -// TODO: Can be removed after SDP 26.7 is released (it's only a migration from 26.3 - 26.7) -// (don't forget about the snafu Error variants). -// Removal is tracked in https://github.com/stackabletech/superset-operator/issues/755 -#[instrument(skip_all)] -async fn migrate_legacy_secret_key_secret_from_26_3( - superset: &SupersetCluster, - validated: &ValidatedCluster, - client: &Client, -) -> Result<()> { - let old_secret_name = &validated.cluster_config.credentials_secret_name; - let new_secret_name = &validated.cluster_config.secret_key_secret_name; - let secret_namespace = &validated.namespace; - - let new_secret = client - .get_opt::(new_secret_name, secret_namespace.as_ref()) - .await - .with_context(|_| RetrieveCredentialsSecretSnafu { - secret_name: new_secret_name, - })?; - if new_secret.is_some() { - tracing::debug!("SECRET_KEY Secret already exists, nothing to migrate"); - return Ok(()); - } - - let old_secret = client - .get_opt::(old_secret_name, secret_namespace.as_ref()) - .await - .with_context(|_| RetrieveCredentialsSecretSnafu { - secret_name: old_secret_name, - })?; - let old_secret_key = old_secret - .and_then(|secret| secret.data) - // Note: We remove the key to take ownership - .and_then(|mut data| data.remove("connections.secretKey")) - .and_then(|key| String::from_utf8(key.0).ok()); - if let Some(old_secret_key) = old_secret_key { - tracing::info!( - old.secret.name = old_secret_name, - old.secret.namespace = %secret_namespace, - new.secret.name = new_secret_name, - new.secret.namespace = %secret_namespace, - "Migrating old SECRET_KEY to new Secret" - ); - - let secret = Secret { - metadata: ObjectMetaBuilder::new() - .name(new_secret_name) - .namespace(secret_namespace) - .ownerreference_from_resource(superset, None, Some(true)) - .context(ObjectMissingMetadataForOwnerRefSnafu)? - .build(), - string_data: Some(BTreeMap::from([( - INTERNAL_SECRET_SECRET_KEY.to_string(), - old_secret_key, - )])), - ..Secret::default() - }; - client - .create(&secret) - .await - .context(CreateRandomSecretSnafu)?; - } - - Ok(()) -} - pub fn error_policy( _obj: Arc>, error: &Error, diff --git a/rust/operator-binary/src/controller/apply.rs b/rust/operator-binary/src/controller/apply.rs new file mode 100644 index 00000000..e311ba9a --- /dev/null +++ b/rust/operator-binary/src/controller/apply.rs @@ -0,0 +1,166 @@ +//! The apply step in the SupersetCluster controller. + +use std::marker::PhantomData; + +use snafu::{ResultExt, Snafu}; +use stackable_operator::{ + client::Client, + cluster_resources::{ClusterResource, ClusterResourceApplyStrategy, ClusterResources}, + commons::random_secret_creation, + deep_merger::ObjectOverrides, + v2::cluster_resources::cluster_resources_new, +}; +use strum::{EnumDiscriminants, IntoStaticStr}; + +use crate::{ + controller::{ + Applied, KubernetesResources, Prepared, ValidatedCluster, controller_name, operator_name, + product_name, + }, + crd::INTERNAL_SECRET_SECRET_KEY, +}; + +#[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 delete orphaned resources"))] + DeleteOrphanedResources { + source: stackable_operator::cluster_resources::Error, + }, + + #[snafu(display("failed to create SECRET_KEY Secret"))] + CreateSecretKeySecret { + source: random_secret_creation::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, deletes the ones that are no longer part of the + /// cluster and marks the result 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 the new resource silently never being applied. + let KubernetesResources { + stateful_sets, + deployments, + services, + listeners, + config_maps, + pod_disruption_budgets, + service_accounts, + role_bindings, + status: _, + } = resources; + + // The ServiceAccount and its RoleBinding come first, because the Pods reference them at + // creation time. The StatefulSets and Deployments come last, so that every ConfigMap and + // Secret they mount already exists, otherwise a changed mount would restart the Pods. + // See https://github.com/stackabletech/commons-operator/issues/111 for details. + 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 listeners = self.add_resources(listeners).await?; + let pod_disruption_budgets = self.add_resources(pod_disruption_budgets).await?; + let stateful_sets = self.add_resources(stateful_sets).await?; + let deployments = self.add_resources(deployments).await?; + + self.cluster_resources + .delete_orphaned_resources(self.client) + .await + .context(DeleteOrphanedResourcesSnafu)?; + + Ok(KubernetesResources { + stateful_sets, + deployments, + services, + listeners, + config_maps, + pod_disruption_budgets, + 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) + } +} + +/// Ensures that the Secret holding the Flask `SECRET_KEY` exists, creating it with a random value +/// if it does not. +/// +/// This is a read-or-create client operation, so it cannot be part of the client-free `build()` +/// step. It is also deliberately not tracked in [`ClusterResources`], so that it survives orphan +/// deletion and an existing Secret is never overwritten (rotating the `SECRET_KEY` would +/// invalidate every session). +pub async fn ensure_secrets(client: &Client, cluster: &ValidatedCluster) -> Result<()> { + random_secret_creation::create_random_secret_if_not_exists( + &cluster.cluster_config.secret_key_secret_name, + INTERNAL_SECRET_SECRET_KEY, + 256, + cluster, + client, + ) + .await + .context(CreateSecretKeySecretSnafu)?; + + Ok(()) +} diff --git a/rust/operator-binary/src/controller/build/mod.rs b/rust/operator-binary/src/controller/build/mod.rs index 72ea8676..cf513a3f 100644 --- a/rust/operator-binary/src/controller/build/mod.rs +++ b/rust/operator-binary/src/controller/build/mod.rs @@ -1,6 +1,6 @@ //! Builders that assemble Kubernetes resources for superset rolegroups. -use std::str::FromStr; +use std::{marker::PhantomData, str::FromStr}; use snafu::{ResultExt, Snafu}; use stackable_operator::{ @@ -13,7 +13,7 @@ use stackable_operator::{ use crate::{ controller::{ - KubernetesResources, ValidatedCluster, + KubernetesResources, Prepared, ValidatedCluster, build::resource::{ config_map::build_rolegroup_config_map, deployment::build_rolegroup_deployment, @@ -61,7 +61,7 @@ pub enum Error { } /// Builds every Kubernetes resource for the given validated cluster. -pub fn build(cluster: &ValidatedCluster) -> Result { +pub fn build(cluster: &ValidatedCluster) -> Result, Error> { let mut stateful_sets = vec![]; let mut deployments = vec![]; let mut services = vec![]; @@ -163,6 +163,7 @@ pub fn build(cluster: &ValidatedCluster) -> Result { pod_disruption_budgets, service_accounts: vec![build_service_account(cluster)], role_bindings: vec![build_role_binding(cluster)], + status: PhantomData, }) } 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..32dfc153 --- /dev/null +++ b/rust/operator-binary/src/controller/update_status.rs @@ -0,0 +1,72 @@ +//! The update_status step in the SupersetCluster controller. + +use snafu::{ResultExt, Snafu}; +use stackable_operator::{ + client::Client, + status::condition::{ + compute_conditions, deployment::DeploymentConditionBuilder, + operations::ClusterOperationsConditionBuilder, statefulset::StatefulSetConditionBuilder, + }, +}; +use strum::{EnumDiscriminants, IntoStaticStr}; + +use crate::{ + OPERATOR_NAME, + controller::{Applied, KubernetesResources}, + crd::v1alpha1::{SupersetCluster, SupersetClusterStatus}, +}; + +#[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 +/// [`SupersetCluster`]. +/// +/// Takes [`KubernetesResources`], so the type system proves that the conditions are +/// derived from the resources the API server returned rather than from the ones that were merely +/// built. The `Node` role contributes StatefulSet conditions and the `Worker`/`Beat` Celery roles +/// contribute Deployment conditions. +pub async fn update_status( + client: &Client, + superset: &SupersetCluster, + applied: &KubernetesResources, +) -> Result<()> { + let mut stateful_set_cond_builder = StatefulSetConditionBuilder::default(); + for stateful_set in &applied.stateful_sets { + stateful_set_cond_builder.add(stateful_set.clone()); + } + + let mut deployment_cond_builder = DeploymentConditionBuilder::default(); + for deployment in &applied.deployments { + deployment_cond_builder.add(deployment.clone()); + } + + let cluster_operation_cond_builder = + ClusterOperationsConditionBuilder::new(&superset.spec.cluster_config.cluster_operation); + + let status = SupersetClusterStatus { + conditions: compute_conditions( + superset, + &[ + &stateful_set_cond_builder, + &deployment_cond_builder, + &cluster_operation_cond_builder, + ], + ), + }; + + client + .apply_patch_status(OPERATOR_NAME, superset, &status) + .await + .context(ApplyStatusSnafu)?; + + Ok(()) +}