implement KubernetesExecutor

deploy/helm/airflow-operator/templates/roles.yaml

@@ -176,6 +176,18 @@ rules:
       - serviceaccounts
     verbs:
       - get
+  - apiGroups:
+      - ""
+    resources:
+      - pods
+    verbs:
+      - create
+      - delete
+      - get
+      - list
+      - patch
+      - update
+      - watch
   - apiGroups:
       - events.k8s.io
     resources:

rust/crd/src/lib.rs

@@ -159,8 +159,7 @@ pub struct AirflowClusterConfig {
     pub dags_git_sync: Vec<GitSync>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub database_initialization: Option<airflowdb::AirflowDbConfigFragment>,
-    #[serde(default, skip_serializing_if = "Option::is_none")]
-    pub executor: Option<String>,
+    pub executor: AirflowExecutor,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub expose_config: Option<bool>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
@@ -303,20 +302,53 @@ pub enum AirflowRole {
     Worker,
 }
 
+#[derive(
+    Clone,
+    Debug,
+    Default,
+    Deserialize,
+    Display,
+    EnumIter,
+    Eq,
+    Hash,
+    JsonSchema,
+    PartialEq,
+    Serialize,
+    EnumString,
+)]
+pub enum AirflowExecutor {
+    #[strum(serialize = "CeleryExecutor")]
+    #[default]
+    CeleryExecutor,
+    #[strum(serialize = "KubernetesExecutor")]
+    KubernetesExecutor,
+    #[strum(serialize = "CeleryKubernetesExecutor")]
+    CeleryKubernetesExecutor,
+}
+
 impl AirflowRole {
     /// Returns the start commands for the different airflow components. Airflow expects all
     /// components to have the same image/configuration (e.g. DAG folder location), even if not all
     /// configuration settings are used everywhere. For this reason we ensure that the webserver
     /// config file is in the Airflow home directory on all pods.
-    pub fn get_commands(&self) -> Vec<String> {
+    pub fn get_commands(&self, executor: &AirflowExecutor) -> Vec<String> {
         let copy_config = format!(
             "cp -RL {CONFIG_PATH}/{AIRFLOW_CONFIG_FILENAME} \
             {AIRFLOW_HOME}/{AIRFLOW_CONFIG_FILENAME}"
         );
         match &self {
             AirflowRole::Webserver => vec![copy_config, "airflow webserver".to_string()],
             AirflowRole::Scheduler => vec![copy_config, "airflow scheduler".to_string()],
-            AirflowRole::Worker => vec![copy_config, "airflow celery worker".to_string()],
+            AirflowRole::Worker => {
+                match executor {
+                    // TODO-ke which command for CeleryKubernetes?
+                    &AirflowExecutor::CeleryExecutor
+                    | &AirflowExecutor::CeleryKubernetesExecutor => {
+                        vec![copy_config, "airflow celery worker".to_string()]
+                    }
+                    &AirflowExecutor::KubernetesExecutor => vec![],
+                }
+            }
         }
     }
 
@@ -706,7 +738,7 @@ mod tests {
         assert_eq!("2.6.1", &resolved_airflow_image.product_version);
         assert_eq!(
             "KubernetesExecutor",
-            cluster.spec.cluster_config.executor.unwrap_or_default()
+            cluster.spec.cluster_config.executor.to_string()
         );
         assert!(cluster.spec.cluster_config.load_examples.unwrap_or(false));
         assert!(cluster.spec.cluster_config.expose_config.unwrap_or(false));

rust/operator-binary/Cargo.toml

@@ -14,6 +14,7 @@ clap = "4.1"
 fnv = "1.0"
 futures = { version = "0.3", features = ["compat"] }
 serde = "1.0"
+serde_yaml = "0.8"
 snafu = "0.7"
 stackable-airflow-crd = { path = "../crd" }
 stackable-operator = { git = "https://github.com/stackabletech/operator-rs.git", tag = "0.44.0" }
@@ -26,5 +27,3 @@ built = { version =  "0.6", features = ["chrono", "git2"] }
 stackable-airflow-crd = { path = "../crd" }
 stackable-operator = { git = "https://github.com/stackabletech/operator-rs.git", tag = "0.44.0" }
 
-[dev-dependencies]
-serde_yaml = "0.8"