refactor: Extract apply and update_status steps

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
 

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<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.
+    ///
+    /// Resources that are owned by this cluster but no longer part of `resources` are deleted
+    /// afterwards.
+    pub async fn apply(
+        mut self,
+        resources: KubernetesResources<Prepared>,
+    ) -> Result<KubernetesResources<Applied>> {
+        // 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<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)
+    }
+}
+
+/// 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(())
+}

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<KubernetesResources, Error> {
+) -> Result<KubernetesResources<Prepared>, 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,
     })
 }
 

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,15 +69,26 @@ 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<T> {
     pub stateful_sets: Vec<StatefulSet>,
     pub services: Vec<Service>,
     pub listeners: Vec<Listener>,
     pub config_maps: Vec<ConfigMap>,
     pub pod_disruption_budgets: Vec<PodDisruptionBudget>,
     pub service_accounts: Vec<ServiceAccount>,
     pub role_bindings: Vec<RoleBinding>,
+    pub status: PhantomData<T>,
 }
 
 #[derive(Clone, Debug)]

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<T, E = Error> = std::result::Result<T, E>;
+
+/// Computes the cluster status from the applied resources and patches it onto the
+/// [`v1alpha1::TrinoCluster`].
+///
+/// Takes [`KubernetesResources<Applied>`], 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<Applied>,
+) -> 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(())
+}