[AIRFLOW-3766] Add support for kubernetes annotations

airflow/config_templates/default_airflow.cfg

@@ -688,6 +688,10 @@ tolerations =
 # The worker pods will be scheduled to the nodes of the specified key-value pairs.
 # Should be supplied in the format: key = value
 
+[kubernetes_annotations]
+# The Key-value annotations pairs to be given to worker pods.
+# Should be supplied in the format: key = value
+
 [kubernetes_secrets]
 # The scheduler mounts the following secrets into your workers as they are launched by the
 # scheduler. You may define as many secrets as needed and the kubernetes launcher will parse the

airflow/contrib/executors/kubernetes_executor.py

@@ -135,6 +135,7 @@ def __init__(self):
             self.kubernetes_section, "worker_container_image_pull_policy"
         )
         self.kube_node_selectors = configuration_dict.get('kubernetes_node_selectors', {})
+        self.kube_annotations = configuration_dict.get('kubernetes_annotations', {})
         self.delete_worker_pods = conf.getboolean(
             self.kubernetes_section, 'delete_worker_pods')
         self.worker_pods_creation_batch_size = conf.getint(

airflow/contrib/kubernetes/worker_configuration.py

@@ -222,7 +222,7 @@ def make_pod(self, namespace, worker_uuid, pod_id, dag_id, task_id, execution_da
             limit_cpu=kube_executor_config.limit_cpu
         )
         gcp_sa_key = kube_executor_config.gcp_service_account_key
-        annotations = dict(kube_executor_config.annotations)
+        annotations = dict(kube_executor_config.annotations) or self.kube_config.kube_annotations
         if gcp_sa_key:
             annotations['iam.cloud.google.com/service-account'] = gcp_sa_key
 

scripts/ci/kubernetes/kube/templates/configmaps.template.yaml

@@ -206,6 +206,10 @@ data:
     # The worker pods will be scheduled to the nodes of the specified key-value pairs.
     # Should be supplied in the format: key = value
 
+    [kubernetes_annotations]
+    # The Key-value annotations pairs to be given to worker pods.
+    # Should be supplied in the format: key = value
+
     [kubernetes_secrets]
     SQL_ALCHEMY_CONN = airflow-secrets=sql_alchemy_conn