refactor: Extract apply and update status steps

CHANGELOG.md

@@ -13,6 +13,13 @@ All notable changes to this project will be documented in this file.
   functions and carry the full set of recommended labels ([#966]).
 - BREAKING: The `nodes` role is now required by the CRD; a NifiCluster without it was
   previously accepted by the API server but failed reconciliation ([#966]).
+- The reconciler now applies resources and derives the cluster status in discrete
+  apply and update_status steps for the `nifi_controller` ([#974]).
+- The sensitive properties key Secret and (for OIDC authentication) the admin password Secret are
+  now dereferenced, built and applied like every other resource, instead of being created
+  out-of-band before the apply step. An existing Secret is re-emitted with its contents unchanged,
+  so applying it is a no-op and the contents are never rotated. The operator therefore now needs
+  the `patch` permission on `secrets` ([#974]).
 - All product containers now run with `securityContext.runAsNonRoot` set to `true` to improve security ([#975]).
 
 ### Fixed
@@ -24,6 +31,7 @@ All notable changes to this project will be documented in this file.
 [#961]: https://github.com/stackabletech/nifi-operator/pull/961
 [#966]: https://github.com/stackabletech/nifi-operator/pull/966
 [#970]: https://github.com/stackabletech/nifi-operator/pull/970
+[#974]: https://github.com/stackabletech/nifi-operator/pull/974
 [#975]: https://github.com/stackabletech/nifi-operator/pull/975
 
 ## [26.7.0] - 2026-07-21

deploy/helm/nifi-operator/templates/clusterrole-operator.yaml

@@ -39,14 +39,17 @@ rules:
       - get
       - list
       - patch
-  # Sensitive properties key and (when OIDC) admin password secret.
+  # Sensitive properties key and (when OIDC) admin password Secret. Applied via SSA like every
+  # other resource, but deliberately not owned by the NifiCluster, so they are never orphan-deleted
+  # (which is also why no `list` is needed here).
   - apiGroups:
       - ""
     resources:
       - secrets
     verbs:
       - get
       - create
+      - patch
   # RoleBinding created per NifiCluster to bind the product ClusterRole to the workload
   # ServiceAccount. Applied via SSA and tracked for orphan cleanup.
   - apiGroups:

extra/crds.yaml

@@ -315,8 +315,8 @@ spec:
                           This setting configures the encryption algorithm to use to encrypt sensitive properties.
                           Valid values are:
 
-                          `nifiPbkdf2AesGcm256` (the default value),
-                          `nifiArgon2AesGcm256`,
+                          `nifiArgon2AesGcm256` (the default value),
+                          `nifiPbkdf2AesGcm256`,
 
                           Learn more about the specifics of the algorithm parameters in the
                           [NiFi documentation](https://nifi.apache.org/docs/nifi-docs/html/administration-guide.html#property-encryption-algorithms).

rust/operator-binary/src/controller/apply.rs

@@ -0,0 +1,139 @@
+//! The apply step in the NifiCluster controller.
+
+use std::marker::PhantomData;
+
+use snafu::{ResultExt, Snafu};
+use stackable_operator::{
+    client::Client,
+    cluster_resources::{ClusterResource, ClusterResourceApplyStrategy, ClusterResources},
+    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,
+};
+
+#[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,
+    },
+}
+
+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.
+    pub async fn apply(
+        mut self,
+        resources: KubernetesResources<Prepared>,
+    ) -> Result<KubernetesResources<Applied>> {
+        // Destructured without `..`, so adding a field to [`KubernetesResources`] fails to
+        // compile here instead of silently never being applied.
+        let KubernetesResources {
+            stateful_sets,
+            services,
+            listeners,
+            config_maps,
+            secrets,
+            pod_disruption_budgets,
+            service_accounts,
+            role_bindings,
+            status: _,
+        } = resources;
+
+        // Apply order is: StatefulSets last (a changed mounted ConfigMap or Secret must exist
+        // first, else the Pods restart unnecessarily, see 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 listeners = self.add_resources(listeners).await?;
+        let config_maps = self.add_resources(config_maps).await?;
+        let secrets = self.add_resources(secrets).await?;
+        let pod_disruption_budgets = self.add_resources(pod_disruption_budgets).await?;
+        let stateful_sets = self.add_resources(stateful_sets).await?;
+
+        // Remove any orphaned resources that still exist in Kubernetes, but have not been added to
+        // the cluster resources during this reconciliation.
+        // TODO: this doesn't cater for a graceful cluster shrink, for that we'd need to predict
+        //  the resources that will be removed and run a disconnect/offload job for those
+        //  see https://github.com/stackabletech/nifi-operator/issues/314
+        self.cluster_resources
+            .delete_orphaned_resources(self.client)
+            .await
+            .context(DeleteOrphanedResourcesSnafu)?;
+
+        Ok(KubernetesResources {
+            stateful_sets,
+            services,
+            listeners,
+            config_maps,
+            secrets,
+            pod_disruption_budgets,
+            service_accounts,
+            role_bindings,
+            status: PhantomData,
+        })
+    }
+
+    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)
+    }
+}

rust/operator-binary/src/controller/build.rs

@@ -2,11 +2,12 @@
 //!
 //! [`ValidatedCluster`]: crate::controller::ValidatedCluster
 
-use std::str::FromStr;
+use std::{marker::PhantomData, str::FromStr};
 
 use snafu::{ResultExt, Snafu};
 use stackable_operator::{
     builder::meta::ObjectMetaBuilder,
+    kvp::Labels,
     v2::{
         builder::meta::ownerreference_from_resource,
         types::{common::Port, operator::RoleGroupName},
@@ -15,12 +16,13 @@ use stackable_operator::{
 
 use crate::{
     controller::{
-        KubernetesResources, ValidatedCluster,
+        KubernetesResources, Prepared, ValidatedCluster,
         build::resource::{
             config_map::build_rolegroup_config_map,
             listener::{build_group_listener, group_listener_name},
             pdb::build_pdb,
             rbac::{build_role_binding, build_service_account},
+            secret::build_secrets,
             service::{build_rolegroup_headless_service, build_rolegroup_metrics_service},
             statefulset::build_node_rolegroup_statefulset,
         },
@@ -49,6 +51,9 @@ pub const BALANCE_PORT: Port = Port(6243);
 // Filesystem paths shared by multiple builders. Single-consumer paths live in their builder.
 pub const NIFI_CONFIG_DIRECTORY: &str = "/stackable/nifi/conf";
 pub const NIFI_PYTHON_WORKING_DIRECTORY: &str = "/nifi-python-working-directory";
+/// Mount path of the sensitive-properties key Secret, whose contents are keyed by
+/// [`SENSITIVE_PROPERTY_KEY_NAME`](resource::secret::SENSITIVE_PROPERTY_KEY_NAME).
+pub const SENSITIVE_PROPERTY_VOLUME_MOUNT: &str = "/stackable/sensitiveproperty";
 
 #[derive(Snafu, Debug)]
 pub enum Error {
@@ -63,14 +68,17 @@ pub enum Error {
         source: resource::statefulset::Error,
         role_group: RoleGroupName,
     },
+
+    #[snafu(display("failed to build the Secrets"))]
+    Secrets { source: resource::secret::Error },
 }
 
 /// Builds every Kubernetes resource for the given validated cluster.
 ///
 /// Does not need a Kubernetes client: every reference to another Kubernetes resource is already
 /// dereferenced and validated by this point, so the errors returned here are resource-assembly
 /// failures only.
-pub fn build(cluster: &ValidatedCluster) -> Result<KubernetesResources, Error> {
+pub fn build(cluster: &ValidatedCluster) -> Result<KubernetesResources<Prepared>, Error> {
     let mut stateful_sets = vec![];
     let mut services = vec![];
     let mut listeners = vec![];
@@ -119,31 +127,34 @@ pub fn build(cluster: &ValidatedCluster) -> Result<KubernetesResources, Error> {
         services,
         listeners,
         config_maps,
+        secrets: build_secrets(cluster).context(SecretsSnafu)?,
         pod_disruption_budgets,
         service_accounts: vec![build_service_account(cluster)],
         role_bindings: vec![build_role_binding(cluster)],
+        status: PhantomData,
     })
 }
 
-/// Returns an [`ObjectMetaBuilder`] pre-filled with the namespace, an owner reference back to
-/// the cluster, and the recommended labels for a resource named `name` in `role_group_name`.
+/// Returns an [`ObjectMetaBuilder`] pre-filled with the cluster's namespace, an owner reference
+/// back to the cluster, the resource `name` and the given `recommended_labels`.
 ///
 /// Consolidates the metadata chain repeated by the child-resource builders. Call sites that
-/// need extra labels/annotations chain them onto the returned builder. Role-level resources
-/// (e.g. the per-role [`Listener`](stackable_operator::crd::listener::v1alpha1::Listener)) pass
-/// the placeholder role-group `none`, preserving the historical
-/// `app.kubernetes.io/role-group: none` label.
+/// need extra labels/annotations chain them onto the returned builder. The labels are passed in
+/// rather than derived here, so callers can pick the variant they need: role-level resources
+/// (e.g. the per-role [`Listener`](stackable_operator::crd::listener::v1alpha1::Listener)) use the
+/// placeholder role group `none`, and resources that must not change after deployment use the
+/// unversioned labels.
 pub(crate) fn object_meta(
     cluster: &ValidatedCluster,
     name: impl Into<String>,
-    role_group_name: &RoleGroupName,
+    recommended_labels: Labels,
 ) -> ObjectMetaBuilder {
     let mut builder = ObjectMetaBuilder::new();
     builder
         .name_and_namespace(cluster)
         .name(name)
         .ownerreference(ownerreference_from_resource(cluster, None, Some(true)))
-        .with_labels(cluster.recommended_labels(role_group_name));
+        .with_labels(recommended_labels);
     builder
 }
 
@@ -184,6 +195,12 @@ mod tests {
             sorted_names(&resources.pod_disruption_budgets),
             ["simple-nifi-node"]
         );
+        // The sensitive-properties key Secret, generated because the fixture has none yet. The
+        // OIDC admin password Secret is absent because the fixture uses SingleUser authentication.
+        assert_eq!(
+            sorted_names(&resources.secrets),
+            ["simple-nifi-sensitive-property-key"]
+        );
         // The cluster-shared RBAC pair.
         assert_eq!(
             sorted_names(&resources.service_accounts),

rust/operator-binary/src/controller/build/properties.rs

@@ -95,7 +95,8 @@ pub(crate) mod test_support {
     use crate::{
         controller::{
             NifiRoleGroupConfig, ValidatedCluster, ValidatedClusterConfig, ValidatedRoleConfig,
-            ValidatedSensitiveProperties, validate::build_role_group_configs,
+            ValidatedSensitiveProperties, dereference::ExistingSecrets,
+            validate::build_role_group_configs,
         },
         crd::{NifiRole, v1alpha1},
         security::{
@@ -179,6 +180,8 @@ pub(crate) mod test_support {
         let uid = Uid::from_str("e6ac237d-a6d4-43a1-8135-f36506110912").expect("valid uid");
         let product_version = ProductVersion::from_str(&image.app_version_label_value)
             .expect("valid product version");
+        let deployed_product_version =
+            ProductVersion::from_str(&image.product_version).expect("valid product version");
 
         ValidatedCluster::new(
             name,
@@ -187,6 +190,7 @@ pub(crate) mod test_support {
             uid,
             image,
             product_version,
+            deployed_product_version,
             role_config,
             role_group_configs,
             ValidatedClusterConfig {
@@ -213,6 +217,8 @@ pub(crate) mod test_support {
                 extra_volumes: nifi.spec.cluster_config.extra_volumes.clone(),
                 host_header_check: nifi.spec.cluster_config.host_header_check.clone(),
             },
+            // As on the first reconcile run: neither Secret exists yet.
+            ExistingSecrets::default(),
         )
     }