added roles for triggerer and dag-processor

Author: adwk67

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

@@ -1169,7 +1169,7 @@ fn build_server_rolegroup_statefulset(
                 }
                 AirflowRole::Webserver
                 | AirflowRole::Worker
-                | AirflowRole::Processor
+                | AirflowRole::DagProcessor
                 | AirflowRole::Triggerer => "Parallel",
             }
             .to_string(),

rust/operator-binary/src/airflow_controller.rs

@@ -205,23 +205,23 @@ pub mod versioned {
         #[serde(default)]
         pub cluster_operation: ClusterOperation,
 
-        /// The `webserver` role provides the main UI for user interaction.
+        /// The `webservers` role provides the main UI for user interaction.
         #[serde(default, skip_serializing_if = "Option::is_none")]
         pub webservers: Option<Role<AirflowConfigFragment, v1alpha1::WebserverRoleConfig>>,
 
-        /// The `scheduler` is responsible for triggering jobs and persisting their metadata to the backend database.
+        /// The `schedulers` is responsible for triggering jobs and persisting their metadata to the backend database.
         /// Jobs are scheduled on the workers/executors.
         #[serde(default, skip_serializing_if = "Option::is_none")]
         pub schedulers: Option<Role<AirflowConfigFragment>>,
 
         #[serde(flatten)]
         pub executor: AirflowExecutor,
 
-        /// The `processor` role runs the DAG processor routine for DAG preparation.
+        /// The `dagProcessors` role runs the DAG processor routine for DAG preparation.
         #[serde(default, skip_serializing_if = "Option::is_none")]
-        pub processors: Option<Role<AirflowConfigFragment>>,
+        pub dag_processors: Option<Role<AirflowConfigFragment>>,
 
-        /// The `triggerer` role runs the triggerer process for use with deferrable DAG operators.
+        /// The `triggerers` role runs the triggerer process for use with deferrable DAG operators.
         #[serde(default, skip_serializing_if = "Option::is_none")]
         pub triggerers: Option<Role<AirflowConfigFragment>>,
     }
@@ -352,7 +352,7 @@ impl v1alpha1::AirflowCluster {
             AirflowRole::Webserver => Some(role_service_name(&self.name_any(), &role.to_string())),
             AirflowRole::Scheduler
             | AirflowRole::Worker
-            | AirflowRole::Processor
+            | AirflowRole::DagProcessor
             | AirflowRole::Triggerer => None,
         }
     }
@@ -367,7 +367,7 @@ 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::DagProcessor => self.spec.dag_processors.to_owned(),
             AirflowRole::Triggerer => self.spec.triggerers.to_owned(),
             AirflowRole::Worker => {
                 if let AirflowExecutor::CeleryExecutor { config } = &self.spec.executor {
@@ -429,9 +429,9 @@ impl v1alpha1::AirflowCluster {
                         roles: AirflowRole::roles(),
                     })?
             }
