diff --git a/CHANGELOG.md b/CHANGELOG.md index 56cd009a..ed06ce44 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,10 +13,13 @@ All notable changes to this project will be documented in this file. functions and carry the full set of recommended labels ([#913]). - BREAKING: The `coordinators` and `workers` roles are now required by the CRD. Previously a TrinoCluster missing either role was accepted by the API server but failed reconciliation ([#913]). +- The reconciler now applies resources and derives the cluster status in discrete + apply and update_status steps for the `trino_controller` ([#923]). [#909]: https://github.com/stackabletech/trino-operator/pull/909 [#913]: https://github.com/stackabletech/trino-operator/pull/913 [#918]: https://github.com/stackabletech/trino-operator/pull/918 +[#923]: https://github.com/stackabletech/trino-operator/pull/923 ## [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..4961f004 --- /dev/null +++ b/rust/operator-binary/src/controller/apply.rs @@ -0,0 +1,179 @@ +//! The apply step in the TrinoCluster 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, shared_internal_secret_name, shared_spooling_secret_name, + }, + crd::{ENV_INTERNAL_SECRET, ENV_SPOOLING_SECRET}, +}; + +#[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 internal secret"))] + CreateInternalSecret { + 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 and marks them as applied. + /// + /// Resources that are owned by this cluster but no longer part of `resources` are deleted + /// afterwards. + pub async fn apply( + mut self, + resources: KubernetesResources, + ) -> Result> { + // Destructured without `..`, so that adding a field to [`KubernetesResources`] fails to + // compile here instead of the new resource silently never being applied. + let KubernetesResources { + stateful_sets, + services, + listeners, + config_maps, + pod_disruption_budgets, + service_accounts, + role_bindings, + status: _, + } = resources; + + // 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 listeners = self.add_resources(listeners).await?; + let config_maps = self.add_resources(config_maps).await?; + let pod_disruption_budgets = self.add_resources(pod_disruption_budgets).await?; + + // Note: The StatefulSet needs to be applied after all ConfigMaps and Secrets it mounts + // to prevent unnecessary Pod restarts. + // See https://github.com/stackabletech/commons-operator/issues/111 for details. + let stateful_sets = self.add_resources(stateful_sets).await?; + + self.cluster_resources + .delete_orphaned_resources(self.client) + .await + .context(DeleteOrphanedResourcesSnafu)?; + + Ok(KubernetesResources { + stateful_sets, + services, + listeners, + config_maps, + pod_disruption_budgets, + service_accounts, + role_bindings, + status: PhantomData, + }) + } + + /// Applies the given resources and returns them as the API server echoed them back. + 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 the two shared random Secrets (internal communication and spooling) exist, creating +/// any that are missing. +/// +/// These are read-or-create client operations, so they cannot be part of the client-free +/// `build()` step. They are also deliberately not tracked in [`ClusterResources`], so that they +/// survive orphan deletion and an existing Secret is never overwritten (rotating them would +/// invalidate all running queries). +pub async fn ensure_random_secrets(client: &Client, cluster: &ValidatedCluster) -> Result<()> { + random_secret_creation::create_random_secret_if_not_exists( + &shared_internal_secret_name(&cluster.name), + ENV_INTERNAL_SECRET, + 512, + cluster, + client, + ) + .await + .context(CreateInternalSecretSnafu)?; + + // This secret is created even if spooling is not configured. + // Trino currently requires the secret to be exactly 256 bits long. + random_secret_creation::create_random_secret_if_not_exists( + &shared_spooling_secret_name(&cluster.name), + ENV_SPOOLING_SECRET, + 32, + cluster, + client, + ) + .await + .context(CreateInternalSecretSnafu)?; + + Ok(()) +} diff --git a/rust/operator-binary/src/controller/build.rs b/rust/operator-binary/src/controller/build.rs index 762b5158..72437d71 100644 --- a/rust/operator-binary/src/controller/build.rs +++ b/rust/operator-binary/src/controller/build.rs @@ -1,6 +1,6 @@ //! Builders that turn a `ValidatedCluster` into Kubernetes resource contents. -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, ValidatedCluster, + KubernetesResources, Prepared, ValidatedCluster, build::resource::{ config_map, listener::{build_group_listener, group_listener_name}, @@ -59,7 +59,7 @@ pub enum Error { pub fn build( cluster: &ValidatedCluster, cluster_info: &KubernetesClusterInfo, -) -> Result { +) -> Result, Error> { let mut stateful_sets = vec![]; let mut services = vec![]; let mut listeners = vec![]; @@ -148,6 +148,7 @@ pub fn build( 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/mod.rs b/rust/operator-binary/src/controller/mod.rs index 551b9737..cd255751 100644 --- a/rust/operator-binary/src/controller/mod.rs +++ b/rust/operator-binary/src/controller/mod.rs @@ -1,4 +1,4 @@ -use std::{collections::BTreeMap, str::FromStr}; +use std::{collections::BTreeMap, marker::PhantomData, str::FromStr}; use stackable_operator::{ commons::{ @@ -43,8 +43,10 @@ use crate::{ trino_controller::{CONTROLLER_NAME, OPERATOR_NAME}, }; +pub(crate) mod apply; pub(crate) mod build; pub(crate) mod dereference; +pub(crate) mod update_status; pub(crate) mod validate; pub use stackable_operator::v2::product_logging::framework::STACKABLE_LOG_DIR; @@ -67,8 +69,18 @@ pub(crate) fn shared_spooling_secret_name(cluster_name: &ClusterName) -> String // Placeholder version label value for resources whose labels must not change after deployment. stackable_operator::constant!(UNVERSIONED_PRODUCT_VERSION: ProductVersion = "none"); +/// Marker for prepared Kubernetes resources which are not applied yet. +pub struct Prepared; + +/// Marker for Kubernetes resources which have been applied to the Kubernetes cluster. +pub struct Applied; + /// Every Kubernetes resource produced by the client-free [`build()`](build::build) step. -pub struct KubernetesResources { +/// +/// `T` marks whether these resources are only [`Prepared`] or already [`Applied`]. The marker +/// lets the type system enforce, for example, that the cluster status is derived from the +/// resources the API server returned rather than from the ones we merely built. +pub struct KubernetesResources { pub stateful_sets: Vec, pub services: Vec, pub listeners: Vec, @@ -76,6 +88,7 @@ pub struct KubernetesResources { pub pod_disruption_budgets: Vec, pub service_accounts: Vec, pub role_bindings: Vec, + pub status: PhantomData, } #[derive(Clone, Debug)] 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..9dc358c8 --- /dev/null +++ b/rust/operator-binary/src/controller/update_status.rs @@ -0,0 +1,61 @@ +//! The update_status step in the TrinoCluster controller. + +use snafu::{ResultExt, Snafu}; +use stackable_operator::{ + client::Client, + status::condition::{ + compute_conditions, operations::ClusterOperationsConditionBuilder, + statefulset::StatefulSetConditionBuilder, + }, +}; +use strum::{EnumDiscriminants, IntoStaticStr}; + +use crate::{ + controller::{Applied, KubernetesResources}, + crd::v1alpha1, + trino_controller::OPERATOR_NAME, +}; + +#[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 +/// [`v1alpha1::TrinoCluster`]. +/// +/// Takes [`KubernetesResources`], so the type system proves that the status is derived +/// from the resources the API server acknowledged and not from the ones we merely built. +pub async fn update_status( + client: &Client, + trino: &v1alpha1::TrinoCluster, + applied: &KubernetesResources, +) -> Result<()> { + let mut sts_cond_builder = StatefulSetConditionBuilder::default(); + for stateful_set in &applied.stateful_sets { + sts_cond_builder.add(stateful_set.clone()); + } + + let cluster_operation_cond_builder = + ClusterOperationsConditionBuilder::new(&trino.spec.cluster_operation); + + let status = v1alpha1::TrinoClusterStatus { + conditions: compute_conditions( + trino, + &[&sts_cond_builder, &cluster_operation_cond_builder], + ), + }; + + client + .apply_patch_status(OPERATOR_NAME, trino, &status) + .await + .context(ApplyStatusSnafu)?; + + Ok(()) +} diff --git a/rust/operator-binary/src/trino_controller.rs b/rust/operator-binary/src/trino_controller.rs index 238c7316..d302e912 100644 --- a/rust/operator-binary/src/trino_controller.rs +++ b/rust/operator-binary/src/trino_controller.rs @@ -1,4 +1,10 @@ -//! Ensures that `Pod`s are configured and running for each [`v1alpha1::TrinoCluster`] +//! Ensures that `Pod`s are configured and running for each [`v1alpha1::TrinoCluster`]. +//! +//! This is the controller driver: it runs the +//! `dereference -> 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; @@ -6,27 +12,23 @@ use snafu::{ResultExt, Snafu}; use stackable_operator::{ cli::OperatorEnvironmentOptions, cluster_resources::ClusterResourceApplyStrategy, - commons::random_secret_creation, kube::{ core::{DeserializeGuard, error_boundary}, runtime::controller::Action, }, logging::controller::ReconcilerError, shared::time::Duration, - status::condition::{ - compute_conditions, operations::ClusterOperationsConditionBuilder, - statefulset::StatefulSetConditionBuilder, - }, - v2::cluster_resources::cluster_resources_new, }; use strum::{EnumDiscriminants, IntoStaticStr}; use crate::{ controller::{ - ValidatedCluster, build, controller_name, dereference, operator_name, product_name, - shared_internal_secret_name, shared_spooling_secret_name, validate, + apply::{self, Applier, ensure_random_secrets}, + build, dereference, + update_status::{self, update_status}, + validate, }, - crd::{ENV_INTERNAL_SECRET, ENV_SPOOLING_SECRET, v1alpha1}, + crd::v1alpha1, }; pub struct Ctx { @@ -43,23 +45,17 @@ pub(crate) const CONTAINER_IMAGE_BASE_NAME: &str = "trino"; #[derive(Snafu, Debug, EnumDiscriminants)] #[strum_discriminants(derive(IntoStaticStr))] pub enum 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 shared random Secrets exist"))] + 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("invalid TrinoCluster object"))] InvalidTrinoCluster { @@ -71,11 +67,6 @@ pub enum Error { #[snafu(display("failed to validate cluster"))] ValidateCluster { source: validate::Error }, - - #[snafu(display("failed to create internal secret"))] - CreateInternalSecret { - source: random_secret_creation::Error, - }, } type Result = std::result::Result; @@ -115,129 +106,31 @@ pub async fn reconcile_trino( "Validated TrinoCluster" ); - let mut cluster_resources = cluster_resources_new( - &product_name(), - &operator_name(), - &controller_name(), - &validated_cluster.name, - &validated_cluster.namespace, - &validated_cluster.uid, - ClusterResourceApplyStrategy::from(&trino.spec.cluster_operation), - &trino.spec.object_overrides, - ); - - ensure_random_secrets(client, &validated_cluster).await?; - + // build (no client required) let resources = build::build(&validated_cluster, &client.kubernetes_cluster_info) .context(BuildResourcesSnafu)?; - let mut sts_cond_builder = StatefulSetConditionBuilder::default(); - - 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 listener in resources.listeners { - cluster_resources - .add(client, listener) - .await - .context(ApplyResourceSnafu)?; - } - - for config_map in resources.config_maps { - cluster_resources - .add(client, config_map) - .await - .context(ApplyResourceSnafu)?; - } - - for pdb in resources.pod_disruption_budgets { - cluster_resources - .add(client, pdb) - .await - .context(ApplyResourceSnafu)?; - } - - // Note: The StatefulSet needs to be applied after all ConfigMaps and Secrets it mounts - // to prevent unnecessary Pod restarts. - // See https://github.com/stackabletech/commons-operator/issues/111 for details. - for stateful_set in resources.stateful_sets { - sts_cond_builder.add( - cluster_resources - .add(client, stateful_set) - .await - .context(ApplyResourceSnafu)?, - ); - } - - let cluster_operation_cond_builder = - ClusterOperationsConditionBuilder::new(&trino.spec.cluster_operation); - - let status = v1alpha1::TrinoClusterStatus { - conditions: compute_conditions( - trino, - &[&sts_cond_builder, &cluster_operation_cond_builder], - ), - }; - - cluster_resources - .delete_orphaned_resources(client) - .await - .context(DeleteOrphanedResourcesSnafu)?; - client - .apply_patch_status(OPERATOR_NAME, trino, &status) + // apply (client required) + ensure_random_secrets(client, &validated_cluster) .await - .context(ApplyStatusSnafu)?; + .context(EnsureSecretsSnafu)?; - Ok(Action::await_change()) -} - -/// Ensures the two shared random Secrets (internal communication and spooling) exist, creating -/// any that are missing. -async fn ensure_random_secrets( - client: &stackable_operator::client::Client, - cluster: &ValidatedCluster, -) -> Result<()> { - random_secret_creation::create_random_secret_if_not_exists( - &shared_internal_secret_name(&cluster.name), - ENV_INTERNAL_SECRET, - 512, - cluster, - client, - ) - .await - .context(CreateInternalSecretSnafu)?; - - // This secret is created even if spooling is not configured. - // Trino currently requires the secret to be exactly 256 bits long. - random_secret_creation::create_random_secret_if_not_exists( - &shared_spooling_secret_name(&cluster.name), - ENV_SPOOLING_SECRET, - 32, - cluster, + let applied = Applier::new( client, + &validated_cluster, + ClusterResourceApplyStrategy::from(&trino.spec.cluster_operation), + &trino.spec.object_overrides, ) + .apply(resources) .await - .context(CreateInternalSecretSnafu)?; + .context(ApplyResourcesSnafu)?; - Ok(()) + // update status (client required) + update_status(client, trino, &applied) + .await + .context(UpdateStatusSnafu)?; + + Ok(Action::await_change()) } pub fn error_policy(