refactor: move secret creation from dereference to reconciler

Secret creation is a side effect that should not happen before the
cluster spec has been validated. Moved it to the reconciler, after
validate_cluster succeeds.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: adwk67

rust/operator-binary/src/airflow_controller.rs

@@ -30,7 +30,10 @@ use stackable_operator::{
     },
     cli::OperatorEnvironmentOptions,
     cluster_resources::{ClusterResourceApplyStrategy, ClusterResources},
-    commons::{product_image_selection::ResolvedProductImage, rbac::build_rbac_resources},
+    commons::{
+        product_image_selection::ResolvedProductImage, random_secret_creation,
+        rbac::build_rbac_resources,
+    },
     crd::{authentication::ldap, git_sync, listener},
     database_connections::{
         TemplatingMechanism,
@@ -90,7 +93,11 @@ use crate::{
             AirflowAuthenticationClassResolved, AirflowClientAuthenticationDetailsResolved,
         },
         authorization::AirflowAuthorizationResolved,
-        build_recommended_labels, v1alpha2,
+        build_recommended_labels,
+        internal_secret::{
+            FERNET_KEY_SECRET_KEY, INTERNAL_SECRET_SECRET_KEY, JWT_SECRET_SECRET_KEY,
+        },
+        v1alpha2,
     },
     env_vars::{self, build_airflow_template_envs},
     operations::{
@@ -208,6 +215,11 @@ pub enum Error {
         source: stackable_operator::client::Error,
     },
 
+    #[snafu(display("failed to create internal secret"))]
+    InternalSecret {
+        source: random_secret_creation::Error,
+    },
+
     #[snafu(display("failed to dereference cluster resources"))]
     Dereference {
         source: crate::controller::dereference::Error,
@@ -378,6 +390,41 @@ pub async fn reconcile_airflow(
     )
     .context(ValidateSnafu)?;
 
+    // TODO: Move secret creation to a dedicated apply step once it exists.
+    random_secret_creation::create_random_secret_if_not_exists(
+        &airflow.shared_internal_secret_secret_name(),
+        INTERNAL_SECRET_SECRET_KEY,
+        256,
+        airflow,
+        client,
+    )
+    .await
+    .context(InternalSecretSnafu)?;
+
+    random_secret_creation::create_random_secret_if_not_exists(
+        &airflow.shared_jwt_secret_secret_name(),
+        JWT_SECRET_SECRET_KEY,
+        256,
+        airflow,
+        client,
+    )
+    .await
+    .context(InternalSecretSnafu)?;
+
+    // https://airflow.apache.org/docs/apache-airflow/stable/security/secrets/fernet.html#security-fernet
+    // does not document how long the fernet key should be, but recommends using
+    // python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())"
+    // which returns `jUm21LuA76YZmrIa9u4eXRg0h0P24MDC9IDOmDvJbfw=`, which has 44 characters, which makes 32 bytes.
+    random_secret_creation::create_random_secret_if_not_exists(
+        &airflow.shared_fernet_key_secret_name(),
+        FERNET_KEY_SECRET_KEY,
+        32,
+        airflow,
+        client,
+    )
+    .await
+    .context(InternalSecretSnafu)?;
+
     let mut cluster_resources = ClusterResources::new(
         APP_NAME,
         OPERATOR_NAME,

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

@@ -1,11 +1,8 @@
 use snafu::{ResultExt, Snafu};
-use stackable_operator::commons::random_secret_creation;
 
 use crate::crd::{
     authentication::AirflowClientAuthenticationDetailsResolved,
-    authorization::AirflowAuthorizationResolved,
-    internal_secret::{FERNET_KEY_SECRET_KEY, INTERNAL_SECRET_SECRET_KEY, JWT_SECRET_SECRET_KEY},
-    v1alpha2,
+    authorization::AirflowAuthorizationResolved, v1alpha2,
 };
 
 #[derive(Snafu, Debug)]
@@ -19,11 +16,6 @@ pub enum Error {
     AuthorizationConfig {
         source: stackable_operator::commons::opa::Error,
     },
-
-    #[snafu(display("failed to create internal secret"))]
-    InternalSecret {
-        source: random_secret_creation::Error,
-    },
 }
 
 /// External references resolved during the dereference step.
@@ -51,40 +43,6 @@ pub async fn dereference(
     .await
     .context(AuthorizationConfigSnafu)?;
 
-    random_secret_creation::create_random_secret_if_not_exists(
-        &airflow.shared_internal_secret_secret_name(),
-        INTERNAL_SECRET_SECRET_KEY,
-        256,
-        airflow,
-        client,
-    )
-    .await
-    .context(InternalSecretSnafu)?;
-
-    random_secret_creation::create_random_secret_if_not_exists(
-        &airflow.shared_jwt_secret_secret_name(),
-        JWT_SECRET_SECRET_KEY,
-        256,
-        airflow,
-        client,
-    )
-    .await
-    .context(InternalSecretSnafu)?;
-
-    // https://airflow.apache.org/docs/apache-airflow/stable/security/secrets/fernet.html#security-fernet
-    // does not document how long the fernet key should be, but recommends using
-    // python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())"
-    // which returns `jUm21LuA76YZmrIa9u4eXRg0h0P24MDC9IDOmDvJbfw=`, which has 44 characters, which makes 32 bytes.
-    random_secret_creation::create_random_secret_if_not_exists(
-        &airflow.shared_fernet_key_secret_name(),
-        FERNET_KEY_SECRET_KEY,
-        32,
-        airflow,
-        client,
-    )
-    .await
-    .context(InternalSecretSnafu)?;
-
     Ok(DereferencedObjects {
         authentication_config,
         authorization_config,