[AIRFLOW-5681] Allow specification of a tag or hash for the git_sync …

…init container (apache#6350)

Co-authored-by: Ash Berlin-Taylor <[email protected]>

airflow/config_templates/default_airflow.cfg

@@ -691,6 +691,11 @@ env_from_secret_ref =
 git_repo =
 git_branch =
 git_subpath =
+
+# The specific rev or hash the git_sync init container will checkout
+# This becomes GIT_SYNC_REV environment variable in the git_sync init container for worker pods
+git_sync_rev =
+
 # Use git_user and git_password for user authentication or git_ssh_key_secret_name and git_ssh_key_secret_key
 # for SSH authentication
 git_user =

airflow/executors/kubernetes_executor.py

@@ -120,6 +120,8 @@ def __init__(self):  # pylint: disable=too-many-statements
         self.git_sync_root = conf.get(self.kubernetes_section, 'git_sync_root')
         # Optionally, the name at which to publish the checked-out files under --root
         self.git_sync_dest = conf.get(self.kubernetes_section, 'git_sync_dest')
+        # Optionally, the tag or hash to checkout
+        self.git_sync_rev = conf.get(self.kubernetes_section, 'git_sync_rev')
         # Optionally, if git_dags_folder_mount_point is set the worker will use
         # {git_dags_folder_mount_point}/{git_sync_dest}/{git_subpath} as dags_folder
         self.git_dags_folder_mount_point = conf.get(self.kubernetes_section,

airflow/kubernetes/worker_configuration.py

@@ -65,6 +65,9 @@ def _get_init_containers(self) -> List[k8s.V1Container]:
         ), k8s.V1EnvVar(
             name='GIT_SYNC_DEST',
             value=self.kube_config.git_sync_dest
+        ), k8s.V1EnvVar(
+            name='GIT_SYNC_REV',
+            value=self.kube_config.git_sync_rev
         ), k8s.V1EnvVar(
             name='GIT_SYNC_DEPTH',
             value='1'

tests/kubernetes/test_worker_configuration.py

@@ -438,6 +438,30 @@ def test_make_pod_git_sync_credentials_secret(self):
         self.assertIn(password_env, pod.spec.init_containers[0].env,
                       'The password env for git credentials did not get into the init container')
 
+    def test_make_pod_git_sync_rev(self):
+        # Tests the pod created with git_sync_credentials_secret will get into the init container
+        self.kube_config.git_sync_rev = 'sampletag'
+        self.kube_config.dags_volume_claim = None
+        self.kube_config.dags_volume_host = None
+        self.kube_config.dags_in_image = None
+        self.kube_config.worker_fs_group = None
+        self.kube_config.git_dags_folder_mount_point = 'dags'
+        self.kube_config.git_sync_dest = 'repo'
+        self.kube_config.git_subpath = 'path'
+
+        worker_config = WorkerConfiguration(self.kube_config)
+
+        pod = worker_config.make_pod("default", str(uuid.uuid4()), "test_pod_id", "test_dag_id",
+                                     "test_task_id", str(datetime.utcnow()), 1, "bash -c 'ls /'")
+
+        rev_env = k8s.V1EnvVar(
+            name='GIT_SYNC_REV',
+            value=self.kube_config.git_sync_rev,
+        )
+
+        self.assertIn(rev_env, pod.spec.init_containers[0].env,
+                      'The git_sync_rev env did not get into the init container')
+
     def test_make_pod_git_sync_ssh_with_known_hosts(self):
         # Tests the pod created with git-sync SSH authentication option is correct with known hosts
         self.kube_config.airflow_configmap = 'airflow-configmap'