[astro] Fix istio sidecar shutdown on newer GKE

Newer GKE verions have started to emit multiple running events for a
given pod with the sidecar still being shown as running. We will put
retries around shutting down the sidecar and also check the current
status of the sidecar, not just the status at the time of the event.

e.g: GKE > 1.18.20.901

(cherry picked from commit cbd50ef)

airflow/kubernetes/istio.py

@@ -15,6 +15,7 @@
 # specific language governing permissions and limitations
 # under the License.
 
+import tenacity
 from kubernetes.client.rest import ApiException
 from kubernetes.stream import stream
 from packaging.version import parse as semantic_version
@@ -132,17 +133,38 @@ def _shutdown_istio_proxy(self, pod):
             self.log.info("Shutting down istio-proxy in pod %s", pod.metadata.name)
             self._post_quitquitquit(pod, container, status_port)
 
+    @tenacity.retry(
+        stop=tenacity.stop_after_attempt(3),
+        wait=tenacity.wait_fixed(0.5),
+        reraise=True,
+        retry=tenacity.retry_if_exception_type(ApiException),
+    )
     def _post_quitquitquit(self, pod, container, status_port):
         """Send the curl to shutdown the isto-proxy container"""
         # Use exec to curl localhost inside of the sidecar.
-        _ = stream(
-            self._client.connect_get_namespaced_pod_exec,
-            pod.metadata.name,
-            pod.metadata.namespace,
-            tty=False,
-            stderr=True,
-            stdin=False,
-            stdout=True,
-            container=container.name,
-            command=['/bin/sh', '-c', f'curl -XPOST http://127.0.0.1:{status_port}/quitquitquit'],
-        )
+        try:
+            _ = stream(
+                self._client.connect_get_namespaced_pod_exec,
+                pod.metadata.name,
+                pod.metadata.namespace,
+                tty=False,
+                stderr=True,
+                stdin=False,
+                stdout=True,
+                container=container.name,
+                command=['/bin/sh', '-c', f'curl -XPOST http://127.0.0.1:{status_port}/quitquitquit'],
+            )
+            return
+        except ApiException:
+            # Check if the istio sidecar has already been shut down
+            current_pod = self._client.read_namespaced_pod(
+                name=pod.metadata.name,
+                namespace=pod.metadata.namespace,
+            )
+            if not self._should_shutdown_istio_proxy(current_pod):
+                self.log.info(
+                    "Istio sidecar is already shut down in %s, so continuing on",
+                    pod.metadata.name,
+                )
+                return
+            raise