Bugfix: Deepcopying Kubernetes Secrets attributes causing issues (#20318

)

Encountered a nasty bug where somebody basically implemented their own KubernetesPodSensor, which failed after more than one attempt when using mode="poke" + a volume + a secret.

Root cause turned out to be in `secret.attach_to_pod()`. In here, a volume and volumemount is created to mount the secret. A deepcopy() is made of the given Pod spec. In order to avoid appending to None, there is this line: `cp_pod.spec.volumes = pod.spec.volumes or []`. In case a volume is set on the Pod spec, a reference is created to the original pod spec volumes, which in turn was a reference to `self.volumes`. As a result, each secret resulted in a volume added to `self.volumes`, which resulted in an error when running the sensor a second time because the secret volume was already mounted during the first sensor attempt.

This PR references the deepcopied object instead, and creates a new list if pod.spec.volumes is None.

Co-authored-by: Bas Harenslak <bas@astronomer.io>

airflow/kubernetes/secret.py

@@ -91,20 +91,28 @@ def to_volume_secret(self) -> Tuple[k8s.V1Volume, k8s.V1VolumeMount]:
     def attach_to_pod(self, pod: k8s.V1Pod) -> k8s.V1Pod:
         """Attaches to pod"""
         cp_pod = copy.deepcopy(pod)
+
         if self.deploy_type == 'volume':
             volume, volume_mount = self.to_volume_secret()
-            cp_pod.spec.volumes = pod.spec.volumes or []
+            if cp_pod.spec.volumes is None:
+                cp_pod.spec.volumes = []
             cp_pod.spec.volumes.append(volume)
-            cp_pod.spec.containers[0].volume_mounts = pod.spec.containers[0].volume_mounts or []
+            if cp_pod.spec.containers[0].volume_mounts is None:
+                cp_pod.spec.containers[0].volume_mounts = []
             cp_pod.spec.containers[0].volume_mounts.append(volume_mount)
+
         if self.deploy_type == 'env' and self.key is not None:
             env = self.to_env_secret()
-            cp_pod.spec.containers[0].env = cp_pod.spec.containers[0].env or []
+            if cp_pod.spec.containers[0].env is None:
+                cp_pod.spec.containers[0].env = []
             cp_pod.spec.containers[0].env.append(env)
+
         if self.deploy_type == 'env' and self.key is None:
             env_from = self.to_env_from_secret()
-            cp_pod.spec.containers[0].env_from = cp_pod.spec.containers[0].env_from or []
+            if cp_pod.spec.containers[0].env_from is None:
+                cp_pod.spec.containers[0].env_from = []
             cp_pod.spec.containers[0].env_from.append(env_from)
+
         return cp_pod
 
     def __eq__(self, other):