-            AirflowRole::Processor => {
+            AirflowRole::DagProcessor => {
                 self.spec
-                    .processors
+                    .dag_processors
                     .as_ref()
                     .context(UnknownAirflowRoleSnafu {
                         role: role.to_string(),
@@ -596,8 +596,8 @@ pub enum AirflowRole {
     #[strum(serialize = "worker")]
     Worker,
 
-    #[strum(serialize = "processor")]
-    Processor,
+    #[strum(serialize = "dagprocessor")]
+    DagProcessor,
 
     #[strum(serialize = "triggerer")]
     Triggerer,
@@ -664,16 +664,16 @@ impl AirflowRole {
                         container_debug_command(),
                         "airflow scheduler &".to_string(),
                     ]);
-                    if airflow.spec.processors.is_none() {
-                        // If no processors role has been specified, the
+                    if airflow.spec.dag_processors.is_none() {
+                        // If no dag_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![
+                AirflowRole::DagProcessor => command.extend(vec![
                     "prepare_signal_handlers".to_string(),
                     container_debug_command(),
                     "airflow dag-processor &".to_string(),
@@ -725,7 +725,7 @@ impl AirflowRole {
                         "airflow scheduler &".to_string(),
                     ]);
                 }
-                AirflowRole::Processor => command.extend(vec![
+                AirflowRole::DagProcessor => command.extend(vec![
                     "prepare_signal_handlers".to_string(),
                     container_debug_command(),
                     "airflow dag-processor &".to_string(),
@@ -791,7 +791,7 @@ impl AirflowRole {
             AirflowRole::Webserver => Some(HTTP_PORT),
             AirflowRole::Scheduler => None,
             AirflowRole::Worker => None,
-            AirflowRole::Processor => None,
+            AirflowRole::DagProcessor => None,
             AirflowRole::Triggerer => None,
         }
     }
@@ -811,7 +811,7 @@ impl AirflowRole {
                 .webservers
                 .to_owned()
                 .map(|webserver| webserver.role_config.listener_class),
-            Self::Worker | Self::Scheduler | Self::Processor | Self::Triggerer => None,
+            Self::Worker | Self::Scheduler | Self::DagProcessor | Self::Triggerer => None,
         }
     }
 }
@@ -949,7 +949,7 @@ impl AirflowConfig {
             graceful_shutdown_timeout: Some(match role {
                 AirflowRole::Webserver
                 | AirflowRole::Scheduler
-                | AirflowRole::Processor
+                | AirflowRole::DagProcessor
                 | AirflowRole::Triggerer => DEFAULT_AIRFLOW_GRACEFUL_SHUTDOWN_TIMEOUT,
                 AirflowRole::Worker => DEFAULT_WORKER_GRACEFUL_SHUTDOWN_TIMEOUT,
             }),
@@ -1023,7 +1023,7 @@ fn default_resources(role: &AirflowRole) -> ResourcesFragment<AirflowStorageConf
                 runtime_limits: NoRuntimeLimitsFragment {},
             },
         ),
-        AirflowRole::Processor => (
+        AirflowRole::DagProcessor => (
             CpuLimitsFragment {
                 min: Some(Quantity("1".to_owned())),
                 max: Some(Quantity("2".to_owned())),

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

@@ -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 {
@@ -450,8 +450,6 @@ 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(
@@ -481,17 +479,13 @@ 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(),
-            env_var_from_secret("AIRFLOW__API_AUTH__JWT_SECRET", secret, "jwt.key"),
+            EnvVar {
+                name: "AIRFLOW__API_AUTH__JWT_SECRET".into(),
+                value: Some(JWT_KEY.into()),
+                ..Default::default()
+            },
         );
         if airflow_role == &AirflowRole::Webserver {
             // Sometimes a race condition can arise when both scheduler and
@@ -529,10 +523,10 @@ fn add_version_specific_env_vars(
                 ..Default::default()
             },
         );
-        if airflow.spec.processors.is_some() {
+        if airflow.spec.dag_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
+            // accompanied by this env-var being set to True.
             env.insert(
                 "AIRFLOW__SCHEDULER__STANDALONE_DAG_PROCESSOR".into(),
                 EnvVar {

rust/operator-binary/src/env_vars.rs

@@ -37,7 +37,7 @@ 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::DagProcessor => max_unavailable_dag_processors(),
         AirflowRole::Triggerer => max_unavailable_triggerers(),
         AirflowRole::Worker => match airflow.spec.executor {
             AirflowExecutor::CeleryExecutor { .. } => max_unavailable_workers(),
@@ -80,7 +80,7 @@ fn max_unavailable_webservers() -> u16 {
     1
 }
 
-fn max_unavailable_processors() -> u16 {
+fn max_unavailable_dag_processors() -> u16 {
     1
 }
 

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

@@ -51,6 +51,30 @@ status:
   readyReplicas: 1
   replicas: 1
 ---
+apiVersion: apps/v1
+kind: StatefulSet
+metadata:
+  name: airflow-dagprocessor-default
+spec:
+  template:
+    spec:
+      terminationGracePeriodSeconds: 120
+status:
+  readyReplicas: 1
+  replicas: 1
+---
+apiVersion: apps/v1
+kind: StatefulSet
+metadata:
+  name: airflow-triggerer-default
+spec:
+  template:
+    spec:
+      terminationGracePeriodSeconds: 120
+status:
+  readyReplicas: 1
+  replicas: 1
+---
 apiVersion: policy/v1
 kind: PodDisruptionBudget
 metadata:
@@ -79,3 +103,21 @@ status:
   expectedPods: 1
   currentHealthy: 1
   disruptionsAllowed: 1
+---
+apiVersion: policy/v1
+kind: PodDisruptionBudget
+metadata:
+  name: airflow-dagprocessor
+status:
+  expectedPods: 1
+  currentHealthy: 1
+  disruptionsAllowed: 1
+---
+apiVersion: policy/v1
+kind: PodDisruptionBudget
+metadata:
+  name: airflow-triggerer
+status:
+  expectedPods: 1
+  currentHealthy: 1
+  disruptionsAllowed: 1