wip: for 3.x stabilisation changes

Author: adwk67

deploy/helm/airflow-operator/crds/crds.yaml

@@ -1168,7 +1168,10 @@ fn build_server_rolegroup_statefulset(
                 AirflowRole::Scheduler => {
                     "OrderedReady" // Scheduler pods should start after another, since part of their startup phase is initializing the database, see crd/src/lib.rs
                 }
-                AirflowRole::Webserver | AirflowRole::Worker => "Parallel",
+                AirflowRole::Webserver
+                | AirflowRole::Worker
+                | AirflowRole::Processor
+                | AirflowRole::Triggerer => "Parallel",
             }
             .to_string(),
         ),

rust/operator-binary/src/airflow_controller.rs

@@ -217,6 +217,14 @@ pub mod versioned {
 
         #[serde(flatten)]
         pub executor: AirflowExecutor,
+
+        /// The `processor` role runs the DAG processor routine for DAG preparation.
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        pub processors: Option<Role<AirflowConfigFragment>>,
+
+        /// The `triggerer` role runs the triggerer process for use with deferrable DAG operators.
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        pub triggerers: Option<Role<AirflowConfigFragment>>,
     }
 
     #[derive(Clone, Deserialize, Debug, JsonSchema, PartialEq, Serialize)]
@@ -343,7 +351,10 @@ impl v1alpha1::AirflowCluster {
     pub fn group_listener_name(&self, role: &AirflowRole) -> Option<String> {
         match role {
             AirflowRole::Webserver => Some(role_service_name(&self.name_any(), &role.to_string())),
-            AirflowRole::Scheduler | AirflowRole::Worker => None,
+            AirflowRole::Scheduler
+            | AirflowRole::Worker
+            | AirflowRole::Processor
+            | AirflowRole::Triggerer => None,
         }
     }
 
@@ -357,6 +368,8 @@ impl v1alpha1::AirflowCluster {
                 .to_owned()
                 .map(extract_role_from_webserver_config),
             AirflowRole::Scheduler => self.spec.schedulers.to_owned(),
+            AirflowRole::Processor => self.spec.processors.to_owned(),
+            AirflowRole::Triggerer => self.spec.triggerers.to_owned(),
             AirflowRole::Worker => {
                 if let AirflowExecutor::CeleryExecutor { config } = &self.spec.executor {
                     Some(config.clone())
@@ -413,6 +426,24 @@ impl v1alpha1::AirflowCluster {
                         roles: AirflowRole::roles(),
                     })?
             }
+            AirflowRole::Processor => {
+                self.spec
+                    .processors
+                    .as_ref()
+                    .context(UnknownAirflowRoleSnafu {
+                        role: role.to_string(),
+                        roles: AirflowRole::roles(),
+                    })?
+            }
+            AirflowRole::Triggerer => {
+                self.spec
+                    .triggerers
+                    .as_ref()
+                    .context(UnknownAirflowRoleSnafu {
+                        role: role.to_string(),
+                        roles: AirflowRole::roles(),
+                    })?
+            }
         };
 
         // Retrieve role resource config
@@ -561,6 +592,12 @@ pub enum AirflowRole {
 
     #[strum(serialize = "worker")]
     Worker,
+
+    #[strum(serialize = "processor")]
+    Processor,
+
+    #[strum(serialize = "triggerer")]
+    Triggerer,
 }
 
 impl AirflowRole {
@@ -622,10 +659,27 @@ impl AirflowRole {
                     command.extend(vec![
                         "prepare_signal_handlers".to_string(),
                         container_debug_command(),
-                        "airflow dag-processor &".to_string(),
                         "airflow scheduler &".to_string(),
                     ]);
+                    if airflow.spec.processors.is_none() {
+                        // If no processors role has been specified, the
+                        // process needs to be included with the scheduler
+                        // (with 3.x there is no longer the possibility of
+                        // starting it as a subprocess, so it has to be
+                        // explicitly started *somewhere*)
+                        command.extend(vec!["airflow dag-processor &".to_string()]);
+                    }
                 }
+                AirflowRole::Processor => command.extend(vec![
+                    "prepare_signal_handlers".to_string(),
+                    container_debug_command(),
+                    "airflow dag-processor &".to_string(),
+                ]),
+                AirflowRole::Triggerer => command.extend(vec![
+                    "prepare_signal_handlers".to_string(),
+                    container_debug_command(),
+                    "airflow triggerer &".to_string(),
+                ]),
                 AirflowRole::Worker => command.extend(vec![
                     "prepare_signal_handlers".to_string(),
                     container_debug_command(),
@@ -668,6 +722,16 @@ impl AirflowRole {
                         "airflow scheduler &".to_string(),
                     ]);
                 }
+                AirflowRole::Processor => command.extend(vec![
+                    "prepare_signal_handlers".to_string(),
+                    container_debug_command(),
+                    "airflow dag-processor &".to_string(),
+                ]),
+                AirflowRole::Triggerer => command.extend(vec![
+                    "prepare_signal_handlers".to_string(),
+                    container_debug_command(),
+                    "airflow triggerer &".to_string(),
+                ]),
                 AirflowRole::Worker => command.extend(vec![
                     "prepare_signal_handlers".to_string(),
                     container_debug_command(),
@@ -724,6 +788,8 @@ impl AirflowRole {
             AirflowRole::Webserver => Some(HTTP_PORT),
             AirflowRole::Scheduler => None,
             AirflowRole::Worker => None,
+            AirflowRole::Processor => None,
+            AirflowRole::Triggerer => None,
         }
     }
 
@@ -742,7 +808,7 @@ impl AirflowRole {
                 .webservers
                 .to_owned()
                 .map(|webserver| webserver.role_config.listener_class),
-            Self::Worker | Self::Scheduler => None,
+            Self::Worker | Self::Scheduler | Self::Processor | Self::Triggerer => None,
         }
     }
 }
@@ -878,9 +944,10 @@ impl AirflowConfig {
             logging: product_logging::spec::default_logging(),
             affinity: get_affinity(cluster_name, role),
             graceful_shutdown_timeout: Some(match role {
-                AirflowRole::Webserver | AirflowRole::Scheduler => {
-                    DEFAULT_AIRFLOW_GRACEFUL_SHUTDOWN_TIMEOUT
-                }
+                AirflowRole::Webserver
+                | AirflowRole::Scheduler
+                | AirflowRole::Processor
+                | AirflowRole::Triggerer => DEFAULT_AIRFLOW_GRACEFUL_SHUTDOWN_TIMEOUT,
                 AirflowRole::Worker => DEFAULT_WORKER_GRACEFUL_SHUTDOWN_TIMEOUT,
             }),
         }
@@ -953,6 +1020,26 @@ fn default_resources(role: &AirflowRole) -> ResourcesFragment<AirflowStorageConf
                 runtime_limits: NoRuntimeLimitsFragment {},
             },
         ),
