Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[AIRFLOW-6040] Fix KubernetesJobWatcher Read time out error #6643

Closed
wants to merge 13 commits into from
28 changes: 18 additions & 10 deletions airflow/executors/kubernetes_executor.py
Original file line number Diff line number Diff line change
Expand Up @@ -293,17 +293,10 @@ def _run(self,
worker_uuid: str,
kube_config: Any) -> Optional[str]:
self.log.info(
'Event: and now my watch begins starting at resource_version: %s',
resource_version
)
'Event: and now my watch begins starting at resource_version: %s, '
'worker_uuid: %s', resource_version, worker_uuid)
watcher = watch.Watch()

kwargs = {'label_selector': 'airflow-worker={}'.format(worker_uuid)}
if resource_version:
kwargs['resource_version'] = resource_version
if kube_config.kube_client_request_args:
for key, value in kube_config.kube_client_request_args.items():
kwargs[key] = value
kwargs = self._get_watcher_args(resource_version, worker_uuid, kube_config)

last_resource_version: Optional[str] = None
for event in watcher.stream(kube_client.list_namespaced_pod, self.namespace,
Expand All @@ -323,6 +316,21 @@ def _run(self,

return last_resource_version

def _get_watcher_args(self,
resource_version: Optional[str],
worker_uuid: str,
kube_config: Any) -> dict:
"""Builds and returns the kwargs necessary for Watcher"""
kwargs = {'label_selector': 'airflow-worker={}'.format(worker_uuid)}
if resource_version:
kwargs['resource_version'] = resource_version
if kube_config.kube_client_request_args:
for key, value in kube_config.kube_client_request_args.iteritems():
kwargs[key] = value
conn_timeout = kube_config.kube_client_request_args.get('_request_timeout', [60, 60])[0]
kwargs['timeout_seconds'] = max(conn_timeout - 1, 1)
return kwargs

def process_error(self, event: Any) -> str:
"""Process error response"""
self.log.error(
Expand Down