diff --git a/airflow/executors/kubernetes_executor.py b/airflow/executors/kubernetes_executor.py index 707fe4fcdc4b8..c3475f28e3700 100644 --- a/airflow/executors/kubernetes_executor.py +++ b/airflow/executors/kubernetes_executor.py @@ -25,6 +25,7 @@ import functools import json import multiprocessing +import re import time from queue import Empty, Queue # pylint: disable=unused-import from typing import Any, Dict, List, Optional, Tuple @@ -173,11 +174,17 @@ def process_error(self, event: Any) -> str: self.log.error('Encountered Error response from k8s list namespaced pod stream => %s', event) raw_object = event['raw_object'] if raw_object['code'] == 410: - self.log.info( - 'Kubernetes resource version is too old, must reset to 0 => %s', (raw_object['message'],) + message = raw_object['message'] + latest_resource_version = self._parse_too_old_failure(message) + if latest_resource_version is None: + # Return resource version 0 + return '0' + self.log.warning( + "Updated to new resource version: %s due to 'too old' error: %s", + latest_resource_version, + raw_object, ) - # Return resource version 0 - return '0' + return latest_resource_version raise AirflowException( 'Kubernetes failure for %s with code %s and message: %s' % (raw_object['reason'], raw_object['code'], raw_object['message']) @@ -218,6 +225,25 @@ def process_status( resource_version, ) + def _parse_too_old_failure(self, message): + """ + Parse stream watcher 410 'too old resource version' error + to get the latest resource version from the message + + See https://github.com/kubernetes-client/python/issues/609 + """ + regex = r"too old resource version: .* \((.*)\)" + result = re.search(regex, message) + if result is None: + return None + match = result.group(1) + if match is None: + return None + try: + return match + except (ValueError, TypeError): + return None + class AirflowKubernetesScheduler(LoggingMixin): """Airflow Scheduler for Kubernetes""" diff --git a/tests/executors/test_kubernetes_executor.py b/tests/executors/test_kubernetes_executor.py index 0129b3af5cdd4..468ebb7134e63 100644 --- a/tests/executors/test_kubernetes_executor.py +++ b/tests/executors/test_kubernetes_executor.py @@ -507,3 +507,19 @@ def test_process_status_catchall(self): self._run() self.watcher.watcher_queue.put.assert_not_called() + + @mock.patch.object(KubernetesJobWatcher, '_parse_too_old_failure') + def test_process_error_event(self, mock_too_old_failure): + message = "too old resource version: 27272 (43334)" + mock_too_old_failure.return_value = '43334' + self.pod.status.phase = 'Pending' + self.pod.metadata.resource_version = '43334' + raw_object = {"code": 410, "message": message} + self.events.append({"type": "ERROR", "object": self.pod, "raw_object": raw_object}) + self._run() + mock_too_old_failure.assert_called_once_with(message) + + def test_parse_too_old_failure(self): + message = "too old resource version: 27272 (43334)" + latest_version = self.watcher._parse_too_old_failure(message) + assert latest_version == '43334'