+        AirflowRole::Processor => (
+            CpuLimitsFragment {
+                min: Some(Quantity("1".to_owned())),
+                max: Some(Quantity("2".to_owned())),
+            },
+            MemoryLimitsFragment {
+                limit: Some(Quantity("1Gi".to_owned())),
+                runtime_limits: NoRuntimeLimitsFragment {},
+            },
+        ),
+        AirflowRole::Triggerer => (
+            CpuLimitsFragment {
+                min: Some(Quantity("1".to_owned())),
+                max: Some(Quantity("2".to_owned())),
+            },
+            MemoryLimitsFragment {
+                limit: Some(Quantity("1Gi".to_owned())),
+                runtime_limits: NoRuntimeLimitsFragment {},
+            },
+        ),
     };
 
     ResourcesFragment {

rust/operator-binary/src/crd/mod.rs

@@ -15,8 +15,8 @@ use stackable_operator::{
 
 use crate::{
     crd::{
-        AirflowConfig, AirflowExecutor, AirflowRole, ExecutorConfig, LOG_CONFIG_DIR,
-        STACKABLE_LOG_DIR, TEMPLATE_LOCATION, TEMPLATE_NAME,
+        AirflowExecutor, AirflowRole, ExecutorConfig, LOG_CONFIG_DIR, STACKABLE_LOG_DIR,
+        TEMPLATE_LOCATION, TEMPLATE_NAME,
         authentication::{
             AirflowAuthenticationClassResolved, AirflowClientAuthenticationDetailsResolved,
         },
@@ -61,7 +61,7 @@ const PYTHONPATH: &str = "PYTHONPATH";
 /// This key is only intended for use during experimental support and will
 /// be replaced with a secret at a later stage. See the issue covering
 /// this at <https://github.com/stackabletech/airflow-operator/issues/639>.
-const JWT_KEY: &str = "ThisKeyIsNotIntendedForProduction!";
+//const JWT_KEY: &str = "ThisKeyIsNotIntendedForProduction!";
 
 #[derive(Snafu, Debug)]
 pub enum Error {
@@ -86,57 +86,53 @@ pub fn build_airflow_statefulset_envs(
     resolved_product_image: &ResolvedProductImage,
 ) -> Result<Vec<EnvVar>, Error> {
     let mut env: BTreeMap<String, EnvVar> = BTreeMap::new();
+    let secret = airflow.spec.cluster_config.credentials_secret.as_str();
 
     env.extend(static_envs(git_sync_resources));
 
-    add_version_specific_env_vars(airflow, airflow_role, resolved_product_image, &mut env);
-
     // environment variables
     let env_vars = rolegroup_config.get(&PropertyNameKind::Env);
 
-    let secret_prop =
-        env_vars.and_then(|vars| vars.get(AirflowConfig::CREDENTIALS_SECRET_PROPERTY));
+    add_version_specific_env_vars(airflow, airflow_role, resolved_product_image, &mut env);
 
-    if let Some(secret) = secret_prop {
+    env.insert(
+        AIRFLOW_WEBSERVER_SECRET_KEY.into(),
+        // The secret key is used to run the webserver flask app and also used to authorize
+        // requests to Celery workers when logs are retrieved.
+        env_var_from_secret(
+            AIRFLOW_WEBSERVER_SECRET_KEY,
+            secret,
+            "connections.secretKey",
+        ),
+    );
+    env.insert(
+        AIRFLOW_DATABASE_SQL_ALCHEMY_CONN.into(),
+        env_var_from_secret(
+            AIRFLOW_DATABASE_SQL_ALCHEMY_CONN,
+            secret,
+            "connections.sqlalchemyDatabaseUri",
+        ),
+    );
+
+    // Redis is only needed when celery executors are used
+    // see https://github.com/stackabletech/airflow-operator/issues/424 for details
+    if matches!(executor, AirflowExecutor::CeleryExecutor { .. }) {
         env.insert(
-            AIRFLOW_WEBSERVER_SECRET_KEY.into(),
-            // The secret key is used to run the webserver flask app and also used to authorize
-            // requests to Celery workers when logs are retrieved.
+            AIRFLOW_CELERY_RESULT_BACKEND.into(),
             env_var_from_secret(
-                AIRFLOW_WEBSERVER_SECRET_KEY,
+                AIRFLOW_CELERY_RESULT_BACKEND,
                 secret,
-                "connections.secretKey",
+                "connections.celeryResultBackend",
             ),
         );
         env.insert(
-            AIRFLOW_DATABASE_SQL_ALCHEMY_CONN.into(),
+            AIRFLOW_CELERY_BROKER_URL.into(),
             env_var_from_secret(
-                AIRFLOW_DATABASE_SQL_ALCHEMY_CONN,
+                AIRFLOW_CELERY_BROKER_URL,
                 secret,
-                "connections.sqlalchemyDatabaseUri",
+                "connections.celeryBrokerUrl",
             ),
         );
-
-        // Redis is only needed when celery executors are used
-        // see https://github.com/stackabletech/airflow-operator/issues/424 for details
-        if matches!(executor, AirflowExecutor::CeleryExecutor { .. }) {
-            env.insert(
-                AIRFLOW_CELERY_RESULT_BACKEND.into(),
-                env_var_from_secret(
-                    AIRFLOW_CELERY_RESULT_BACKEND,
-                    secret,
-                    "connections.celeryResultBackend",
-                ),
-            );
-            env.insert(
-                AIRFLOW_CELERY_BROKER_URL.into(),
-                env_var_from_secret(
-                    AIRFLOW_CELERY_BROKER_URL,
-                    secret,
-                    "connections.celeryBrokerUrl",
-                ),
-            );
-        }
     }
 
     let dags_folder = get_dags_folder(git_sync_resources);
@@ -454,6 +450,8 @@ fn add_version_specific_env_vars(
     resolved_product_image: &ResolvedProductImage,
     env: &mut BTreeMap<String, EnvVar>,
 ) {
+    let secret = airflow.spec.cluster_config.credentials_secret.as_str();
+
     if resolved_product_image.product_version.starts_with("3.") {
         env.extend(execution_server_env_vars(airflow));
         env.insert(
@@ -483,13 +481,17 @@ fn add_version_specific_env_vars(
         // See issue <https://github.com/stackabletech/airflow-operator/issues/639>:
         // later it will be accessed from a secret to avoid cluster restarts
         // being triggered by an operator restart.
+        // env.insert(
+        //     "AIRFLOW__API_AUTH__JWT_SECRET".into(),
+        //     EnvVar {
+        //         name: "AIRFLOW__API_AUTH__JWT_SECRET".into(),
+        //         value: Some(JWT_KEY.into()),
+        //         ..Default::default()
+        //     },
+        // );
         env.insert(
             "AIRFLOW__API_AUTH__JWT_SECRET".into(),
-            EnvVar {
-                name: "AIRFLOW__API_AUTH__JWT_SECRET".into(),
-                value: Some(JWT_KEY.into()),
-                ..Default::default()
-            },
+            env_var_from_secret("AIRFLOW__API_AUTH__JWT_SECRET", secret, "jwt.key"),
         );
         if airflow_role == &AirflowRole::Webserver {
             // Sometimes a race condition can arise when both scheduler and
@@ -527,6 +529,19 @@ fn add_version_specific_env_vars(
                 ..Default::default()
             },
         );
+        if airflow.spec.processors.is_some() {
+            // In airflow 2.x the dag-processor can optionally be started as a
+            // standalone process (rather then as a scheduler subprocess),
+            // governed by this env-var
+            env.insert(
+                "AIRFLOW__SCHEDULER__STANDALONE_DAG_PROCESSOR".into(),
+                EnvVar {
+                    name: "AIRFLOW__SCHEDULER__STANDALONE_DAG_PROCESSOR".into(),
+                    value: Some("True".into()),
+                    ..Default::default()
+                },
+            );
+        }
     }
 }
 

rust/operator-binary/src/env_vars.rs

@@ -37,6 +37,8 @@ pub async fn add_pdbs(
     let max_unavailable = pdb.max_unavailable.unwrap_or(match role {
         AirflowRole::Scheduler => max_unavailable_schedulers(),
         AirflowRole::Webserver => max_unavailable_webservers(),
+        AirflowRole::Processor => max_unavailable_processors(),
+        AirflowRole::Triggerer => max_unavailable_triggerers(),
         AirflowRole::Worker => match airflow.spec.executor {
             AirflowExecutor::CeleryExecutor { .. } => max_unavailable_workers(),
             AirflowExecutor::KubernetesExecutor { .. } => {
@@ -77,3 +79,11 @@ fn max_unavailable_workers() -> u16 {
 fn max_unavailable_webservers() -> u16 {
     1
 }
+
+fn max_unavailable_processors() -> u16 {
+    1
+}
+
+fn max_unavailable_triggerers() -> u16 {
+    1
+}

rust/operator-binary/src/operations/pdb.rs

@@ -0,0 +1,25 @@
+---
+apiVersion: v1
+kind: Secret
+metadata:
+  name: minio-credentials
+  labels:
+    secrets.stackable.tech/class: spark-pi-private-s3-credentials-class
+timeout: 240
+stringData:
+  accessKey: minioAccessKey
+  secretKey: minioSecretKey
+  # The following two entries are used by the Bitnami chart for MinIO to
+  # set up credentials for accessing buckets managed by the MinIO tenant.
+  root-user: minioAccessKey
+  root-password: minioSecretKey
+---
+apiVersion: secrets.stackable.tech/v1alpha1
+kind: SecretClass
+metadata:
+  name: spark-pi-private-s3-credentials-class
+spec:
+  backend:
+    k8sSearch:
+      searchNamespace:
+        pod: {}