stackabletech · adwk67 · Aug 23, 2023 · Jul 28, 2023 · Aug 1, 2023 · Aug 1, 2023
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -4,6 +4,7 @@
 
 ### Added
 
+- [BREAKING] Implement KubernetesExecutor ([#311]).
 - Default stackableVersion to operator version ([#312]).
 
 ### Changed
@@ -18,6 +19,7 @@
 
 [#303]: https://github.com/stackabletech/airflow-operator/pull/303
 [#308]: https://github.com/stackabletech/airflow-operator/pull/308
+[#311]: https://github.com/stackabletech/airflow-operator/pull/311
 [#312]: https://github.com/stackabletech/airflow-operator/pull/312
 [#316]: https://github.com/stackabletech/airflow-operator/pull/316
 

diff --git a/deploy/helm/airflow-operator/crds/crds.yaml b/deploy/helm/airflow-operator/crds/crds.yaml
diff --git a/deploy/helm/airflow-operator/templates/roles.yaml b/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:

diff --git a/docs/modules/airflow/examples/example-airflow-dags-configmap.yaml b/docs/modules/airflow/examples/example-airflow-dags-configmap.yaml
@@ -7,7 +7,6 @@ spec:
   image:
     productVersion: 2.6.1
   clusterConfig:
-    executor: CeleryExecutor
     loadExamples: false
     exposeConfig: false
     credentialsSecret: simple-airflow-credentials
@@ -25,7 +24,7 @@ spec:
         envOverrides:
           AIRFLOW__CORE__DAGS_FOLDER: "/dags" # <8>
         replicas: 1
-  workers:
+  celeryExecutors:
     roleGroups:
       default:
         envOverrides:

diff --git a/docs/modules/airflow/examples/example-airflow-gitsync.yaml b/docs/modules/airflow/examples/example-airflow-gitsync.yaml
@@ -7,7 +7,6 @@ spec:
   image:
     productVersion: "2.6.1"
   clusterConfig:
-    executor: CeleryExecutor
     loadExamples: false
     exposeConfig: false
     credentialsSecret: test-airflow-credentials # <1>

diff --git a/docs/modules/airflow/examples/example-airflow-incluster.yaml b/docs/modules/airflow/examples/example-airflow-incluster.yaml
@@ -7,7 +7,6 @@ spec:
   image:
     productVersion: 2.6.1
   clusterConfig:
-    executor: CeleryExecutor
     loadExamples: false
     exposeConfig: false
     credentialsSecret: simple-airflow-credentials
@@ -17,7 +16,7 @@ spec:
         envOverrides:
           AIRFLOW_CONN_KUBERNETES_IN_CLUSTER: "kubernetes://?__extra__=%7B%22extra__kubernetes__in_cluster%22%3A+true%2C+%22extra__kubernetes__kube_config%22%3A+%22%22%2C+%22extra__kubernetes__kube_config_path%22%3A+%22%22%2C+%22extra__kubernetes__namespace%22%3A+%22%22%7D"
         replicas: 1
-  workers:
+  celeryExecutors:
     roleGroups:
       default:
         envOverrides:

diff --git a/docs/modules/airflow/examples/getting_started/code/airflow.yaml b/docs/modules/airflow/examples/getting_started/code/airflow.yaml
@@ -7,18 +7,24 @@ spec:
   image:
     productVersion: 2.6.1
   clusterConfig:
-    executor: CeleryExecutor
     loadExamples: true
     exposeConfig: false
     credentialsSecret: simple-airflow-credentials
   webservers:
     roleGroups:
       default:
         replicas: 1
-  workers:
+  celeryExecutors:
     roleGroups:
       default:
         replicas: 2
+    config:
+      resources:
+        cpu:
+          min: 400m
+          max: 800m
+        memory:
+          limit: 2Gi
   schedulers:
     roleGroups:
       default:

diff --git a/docs/modules/airflow/examples/getting_started/code/airflow.yaml.j2 b/docs/modules/airflow/examples/getting_started/code/airflow.yaml.j2
@@ -7,18 +7,24 @@ spec:
   image:
     productVersion: 2.6.1
   clusterConfig:
-    executor: CeleryExecutor
     loadExamples: true
     exposeConfig: false
     credentialsSecret: simple-airflow-credentials
   webservers:
     roleGroups:
       default:
         replicas: 1
-  workers:
+  celeryExecutors:
     roleGroups:
       default:
         replicas: 2
+    config:
+      resources:
+        cpu:
+          min: 400m
+          max: 800m
+        memory:
+          limit: 2Gi
   schedulers:
     roleGroups:
       default:

diff --git a/docs/modules/airflow/pages/getting_started/first_steps.adoc b/docs/modules/airflow/pages/getting_started/first_steps.adoc
@@ -56,14 +56,11 @@ Where:
 
 - `metadata.name` contains the name of the Airflow cluster
 - the label of the Docker image provided by Stackable must be set in `spec.version`
-- `spec.executor`: this setting determines how the cluster will run (for more information see https://airflow.apache.org/docs/apache-airflow/stable/executor/index.html#executor-types): the `CeleryExecutor`
-is the recommended setting although `SequentialExecutor` (all jobs run in one process in series) and `LocalExecutor`
-(whereby all jobs are run on one node, using whatever parallelism is possible) are also supported
+- `spec.executor`: this setting determines how the cluster will run (for more information see https://airflow.apache.org/docs/apache-airflow/stable/executor/index.html#executor-types). This field takes one of two values, `celery` or `kubernetes`: `celery` will deploy workers as defined by the role and specify `CeleryExecutor` internally. `kubernetes` will set `KubernetesExecutor` internally, and the scheduler will take responsibility for creating (and terminating) pods for the DAG individual tasks.
 - the `spec.loadExamples` key is optional and defaults to `false`. It is set to `true` here as the example DAGs will be used when verifying the installation.
 - the `spec.exposeConfig` key is optional and defaults to `false`. It is set to `true` only as an aid to verify the configuration and should never be used as such in anything other than test or demo clusters.
 - the previously created secret must be referenced in `spec.credentialsSecret`
 
-
 NOTE: Please note that the version you need to specify for `spec.version` is not only the version of Apache Airflow which you want to roll out, but has to be amended with a Stackable version as shown. This Stackable version is the version of the underlying container image which is used to execute the processes. For a list of available versions please check our
 https://repo.stackable.tech/#browse/browse:docker:v2%2Fstackable%airflow%2Ftags[image registry].
 It should generally be safe to simply use the latest image version that is available.

diff --git a/docs/modules/airflow/pages/getting_started/installation.adoc b/docs/modules/airflow/pages/getting_started/installation.adoc
@@ -4,7 +4,7 @@ On this page you will install the Stackable Airflow Operator, the software that
 
 == Required external components: Postgresql and Redis
 
-Postgresql and Redis are required by Airflow: Postgresql to store metadata about DAG runs, and Redis to schedule and/or queue DAG jobs. They are components that may well already be available for customers, in which case we treat them here as pre-requisites for an airflow cluster and hence as part of the installation process. These components will be installed using Helm. Note that specific versions are declared:
+Postgresql is required by Airflow to store metadata about DAG runs, and Redis is required by the Celery executor to schedule and/or queue DAG jobs. They are components that may well already be available for customers, in which case we treat them here as pre-requisites for an airflow cluster and hence as part of the installation process. These components will be installed using Helm. Note that specific versions are declared:
 
 [source,bash]
 ----

diff --git a/docs/modules/airflow/pages/index.adoc b/docs/modules/airflow/pages/index.adoc
@@ -15,10 +15,47 @@ The Operator manages three https://kubernetes.io/docs/concepts/extend-kubernetes
 
 === Custom resources
 
-The AirflowCluster is the main resource for the configuration of the Airflow instance. The resource defines three xref:concepts:roles-and-role-groups.adoc[roles]: `webserver`, `worker` and `scheduler`. The various configuration options are explained in the xref:usage-guide/index.adoc[]. It helps you tune your cluster to your needs by configuring xref:usage-guide/storage-resources.adoc[resource usage], xref:usage-guide/security.adoc[security], xref:usage-guide/logging.adoc[logging] and more.
+The AirflowCluster is the main resource for the configuration of the Airflow instance. The resource defines three xref:concepts:roles-and-role-groups.adoc[roles]: `webserver`, `worker` and `scheduler` (the `worker` role is embedded within `spec.celeryExecutors`: this is described in the next section). The various configuration options are explained in the xref:usage-guide/index.adoc[]. It helps you tune your cluster to your needs by configuring xref:usage-guide/storage-resources.adoc[resource usage], xref:usage-guide/security.adoc[security], xref:usage-guide/logging.adoc[logging] and more.
 
 When an AirflowCluster is first deployed, an AirflowDB resource is created. The AirflowDB resource is a wrapper resource for the metadata SQL database that is used by Airflow to store information on users and permissions as well as workflows, task instances and their execution. The resource contains some configuration but also keeps track of whether the database has been initialized or not. It is not deleted automatically if a AirflowCluster is deleted, and so can be reused.
 
+=== Executors
+
+The `worker` role is deployed when `spec.celeryExecutors` is specified (the alternative is `spec.kubernetesExecutors`, whereby pods are created dynamically as needed without jobs being routed through a redis queue to the workers). This means that for `celeryExecutors` there exists an implicit single role which does not appear in resource definition. This is illustrated below:
+
+==== `celeryExecutors`
+
+[source,yaml]
+----
+spec:
+...
+celeryExecutors:
+  roleGroups:
+    default:
+      envOverrides:
+        ...
+      configOverrides:
+        ...
+      replicas: 2
+  config:
+    logging:
+          ...
+----
+
+==== `kubernetesExecutors`
+
+[source,yaml]
+----
+spec:
+...
+kubernetesExecutors:
+  config:
+    logging:
+      ...
+    resources:
+      ...
+----
+
 === Kubernetes resources
 
 Based on the custom resources you define, the Operator creates ConfigMaps, StatefulSets and Services.
@@ -35,6 +72,8 @@ ConfigMaps are created, one per RoleGroup and also one for the AirflowDB. Both C
 
 Airflow requires an SQL database in which to store its metadata as well as Redis for job execution. The xref:required-external-components.adoc[required external components] page lists all supported databases and Redis versions to use in production. You need to provide these components for production use, but the xref:getting_started/index.adoc[] guides you through installing an example database and Redis instance with an Airflow instance that you can use to get started.
 
+NOTE: Redis is only needed if the executors have been set to `spec.celeryExecutors` as the jobs will be queued via Redis before being assigned to a `worker` pod. When using `spec.kubernetesExecutors` the scheduler will take direct responsibility for this.
+
 == Using custom workflows/DAGs
 
 https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/dags.html[Direct acyclic graphs (DAGs) of tasks] are the core entities you will use in Airflow. Have a look at the page on xref:usage-guide/mounting-dags.adoc[] to learn about the different ways of loading your custom DAGs into Airflow.

diff --git a/examples/simple-airflow-cluster-dags-cmap.yaml b/examples/simple-airflow-cluster-dags-cmap.yaml
@@ -84,7 +84,6 @@ spec:
     productVersion: 2.6.1
     stackableVersion: 0.0.0-dev
   clusterConfig:
-    executor: CeleryExecutor
     loadExamples: false
     exposeConfig: false
     credentialsSecret: simple-airflow-credentials
@@ -102,7 +101,7 @@ spec:
         envOverrides:
           AIRFLOW__CORE__DAGS_FOLDER: "/dags"
         replicas: 1
-  workers:
+  celeryExecutors:
     roleGroups:
       default:
         envOverrides:

diff --git a/examples/simple-airflow-cluster-ldap-insecure-tls.yaml b/examples/simple-airflow-cluster-ldap-insecure-tls.yaml
@@ -150,7 +150,6 @@ spec:
     productVersion: 2.6.1
     stackableVersion: 0.0.0-dev
   clusterConfig:
-    executor: CeleryExecutor
     loadExamples: true
     exposeConfig: true
     credentialsSecret: airflow-with-ldap-server-veri-tls-credentials
@@ -161,7 +160,7 @@ spec:
     roleGroups:
       default:
         replicas: 1
-  workers:
+  celeryExecutors:
     roleGroups:
       default:
         replicas: 1

diff --git a/examples/simple-airflow-cluster-ldap.yaml b/examples/simple-airflow-cluster-ldap.yaml
@@ -148,7 +148,6 @@ spec:
     productVersion: 2.6.1
     stackableVersion: 0.0.0-dev
   clusterConfig:
-    executor: CeleryExecutor
     loadExamples: true
     exposeConfig: true
     credentialsSecret: airflow-with-ldap-server-veri-tls-credentials
@@ -159,7 +158,7 @@ spec:
     roleGroups:
       default:
         replicas: 1
-  workers:
+  celeryExecutors:
     roleGroups:
       default:
         replicas: 1

diff --git a/examples/simple-airflow-cluster.yaml b/examples/simple-airflow-cluster.yaml
@@ -24,15 +24,14 @@ spec:
     productVersion: 2.6.1
     stackableVersion: 0.0.0-dev
   clusterConfig:
-    executor: CeleryExecutor
     loadExamples: true
     exposeConfig: false
     credentialsSecret: simple-airflow-credentials
   webservers:
     roleGroups:
       default:
         replicas: 1
-  workers:
+  celeryExecutors:
     roleGroups:
       default:
         replicas: 2

diff --git a/rust/crd/src/affinity.rs b/rust/crd/src/affinity.rs
@@ -51,36 +51,38 @@ mod tests {
     use crate::{AirflowCluster, AirflowRole};
 
     #[rstest]
-    #[case(AirflowRole::Worker)]
+    // #[case(AirflowRole::Worker)]
     #[case(AirflowRole::Scheduler)]
     #[case(AirflowRole::Webserver)]
     fn test_affinity_defaults(#[case] role: AirflowRole) {
-        let input = r#"
+        let cluster = "
         apiVersion: airflow.stackable.tech/v1alpha1
         kind: AirflowCluster
         metadata:
           name: airflow
         spec:
           image:
             productVersion: 2.6.1
-          executor: CeleryExecutor
           loadExamples: true
           exposeConfig: false
           credentialsSecret: simple-airflow-credentials
           webservers:
             roleGroups:
               default:
                 replicas: 1
-          workers:
+          celeryExecutors:
             roleGroups:
               default:
                 replicas: 2
           schedulers:
             roleGroups:
               default:
                 replicas: 1
-        "#;
-        let airflow: AirflowCluster = serde_yaml::from_str(input).expect("illegal test input");
+        ";
+
+        let deserializer = serde_yaml::Deserializer::from_str(cluster);
+        let airflow: AirflowCluster =
+            serde_yaml::with::singleton_map_recursive::deserialize(deserializer).unwrap();
 
         let rolegroup_ref = RoleGroupRef {
             cluster: ObjectRef::from_obj(&airflow),
@@ -150,23 +152,22 @@ mod tests {
 
     #[test]
     fn test_affinity_legacy_node_selector() {
-        let input = r#"
+        let cluster = "
         apiVersion: airflow.stackable.tech/v1alpha1
         kind: AirflowCluster
         metadata:
           name: airflow
         spec:
           image:
             productVersion: 2.6.1
-          executor: CeleryExecutor
           loadExamples: true
           exposeConfig: false
           credentialsSecret: simple-airflow-credentials
           webservers:
             roleGroups:
               default:
                 replicas: 1
-          workers:
+          celeryExecutors:
             roleGroups:
               default:
                 replicas: 2
@@ -183,9 +184,11 @@ mod tests {
                       values:
                         - antarctica-east1
                         - antarctica-west1
-        "#;
+        ";
 
-        let airflow: AirflowCluster = serde_yaml::from_str(input).expect("illegal test input");
+        let deserializer = serde_yaml::Deserializer::from_str(cluster);
+        let airflow: AirflowCluster =
+            serde_yaml::with::singleton_map_recursive::deserialize(deserializer).unwrap();
 
         let expected: StackableAffinity = StackableAffinity {
             node_affinity: Some(NodeAffinity {