You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by GitBox <gi...@apache.org> on 2021/04/02 20:11:15 UTC
[GitHub] [airflow] dimberman opened a new pull request #15165: Separate pod_launcher from core airflow
dimberman opened a new pull request #15165:
URL: https://github.com/apache/airflow/pull/15165
Currently, the KubernetesPodOperator uses the pod_launcher class in airflow core. This means that if we need to fix a bug in the KubernetesPodOperator such as #15137 then the new cncf.kubernetes package will require an Airflow upgrade. Since we hope to release providers in a much faster cadence than Airflow core releases, we should separate this dependency.
<!--
Thank you for contributing! Please make sure that your code changes
are covered with tests. And in case of new features or big changes
remember to adjust the documentation.
Feel free to ping committers for the review!
In case of existing issue, reference it using one of the following:
closes: #ISSUE
related: #ISSUE
How to write a good git commit message:
http://chris.beams.io/posts/git-commit/
-->
---
**^ Add meaningful description above**
Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/master/CONTRIBUTING.rst#pull-request-guidelines)** for more information.
In case of fundamental code change, Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvements+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in [UPDATING.md](https://github.com/apache/airflow/blob/master/UPDATING.md).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] dimberman commented on pull request #15165: Separate pod_launcher from core airflow
Posted by GitBox <gi...@apache.org>.
dimberman commented on pull request #15165:
URL: https://github.com/apache/airflow/pull/15165#issuecomment-812703349
cc: @SamWheating
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] dimberman merged pull request #15165: Separate Kubernetes pod_launcher from core airflow
Posted by GitBox <gi...@apache.org>.
dimberman merged pull request #15165:
URL: https://github.com/apache/airflow/pull/15165
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on a change in pull request #15165: Separate pod_launcher from core airflow
Posted by GitBox <gi...@apache.org>.
potiuk commented on a change in pull request #15165:
URL: https://github.com/apache/airflow/pull/15165#discussion_r607214375
##########
File path: airflow/kubernetes/pod_launcher_deprecated.py
##########
@@ -0,0 +1,300 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Launches PODs"""
+import json
+import math
+import time
+from datetime import datetime as dt
+from typing import Optional, Tuple
+
+import pendulum
+import tenacity
+from kubernetes import client, watch
+from kubernetes.client.models.v1_pod import V1Pod
+from kubernetes.client.rest import ApiException
+from kubernetes.stream import stream as kubernetes_stream
+from requests.exceptions import BaseHTTPError
+
+from airflow.exceptions import AirflowException
+from airflow.kubernetes.kube_client import get_kube_client
+from airflow.kubernetes.pod_generator import PodDefaults
+from airflow.settings import pod_mutation_hook
+from airflow.utils.log.logging_mixin import LoggingMixin
+from airflow.utils.state import State
+
+
Review comment:
Should we also a the deprecation warning directly here (or even move the warning from the old pod_launcher)? This way even if someone by mistake imports PodLauncher/PodStatus directly from there (because IDE will provide this as first-class option), will still get the warning.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] kaxil commented on a change in pull request #15165: Separate pod_launcher from core airflow
Posted by GitBox <gi...@apache.org>.
kaxil commented on a change in pull request #15165:
URL: https://github.com/apache/airflow/pull/15165#discussion_r607178652
##########
File path: airflow/kubernetes/pod_launcher.py
##########
@@ -14,283 +15,15 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
-"""Launches PODs"""
-import json
-import math
-import time
-from datetime import datetime as dt
-from typing import Optional, Tuple
-
-import pendulum
-import tenacity
-from kubernetes import client, watch
-from kubernetes.client.models.v1_pod import V1Pod
-from kubernetes.client.rest import ApiException
-from kubernetes.stream import stream as kubernetes_stream
-from requests.exceptions import BaseHTTPError
-
-from airflow.exceptions import AirflowException
-from airflow.kubernetes.kube_client import get_kube_client
-from airflow.kubernetes.pod_generator import PodDefaults
-from airflow.settings import pod_mutation_hook
-from airflow.utils.log.logging_mixin import LoggingMixin
-from airflow.utils.state import State
-
-
-class PodStatus:
- """Status of the PODs"""
-
- PENDING = 'pending'
- RUNNING = 'running'
- FAILED = 'failed'
- SUCCEEDED = 'succeeded'
-
-
-class PodLauncher(LoggingMixin):
- """Launches PODS"""
-
- def __init__(
- self,
- kube_client: client.CoreV1Api = None,
- in_cluster: bool = True,
- cluster_context: Optional[str] = None,
- extract_xcom: bool = False,
- ):
- """
- Creates the launcher.
-
- :param kube_client: kubernetes client
- :param in_cluster: whether we are in cluster
- :param cluster_context: context of the cluster
- :param extract_xcom: whether we should extract xcom
- """
- super().__init__()
- self._client = kube_client or get_kube_client(in_cluster=in_cluster, cluster_context=cluster_context)
- self._watch = watch.Watch()
- self.extract_xcom = extract_xcom
-
- def run_pod_async(self, pod: V1Pod, **kwargs):
- """Runs POD asynchronously"""
- pod_mutation_hook(pod)
-
- sanitized_pod = self._client.api_client.sanitize_for_serialization(pod)
- json_pod = json.dumps(sanitized_pod, indent=2)
-
- self.log.debug('Pod Creation Request: \n%s', json_pod)
- try:
- resp = self._client.create_namespaced_pod(
- body=sanitized_pod, namespace=pod.metadata.namespace, **kwargs
- )
- self.log.debug('Pod Creation Response: %s', resp)
- except Exception as e:
- self.log.exception('Exception when attempting to create Namespaced Pod: %s', json_pod)
- raise e
- return resp
-
- def delete_pod(self, pod: V1Pod):
- """Deletes POD"""
- try:
- self._client.delete_namespaced_pod(
- pod.metadata.name, pod.metadata.namespace, body=client.V1DeleteOptions()
- )
- except ApiException as e:
- # If the pod is already deleted
- if e.status != 404:
- raise
-
- def start_pod(self, pod: V1Pod, startup_timeout: int = 120):
- """
- Launches the pod synchronously and waits for completion.
-
- :param pod:
- :param startup_timeout: Timeout for startup of the pod (if pod is pending for too long, fails task)
- :return:
- """
- resp = self.run_pod_async(pod)
- curr_time = dt.now()
- if resp.status.start_time is None:
- while self.pod_not_started(pod):
- self.log.warning("Pod not yet started: %s", pod.metadata.name)
- delta = dt.now() - curr_time
- if delta.total_seconds() >= startup_timeout:
- raise AirflowException("Pod took too long to start")
- time.sleep(1)
-
- def monitor_pod(self, pod: V1Pod, get_logs: bool) -> Tuple[State, Optional[str]]:
- """
- Monitors a pod and returns the final state
-
- :param pod: pod spec that will be monitored
- :type pod : V1Pod
- :param get_logs: whether to read the logs locally
- :return: Tuple[State, Optional[str]]
- """
- if get_logs:
- read_logs_since_sec = None
- last_log_time = None
- while True:
- logs = self.read_pod_logs(pod, timestamps=True, since_seconds=read_logs_since_sec)
- for line in logs:
- timestamp, message = self.parse_log_line(line.decode('utf-8'))
- last_log_time = pendulum.parse(timestamp)
- self.log.info(message)
- time.sleep(1)
-
- if not self.base_container_is_running(pod):
- break
-
- self.log.warning('Pod %s log read interrupted', pod.metadata.name)
- if last_log_time:
- delta = pendulum.now() - last_log_time
- # Prefer logs duplication rather than loss
- read_logs_since_sec = math.ceil(delta.total_seconds())
- result = None
- if self.extract_xcom:
- while self.base_container_is_running(pod):
- self.log.info('Container %s has state %s', pod.metadata.name, State.RUNNING)
- time.sleep(2)
- result = self._extract_xcom(pod)
- self.log.info(result)
- result = json.loads(result)
- while self.pod_is_running(pod):
- self.log.info('Pod %s has state %s', pod.metadata.name, State.RUNNING)
- time.sleep(2)
- return self._task_status(self.read_pod(pod)), result
-
- def parse_log_line(self, line: str) -> Tuple[str, str]:
- """
- Parse K8s log line and returns the final state
-
- :param line: k8s log line
- :type line: str
- :return: timestamp and log message
- :rtype: Tuple[str, str]
- """
- split_at = line.find(' ')
- if split_at == -1:
- raise Exception(f'Log not in "{{timestamp}} {{log}}" format. Got: {line}')
- timestamp = line[:split_at]
- message = line[split_at + 1 :].rstrip()
- return timestamp, message
-
- def _task_status(self, event):
- self.log.info('Event: %s had an event of type %s', event.metadata.name, event.status.phase)
- status = self.process_status(event.metadata.name, event.status.phase)
- return status
-
- def pod_not_started(self, pod: V1Pod):
- """Tests if pod has not started"""
- state = self._task_status(self.read_pod(pod))
- return state == State.QUEUED
-
- def pod_is_running(self, pod: V1Pod):
- """Tests if pod is running"""
- state = self._task_status(self.read_pod(pod))
- return state not in (State.SUCCESS, State.FAILED)
-
- def base_container_is_running(self, pod: V1Pod):
- """Tests if base container is running"""
- event = self.read_pod(pod)
- status = next(iter(filter(lambda s: s.name == 'base', event.status.container_statuses)), None)
- if not status:
- return False
- return status.state.running is not None
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod_logs(
- self,
- pod: V1Pod,
- tail_lines: Optional[int] = None,
- timestamps: bool = False,
- since_seconds: Optional[int] = None,
- ):
- """Reads log from the POD"""
- additional_kwargs = {}
- if since_seconds:
- additional_kwargs['since_seconds'] = since_seconds
-
- if tail_lines:
- additional_kwargs['tail_lines'] = tail_lines
-
- try:
- return self._client.read_namespaced_pod_log(
- name=pod.metadata.name,
- namespace=pod.metadata.namespace,
- container='base',
- follow=True,
- timestamps=timestamps,
- _preload_content=False,
- **additional_kwargs,
- )
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod_events(self, pod):
- """Reads events from the POD"""
- try:
- return self._client.list_namespaced_event(
- namespace=pod.metadata.namespace, field_selector=f"involvedObject.name={pod.metadata.name}"
- )
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod(self, pod: V1Pod):
- """Read POD information"""
- try:
- return self._client.read_namespaced_pod(pod.metadata.name, pod.metadata.namespace)
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- def _extract_xcom(self, pod: V1Pod):
- resp = kubernetes_stream(
- self._client.connect_get_namespaced_pod_exec,
- pod.metadata.name,
- pod.metadata.namespace,
- container=PodDefaults.SIDECAR_CONTAINER_NAME,
- command=['/bin/sh'],
- stdin=True,
- stdout=True,
- stderr=True,
- tty=False,
- _preload_content=False,
- )
- try:
- result = self._exec_pod_command(resp, f'cat {PodDefaults.XCOM_MOUNT_PATH}/return.json')
- self._exec_pod_command(resp, 'kill -s SIGINT 1')
- finally:
- resp.close()
- if result is None:
- raise AirflowException(f'Failed to extract xcom from pod: {pod.metadata.name}')
- return result
-
- def _exec_pod_command(self, resp, command):
- if resp.is_open():
- self.log.info('Running command... %s\n', command)
- resp.write_stdin(command + '\n')
- while resp.is_open():
- resp.update(timeout=1)
- if resp.peek_stdout():
- return resp.read_stdout()
- if resp.peek_stderr():
- self.log.info(resp.read_stderr())
- break
- return None
-
- def process_status(self, job_id, status):
- """Process status information for the JOB"""
- status = status.lower()
- if status == PodStatus.PENDING:
- return State.QUEUED
- elif status == PodStatus.FAILED:
- self.log.error('Event with job id %s Failed', job_id)
- return State.FAILED
- elif status == PodStatus.SUCCEEDED:
- self.log.info('Event with job id %s Succeeded', job_id)
- return State.SUCCESS
- elif status == PodStatus.RUNNING:
- return State.RUNNING
- else:
- self.log.error('Event: Invalid state %s on job %s', status, job_id)
- return State.FAILED
+"""This module is deprecated. Please use `kubernetes.client.models for V1ResourceRequirements and Port."""
+# flake8: noqa
+# pylint: disable=unused-import
+import warnings
+
+from airflow.kubernetes import pod_launcher_deprecated # pylint: disable=unused-import
Review comment:
Please also rebase on latest master, some k8s related changes were merged that updated tests too.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on a change in pull request #15165: Separate pod_launcher from core airflow
Posted by GitBox <gi...@apache.org>.
potiuk commented on a change in pull request #15165:
URL: https://github.com/apache/airflow/pull/15165#discussion_r607210186
##########
File path: airflow/kubernetes/pod_launcher.py
##########
@@ -14,283 +15,15 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
-"""Launches PODs"""
-import json
-import math
-import time
-from datetime import datetime as dt
-from typing import Optional, Tuple
-
-import pendulum
-import tenacity
-from kubernetes import client, watch
-from kubernetes.client.models.v1_pod import V1Pod
-from kubernetes.client.rest import ApiException
-from kubernetes.stream import stream as kubernetes_stream
-from requests.exceptions import BaseHTTPError
-
-from airflow.exceptions import AirflowException
-from airflow.kubernetes.kube_client import get_kube_client
-from airflow.kubernetes.pod_generator import PodDefaults
-from airflow.settings import pod_mutation_hook
-from airflow.utils.log.logging_mixin import LoggingMixin
-from airflow.utils.state import State
-
-
-class PodStatus:
- """Status of the PODs"""
-
- PENDING = 'pending'
- RUNNING = 'running'
- FAILED = 'failed'
- SUCCEEDED = 'succeeded'
-
-
-class PodLauncher(LoggingMixin):
- """Launches PODS"""
-
- def __init__(
- self,
- kube_client: client.CoreV1Api = None,
- in_cluster: bool = True,
- cluster_context: Optional[str] = None,
- extract_xcom: bool = False,
- ):
- """
- Creates the launcher.
-
- :param kube_client: kubernetes client
- :param in_cluster: whether we are in cluster
- :param cluster_context: context of the cluster
- :param extract_xcom: whether we should extract xcom
- """
- super().__init__()
- self._client = kube_client or get_kube_client(in_cluster=in_cluster, cluster_context=cluster_context)
- self._watch = watch.Watch()
- self.extract_xcom = extract_xcom
-
- def run_pod_async(self, pod: V1Pod, **kwargs):
- """Runs POD asynchronously"""
- pod_mutation_hook(pod)
-
- sanitized_pod = self._client.api_client.sanitize_for_serialization(pod)
- json_pod = json.dumps(sanitized_pod, indent=2)
-
- self.log.debug('Pod Creation Request: \n%s', json_pod)
- try:
- resp = self._client.create_namespaced_pod(
- body=sanitized_pod, namespace=pod.metadata.namespace, **kwargs
- )
- self.log.debug('Pod Creation Response: %s', resp)
- except Exception as e:
- self.log.exception('Exception when attempting to create Namespaced Pod: %s', json_pod)
- raise e
- return resp
-
- def delete_pod(self, pod: V1Pod):
- """Deletes POD"""
- try:
- self._client.delete_namespaced_pod(
- pod.metadata.name, pod.metadata.namespace, body=client.V1DeleteOptions()
- )
- except ApiException as e:
- # If the pod is already deleted
- if e.status != 404:
- raise
-
- def start_pod(self, pod: V1Pod, startup_timeout: int = 120):
- """
- Launches the pod synchronously and waits for completion.
-
- :param pod:
- :param startup_timeout: Timeout for startup of the pod (if pod is pending for too long, fails task)
- :return:
- """
- resp = self.run_pod_async(pod)
- curr_time = dt.now()
- if resp.status.start_time is None:
- while self.pod_not_started(pod):
- self.log.warning("Pod not yet started: %s", pod.metadata.name)
- delta = dt.now() - curr_time
- if delta.total_seconds() >= startup_timeout:
- raise AirflowException("Pod took too long to start")
- time.sleep(1)
-
- def monitor_pod(self, pod: V1Pod, get_logs: bool) -> Tuple[State, Optional[str]]:
- """
- Monitors a pod and returns the final state
-
- :param pod: pod spec that will be monitored
- :type pod : V1Pod
- :param get_logs: whether to read the logs locally
- :return: Tuple[State, Optional[str]]
- """
- if get_logs:
- read_logs_since_sec = None
- last_log_time = None
- while True:
- logs = self.read_pod_logs(pod, timestamps=True, since_seconds=read_logs_since_sec)
- for line in logs:
- timestamp, message = self.parse_log_line(line.decode('utf-8'))
- last_log_time = pendulum.parse(timestamp)
- self.log.info(message)
- time.sleep(1)
-
- if not self.base_container_is_running(pod):
- break
-
- self.log.warning('Pod %s log read interrupted', pod.metadata.name)
- if last_log_time:
- delta = pendulum.now() - last_log_time
- # Prefer logs duplication rather than loss
- read_logs_since_sec = math.ceil(delta.total_seconds())
- result = None
- if self.extract_xcom:
- while self.base_container_is_running(pod):
- self.log.info('Container %s has state %s', pod.metadata.name, State.RUNNING)
- time.sleep(2)
- result = self._extract_xcom(pod)
- self.log.info(result)
- result = json.loads(result)
- while self.pod_is_running(pod):
- self.log.info('Pod %s has state %s', pod.metadata.name, State.RUNNING)
- time.sleep(2)
- return self._task_status(self.read_pod(pod)), result
-
- def parse_log_line(self, line: str) -> Tuple[str, str]:
- """
- Parse K8s log line and returns the final state
-
- :param line: k8s log line
- :type line: str
- :return: timestamp and log message
- :rtype: Tuple[str, str]
- """
- split_at = line.find(' ')
- if split_at == -1:
- raise Exception(f'Log not in "{{timestamp}} {{log}}" format. Got: {line}')
- timestamp = line[:split_at]
- message = line[split_at + 1 :].rstrip()
- return timestamp, message
-
- def _task_status(self, event):
- self.log.info('Event: %s had an event of type %s', event.metadata.name, event.status.phase)
- status = self.process_status(event.metadata.name, event.status.phase)
- return status
-
- def pod_not_started(self, pod: V1Pod):
- """Tests if pod has not started"""
- state = self._task_status(self.read_pod(pod))
- return state == State.QUEUED
-
- def pod_is_running(self, pod: V1Pod):
- """Tests if pod is running"""
- state = self._task_status(self.read_pod(pod))
- return state not in (State.SUCCESS, State.FAILED)
-
- def base_container_is_running(self, pod: V1Pod):
- """Tests if base container is running"""
- event = self.read_pod(pod)
- status = next(iter(filter(lambda s: s.name == 'base', event.status.container_statuses)), None)
- if not status:
- return False
- return status.state.running is not None
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod_logs(
- self,
- pod: V1Pod,
- tail_lines: Optional[int] = None,
- timestamps: bool = False,
- since_seconds: Optional[int] = None,
- ):
- """Reads log from the POD"""
- additional_kwargs = {}
- if since_seconds:
- additional_kwargs['since_seconds'] = since_seconds
-
- if tail_lines:
- additional_kwargs['tail_lines'] = tail_lines
-
- try:
- return self._client.read_namespaced_pod_log(
- name=pod.metadata.name,
- namespace=pod.metadata.namespace,
- container='base',
- follow=True,
- timestamps=timestamps,
- _preload_content=False,
- **additional_kwargs,
- )
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod_events(self, pod):
- """Reads events from the POD"""
- try:
- return self._client.list_namespaced_event(
- namespace=pod.metadata.namespace, field_selector=f"involvedObject.name={pod.metadata.name}"
- )
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod(self, pod: V1Pod):
- """Read POD information"""
- try:
- return self._client.read_namespaced_pod(pod.metadata.name, pod.metadata.namespace)
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- def _extract_xcom(self, pod: V1Pod):
- resp = kubernetes_stream(
- self._client.connect_get_namespaced_pod_exec,
- pod.metadata.name,
- pod.metadata.namespace,
- container=PodDefaults.SIDECAR_CONTAINER_NAME,
- command=['/bin/sh'],
- stdin=True,
- stdout=True,
- stderr=True,
- tty=False,
- _preload_content=False,
- )
- try:
- result = self._exec_pod_command(resp, f'cat {PodDefaults.XCOM_MOUNT_PATH}/return.json')
- self._exec_pod_command(resp, 'kill -s SIGINT 1')
- finally:
- resp.close()
- if result is None:
- raise AirflowException(f'Failed to extract xcom from pod: {pod.metadata.name}')
- return result
-
- def _exec_pod_command(self, resp, command):
- if resp.is_open():
- self.log.info('Running command... %s\n', command)
- resp.write_stdin(command + '\n')
- while resp.is_open():
- resp.update(timeout=1)
- if resp.peek_stdout():
- return resp.read_stdout()
- if resp.peek_stderr():
- self.log.info(resp.read_stderr())
- break
- return None
-
- def process_status(self, job_id, status):
- """Process status information for the JOB"""
- status = status.lower()
- if status == PodStatus.PENDING:
- return State.QUEUED
- elif status == PodStatus.FAILED:
- self.log.error('Event with job id %s Failed', job_id)
- return State.FAILED
- elif status == PodStatus.SUCCEEDED:
- self.log.info('Event with job id %s Succeeded', job_id)
- return State.SUCCESS
- elif status == PodStatus.RUNNING:
- return State.RUNNING
- else:
- self.log.error('Event: Invalid state %s on job %s', status, job_id)
- return State.FAILED
+"""This module is deprecated. Please use `kubernetes.client.models for V1ResourceRequirements and Port."""
Review comment:
That works for me @dimberman
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] dimberman commented on a change in pull request #15165: Separate pod_launcher from core airflow
Posted by GitBox <gi...@apache.org>.
dimberman commented on a change in pull request #15165:
URL: https://github.com/apache/airflow/pull/15165#discussion_r606730124
##########
File path: airflow/executors/kubernetes_executor.py
##########
@@ -237,12 +238,29 @@ def __init__(
self.namespace = self.kube_config.kube_namespace
self.log.debug("Kubernetes using namespace %s", self.namespace)
self.kube_client = kube_client
- self.launcher = PodLauncher(kube_client=self.kube_client)
self._manager = multiprocessing.Manager()
self.watcher_queue = self._manager.Queue()
self.scheduler_job_id = scheduler_job_id
self.kube_watcher = self._make_kube_watcher()
+ def run_pod_async(self, pod: V1Pod, **kwargs):
+ """Runs POD asynchronously"""
+ pod_mutation_hook(pod)
+
+ sanitized_pod = self.kube_client.api_client.sanitize_for_serialization(pod)
+ json_pod = json.dumps(sanitized_pod, indent=2)
+
+ self.log.debug('Pod Creation Request: \n%s', json_pod)
+ try:
+ resp = self.kube_client.create_namespaced_pod(
+ body=sanitized_pod, namespace=pod.metadata.namespace, **kwargs
+ )
+ self.log.debug('Pod Creation Response: %s', resp)
+ except Exception as e:
+ self.log.exception('Exception when attempting to create Namespaced Pod: %s', json_pod)
+ raise e
+ return resp
Review comment:
@ashb It makes sense. The KubernetesExecutor is fire and forget. We don't monitor the task via the pod launcher for the KubernetesExecutor, just monitor the task state via the job watcher.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] kaxil commented on a change in pull request #15165: Separate pod_launcher from core airflow
Posted by GitBox <gi...@apache.org>.
kaxil commented on a change in pull request #15165:
URL: https://github.com/apache/airflow/pull/15165#discussion_r607213211
##########
File path: airflow/kubernetes/pod_launcher.py
##########
@@ -14,283 +15,15 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
-"""Launches PODs"""
-import json
-import math
-import time
-from datetime import datetime as dt
-from typing import Optional, Tuple
-
-import pendulum
-import tenacity
-from kubernetes import client, watch
-from kubernetes.client.models.v1_pod import V1Pod
-from kubernetes.client.rest import ApiException
-from kubernetes.stream import stream as kubernetes_stream
-from requests.exceptions import BaseHTTPError
-
-from airflow.exceptions import AirflowException
-from airflow.kubernetes.kube_client import get_kube_client
-from airflow.kubernetes.pod_generator import PodDefaults
-from airflow.settings import pod_mutation_hook
-from airflow.utils.log.logging_mixin import LoggingMixin
-from airflow.utils.state import State
-
-
-class PodStatus:
- """Status of the PODs"""
-
- PENDING = 'pending'
- RUNNING = 'running'
- FAILED = 'failed'
- SUCCEEDED = 'succeeded'
-
-
-class PodLauncher(LoggingMixin):
- """Launches PODS"""
-
- def __init__(
- self,
- kube_client: client.CoreV1Api = None,
- in_cluster: bool = True,
- cluster_context: Optional[str] = None,
- extract_xcom: bool = False,
- ):
- """
- Creates the launcher.
-
- :param kube_client: kubernetes client
- :param in_cluster: whether we are in cluster
- :param cluster_context: context of the cluster
- :param extract_xcom: whether we should extract xcom
- """
- super().__init__()
- self._client = kube_client or get_kube_client(in_cluster=in_cluster, cluster_context=cluster_context)
- self._watch = watch.Watch()
- self.extract_xcom = extract_xcom
-
- def run_pod_async(self, pod: V1Pod, **kwargs):
- """Runs POD asynchronously"""
- pod_mutation_hook(pod)
-
- sanitized_pod = self._client.api_client.sanitize_for_serialization(pod)
- json_pod = json.dumps(sanitized_pod, indent=2)
-
- self.log.debug('Pod Creation Request: \n%s', json_pod)
- try:
- resp = self._client.create_namespaced_pod(
- body=sanitized_pod, namespace=pod.metadata.namespace, **kwargs
- )
- self.log.debug('Pod Creation Response: %s', resp)
- except Exception as e:
- self.log.exception('Exception when attempting to create Namespaced Pod: %s', json_pod)
- raise e
- return resp
-
- def delete_pod(self, pod: V1Pod):
- """Deletes POD"""
- try:
- self._client.delete_namespaced_pod(
- pod.metadata.name, pod.metadata.namespace, body=client.V1DeleteOptions()
- )
- except ApiException as e:
- # If the pod is already deleted
- if e.status != 404:
- raise
-
- def start_pod(self, pod: V1Pod, startup_timeout: int = 120):
- """
- Launches the pod synchronously and waits for completion.
-
- :param pod:
- :param startup_timeout: Timeout for startup of the pod (if pod is pending for too long, fails task)
- :return:
- """
- resp = self.run_pod_async(pod)
- curr_time = dt.now()
- if resp.status.start_time is None:
- while self.pod_not_started(pod):
- self.log.warning("Pod not yet started: %s", pod.metadata.name)
- delta = dt.now() - curr_time
- if delta.total_seconds() >= startup_timeout:
- raise AirflowException("Pod took too long to start")
- time.sleep(1)
-
- def monitor_pod(self, pod: V1Pod, get_logs: bool) -> Tuple[State, Optional[str]]:
- """
- Monitors a pod and returns the final state
-
- :param pod: pod spec that will be monitored
- :type pod : V1Pod
- :param get_logs: whether to read the logs locally
- :return: Tuple[State, Optional[str]]
- """
- if get_logs:
- read_logs_since_sec = None
- last_log_time = None
- while True:
- logs = self.read_pod_logs(pod, timestamps=True, since_seconds=read_logs_since_sec)
- for line in logs:
- timestamp, message = self.parse_log_line(line.decode('utf-8'))
- last_log_time = pendulum.parse(timestamp)
- self.log.info(message)
- time.sleep(1)
-
- if not self.base_container_is_running(pod):
- break
-
- self.log.warning('Pod %s log read interrupted', pod.metadata.name)
- if last_log_time:
- delta = pendulum.now() - last_log_time
- # Prefer logs duplication rather than loss
- read_logs_since_sec = math.ceil(delta.total_seconds())
- result = None
- if self.extract_xcom:
- while self.base_container_is_running(pod):
- self.log.info('Container %s has state %s', pod.metadata.name, State.RUNNING)
- time.sleep(2)
- result = self._extract_xcom(pod)
- self.log.info(result)
- result = json.loads(result)
- while self.pod_is_running(pod):
- self.log.info('Pod %s has state %s', pod.metadata.name, State.RUNNING)
- time.sleep(2)
- return self._task_status(self.read_pod(pod)), result
-
- def parse_log_line(self, line: str) -> Tuple[str, str]:
- """
- Parse K8s log line and returns the final state
-
- :param line: k8s log line
- :type line: str
- :return: timestamp and log message
- :rtype: Tuple[str, str]
- """
- split_at = line.find(' ')
- if split_at == -1:
- raise Exception(f'Log not in "{{timestamp}} {{log}}" format. Got: {line}')
- timestamp = line[:split_at]
- message = line[split_at + 1 :].rstrip()
- return timestamp, message
-
- def _task_status(self, event):
- self.log.info('Event: %s had an event of type %s', event.metadata.name, event.status.phase)
- status = self.process_status(event.metadata.name, event.status.phase)
- return status
-
- def pod_not_started(self, pod: V1Pod):
- """Tests if pod has not started"""
- state = self._task_status(self.read_pod(pod))
- return state == State.QUEUED
-
- def pod_is_running(self, pod: V1Pod):
- """Tests if pod is running"""
- state = self._task_status(self.read_pod(pod))
- return state not in (State.SUCCESS, State.FAILED)
-
- def base_container_is_running(self, pod: V1Pod):
- """Tests if base container is running"""
- event = self.read_pod(pod)
- status = next(iter(filter(lambda s: s.name == 'base', event.status.container_statuses)), None)
- if not status:
- return False
- return status.state.running is not None
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod_logs(
- self,
- pod: V1Pod,
- tail_lines: Optional[int] = None,
- timestamps: bool = False,
- since_seconds: Optional[int] = None,
- ):
- """Reads log from the POD"""
- additional_kwargs = {}
- if since_seconds:
- additional_kwargs['since_seconds'] = since_seconds
-
- if tail_lines:
- additional_kwargs['tail_lines'] = tail_lines
-
- try:
- return self._client.read_namespaced_pod_log(
- name=pod.metadata.name,
- namespace=pod.metadata.namespace,
- container='base',
- follow=True,
- timestamps=timestamps,
- _preload_content=False,
- **additional_kwargs,
- )
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod_events(self, pod):
- """Reads events from the POD"""
- try:
- return self._client.list_namespaced_event(
- namespace=pod.metadata.namespace, field_selector=f"involvedObject.name={pod.metadata.name}"
- )
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod(self, pod: V1Pod):
- """Read POD information"""
- try:
- return self._client.read_namespaced_pod(pod.metadata.name, pod.metadata.namespace)
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- def _extract_xcom(self, pod: V1Pod):
- resp = kubernetes_stream(
- self._client.connect_get_namespaced_pod_exec,
- pod.metadata.name,
- pod.metadata.namespace,
- container=PodDefaults.SIDECAR_CONTAINER_NAME,
- command=['/bin/sh'],
- stdin=True,
- stdout=True,
- stderr=True,
- tty=False,
- _preload_content=False,
- )
- try:
- result = self._exec_pod_command(resp, f'cat {PodDefaults.XCOM_MOUNT_PATH}/return.json')
- self._exec_pod_command(resp, 'kill -s SIGINT 1')
- finally:
- resp.close()
- if result is None:
- raise AirflowException(f'Failed to extract xcom from pod: {pod.metadata.name}')
- return result
-
- def _exec_pod_command(self, resp, command):
- if resp.is_open():
- self.log.info('Running command... %s\n', command)
- resp.write_stdin(command + '\n')
- while resp.is_open():
- resp.update(timeout=1)
- if resp.peek_stdout():
- return resp.read_stdout()
- if resp.peek_stderr():
- self.log.info(resp.read_stderr())
- break
- return None
-
- def process_status(self, job_id, status):
- """Process status information for the JOB"""
- status = status.lower()
- if status == PodStatus.PENDING:
- return State.QUEUED
- elif status == PodStatus.FAILED:
- self.log.error('Event with job id %s Failed', job_id)
- return State.FAILED
- elif status == PodStatus.SUCCEEDED:
- self.log.info('Event with job id %s Succeeded', job_id)
- return State.SUCCESS
- elif status == PodStatus.RUNNING:
- return State.RUNNING
- else:
- self.log.error('Event: Invalid state %s on job %s', status, job_id)
- return State.FAILED
+"""This module is deprecated. Please use `kubernetes.client.models for V1ResourceRequirements and Port."""
+# flake8: noqa
+# pylint: disable=unused-import
+import warnings
+
+from airflow.kubernetes import pod_launcher_deprecated # pylint: disable=unused-import
Review comment:
Cool
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] dimberman commented on a change in pull request #15165: Separate Kubernetes pod_launcher from core airflow
Posted by GitBox <gi...@apache.org>.
dimberman commented on a change in pull request #15165:
URL: https://github.com/apache/airflow/pull/15165#discussion_r607314879
##########
File path: UPDATING.md
##########
@@ -70,6 +70,12 @@ https://developers.google.com/style/inclusive-documentation
-->
+### Removed pod_launcher from core airflow
+
+Moved the pod launcher from `airflow.kubernetes.pod_launcher` to `airflow.providers.cncf.kubernetes.utils.pod_launcher`
+
+This will alow users to fix the KubernetesPodOperator without requiring an airflow upgrade
Review comment:
I fixed this doc with slightly different wording.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] dimberman commented on a change in pull request #15165: Separate pod_launcher from core airflow
Posted by GitBox <gi...@apache.org>.
dimberman commented on a change in pull request #15165:
URL: https://github.com/apache/airflow/pull/15165#discussion_r606418206
##########
File path: airflow/kubernetes/pod_launcher.py
##########
@@ -14,283 +15,16 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
-"""Launches PODs"""
-import json
-import math
-import time
-from datetime import datetime as dt
-from typing import Optional, Tuple
-
-import pendulum
-import tenacity
-from kubernetes import client, watch
-from kubernetes.client.models.v1_pod import V1Pod
-from kubernetes.client.rest import ApiException
-from kubernetes.stream import stream as kubernetes_stream
-from requests.exceptions import BaseHTTPError
-
-from airflow.exceptions import AirflowException
-from airflow.kubernetes.kube_client import get_kube_client
-from airflow.kubernetes.pod_generator import PodDefaults
-from airflow.settings import pod_mutation_hook
-from airflow.utils.log.logging_mixin import LoggingMixin
-from airflow.utils.state import State
-
-
-class PodStatus:
- """Status of the PODs"""
-
- PENDING = 'pending'
- RUNNING = 'running'
- FAILED = 'failed'
- SUCCEEDED = 'succeeded'
-
-
-class PodLauncher(LoggingMixin):
- """Launches PODS"""
-
- def __init__(
- self,
- kube_client: client.CoreV1Api = None,
- in_cluster: bool = True,
- cluster_context: Optional[str] = None,
- extract_xcom: bool = False,
- ):
- """
- Creates the launcher.
-
- :param kube_client: kubernetes client
- :param in_cluster: whether we are in cluster
- :param cluster_context: context of the cluster
- :param extract_xcom: whether we should extract xcom
- """
- super().__init__()
- self._client = kube_client or get_kube_client(in_cluster=in_cluster, cluster_context=cluster_context)
- self._watch = watch.Watch()
- self.extract_xcom = extract_xcom
-
- def run_pod_async(self, pod: V1Pod, **kwargs):
- """Runs POD asynchronously"""
- pod_mutation_hook(pod)
-
- sanitized_pod = self._client.api_client.sanitize_for_serialization(pod)
- json_pod = json.dumps(sanitized_pod, indent=2)
-
- self.log.debug('Pod Creation Request: \n%s', json_pod)
- try:
- resp = self._client.create_namespaced_pod(
- body=sanitized_pod, namespace=pod.metadata.namespace, **kwargs
- )
- self.log.debug('Pod Creation Response: %s', resp)
- except Exception as e:
- self.log.exception('Exception when attempting to create Namespaced Pod: %s', json_pod)
- raise e
- return resp
-
- def delete_pod(self, pod: V1Pod):
- """Deletes POD"""
- try:
- self._client.delete_namespaced_pod(
- pod.metadata.name, pod.metadata.namespace, body=client.V1DeleteOptions()
- )
- except ApiException as e:
- # If the pod is already deleted
- if e.status != 404:
- raise
-
- def start_pod(self, pod: V1Pod, startup_timeout: int = 120):
- """
- Launches the pod synchronously and waits for completion.
-
- :param pod:
- :param startup_timeout: Timeout for startup of the pod (if pod is pending for too long, fails task)
- :return:
- """
- resp = self.run_pod_async(pod)
- curr_time = dt.now()
- if resp.status.start_time is None:
- while self.pod_not_started(pod):
- self.log.warning("Pod not yet started: %s", pod.metadata.name)
- delta = dt.now() - curr_time
- if delta.total_seconds() >= startup_timeout:
- raise AirflowException("Pod took too long to start")
- time.sleep(1)
-
- def monitor_pod(self, pod: V1Pod, get_logs: bool) -> Tuple[State, Optional[str]]:
- """
- Monitors a pod and returns the final state
-
- :param pod: pod spec that will be monitored
- :type pod : V1Pod
- :param get_logs: whether to read the logs locally
- :return: Tuple[State, Optional[str]]
- """
- if get_logs:
- read_logs_since_sec = None
- last_log_time = None
- while True:
- logs = self.read_pod_logs(pod, timestamps=True, since_seconds=read_logs_since_sec)
- for line in logs:
- timestamp, message = self.parse_log_line(line.decode('utf-8'))
- last_log_time = pendulum.parse(timestamp)
- self.log.info(message)
- time.sleep(1)
-
- if not self.base_container_is_running(pod):
- break
-
- self.log.warning('Pod %s log read interrupted', pod.metadata.name)
- if last_log_time:
- delta = pendulum.now() - last_log_time
- # Prefer logs duplication rather than loss
- read_logs_since_sec = math.ceil(delta.total_seconds())
- result = None
- if self.extract_xcom:
- while self.base_container_is_running(pod):
- self.log.info('Container %s has state %s', pod.metadata.name, State.RUNNING)
- time.sleep(2)
- result = self._extract_xcom(pod)
- self.log.info(result)
- result = json.loads(result)
- while self.pod_is_running(pod):
- self.log.info('Pod %s has state %s', pod.metadata.name, State.RUNNING)
- time.sleep(2)
- return self._task_status(self.read_pod(pod)), result
-
- def parse_log_line(self, line: str) -> Tuple[str, str]:
- """
- Parse K8s log line and returns the final state
-
- :param line: k8s log line
- :type line: str
- :return: timestamp and log message
- :rtype: Tuple[str, str]
- """
- split_at = line.find(' ')
- if split_at == -1:
- raise Exception(f'Log not in "{{timestamp}} {{log}}" format. Got: {line}')
- timestamp = line[:split_at]
- message = line[split_at + 1 :].rstrip()
- return timestamp, message
-
- def _task_status(self, event):
- self.log.info('Event: %s had an event of type %s', event.metadata.name, event.status.phase)
- status = self.process_status(event.metadata.name, event.status.phase)
- return status
-
- def pod_not_started(self, pod: V1Pod):
- """Tests if pod has not started"""
- state = self._task_status(self.read_pod(pod))
- return state == State.QUEUED
-
- def pod_is_running(self, pod: V1Pod):
- """Tests if pod is running"""
- state = self._task_status(self.read_pod(pod))
- return state not in (State.SUCCESS, State.FAILED)
-
- def base_container_is_running(self, pod: V1Pod):
- """Tests if base container is running"""
- event = self.read_pod(pod)
- status = next(iter(filter(lambda s: s.name == 'base', event.status.container_statuses)), None)
- if not status:
- return False
- return status.state.running is not None
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod_logs(
- self,
- pod: V1Pod,
- tail_lines: Optional[int] = None,
- timestamps: bool = False,
- since_seconds: Optional[int] = None,
- ):
- """Reads log from the POD"""
- additional_kwargs = {}
- if since_seconds:
- additional_kwargs['since_seconds'] = since_seconds
-
- if tail_lines:
- additional_kwargs['tail_lines'] = tail_lines
-
- try:
- return self._client.read_namespaced_pod_log(
- name=pod.metadata.name,
- namespace=pod.metadata.namespace,
- container='base',
- follow=True,
- timestamps=timestamps,
- _preload_content=False,
- **additional_kwargs,
- )
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod_events(self, pod):
- """Reads events from the POD"""
- try:
- return self._client.list_namespaced_event(
- namespace=pod.metadata.namespace, field_selector=f"involvedObject.name={pod.metadata.name}"
- )
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod(self, pod: V1Pod):
- """Read POD information"""
- try:
- return self._client.read_namespaced_pod(pod.metadata.name, pod.metadata.namespace)
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- def _extract_xcom(self, pod: V1Pod):
- resp = kubernetes_stream(
- self._client.connect_get_namespaced_pod_exec,
- pod.metadata.name,
- pod.metadata.namespace,
- container=PodDefaults.SIDECAR_CONTAINER_NAME,
- command=['/bin/sh'],
- stdin=True,
- stdout=True,
- stderr=True,
- tty=False,
- _preload_content=False,
- )
- try:
- result = self._exec_pod_command(resp, f'cat {PodDefaults.XCOM_MOUNT_PATH}/return.json')
- self._exec_pod_command(resp, 'kill -s SIGINT 1')
- finally:
- resp.close()
- if result is None:
- raise AirflowException(f'Failed to extract xcom from pod: {pod.metadata.name}')
- return result
-
- def _exec_pod_command(self, resp, command):
- if resp.is_open():
- self.log.info('Running command... %s\n', command)
- resp.write_stdin(command + '\n')
- while resp.is_open():
- resp.update(timeout=1)
- if resp.peek_stdout():
- return resp.read_stdout()
- if resp.peek_stderr():
- self.log.info(resp.read_stderr())
- break
- return None
-
- def process_status(self, job_id, status):
- """Process status information for the JOB"""
- status = status.lower()
- if status == PodStatus.PENDING:
- return State.QUEUED
- elif status == PodStatus.FAILED:
- self.log.error('Event with job id %s Failed', job_id)
- return State.FAILED
- elif status == PodStatus.SUCCEEDED:
- self.log.info('Event with job id %s Succeeded', job_id)
- return State.SUCCESS
- elif status == PodStatus.RUNNING:
- return State.RUNNING
- else:
- self.log.error('Event: Invalid state %s on job %s', status, job_id)
- return State.FAILED
+"""This module is deprecated. Please use `kubernetes.client.models for V1ResourceRequirements and Port."""
+# flake8: noqa
+# pylint: disable=unused-import
+import warnings
+
+with warnings.catch_warnings():
+ from airflow.providers.cncf.kubernetes.util import pod_launcher # pylint: disable=unused-import
+
+warnings.warn(
+ "This module is deprecated. Please use `airflow.providers.cncf.kubernetes.util.pod_launcher`",
+ DeprecationWarning,
+ stacklevel=2,
Review comment:
@eladkal added
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] kaxil commented on a change in pull request #15165: Separate pod_launcher from core airflow
Posted by GitBox <gi...@apache.org>.
kaxil commented on a change in pull request #15165:
URL: https://github.com/apache/airflow/pull/15165#discussion_r607178652
##########
File path: airflow/kubernetes/pod_launcher.py
##########
@@ -14,283 +15,15 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
-"""Launches PODs"""
-import json
-import math
-import time
-from datetime import datetime as dt
-from typing import Optional, Tuple
-
-import pendulum
-import tenacity
-from kubernetes import client, watch
-from kubernetes.client.models.v1_pod import V1Pod
-from kubernetes.client.rest import ApiException
-from kubernetes.stream import stream as kubernetes_stream
-from requests.exceptions import BaseHTTPError
-
-from airflow.exceptions import AirflowException
-from airflow.kubernetes.kube_client import get_kube_client
-from airflow.kubernetes.pod_generator import PodDefaults
-from airflow.settings import pod_mutation_hook
-from airflow.utils.log.logging_mixin import LoggingMixin
-from airflow.utils.state import State
-
-
-class PodStatus:
- """Status of the PODs"""
-
- PENDING = 'pending'
- RUNNING = 'running'
- FAILED = 'failed'
- SUCCEEDED = 'succeeded'
-
-
-class PodLauncher(LoggingMixin):
- """Launches PODS"""
-
- def __init__(
- self,
- kube_client: client.CoreV1Api = None,
- in_cluster: bool = True,
- cluster_context: Optional[str] = None,
- extract_xcom: bool = False,
- ):
- """
- Creates the launcher.
-
- :param kube_client: kubernetes client
- :param in_cluster: whether we are in cluster
- :param cluster_context: context of the cluster
- :param extract_xcom: whether we should extract xcom
- """
- super().__init__()
- self._client = kube_client or get_kube_client(in_cluster=in_cluster, cluster_context=cluster_context)
- self._watch = watch.Watch()
- self.extract_xcom = extract_xcom
-
- def run_pod_async(self, pod: V1Pod, **kwargs):
- """Runs POD asynchronously"""
- pod_mutation_hook(pod)
-
- sanitized_pod = self._client.api_client.sanitize_for_serialization(pod)
- json_pod = json.dumps(sanitized_pod, indent=2)
-
- self.log.debug('Pod Creation Request: \n%s', json_pod)
- try:
- resp = self._client.create_namespaced_pod(
- body=sanitized_pod, namespace=pod.metadata.namespace, **kwargs
- )
- self.log.debug('Pod Creation Response: %s', resp)
- except Exception as e:
- self.log.exception('Exception when attempting to create Namespaced Pod: %s', json_pod)
- raise e
- return resp
-
- def delete_pod(self, pod: V1Pod):
- """Deletes POD"""
- try:
- self._client.delete_namespaced_pod(
- pod.metadata.name, pod.metadata.namespace, body=client.V1DeleteOptions()
- )
- except ApiException as e:
- # If the pod is already deleted
- if e.status != 404:
- raise
-
- def start_pod(self, pod: V1Pod, startup_timeout: int = 120):
- """
- Launches the pod synchronously and waits for completion.
-
- :param pod:
- :param startup_timeout: Timeout for startup of the pod (if pod is pending for too long, fails task)
- :return:
- """
- resp = self.run_pod_async(pod)
- curr_time = dt.now()
- if resp.status.start_time is None:
- while self.pod_not_started(pod):
- self.log.warning("Pod not yet started: %s", pod.metadata.name)
- delta = dt.now() - curr_time
- if delta.total_seconds() >= startup_timeout:
- raise AirflowException("Pod took too long to start")
- time.sleep(1)
-
- def monitor_pod(self, pod: V1Pod, get_logs: bool) -> Tuple[State, Optional[str]]:
- """
- Monitors a pod and returns the final state
-
- :param pod: pod spec that will be monitored
- :type pod : V1Pod
- :param get_logs: whether to read the logs locally
- :return: Tuple[State, Optional[str]]
- """
- if get_logs:
- read_logs_since_sec = None
- last_log_time = None
- while True:
- logs = self.read_pod_logs(pod, timestamps=True, since_seconds=read_logs_since_sec)
- for line in logs:
- timestamp, message = self.parse_log_line(line.decode('utf-8'))
- last_log_time = pendulum.parse(timestamp)
- self.log.info(message)
- time.sleep(1)
-
- if not self.base_container_is_running(pod):
- break
-
- self.log.warning('Pod %s log read interrupted', pod.metadata.name)
- if last_log_time:
- delta = pendulum.now() - last_log_time
- # Prefer logs duplication rather than loss
- read_logs_since_sec = math.ceil(delta.total_seconds())
- result = None
- if self.extract_xcom:
- while self.base_container_is_running(pod):
- self.log.info('Container %s has state %s', pod.metadata.name, State.RUNNING)
- time.sleep(2)
- result = self._extract_xcom(pod)
- self.log.info(result)
- result = json.loads(result)
- while self.pod_is_running(pod):
- self.log.info('Pod %s has state %s', pod.metadata.name, State.RUNNING)
- time.sleep(2)
- return self._task_status(self.read_pod(pod)), result
-
- def parse_log_line(self, line: str) -> Tuple[str, str]:
- """
- Parse K8s log line and returns the final state
-
- :param line: k8s log line
- :type line: str
- :return: timestamp and log message
- :rtype: Tuple[str, str]
- """
- split_at = line.find(' ')
- if split_at == -1:
- raise Exception(f'Log not in "{{timestamp}} {{log}}" format. Got: {line}')
- timestamp = line[:split_at]
- message = line[split_at + 1 :].rstrip()
- return timestamp, message
-
- def _task_status(self, event):
- self.log.info('Event: %s had an event of type %s', event.metadata.name, event.status.phase)
- status = self.process_status(event.metadata.name, event.status.phase)
- return status
-
- def pod_not_started(self, pod: V1Pod):
- """Tests if pod has not started"""
- state = self._task_status(self.read_pod(pod))
- return state == State.QUEUED
-
- def pod_is_running(self, pod: V1Pod):
- """Tests if pod is running"""
- state = self._task_status(self.read_pod(pod))
- return state not in (State.SUCCESS, State.FAILED)
-
- def base_container_is_running(self, pod: V1Pod):
- """Tests if base container is running"""
- event = self.read_pod(pod)
- status = next(iter(filter(lambda s: s.name == 'base', event.status.container_statuses)), None)
- if not status:
- return False
- return status.state.running is not None
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod_logs(
- self,
- pod: V1Pod,
- tail_lines: Optional[int] = None,
- timestamps: bool = False,
- since_seconds: Optional[int] = None,
- ):
- """Reads log from the POD"""
- additional_kwargs = {}
- if since_seconds:
- additional_kwargs['since_seconds'] = since_seconds
-
- if tail_lines:
- additional_kwargs['tail_lines'] = tail_lines
-
- try:
- return self._client.read_namespaced_pod_log(
- name=pod.metadata.name,
- namespace=pod.metadata.namespace,
- container='base',
- follow=True,
- timestamps=timestamps,
- _preload_content=False,
- **additional_kwargs,
- )
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod_events(self, pod):
- """Reads events from the POD"""
- try:
- return self._client.list_namespaced_event(
- namespace=pod.metadata.namespace, field_selector=f"involvedObject.name={pod.metadata.name}"
- )
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod(self, pod: V1Pod):
- """Read POD information"""
- try:
- return self._client.read_namespaced_pod(pod.metadata.name, pod.metadata.namespace)
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- def _extract_xcom(self, pod: V1Pod):
- resp = kubernetes_stream(
- self._client.connect_get_namespaced_pod_exec,
- pod.metadata.name,
- pod.metadata.namespace,
- container=PodDefaults.SIDECAR_CONTAINER_NAME,
- command=['/bin/sh'],
- stdin=True,
- stdout=True,
- stderr=True,
- tty=False,
- _preload_content=False,
- )
- try:
- result = self._exec_pod_command(resp, f'cat {PodDefaults.XCOM_MOUNT_PATH}/return.json')
- self._exec_pod_command(resp, 'kill -s SIGINT 1')
- finally:
- resp.close()
- if result is None:
- raise AirflowException(f'Failed to extract xcom from pod: {pod.metadata.name}')
- return result
-
- def _exec_pod_command(self, resp, command):
- if resp.is_open():
- self.log.info('Running command... %s\n', command)
- resp.write_stdin(command + '\n')
- while resp.is_open():
- resp.update(timeout=1)
- if resp.peek_stdout():
- return resp.read_stdout()
- if resp.peek_stderr():
- self.log.info(resp.read_stderr())
- break
- return None
-
- def process_status(self, job_id, status):
- """Process status information for the JOB"""
- status = status.lower()
- if status == PodStatus.PENDING:
- return State.QUEUED
- elif status == PodStatus.FAILED:
- self.log.error('Event with job id %s Failed', job_id)
- return State.FAILED
- elif status == PodStatus.SUCCEEDED:
- self.log.info('Event with job id %s Succeeded', job_id)
- return State.SUCCESS
- elif status == PodStatus.RUNNING:
- return State.RUNNING
- else:
- self.log.error('Event: Invalid state %s on job %s', status, job_id)
- return State.FAILED
+"""This module is deprecated. Please use `kubernetes.client.models for V1ResourceRequirements and Port."""
+# flake8: noqa
+# pylint: disable=unused-import
+import warnings
+
+from airflow.kubernetes import pod_launcher_deprecated # pylint: disable=unused-import
Review comment:
Please also rebase on latest master
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] github-actions[bot] commented on pull request #15165: Separate pod_launcher from core airflow
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #15165:
URL: https://github.com/apache/airflow/pull/15165#issuecomment-812719502
[The Workflow run](https://github.com/apache/airflow/actions/runs/712801528) is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] eladkal commented on a change in pull request #15165: Separate pod_launcher from core airflow
Posted by GitBox <gi...@apache.org>.
eladkal commented on a change in pull request #15165:
URL: https://github.com/apache/airflow/pull/15165#discussion_r606416594
##########
File path: airflow/kubernetes/pod_launcher.py
##########
@@ -14,283 +15,16 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
-"""Launches PODs"""
-import json
-import math
-import time
-from datetime import datetime as dt
-from typing import Optional, Tuple
-
-import pendulum
-import tenacity
-from kubernetes import client, watch
-from kubernetes.client.models.v1_pod import V1Pod
-from kubernetes.client.rest import ApiException
-from kubernetes.stream import stream as kubernetes_stream
-from requests.exceptions import BaseHTTPError
-
-from airflow.exceptions import AirflowException
-from airflow.kubernetes.kube_client import get_kube_client
-from airflow.kubernetes.pod_generator import PodDefaults
-from airflow.settings import pod_mutation_hook
-from airflow.utils.log.logging_mixin import LoggingMixin
-from airflow.utils.state import State
-
-
-class PodStatus:
- """Status of the PODs"""
-
- PENDING = 'pending'
- RUNNING = 'running'
- FAILED = 'failed'
- SUCCEEDED = 'succeeded'
-
-
-class PodLauncher(LoggingMixin):
- """Launches PODS"""
-
- def __init__(
- self,
- kube_client: client.CoreV1Api = None,
- in_cluster: bool = True,
- cluster_context: Optional[str] = None,
- extract_xcom: bool = False,
- ):
- """
- Creates the launcher.
-
- :param kube_client: kubernetes client
- :param in_cluster: whether we are in cluster
- :param cluster_context: context of the cluster
- :param extract_xcom: whether we should extract xcom
- """
- super().__init__()
- self._client = kube_client or get_kube_client(in_cluster=in_cluster, cluster_context=cluster_context)
- self._watch = watch.Watch()
- self.extract_xcom = extract_xcom
-
- def run_pod_async(self, pod: V1Pod, **kwargs):
- """Runs POD asynchronously"""
- pod_mutation_hook(pod)
-
- sanitized_pod = self._client.api_client.sanitize_for_serialization(pod)
- json_pod = json.dumps(sanitized_pod, indent=2)
-
- self.log.debug('Pod Creation Request: \n%s', json_pod)
- try:
- resp = self._client.create_namespaced_pod(
- body=sanitized_pod, namespace=pod.metadata.namespace, **kwargs
- )
- self.log.debug('Pod Creation Response: %s', resp)
- except Exception as e:
- self.log.exception('Exception when attempting to create Namespaced Pod: %s', json_pod)
- raise e
- return resp
-
- def delete_pod(self, pod: V1Pod):
- """Deletes POD"""
- try:
- self._client.delete_namespaced_pod(
- pod.metadata.name, pod.metadata.namespace, body=client.V1DeleteOptions()
- )
- except ApiException as e:
- # If the pod is already deleted
- if e.status != 404:
- raise
-
- def start_pod(self, pod: V1Pod, startup_timeout: int = 120):
- """
- Launches the pod synchronously and waits for completion.
-
- :param pod:
- :param startup_timeout: Timeout for startup of the pod (if pod is pending for too long, fails task)
- :return:
- """
- resp = self.run_pod_async(pod)
- curr_time = dt.now()
- if resp.status.start_time is None:
- while self.pod_not_started(pod):
- self.log.warning("Pod not yet started: %s", pod.metadata.name)
- delta = dt.now() - curr_time
- if delta.total_seconds() >= startup_timeout:
- raise AirflowException("Pod took too long to start")
- time.sleep(1)
-
- def monitor_pod(self, pod: V1Pod, get_logs: bool) -> Tuple[State, Optional[str]]:
- """
- Monitors a pod and returns the final state
-
- :param pod: pod spec that will be monitored
- :type pod : V1Pod
- :param get_logs: whether to read the logs locally
- :return: Tuple[State, Optional[str]]
- """
- if get_logs:
- read_logs_since_sec = None
- last_log_time = None
- while True:
- logs = self.read_pod_logs(pod, timestamps=True, since_seconds=read_logs_since_sec)
- for line in logs:
- timestamp, message = self.parse_log_line(line.decode('utf-8'))
- last_log_time = pendulum.parse(timestamp)
- self.log.info(message)
- time.sleep(1)
-
- if not self.base_container_is_running(pod):
- break
-
- self.log.warning('Pod %s log read interrupted', pod.metadata.name)
- if last_log_time:
- delta = pendulum.now() - last_log_time
- # Prefer logs duplication rather than loss
- read_logs_since_sec = math.ceil(delta.total_seconds())
- result = None
- if self.extract_xcom:
- while self.base_container_is_running(pod):
- self.log.info('Container %s has state %s', pod.metadata.name, State.RUNNING)
- time.sleep(2)
- result = self._extract_xcom(pod)
- self.log.info(result)
- result = json.loads(result)
- while self.pod_is_running(pod):
- self.log.info('Pod %s has state %s', pod.metadata.name, State.RUNNING)
- time.sleep(2)
- return self._task_status(self.read_pod(pod)), result
-
- def parse_log_line(self, line: str) -> Tuple[str, str]:
- """
- Parse K8s log line and returns the final state
-
- :param line: k8s log line
- :type line: str
- :return: timestamp and log message
- :rtype: Tuple[str, str]
- """
- split_at = line.find(' ')
- if split_at == -1:
- raise Exception(f'Log not in "{{timestamp}} {{log}}" format. Got: {line}')
- timestamp = line[:split_at]
- message = line[split_at + 1 :].rstrip()
- return timestamp, message
-
- def _task_status(self, event):
- self.log.info('Event: %s had an event of type %s', event.metadata.name, event.status.phase)
- status = self.process_status(event.metadata.name, event.status.phase)
- return status
-
- def pod_not_started(self, pod: V1Pod):
- """Tests if pod has not started"""
- state = self._task_status(self.read_pod(pod))
- return state == State.QUEUED
-
- def pod_is_running(self, pod: V1Pod):
- """Tests if pod is running"""
- state = self._task_status(self.read_pod(pod))
- return state not in (State.SUCCESS, State.FAILED)
-
- def base_container_is_running(self, pod: V1Pod):
- """Tests if base container is running"""
- event = self.read_pod(pod)
- status = next(iter(filter(lambda s: s.name == 'base', event.status.container_statuses)), None)
- if not status:
- return False
- return status.state.running is not None
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod_logs(
- self,
- pod: V1Pod,
- tail_lines: Optional[int] = None,
- timestamps: bool = False,
- since_seconds: Optional[int] = None,
- ):
- """Reads log from the POD"""
- additional_kwargs = {}
- if since_seconds:
- additional_kwargs['since_seconds'] = since_seconds
-
- if tail_lines:
- additional_kwargs['tail_lines'] = tail_lines
-
- try:
- return self._client.read_namespaced_pod_log(
- name=pod.metadata.name,
- namespace=pod.metadata.namespace,
- container='base',
- follow=True,
- timestamps=timestamps,
- _preload_content=False,
- **additional_kwargs,
- )
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod_events(self, pod):
- """Reads events from the POD"""
- try:
- return self._client.list_namespaced_event(
- namespace=pod.metadata.namespace, field_selector=f"involvedObject.name={pod.metadata.name}"
- )
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod(self, pod: V1Pod):
- """Read POD information"""
- try:
- return self._client.read_namespaced_pod(pod.metadata.name, pod.metadata.namespace)
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- def _extract_xcom(self, pod: V1Pod):
- resp = kubernetes_stream(
- self._client.connect_get_namespaced_pod_exec,
- pod.metadata.name,
- pod.metadata.namespace,
- container=PodDefaults.SIDECAR_CONTAINER_NAME,
- command=['/bin/sh'],
- stdin=True,
- stdout=True,
- stderr=True,
- tty=False,
- _preload_content=False,
- )
- try:
- result = self._exec_pod_command(resp, f'cat {PodDefaults.XCOM_MOUNT_PATH}/return.json')
- self._exec_pod_command(resp, 'kill -s SIGINT 1')
- finally:
- resp.close()
- if result is None:
- raise AirflowException(f'Failed to extract xcom from pod: {pod.metadata.name}')
- return result
-
- def _exec_pod_command(self, resp, command):
- if resp.is_open():
- self.log.info('Running command... %s\n', command)
- resp.write_stdin(command + '\n')
- while resp.is_open():
- resp.update(timeout=1)
- if resp.peek_stdout():
- return resp.read_stdout()
- if resp.peek_stderr():
- self.log.info(resp.read_stderr())
- break
- return None
-
- def process_status(self, job_id, status):
- """Process status information for the JOB"""
- status = status.lower()
- if status == PodStatus.PENDING:
- return State.QUEUED
- elif status == PodStatus.FAILED:
- self.log.error('Event with job id %s Failed', job_id)
- return State.FAILED
- elif status == PodStatus.SUCCEEDED:
- self.log.info('Event with job id %s Succeeded', job_id)
- return State.SUCCESS
- elif status == PodStatus.RUNNING:
- return State.RUNNING
- else:
- self.log.error('Event: Invalid state %s on job %s', status, job_id)
- return State.FAILED
+"""This module is deprecated. Please use `kubernetes.client.models for V1ResourceRequirements and Port."""
+# flake8: noqa
+# pylint: disable=unused-import
+import warnings
+
+with warnings.catch_warnings():
+ from airflow.providers.cncf.kubernetes.util import pod_launcher # pylint: disable=unused-import
+
+warnings.warn(
+ "This module is deprecated. Please use `airflow.providers.cncf.kubernetes.util.pod_launcher`",
+ DeprecationWarning,
+ stacklevel=2,
Review comment:
Should we note in `UPDATING.md` that this class is deprecated?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on a change in pull request #15165: Separate pod_launcher from core airflow
Posted by GitBox <gi...@apache.org>.
potiuk commented on a change in pull request #15165:
URL: https://github.com/apache/airflow/pull/15165#discussion_r607214375
##########
File path: airflow/kubernetes/pod_launcher_deprecated.py
##########
@@ -0,0 +1,300 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Launches PODs"""
+import json
+import math
+import time
+from datetime import datetime as dt
+from typing import Optional, Tuple
+
+import pendulum
+import tenacity
+from kubernetes import client, watch
+from kubernetes.client.models.v1_pod import V1Pod
+from kubernetes.client.rest import ApiException
+from kubernetes.stream import stream as kubernetes_stream
+from requests.exceptions import BaseHTTPError
+
+from airflow.exceptions import AirflowException
+from airflow.kubernetes.kube_client import get_kube_client
+from airflow.kubernetes.pod_generator import PodDefaults
+from airflow.settings import pod_mutation_hook
+from airflow.utils.log.logging_mixin import LoggingMixin
+from airflow.utils.state import State
+
+
Review comment:
Should we also add the deprecation warning directly here (or even move the warning from the old pod_launcher)? This way even if someone by mistake imports PodLauncher/PodStatus directly from there (because IDE will provide this as first-class option), will still get the warning.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a change in pull request #15165: Separate pod_launcher from core airflow
Posted by GitBox <gi...@apache.org>.
ashb commented on a change in pull request #15165:
URL: https://github.com/apache/airflow/pull/15165#discussion_r606712369
##########
File path: airflow/executors/kubernetes_executor.py
##########
@@ -237,12 +238,29 @@ def __init__(
self.namespace = self.kube_config.kube_namespace
self.log.debug("Kubernetes using namespace %s", self.namespace)
self.kube_client = kube_client
- self.launcher = PodLauncher(kube_client=self.kube_client)
self._manager = multiprocessing.Manager()
self.watcher_queue = self._manager.Queue()
self.scheduler_job_id = scheduler_job_id
self.kube_watcher = self._make_kube_watcher()
+ def run_pod_async(self, pod: V1Pod, **kwargs):
+ """Runs POD asynchronously"""
+ pod_mutation_hook(pod)
+
+ sanitized_pod = self.kube_client.api_client.sanitize_for_serialization(pod)
+ json_pod = json.dumps(sanitized_pod, indent=2)
+
+ self.log.debug('Pod Creation Request: \n%s', json_pod)
+ try:
+ resp = self.kube_client.create_namespaced_pod(
+ body=sanitized_pod, namespace=pod.metadata.namespace, **kwargs
+ )
+ self.log.debug('Pod Creation Response: %s', resp)
+ except Exception as e:
+ self.log.exception('Exception when attempting to create Namespaced Pod: %s', json_pod)
+ raise e
+ return resp
Review comment:
This is all the executor used the pod launcher for?!
I was expecting this to be a much bigger change
##########
File path: airflow/kubernetes/pod_launcher.py
##########
@@ -14,283 +15,16 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
-"""Launches PODs"""
-import json
-import math
-import time
-from datetime import datetime as dt
-from typing import Optional, Tuple
-
-import pendulum
-import tenacity
-from kubernetes import client, watch
-from kubernetes.client.models.v1_pod import V1Pod
-from kubernetes.client.rest import ApiException
-from kubernetes.stream import stream as kubernetes_stream
-from requests.exceptions import BaseHTTPError
-
-from airflow.exceptions import AirflowException
-from airflow.kubernetes.kube_client import get_kube_client
-from airflow.kubernetes.pod_generator import PodDefaults
-from airflow.settings import pod_mutation_hook
-from airflow.utils.log.logging_mixin import LoggingMixin
-from airflow.utils.state import State
-
-
-class PodStatus:
- """Status of the PODs"""
-
- PENDING = 'pending'
- RUNNING = 'running'
- FAILED = 'failed'
- SUCCEEDED = 'succeeded'
-
-
-class PodLauncher(LoggingMixin):
- """Launches PODS"""
-
- def __init__(
- self,
- kube_client: client.CoreV1Api = None,
- in_cluster: bool = True,
- cluster_context: Optional[str] = None,
- extract_xcom: bool = False,
- ):
- """
- Creates the launcher.
-
- :param kube_client: kubernetes client
- :param in_cluster: whether we are in cluster
- :param cluster_context: context of the cluster
- :param extract_xcom: whether we should extract xcom
- """
- super().__init__()
- self._client = kube_client or get_kube_client(in_cluster=in_cluster, cluster_context=cluster_context)
- self._watch = watch.Watch()
- self.extract_xcom = extract_xcom
-
- def run_pod_async(self, pod: V1Pod, **kwargs):
- """Runs POD asynchronously"""
- pod_mutation_hook(pod)
-
- sanitized_pod = self._client.api_client.sanitize_for_serialization(pod)
- json_pod = json.dumps(sanitized_pod, indent=2)
-
- self.log.debug('Pod Creation Request: \n%s', json_pod)
- try:
- resp = self._client.create_namespaced_pod(
- body=sanitized_pod, namespace=pod.metadata.namespace, **kwargs
- )
- self.log.debug('Pod Creation Response: %s', resp)
- except Exception as e:
- self.log.exception('Exception when attempting to create Namespaced Pod: %s', json_pod)
- raise e
- return resp
-
- def delete_pod(self, pod: V1Pod):
- """Deletes POD"""
- try:
- self._client.delete_namespaced_pod(
- pod.metadata.name, pod.metadata.namespace, body=client.V1DeleteOptions()
- )
- except ApiException as e:
- # If the pod is already deleted
- if e.status != 404:
- raise
-
- def start_pod(self, pod: V1Pod, startup_timeout: int = 120):
- """
- Launches the pod synchronously and waits for completion.
-
- :param pod:
- :param startup_timeout: Timeout for startup of the pod (if pod is pending for too long, fails task)
- :return:
- """
- resp = self.run_pod_async(pod)
- curr_time = dt.now()
- if resp.status.start_time is None:
- while self.pod_not_started(pod):
- self.log.warning("Pod not yet started: %s", pod.metadata.name)
- delta = dt.now() - curr_time
- if delta.total_seconds() >= startup_timeout:
- raise AirflowException("Pod took too long to start")
- time.sleep(1)
-
- def monitor_pod(self, pod: V1Pod, get_logs: bool) -> Tuple[State, Optional[str]]:
- """
- Monitors a pod and returns the final state
-
- :param pod: pod spec that will be monitored
- :type pod : V1Pod
- :param get_logs: whether to read the logs locally
- :return: Tuple[State, Optional[str]]
- """
- if get_logs:
- read_logs_since_sec = None
- last_log_time = None
- while True:
- logs = self.read_pod_logs(pod, timestamps=True, since_seconds=read_logs_since_sec)
- for line in logs:
- timestamp, message = self.parse_log_line(line.decode('utf-8'))
- last_log_time = pendulum.parse(timestamp)
- self.log.info(message)
- time.sleep(1)
-
- if not self.base_container_is_running(pod):
- break
-
- self.log.warning('Pod %s log read interrupted', pod.metadata.name)
- if last_log_time:
- delta = pendulum.now() - last_log_time
- # Prefer logs duplication rather than loss
- read_logs_since_sec = math.ceil(delta.total_seconds())
- result = None
- if self.extract_xcom:
- while self.base_container_is_running(pod):
- self.log.info('Container %s has state %s', pod.metadata.name, State.RUNNING)
- time.sleep(2)
- result = self._extract_xcom(pod)
- self.log.info(result)
- result = json.loads(result)
- while self.pod_is_running(pod):
- self.log.info('Pod %s has state %s', pod.metadata.name, State.RUNNING)
- time.sleep(2)
- return self._task_status(self.read_pod(pod)), result
-
- def parse_log_line(self, line: str) -> Tuple[str, str]:
- """
- Parse K8s log line and returns the final state
-
- :param line: k8s log line
- :type line: str
- :return: timestamp and log message
- :rtype: Tuple[str, str]
- """
- split_at = line.find(' ')
- if split_at == -1:
- raise Exception(f'Log not in "{{timestamp}} {{log}}" format. Got: {line}')
- timestamp = line[:split_at]
- message = line[split_at + 1 :].rstrip()
- return timestamp, message
-
- def _task_status(self, event):
- self.log.info('Event: %s had an event of type %s', event.metadata.name, event.status.phase)
- status = self.process_status(event.metadata.name, event.status.phase)
- return status
-
- def pod_not_started(self, pod: V1Pod):
- """Tests if pod has not started"""
- state = self._task_status(self.read_pod(pod))
- return state == State.QUEUED
-
- def pod_is_running(self, pod: V1Pod):
- """Tests if pod is running"""
- state = self._task_status(self.read_pod(pod))
- return state not in (State.SUCCESS, State.FAILED)
-
- def base_container_is_running(self, pod: V1Pod):
- """Tests if base container is running"""
- event = self.read_pod(pod)
- status = next(iter(filter(lambda s: s.name == 'base', event.status.container_statuses)), None)
- if not status:
- return False
- return status.state.running is not None
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod_logs(
- self,
- pod: V1Pod,
- tail_lines: Optional[int] = None,
- timestamps: bool = False,
- since_seconds: Optional[int] = None,
- ):
- """Reads log from the POD"""
- additional_kwargs = {}
- if since_seconds:
- additional_kwargs['since_seconds'] = since_seconds
-
- if tail_lines:
- additional_kwargs['tail_lines'] = tail_lines
-
- try:
- return self._client.read_namespaced_pod_log(
- name=pod.metadata.name,
- namespace=pod.metadata.namespace,
- container='base',
- follow=True,
- timestamps=timestamps,
- _preload_content=False,
- **additional_kwargs,
- )
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod_events(self, pod):
- """Reads events from the POD"""
- try:
- return self._client.list_namespaced_event(
- namespace=pod.metadata.namespace, field_selector=f"involvedObject.name={pod.metadata.name}"
- )
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod(self, pod: V1Pod):
- """Read POD information"""
- try:
- return self._client.read_namespaced_pod(pod.metadata.name, pod.metadata.namespace)
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- def _extract_xcom(self, pod: V1Pod):
- resp = kubernetes_stream(
- self._client.connect_get_namespaced_pod_exec,
- pod.metadata.name,
- pod.metadata.namespace,
- container=PodDefaults.SIDECAR_CONTAINER_NAME,
- command=['/bin/sh'],
- stdin=True,
- stdout=True,
- stderr=True,
- tty=False,
- _preload_content=False,
- )
- try:
- result = self._exec_pod_command(resp, f'cat {PodDefaults.XCOM_MOUNT_PATH}/return.json')
- self._exec_pod_command(resp, 'kill -s SIGINT 1')
- finally:
- resp.close()
- if result is None:
- raise AirflowException(f'Failed to extract xcom from pod: {pod.metadata.name}')
- return result
-
- def _exec_pod_command(self, resp, command):
- if resp.is_open():
- self.log.info('Running command... %s\n', command)
- resp.write_stdin(command + '\n')
- while resp.is_open():
- resp.update(timeout=1)
- if resp.peek_stdout():
- return resp.read_stdout()
- if resp.peek_stderr():
- self.log.info(resp.read_stderr())
- break
- return None
-
- def process_status(self, job_id, status):
- """Process status information for the JOB"""
- status = status.lower()
- if status == PodStatus.PENDING:
- return State.QUEUED
- elif status == PodStatus.FAILED:
- self.log.error('Event with job id %s Failed', job_id)
- return State.FAILED
- elif status == PodStatus.SUCCEEDED:
- self.log.info('Event with job id %s Succeeded', job_id)
- return State.SUCCESS
- elif status == PodStatus.RUNNING:
- return State.RUNNING
- else:
- self.log.error('Event: Invalid state %s on job %s', status, job_id)
- return State.FAILED
+"""This module is deprecated. Please use `kubernetes.client.models for V1ResourceRequirements and Port."""
+# flake8: noqa
+# pylint: disable=unused-import
+import warnings
+
+with warnings.catch_warnings():
+ from airflow.providers.cncf.kubernetes.utils import pod_launcher # pylint: disable=unused-import
Review comment:
This shouldn't warm, surely?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] kaxil commented on a change in pull request #15165: Separate Kubernetes pod_launcher from core airflow
Posted by GitBox <gi...@apache.org>.
kaxil commented on a change in pull request #15165:
URL: https://github.com/apache/airflow/pull/15165#discussion_r607253114
##########
File path: UPDATING.md
##########
@@ -70,6 +70,12 @@ https://developers.google.com/style/inclusive-documentation
-->
+### Removed pod_launcher from core airflow
Review comment:
```suggestion
### Removed Kubernete `pod_launcher` from core airflow
```
##########
File path: airflow/kubernetes/pod_launcher.py
##########
@@ -14,283 +15,7 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
-"""Launches PODs"""
-import json
-import math
-import time
-from datetime import datetime as dt
-from typing import Optional, Tuple
-
-import pendulum
-import tenacity
-from kubernetes import client, watch
-from kubernetes.client.models.v1_pod import V1Pod
-from kubernetes.client.rest import ApiException
-from kubernetes.stream import stream as kubernetes_stream
-from requests.exceptions import BaseHTTPError
-
-from airflow.exceptions import AirflowException
-from airflow.kubernetes.kube_client import get_kube_client
-from airflow.kubernetes.pod_generator import PodDefaults
-from airflow.settings import pod_mutation_hook
-from airflow.utils.log.logging_mixin import LoggingMixin
-from airflow.utils.state import State
-
-
-class PodStatus:
- """Status of the PODs"""
-
- PENDING = 'pending'
- RUNNING = 'running'
- FAILED = 'failed'
- SUCCEEDED = 'succeeded'
-
-
-class PodLauncher(LoggingMixin):
- """Launches PODS"""
-
- def __init__(
- self,
- kube_client: client.CoreV1Api = None,
- in_cluster: bool = True,
- cluster_context: Optional[str] = None,
- extract_xcom: bool = False,
- ):
- """
- Creates the launcher.
-
- :param kube_client: kubernetes client
- :param in_cluster: whether we are in cluster
- :param cluster_context: context of the cluster
- :param extract_xcom: whether we should extract xcom
- """
- super().__init__()
- self._client = kube_client or get_kube_client(in_cluster=in_cluster, cluster_context=cluster_context)
- self._watch = watch.Watch()
- self.extract_xcom = extract_xcom
-
- def run_pod_async(self, pod: V1Pod, **kwargs):
- """Runs POD asynchronously"""
- pod_mutation_hook(pod)
-
- sanitized_pod = self._client.api_client.sanitize_for_serialization(pod)
- json_pod = json.dumps(sanitized_pod, indent=2)
-
- self.log.debug('Pod Creation Request: \n%s', json_pod)
- try:
- resp = self._client.create_namespaced_pod(
- body=sanitized_pod, namespace=pod.metadata.namespace, **kwargs
- )
- self.log.debug('Pod Creation Response: %s', resp)
- except Exception as e:
- self.log.exception('Exception when attempting to create Namespaced Pod: %s', json_pod)
- raise e
- return resp
-
- def delete_pod(self, pod: V1Pod):
- """Deletes POD"""
- try:
- self._client.delete_namespaced_pod(
- pod.metadata.name, pod.metadata.namespace, body=client.V1DeleteOptions()
- )
- except ApiException as e:
- # If the pod is already deleted
- if e.status != 404:
- raise
-
- def start_pod(self, pod: V1Pod, startup_timeout: int = 120):
- """
- Launches the pod synchronously and waits for completion.
-
- :param pod:
- :param startup_timeout: Timeout for startup of the pod (if pod is pending for too long, fails task)
- :return:
- """
- resp = self.run_pod_async(pod)
- curr_time = dt.now()
- if resp.status.start_time is None:
- while self.pod_not_started(pod):
- self.log.warning("Pod not yet started: %s", pod.metadata.name)
- delta = dt.now() - curr_time
- if delta.total_seconds() >= startup_timeout:
- raise AirflowException("Pod took too long to start")
- time.sleep(1)
-
- def monitor_pod(self, pod: V1Pod, get_logs: bool) -> Tuple[State, Optional[str]]:
- """
- Monitors a pod and returns the final state
-
- :param pod: pod spec that will be monitored
- :type pod : V1Pod
- :param get_logs: whether to read the logs locally
- :return: Tuple[State, Optional[str]]
- """
- if get_logs:
- read_logs_since_sec = None
- last_log_time = None
- while True:
- logs = self.read_pod_logs(pod, timestamps=True, since_seconds=read_logs_since_sec)
- for line in logs:
- timestamp, message = self.parse_log_line(line.decode('utf-8'))
- last_log_time = pendulum.parse(timestamp)
- self.log.info(message)
- time.sleep(1)
-
- if not self.base_container_is_running(pod):
- break
-
- self.log.warning('Pod %s log read interrupted', pod.metadata.name)
- if last_log_time:
- delta = pendulum.now() - last_log_time
- # Prefer logs duplication rather than loss
- read_logs_since_sec = math.ceil(delta.total_seconds())
- result = None
- if self.extract_xcom:
- while self.base_container_is_running(pod):
- self.log.info('Container %s has state %s', pod.metadata.name, State.RUNNING)
- time.sleep(2)
- result = self._extract_xcom(pod)
- self.log.info(result)
- result = json.loads(result)
- while self.pod_is_running(pod):
- self.log.info('Pod %s has state %s', pod.metadata.name, State.RUNNING)
- time.sleep(2)
- return self._task_status(self.read_pod(pod)), result
-
- def parse_log_line(self, line: str) -> Tuple[str, str]:
- """
- Parse K8s log line and returns the final state
-
- :param line: k8s log line
- :type line: str
- :return: timestamp and log message
- :rtype: Tuple[str, str]
- """
- split_at = line.find(' ')
- if split_at == -1:
- raise Exception(f'Log not in "{{timestamp}} {{log}}" format. Got: {line}')
- timestamp = line[:split_at]
- message = line[split_at + 1 :].rstrip()
- return timestamp, message
-
- def _task_status(self, event):
- self.log.info('Event: %s had an event of type %s', event.metadata.name, event.status.phase)
- status = self.process_status(event.metadata.name, event.status.phase)
- return status
-
- def pod_not_started(self, pod: V1Pod):
- """Tests if pod has not started"""
- state = self._task_status(self.read_pod(pod))
- return state == State.QUEUED
-
- def pod_is_running(self, pod: V1Pod):
- """Tests if pod is running"""
- state = self._task_status(self.read_pod(pod))
- return state not in (State.SUCCESS, State.FAILED)
-
- def base_container_is_running(self, pod: V1Pod):
- """Tests if base container is running"""
- event = self.read_pod(pod)
- status = next(iter(filter(lambda s: s.name == 'base', event.status.container_statuses)), None)
- if not status:
- return False
- return status.state.running is not None
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod_logs(
- self,
- pod: V1Pod,
- tail_lines: Optional[int] = None,
- timestamps: bool = False,
- since_seconds: Optional[int] = None,
- ):
- """Reads log from the POD"""
- additional_kwargs = {}
- if since_seconds:
- additional_kwargs['since_seconds'] = since_seconds
-
- if tail_lines:
- additional_kwargs['tail_lines'] = tail_lines
-
- try:
- return self._client.read_namespaced_pod_log(
- name=pod.metadata.name,
- namespace=pod.metadata.namespace,
- container='base',
- follow=True,
- timestamps=timestamps,
- _preload_content=False,
- **additional_kwargs,
- )
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod_events(self, pod):
- """Reads events from the POD"""
- try:
- return self._client.list_namespaced_event(
- namespace=pod.metadata.namespace, field_selector=f"involvedObject.name={pod.metadata.name}"
- )
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod(self, pod: V1Pod):
- """Read POD information"""
- try:
- return self._client.read_namespaced_pod(pod.metadata.name, pod.metadata.namespace)
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- def _extract_xcom(self, pod: V1Pod):
- resp = kubernetes_stream(
- self._client.connect_get_namespaced_pod_exec,
- pod.metadata.name,
- pod.metadata.namespace,
- container=PodDefaults.SIDECAR_CONTAINER_NAME,
- command=['/bin/sh'],
- stdin=True,
- stdout=True,
- stderr=True,
- tty=False,
- _preload_content=False,
- )
- try:
- result = self._exec_pod_command(resp, f'cat {PodDefaults.XCOM_MOUNT_PATH}/return.json')
- self._exec_pod_command(resp, 'kill -s SIGINT 1')
- finally:
- resp.close()
- if result is None:
- raise AirflowException(f'Failed to extract xcom from pod: {pod.metadata.name}')
- return result
-
- def _exec_pod_command(self, resp, command):
- if resp.is_open():
- self.log.info('Running command... %s\n', command)
- resp.write_stdin(command + '\n')
- while resp.is_open():
- resp.update(timeout=1)
- if resp.peek_stdout():
- return resp.read_stdout()
- if resp.peek_stderr():
- self.log.info(resp.read_stderr())
- break
- return None
-
- def process_status(self, job_id, status):
- """Process status information for the JOB"""
- status = status.lower()
- if status == PodStatus.PENDING:
- return State.QUEUED
- elif status == PodStatus.FAILED:
- self.log.error('Event with job id %s Failed', job_id)
- return State.FAILED
- elif status == PodStatus.SUCCEEDED:
- self.log.info('Event with job id %s Succeeded', job_id)
- return State.SUCCESS
- elif status == PodStatus.RUNNING:
- return State.RUNNING
- else:
- self.log.error('Event: Invalid state %s on job %s', status, job_id)
- return State.FAILED
+"""This module is deprecated. Please use `kubernetes.client.models for V1ResourceRequirements and Port."""
Review comment:
The message looks incorrect, maybe add something about using the cncf.kubernetes provider to use PodLauncher
##########
File path: UPDATING.md
##########
@@ -70,6 +70,12 @@ https://developers.google.com/style/inclusive-documentation
-->
+### Removed pod_launcher from core airflow
+
+Moved the pod launcher from `airflow.kubernetes.pod_launcher` to `airflow.providers.cncf.kubernetes.utils.pod_launcher`
+
+This will alow users to fix the KubernetesPodOperator without requiring an airflow upgrade
Review comment:
```suggestion
This will allow users to release new versions of Kubernetes provider without requiring an airflow upgrade
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] SamWheating commented on a change in pull request #15165: Separate pod_launcher from core airflow
Posted by GitBox <gi...@apache.org>.
SamWheating commented on a change in pull request #15165:
URL: https://github.com/apache/airflow/pull/15165#discussion_r607127192
##########
File path: airflow/kubernetes/pod_launcher.py
##########
@@ -14,283 +15,15 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
-"""Launches PODs"""
-import json
-import math
-import time
-from datetime import datetime as dt
-from typing import Optional, Tuple
-
-import pendulum
-import tenacity
-from kubernetes import client, watch
-from kubernetes.client.models.v1_pod import V1Pod
-from kubernetes.client.rest import ApiException
-from kubernetes.stream import stream as kubernetes_stream
-from requests.exceptions import BaseHTTPError
-
-from airflow.exceptions import AirflowException
-from airflow.kubernetes.kube_client import get_kube_client
-from airflow.kubernetes.pod_generator import PodDefaults
-from airflow.settings import pod_mutation_hook
-from airflow.utils.log.logging_mixin import LoggingMixin
-from airflow.utils.state import State
-
-
-class PodStatus:
- """Status of the PODs"""
-
- PENDING = 'pending'
- RUNNING = 'running'
- FAILED = 'failed'
- SUCCEEDED = 'succeeded'
-
-
-class PodLauncher(LoggingMixin):
- """Launches PODS"""
-
- def __init__(
- self,
- kube_client: client.CoreV1Api = None,
- in_cluster: bool = True,
- cluster_context: Optional[str] = None,
- extract_xcom: bool = False,
- ):
- """
- Creates the launcher.
-
- :param kube_client: kubernetes client
- :param in_cluster: whether we are in cluster
- :param cluster_context: context of the cluster
- :param extract_xcom: whether we should extract xcom
- """
- super().__init__()
- self._client = kube_client or get_kube_client(in_cluster=in_cluster, cluster_context=cluster_context)
- self._watch = watch.Watch()
- self.extract_xcom = extract_xcom
-
- def run_pod_async(self, pod: V1Pod, **kwargs):
- """Runs POD asynchronously"""
- pod_mutation_hook(pod)
-
- sanitized_pod = self._client.api_client.sanitize_for_serialization(pod)
- json_pod = json.dumps(sanitized_pod, indent=2)
-
- self.log.debug('Pod Creation Request: \n%s', json_pod)
- try:
- resp = self._client.create_namespaced_pod(
- body=sanitized_pod, namespace=pod.metadata.namespace, **kwargs
- )
- self.log.debug('Pod Creation Response: %s', resp)
- except Exception as e:
- self.log.exception('Exception when attempting to create Namespaced Pod: %s', json_pod)
- raise e
- return resp
-
- def delete_pod(self, pod: V1Pod):
- """Deletes POD"""
- try:
- self._client.delete_namespaced_pod(
- pod.metadata.name, pod.metadata.namespace, body=client.V1DeleteOptions()
- )
- except ApiException as e:
- # If the pod is already deleted
- if e.status != 404:
- raise
-
- def start_pod(self, pod: V1Pod, startup_timeout: int = 120):
- """
- Launches the pod synchronously and waits for completion.
-
- :param pod:
- :param startup_timeout: Timeout for startup of the pod (if pod is pending for too long, fails task)
- :return:
- """
- resp = self.run_pod_async(pod)
- curr_time = dt.now()
- if resp.status.start_time is None:
- while self.pod_not_started(pod):
- self.log.warning("Pod not yet started: %s", pod.metadata.name)
- delta = dt.now() - curr_time
- if delta.total_seconds() >= startup_timeout:
- raise AirflowException("Pod took too long to start")
- time.sleep(1)
-
- def monitor_pod(self, pod: V1Pod, get_logs: bool) -> Tuple[State, Optional[str]]:
- """
- Monitors a pod and returns the final state
-
- :param pod: pod spec that will be monitored
- :type pod : V1Pod
- :param get_logs: whether to read the logs locally
- :return: Tuple[State, Optional[str]]
- """
- if get_logs:
- read_logs_since_sec = None
- last_log_time = None
- while True:
- logs = self.read_pod_logs(pod, timestamps=True, since_seconds=read_logs_since_sec)
- for line in logs:
- timestamp, message = self.parse_log_line(line.decode('utf-8'))
- last_log_time = pendulum.parse(timestamp)
- self.log.info(message)
- time.sleep(1)
-
- if not self.base_container_is_running(pod):
- break
-
- self.log.warning('Pod %s log read interrupted', pod.metadata.name)
- if last_log_time:
- delta = pendulum.now() - last_log_time
- # Prefer logs duplication rather than loss
- read_logs_since_sec = math.ceil(delta.total_seconds())
- result = None
- if self.extract_xcom:
- while self.base_container_is_running(pod):
- self.log.info('Container %s has state %s', pod.metadata.name, State.RUNNING)
- time.sleep(2)
- result = self._extract_xcom(pod)
- self.log.info(result)
- result = json.loads(result)
- while self.pod_is_running(pod):
- self.log.info('Pod %s has state %s', pod.metadata.name, State.RUNNING)
- time.sleep(2)
- return self._task_status(self.read_pod(pod)), result
-
- def parse_log_line(self, line: str) -> Tuple[str, str]:
- """
- Parse K8s log line and returns the final state
-
- :param line: k8s log line
- :type line: str
- :return: timestamp and log message
- :rtype: Tuple[str, str]
- """
- split_at = line.find(' ')
- if split_at == -1:
- raise Exception(f'Log not in "{{timestamp}} {{log}}" format. Got: {line}')
- timestamp = line[:split_at]
- message = line[split_at + 1 :].rstrip()
- return timestamp, message
-
- def _task_status(self, event):
- self.log.info('Event: %s had an event of type %s', event.metadata.name, event.status.phase)
- status = self.process_status(event.metadata.name, event.status.phase)
- return status
-
- def pod_not_started(self, pod: V1Pod):
- """Tests if pod has not started"""
- state = self._task_status(self.read_pod(pod))
- return state == State.QUEUED
-
- def pod_is_running(self, pod: V1Pod):
- """Tests if pod is running"""
- state = self._task_status(self.read_pod(pod))
- return state not in (State.SUCCESS, State.FAILED)
-
- def base_container_is_running(self, pod: V1Pod):
- """Tests if base container is running"""
- event = self.read_pod(pod)
- status = next(iter(filter(lambda s: s.name == 'base', event.status.container_statuses)), None)
- if not status:
- return False
- return status.state.running is not None
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod_logs(
- self,
- pod: V1Pod,
- tail_lines: Optional[int] = None,
- timestamps: bool = False,
- since_seconds: Optional[int] = None,
- ):
- """Reads log from the POD"""
- additional_kwargs = {}
- if since_seconds:
- additional_kwargs['since_seconds'] = since_seconds
-
- if tail_lines:
- additional_kwargs['tail_lines'] = tail_lines
-
- try:
- return self._client.read_namespaced_pod_log(
- name=pod.metadata.name,
- namespace=pod.metadata.namespace,
- container='base',
- follow=True,
- timestamps=timestamps,
- _preload_content=False,
- **additional_kwargs,
- )
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod_events(self, pod):
- """Reads events from the POD"""
- try:
- return self._client.list_namespaced_event(
- namespace=pod.metadata.namespace, field_selector=f"involvedObject.name={pod.metadata.name}"
- )
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod(self, pod: V1Pod):
- """Read POD information"""
- try:
- return self._client.read_namespaced_pod(pod.metadata.name, pod.metadata.namespace)
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- def _extract_xcom(self, pod: V1Pod):
- resp = kubernetes_stream(
- self._client.connect_get_namespaced_pod_exec,
- pod.metadata.name,
- pod.metadata.namespace,
- container=PodDefaults.SIDECAR_CONTAINER_NAME,
- command=['/bin/sh'],
- stdin=True,
- stdout=True,
- stderr=True,
- tty=False,
- _preload_content=False,
- )
- try:
- result = self._exec_pod_command(resp, f'cat {PodDefaults.XCOM_MOUNT_PATH}/return.json')
- self._exec_pod_command(resp, 'kill -s SIGINT 1')
- finally:
- resp.close()
- if result is None:
- raise AirflowException(f'Failed to extract xcom from pod: {pod.metadata.name}')
- return result
-
- def _exec_pod_command(self, resp, command):
- if resp.is_open():
- self.log.info('Running command... %s\n', command)
- resp.write_stdin(command + '\n')
- while resp.is_open():
- resp.update(timeout=1)
- if resp.peek_stdout():
- return resp.read_stdout()
- if resp.peek_stderr():
- self.log.info(resp.read_stderr())
- break
- return None
-
- def process_status(self, job_id, status):
- """Process status information for the JOB"""
- status = status.lower()
- if status == PodStatus.PENDING:
- return State.QUEUED
- elif status == PodStatus.FAILED:
- self.log.error('Event with job id %s Failed', job_id)
- return State.FAILED
- elif status == PodStatus.SUCCEEDED:
- self.log.info('Event with job id %s Succeeded', job_id)
- return State.SUCCESS
- elif status == PodStatus.RUNNING:
- return State.RUNNING
- else:
- self.log.error('Event: Invalid state %s on job %s', status, job_id)
- return State.FAILED
+"""This module is deprecated. Please use `kubernetes.client.models for V1ResourceRequirements and Port."""
Review comment:
This is a really good point. I'd be in favour of the third option here, especially as the CNCF provider doesn't introduce many additional installation requirements (though I suppose this is subject to change if other functionality is added to the provider)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] SamWheating commented on a change in pull request #15165: Separate pod_launcher from core airflow
Posted by GitBox <gi...@apache.org>.
SamWheating commented on a change in pull request #15165:
URL: https://github.com/apache/airflow/pull/15165#discussion_r607128196
##########
File path: airflow/kubernetes/pod_launcher.py
##########
@@ -14,283 +15,15 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
-"""Launches PODs"""
-import json
-import math
-import time
-from datetime import datetime as dt
-from typing import Optional, Tuple
-
-import pendulum
-import tenacity
-from kubernetes import client, watch
-from kubernetes.client.models.v1_pod import V1Pod
-from kubernetes.client.rest import ApiException
-from kubernetes.stream import stream as kubernetes_stream
-from requests.exceptions import BaseHTTPError
-
-from airflow.exceptions import AirflowException
-from airflow.kubernetes.kube_client import get_kube_client
-from airflow.kubernetes.pod_generator import PodDefaults
-from airflow.settings import pod_mutation_hook
-from airflow.utils.log.logging_mixin import LoggingMixin
-from airflow.utils.state import State
-
-
-class PodStatus:
- """Status of the PODs"""
-
- PENDING = 'pending'
- RUNNING = 'running'
- FAILED = 'failed'
- SUCCEEDED = 'succeeded'
-
-
-class PodLauncher(LoggingMixin):
- """Launches PODS"""
-
- def __init__(
- self,
- kube_client: client.CoreV1Api = None,
- in_cluster: bool = True,
- cluster_context: Optional[str] = None,
- extract_xcom: bool = False,
- ):
- """
- Creates the launcher.
-
- :param kube_client: kubernetes client
- :param in_cluster: whether we are in cluster
- :param cluster_context: context of the cluster
- :param extract_xcom: whether we should extract xcom
- """
- super().__init__()
- self._client = kube_client or get_kube_client(in_cluster=in_cluster, cluster_context=cluster_context)
- self._watch = watch.Watch()
- self.extract_xcom = extract_xcom
-
- def run_pod_async(self, pod: V1Pod, **kwargs):
- """Runs POD asynchronously"""
- pod_mutation_hook(pod)
-
- sanitized_pod = self._client.api_client.sanitize_for_serialization(pod)
- json_pod = json.dumps(sanitized_pod, indent=2)
-
- self.log.debug('Pod Creation Request: \n%s', json_pod)
- try:
- resp = self._client.create_namespaced_pod(
- body=sanitized_pod, namespace=pod.metadata.namespace, **kwargs
- )
- self.log.debug('Pod Creation Response: %s', resp)
- except Exception as e:
- self.log.exception('Exception when attempting to create Namespaced Pod: %s', json_pod)
- raise e
- return resp
-
- def delete_pod(self, pod: V1Pod):
- """Deletes POD"""
- try:
- self._client.delete_namespaced_pod(
- pod.metadata.name, pod.metadata.namespace, body=client.V1DeleteOptions()
- )
- except ApiException as e:
- # If the pod is already deleted
- if e.status != 404:
- raise
-
- def start_pod(self, pod: V1Pod, startup_timeout: int = 120):
- """
- Launches the pod synchronously and waits for completion.
-
- :param pod:
- :param startup_timeout: Timeout for startup of the pod (if pod is pending for too long, fails task)
- :return:
- """
- resp = self.run_pod_async(pod)
- curr_time = dt.now()
- if resp.status.start_time is None:
- while self.pod_not_started(pod):
- self.log.warning("Pod not yet started: %s", pod.metadata.name)
- delta = dt.now() - curr_time
- if delta.total_seconds() >= startup_timeout:
- raise AirflowException("Pod took too long to start")
- time.sleep(1)
-
- def monitor_pod(self, pod: V1Pod, get_logs: bool) -> Tuple[State, Optional[str]]:
- """
- Monitors a pod and returns the final state
-
- :param pod: pod spec that will be monitored
- :type pod : V1Pod
- :param get_logs: whether to read the logs locally
- :return: Tuple[State, Optional[str]]
- """
- if get_logs:
- read_logs_since_sec = None
- last_log_time = None
- while True:
- logs = self.read_pod_logs(pod, timestamps=True, since_seconds=read_logs_since_sec)
- for line in logs:
- timestamp, message = self.parse_log_line(line.decode('utf-8'))
- last_log_time = pendulum.parse(timestamp)
- self.log.info(message)
- time.sleep(1)
-
- if not self.base_container_is_running(pod):
- break
-
- self.log.warning('Pod %s log read interrupted', pod.metadata.name)
- if last_log_time:
- delta = pendulum.now() - last_log_time
- # Prefer logs duplication rather than loss
- read_logs_since_sec = math.ceil(delta.total_seconds())
- result = None
- if self.extract_xcom:
- while self.base_container_is_running(pod):
- self.log.info('Container %s has state %s', pod.metadata.name, State.RUNNING)
- time.sleep(2)
- result = self._extract_xcom(pod)
- self.log.info(result)
- result = json.loads(result)
- while self.pod_is_running(pod):
- self.log.info('Pod %s has state %s', pod.metadata.name, State.RUNNING)
- time.sleep(2)
- return self._task_status(self.read_pod(pod)), result
-
- def parse_log_line(self, line: str) -> Tuple[str, str]:
- """
- Parse K8s log line and returns the final state
-
- :param line: k8s log line
- :type line: str
- :return: timestamp and log message
- :rtype: Tuple[str, str]
- """
- split_at = line.find(' ')
- if split_at == -1:
- raise Exception(f'Log not in "{{timestamp}} {{log}}" format. Got: {line}')
- timestamp = line[:split_at]
- message = line[split_at + 1 :].rstrip()
- return timestamp, message
-
- def _task_status(self, event):
- self.log.info('Event: %s had an event of type %s', event.metadata.name, event.status.phase)
- status = self.process_status(event.metadata.name, event.status.phase)
- return status
-
- def pod_not_started(self, pod: V1Pod):
- """Tests if pod has not started"""
- state = self._task_status(self.read_pod(pod))
- return state == State.QUEUED
-
- def pod_is_running(self, pod: V1Pod):
- """Tests if pod is running"""
- state = self._task_status(self.read_pod(pod))
- return state not in (State.SUCCESS, State.FAILED)
-
- def base_container_is_running(self, pod: V1Pod):
- """Tests if base container is running"""
- event = self.read_pod(pod)
- status = next(iter(filter(lambda s: s.name == 'base', event.status.container_statuses)), None)
- if not status:
- return False
- return status.state.running is not None
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod_logs(
- self,
- pod: V1Pod,
- tail_lines: Optional[int] = None,
- timestamps: bool = False,
- since_seconds: Optional[int] = None,
- ):
- """Reads log from the POD"""
- additional_kwargs = {}
- if since_seconds:
- additional_kwargs['since_seconds'] = since_seconds
-
- if tail_lines:
- additional_kwargs['tail_lines'] = tail_lines
-
- try:
- return self._client.read_namespaced_pod_log(
- name=pod.metadata.name,
- namespace=pod.metadata.namespace,
- container='base',
- follow=True,
- timestamps=timestamps,
- _preload_content=False,
- **additional_kwargs,
- )
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod_events(self, pod):
- """Reads events from the POD"""
- try:
- return self._client.list_namespaced_event(
- namespace=pod.metadata.namespace, field_selector=f"involvedObject.name={pod.metadata.name}"
- )
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod(self, pod: V1Pod):
- """Read POD information"""
- try:
- return self._client.read_namespaced_pod(pod.metadata.name, pod.metadata.namespace)
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- def _extract_xcom(self, pod: V1Pod):
- resp = kubernetes_stream(
- self._client.connect_get_namespaced_pod_exec,
- pod.metadata.name,
- pod.metadata.namespace,
- container=PodDefaults.SIDECAR_CONTAINER_NAME,
- command=['/bin/sh'],
- stdin=True,
- stdout=True,
- stderr=True,
- tty=False,
- _preload_content=False,
- )
- try:
- result = self._exec_pod_command(resp, f'cat {PodDefaults.XCOM_MOUNT_PATH}/return.json')
- self._exec_pod_command(resp, 'kill -s SIGINT 1')
- finally:
- resp.close()
- if result is None:
- raise AirflowException(f'Failed to extract xcom from pod: {pod.metadata.name}')
- return result
-
- def _exec_pod_command(self, resp, command):
- if resp.is_open():
- self.log.info('Running command... %s\n', command)
- resp.write_stdin(command + '\n')
- while resp.is_open():
- resp.update(timeout=1)
- if resp.peek_stdout():
- return resp.read_stdout()
- if resp.peek_stderr():
- self.log.info(resp.read_stderr())
- break
- return None
-
- def process_status(self, job_id, status):
- """Process status information for the JOB"""
- status = status.lower()
- if status == PodStatus.PENDING:
- return State.QUEUED
- elif status == PodStatus.FAILED:
- self.log.error('Event with job id %s Failed', job_id)
- return State.FAILED
- elif status == PodStatus.SUCCEEDED:
- self.log.info('Event with job id %s Succeeded', job_id)
- return State.SUCCESS
- elif status == PodStatus.RUNNING:
- return State.RUNNING
- else:
- self.log.error('Event: Invalid state %s on job %s', status, job_id)
- return State.FAILED
+"""This module is deprecated. Please use `kubernetes.client.models for V1ResourceRequirements and Port."""
+# flake8: noqa
+# pylint: disable=unused-import
+import warnings
+
+from airflow.providers.cncf.kubernetes.utils import pod_launcher # pylint: disable=unused-import
Review comment:
I believe that this will cause ImportErrors if the CNCF provider isn't installed.
Should we catch the exception and provide some clearer messaging to tell users that they need to install the provider?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] kaxil commented on a change in pull request #15165: Separate pod_launcher from core airflow
Posted by GitBox <gi...@apache.org>.
kaxil commented on a change in pull request #15165:
URL: https://github.com/apache/airflow/pull/15165#discussion_r607177156
##########
File path: airflow/kubernetes/pod_launcher.py
##########
@@ -14,283 +15,15 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
-"""Launches PODs"""
-import json
-import math
-import time
-from datetime import datetime as dt
-from typing import Optional, Tuple
-
-import pendulum
-import tenacity
-from kubernetes import client, watch
-from kubernetes.client.models.v1_pod import V1Pod
-from kubernetes.client.rest import ApiException
-from kubernetes.stream import stream as kubernetes_stream
-from requests.exceptions import BaseHTTPError
-
-from airflow.exceptions import AirflowException
-from airflow.kubernetes.kube_client import get_kube_client
-from airflow.kubernetes.pod_generator import PodDefaults
-from airflow.settings import pod_mutation_hook
-from airflow.utils.log.logging_mixin import LoggingMixin
-from airflow.utils.state import State
-
-
-class PodStatus:
- """Status of the PODs"""
-
- PENDING = 'pending'
- RUNNING = 'running'
- FAILED = 'failed'
- SUCCEEDED = 'succeeded'
-
-
-class PodLauncher(LoggingMixin):
- """Launches PODS"""
-
- def __init__(
- self,
- kube_client: client.CoreV1Api = None,
- in_cluster: bool = True,
- cluster_context: Optional[str] = None,
- extract_xcom: bool = False,
- ):
- """
- Creates the launcher.
-
- :param kube_client: kubernetes client
- :param in_cluster: whether we are in cluster
- :param cluster_context: context of the cluster
- :param extract_xcom: whether we should extract xcom
- """
- super().__init__()
- self._client = kube_client or get_kube_client(in_cluster=in_cluster, cluster_context=cluster_context)
- self._watch = watch.Watch()
- self.extract_xcom = extract_xcom
-
- def run_pod_async(self, pod: V1Pod, **kwargs):
- """Runs POD asynchronously"""
- pod_mutation_hook(pod)
-
- sanitized_pod = self._client.api_client.sanitize_for_serialization(pod)
- json_pod = json.dumps(sanitized_pod, indent=2)
-
- self.log.debug('Pod Creation Request: \n%s', json_pod)
- try:
- resp = self._client.create_namespaced_pod(
- body=sanitized_pod, namespace=pod.metadata.namespace, **kwargs
- )
- self.log.debug('Pod Creation Response: %s', resp)
- except Exception as e:
- self.log.exception('Exception when attempting to create Namespaced Pod: %s', json_pod)
- raise e
- return resp
-
- def delete_pod(self, pod: V1Pod):
- """Deletes POD"""
- try:
- self._client.delete_namespaced_pod(
- pod.metadata.name, pod.metadata.namespace, body=client.V1DeleteOptions()
- )
- except ApiException as e:
- # If the pod is already deleted
- if e.status != 404:
- raise
-
- def start_pod(self, pod: V1Pod, startup_timeout: int = 120):
- """
- Launches the pod synchronously and waits for completion.
-
- :param pod:
- :param startup_timeout: Timeout for startup of the pod (if pod is pending for too long, fails task)
- :return:
- """
- resp = self.run_pod_async(pod)
- curr_time = dt.now()
- if resp.status.start_time is None:
- while self.pod_not_started(pod):
- self.log.warning("Pod not yet started: %s", pod.metadata.name)
- delta = dt.now() - curr_time
- if delta.total_seconds() >= startup_timeout:
- raise AirflowException("Pod took too long to start")
- time.sleep(1)
-
- def monitor_pod(self, pod: V1Pod, get_logs: bool) -> Tuple[State, Optional[str]]:
- """
- Monitors a pod and returns the final state
-
- :param pod: pod spec that will be monitored
- :type pod : V1Pod
- :param get_logs: whether to read the logs locally
- :return: Tuple[State, Optional[str]]
- """
- if get_logs:
- read_logs_since_sec = None
- last_log_time = None
- while True:
- logs = self.read_pod_logs(pod, timestamps=True, since_seconds=read_logs_since_sec)
- for line in logs:
- timestamp, message = self.parse_log_line(line.decode('utf-8'))
- last_log_time = pendulum.parse(timestamp)
- self.log.info(message)
- time.sleep(1)
-
- if not self.base_container_is_running(pod):
- break
-
- self.log.warning('Pod %s log read interrupted', pod.metadata.name)
- if last_log_time:
- delta = pendulum.now() - last_log_time
- # Prefer logs duplication rather than loss
- read_logs_since_sec = math.ceil(delta.total_seconds())
- result = None
- if self.extract_xcom:
- while self.base_container_is_running(pod):
- self.log.info('Container %s has state %s', pod.metadata.name, State.RUNNING)
- time.sleep(2)
- result = self._extract_xcom(pod)
- self.log.info(result)
- result = json.loads(result)
- while self.pod_is_running(pod):
- self.log.info('Pod %s has state %s', pod.metadata.name, State.RUNNING)
- time.sleep(2)
- return self._task_status(self.read_pod(pod)), result
-
- def parse_log_line(self, line: str) -> Tuple[str, str]:
- """
- Parse K8s log line and returns the final state
-
- :param line: k8s log line
- :type line: str
- :return: timestamp and log message
- :rtype: Tuple[str, str]
- """
- split_at = line.find(' ')
- if split_at == -1:
- raise Exception(f'Log not in "{{timestamp}} {{log}}" format. Got: {line}')
- timestamp = line[:split_at]
- message = line[split_at + 1 :].rstrip()
- return timestamp, message
-
- def _task_status(self, event):
- self.log.info('Event: %s had an event of type %s', event.metadata.name, event.status.phase)
- status = self.process_status(event.metadata.name, event.status.phase)
- return status
-
- def pod_not_started(self, pod: V1Pod):
- """Tests if pod has not started"""
- state = self._task_status(self.read_pod(pod))
- return state == State.QUEUED
-
- def pod_is_running(self, pod: V1Pod):
- """Tests if pod is running"""
- state = self._task_status(self.read_pod(pod))
- return state not in (State.SUCCESS, State.FAILED)
-
- def base_container_is_running(self, pod: V1Pod):
- """Tests if base container is running"""
- event = self.read_pod(pod)
- status = next(iter(filter(lambda s: s.name == 'base', event.status.container_statuses)), None)
- if not status:
- return False
- return status.state.running is not None
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod_logs(
- self,
- pod: V1Pod,
- tail_lines: Optional[int] = None,
- timestamps: bool = False,
- since_seconds: Optional[int] = None,
- ):
- """Reads log from the POD"""
- additional_kwargs = {}
- if since_seconds:
- additional_kwargs['since_seconds'] = since_seconds
-
- if tail_lines:
- additional_kwargs['tail_lines'] = tail_lines
-
- try:
- return self._client.read_namespaced_pod_log(
- name=pod.metadata.name,
- namespace=pod.metadata.namespace,
- container='base',
- follow=True,
- timestamps=timestamps,
- _preload_content=False,
- **additional_kwargs,
- )
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod_events(self, pod):
- """Reads events from the POD"""
- try:
- return self._client.list_namespaced_event(
- namespace=pod.metadata.namespace, field_selector=f"involvedObject.name={pod.metadata.name}"
- )
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod(self, pod: V1Pod):
- """Read POD information"""
- try:
- return self._client.read_namespaced_pod(pod.metadata.name, pod.metadata.namespace)
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- def _extract_xcom(self, pod: V1Pod):
- resp = kubernetes_stream(
- self._client.connect_get_namespaced_pod_exec,
- pod.metadata.name,
- pod.metadata.namespace,
- container=PodDefaults.SIDECAR_CONTAINER_NAME,
- command=['/bin/sh'],
- stdin=True,
- stdout=True,
- stderr=True,
- tty=False,
- _preload_content=False,
- )
- try:
- result = self._exec_pod_command(resp, f'cat {PodDefaults.XCOM_MOUNT_PATH}/return.json')
- self._exec_pod_command(resp, 'kill -s SIGINT 1')
- finally:
- resp.close()
- if result is None:
- raise AirflowException(f'Failed to extract xcom from pod: {pod.metadata.name}')
- return result
-
- def _exec_pod_command(self, resp, command):
- if resp.is_open():
- self.log.info('Running command... %s\n', command)
- resp.write_stdin(command + '\n')
- while resp.is_open():
- resp.update(timeout=1)
- if resp.peek_stdout():
- return resp.read_stdout()
- if resp.peek_stderr():
- self.log.info(resp.read_stderr())
- break
- return None
-
- def process_status(self, job_id, status):
- """Process status information for the JOB"""
- status = status.lower()
- if status == PodStatus.PENDING:
- return State.QUEUED
- elif status == PodStatus.FAILED:
- self.log.error('Event with job id %s Failed', job_id)
- return State.FAILED
- elif status == PodStatus.SUCCEEDED:
- self.log.info('Event with job id %s Succeeded', job_id)
- return State.SUCCESS
- elif status == PodStatus.RUNNING:
- return State.RUNNING
- else:
- self.log.error('Event: Invalid state %s on job %s', status, job_id)
- return State.FAILED
+"""This module is deprecated. Please use `kubernetes.client.models for V1ResourceRequirements and Port."""
+# flake8: noqa
+# pylint: disable=unused-import
+import warnings
+
+from airflow.kubernetes import pod_launcher_deprecated # pylint: disable=unused-import
Review comment:
This does not take care of the case if someone would have been importing class from `airflow/kubernetes/pod_launcher.py`
example:
```python
>>> from airflow.kubernetes.pod_launcher import PodStatus, PodLauncher
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
ImportError: cannot import name 'PodStatus'
>>> from airflow.kubernetes.pod_launcher import PodLauncher
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
ImportError: cannot import name 'PodLauncher'
```
This will now raise `ImportError`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] github-actions[bot] commented on pull request #15165: Separate Kubernetes pod_launcher from core airflow
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #15165:
URL: https://github.com/apache/airflow/pull/15165#issuecomment-813661035
The PR is likely OK to be merged with just subset of tests for default Python and Database versions without running the full matrix of tests, because it does not modify the core of Airflow. If the committers decide that the full tests matrix is needed, they will add the label 'full tests needed'. Then you should rebase to the latest master or amend the last commit of the PR, and push it with --force-with-lease.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] dimberman commented on a change in pull request #15165: Separate pod_launcher from core airflow
Posted by GitBox <gi...@apache.org>.
dimberman commented on a change in pull request #15165:
URL: https://github.com/apache/airflow/pull/15165#discussion_r607142879
##########
File path: airflow/kubernetes/pod_launcher.py
##########
@@ -14,283 +15,15 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
-"""Launches PODs"""
-import json
-import math
-import time
-from datetime import datetime as dt
-from typing import Optional, Tuple
-
-import pendulum
-import tenacity
-from kubernetes import client, watch
-from kubernetes.client.models.v1_pod import V1Pod
-from kubernetes.client.rest import ApiException
-from kubernetes.stream import stream as kubernetes_stream
-from requests.exceptions import BaseHTTPError
-
-from airflow.exceptions import AirflowException
-from airflow.kubernetes.kube_client import get_kube_client
-from airflow.kubernetes.pod_generator import PodDefaults
-from airflow.settings import pod_mutation_hook
-from airflow.utils.log.logging_mixin import LoggingMixin
-from airflow.utils.state import State
-
-
-class PodStatus:
- """Status of the PODs"""
-
- PENDING = 'pending'
- RUNNING = 'running'
- FAILED = 'failed'
- SUCCEEDED = 'succeeded'
-
-
-class PodLauncher(LoggingMixin):
- """Launches PODS"""
-
- def __init__(
- self,
- kube_client: client.CoreV1Api = None,
- in_cluster: bool = True,
- cluster_context: Optional[str] = None,
- extract_xcom: bool = False,
- ):
- """
- Creates the launcher.
-
- :param kube_client: kubernetes client
- :param in_cluster: whether we are in cluster
- :param cluster_context: context of the cluster
- :param extract_xcom: whether we should extract xcom
- """
- super().__init__()
- self._client = kube_client or get_kube_client(in_cluster=in_cluster, cluster_context=cluster_context)
- self._watch = watch.Watch()
- self.extract_xcom = extract_xcom
-
- def run_pod_async(self, pod: V1Pod, **kwargs):
- """Runs POD asynchronously"""
- pod_mutation_hook(pod)
-
- sanitized_pod = self._client.api_client.sanitize_for_serialization(pod)
- json_pod = json.dumps(sanitized_pod, indent=2)
-
- self.log.debug('Pod Creation Request: \n%s', json_pod)
- try:
- resp = self._client.create_namespaced_pod(
- body=sanitized_pod, namespace=pod.metadata.namespace, **kwargs
- )
- self.log.debug('Pod Creation Response: %s', resp)
- except Exception as e:
- self.log.exception('Exception when attempting to create Namespaced Pod: %s', json_pod)
- raise e
- return resp
-
- def delete_pod(self, pod: V1Pod):
- """Deletes POD"""
- try:
- self._client.delete_namespaced_pod(
- pod.metadata.name, pod.metadata.namespace, body=client.V1DeleteOptions()
- )
- except ApiException as e:
- # If the pod is already deleted
- if e.status != 404:
- raise
-
- def start_pod(self, pod: V1Pod, startup_timeout: int = 120):
- """
- Launches the pod synchronously and waits for completion.
-
- :param pod:
- :param startup_timeout: Timeout for startup of the pod (if pod is pending for too long, fails task)
- :return:
- """
- resp = self.run_pod_async(pod)
- curr_time = dt.now()
- if resp.status.start_time is None:
- while self.pod_not_started(pod):
- self.log.warning("Pod not yet started: %s", pod.metadata.name)
- delta = dt.now() - curr_time
- if delta.total_seconds() >= startup_timeout:
- raise AirflowException("Pod took too long to start")
- time.sleep(1)
-
- def monitor_pod(self, pod: V1Pod, get_logs: bool) -> Tuple[State, Optional[str]]:
- """
- Monitors a pod and returns the final state
-
- :param pod: pod spec that will be monitored
- :type pod : V1Pod
- :param get_logs: whether to read the logs locally
- :return: Tuple[State, Optional[str]]
- """
- if get_logs:
- read_logs_since_sec = None
- last_log_time = None
- while True:
- logs = self.read_pod_logs(pod, timestamps=True, since_seconds=read_logs_since_sec)
- for line in logs:
- timestamp, message = self.parse_log_line(line.decode('utf-8'))
- last_log_time = pendulum.parse(timestamp)
- self.log.info(message)
- time.sleep(1)
-
- if not self.base_container_is_running(pod):
- break
-
- self.log.warning('Pod %s log read interrupted', pod.metadata.name)
- if last_log_time:
- delta = pendulum.now() - last_log_time
- # Prefer logs duplication rather than loss
- read_logs_since_sec = math.ceil(delta.total_seconds())
- result = None
- if self.extract_xcom:
- while self.base_container_is_running(pod):
- self.log.info('Container %s has state %s', pod.metadata.name, State.RUNNING)
- time.sleep(2)
- result = self._extract_xcom(pod)
- self.log.info(result)
- result = json.loads(result)
- while self.pod_is_running(pod):
- self.log.info('Pod %s has state %s', pod.metadata.name, State.RUNNING)
- time.sleep(2)
- return self._task_status(self.read_pod(pod)), result
-
- def parse_log_line(self, line: str) -> Tuple[str, str]:
- """
- Parse K8s log line and returns the final state
-
- :param line: k8s log line
- :type line: str
- :return: timestamp and log message
- :rtype: Tuple[str, str]
- """
- split_at = line.find(' ')
- if split_at == -1:
- raise Exception(f'Log not in "{{timestamp}} {{log}}" format. Got: {line}')
- timestamp = line[:split_at]
- message = line[split_at + 1 :].rstrip()
- return timestamp, message
-
- def _task_status(self, event):
- self.log.info('Event: %s had an event of type %s', event.metadata.name, event.status.phase)
- status = self.process_status(event.metadata.name, event.status.phase)
- return status
-
- def pod_not_started(self, pod: V1Pod):
- """Tests if pod has not started"""
- state = self._task_status(self.read_pod(pod))
- return state == State.QUEUED
-
- def pod_is_running(self, pod: V1Pod):
- """Tests if pod is running"""
- state = self._task_status(self.read_pod(pod))
- return state not in (State.SUCCESS, State.FAILED)
-
- def base_container_is_running(self, pod: V1Pod):
- """Tests if base container is running"""
- event = self.read_pod(pod)
- status = next(iter(filter(lambda s: s.name == 'base', event.status.container_statuses)), None)
- if not status:
- return False
- return status.state.running is not None
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod_logs(
- self,
- pod: V1Pod,
- tail_lines: Optional[int] = None,
- timestamps: bool = False,
- since_seconds: Optional[int] = None,
- ):
- """Reads log from the POD"""
- additional_kwargs = {}
- if since_seconds:
- additional_kwargs['since_seconds'] = since_seconds
-
- if tail_lines:
- additional_kwargs['tail_lines'] = tail_lines
-
- try:
- return self._client.read_namespaced_pod_log(
- name=pod.metadata.name,
- namespace=pod.metadata.namespace,
- container='base',
- follow=True,
- timestamps=timestamps,
- _preload_content=False,
- **additional_kwargs,
- )
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod_events(self, pod):
- """Reads events from the POD"""
- try:
- return self._client.list_namespaced_event(
- namespace=pod.metadata.namespace, field_selector=f"involvedObject.name={pod.metadata.name}"
- )
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod(self, pod: V1Pod):
- """Read POD information"""
- try:
- return self._client.read_namespaced_pod(pod.metadata.name, pod.metadata.namespace)
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- def _extract_xcom(self, pod: V1Pod):
- resp = kubernetes_stream(
- self._client.connect_get_namespaced_pod_exec,
- pod.metadata.name,
- pod.metadata.namespace,
- container=PodDefaults.SIDECAR_CONTAINER_NAME,
- command=['/bin/sh'],
- stdin=True,
- stdout=True,
- stderr=True,
- tty=False,
- _preload_content=False,
- )
- try:
- result = self._exec_pod_command(resp, f'cat {PodDefaults.XCOM_MOUNT_PATH}/return.json')
- self._exec_pod_command(resp, 'kill -s SIGINT 1')
- finally:
- resp.close()
- if result is None:
- raise AirflowException(f'Failed to extract xcom from pod: {pod.metadata.name}')
- return result
-
- def _exec_pod_command(self, resp, command):
- if resp.is_open():
- self.log.info('Running command... %s\n', command)
- resp.write_stdin(command + '\n')
- while resp.is_open():
- resp.update(timeout=1)
- if resp.peek_stdout():
- return resp.read_stdout()
- if resp.peek_stderr():
- self.log.info(resp.read_stderr())
- break
- return None
-
- def process_status(self, job_id, status):
- """Process status information for the JOB"""
- status = status.lower()
- if status == PodStatus.PENDING:
- return State.QUEUED
- elif status == PodStatus.FAILED:
- self.log.error('Event with job id %s Failed', job_id)
- return State.FAILED
- elif status == PodStatus.SUCCEEDED:
- self.log.info('Event with job id %s Succeeded', job_id)
- return State.SUCCESS
- elif status == PodStatus.RUNNING:
- return State.RUNNING
- else:
- self.log.error('Event: Invalid state %s on job %s', status, job_id)
- return State.FAILED
+"""This module is deprecated. Please use `kubernetes.client.models for V1ResourceRequirements and Port."""
Review comment:
@potiuk we can have a pod_launcher_deprecated.py file with the old code. We did something similar when upgrading the kubernetesexecutor for 2.0
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] kaxil commented on a change in pull request #15165: Separate pod_launcher from core airflow
Posted by GitBox <gi...@apache.org>.
kaxil commented on a change in pull request #15165:
URL: https://github.com/apache/airflow/pull/15165#discussion_r607177156
##########
File path: airflow/kubernetes/pod_launcher.py
##########
@@ -14,283 +15,15 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
-"""Launches PODs"""
-import json
-import math
-import time
-from datetime import datetime as dt
-from typing import Optional, Tuple
-
-import pendulum
-import tenacity
-from kubernetes import client, watch
-from kubernetes.client.models.v1_pod import V1Pod
-from kubernetes.client.rest import ApiException
-from kubernetes.stream import stream as kubernetes_stream
-from requests.exceptions import BaseHTTPError
-
-from airflow.exceptions import AirflowException
-from airflow.kubernetes.kube_client import get_kube_client
-from airflow.kubernetes.pod_generator import PodDefaults
-from airflow.settings import pod_mutation_hook
-from airflow.utils.log.logging_mixin import LoggingMixin
-from airflow.utils.state import State
-
-
-class PodStatus:
- """Status of the PODs"""
-
- PENDING = 'pending'
- RUNNING = 'running'
- FAILED = 'failed'
- SUCCEEDED = 'succeeded'
-
-
-class PodLauncher(LoggingMixin):
- """Launches PODS"""
-
- def __init__(
- self,
- kube_client: client.CoreV1Api = None,
- in_cluster: bool = True,
- cluster_context: Optional[str] = None,
- extract_xcom: bool = False,
- ):
- """
- Creates the launcher.
-
- :param kube_client: kubernetes client
- :param in_cluster: whether we are in cluster
- :param cluster_context: context of the cluster
- :param extract_xcom: whether we should extract xcom
- """
- super().__init__()
- self._client = kube_client or get_kube_client(in_cluster=in_cluster, cluster_context=cluster_context)
- self._watch = watch.Watch()
- self.extract_xcom = extract_xcom
-
- def run_pod_async(self, pod: V1Pod, **kwargs):
- """Runs POD asynchronously"""
- pod_mutation_hook(pod)
-
- sanitized_pod = self._client.api_client.sanitize_for_serialization(pod)
- json_pod = json.dumps(sanitized_pod, indent=2)
-
- self.log.debug('Pod Creation Request: \n%s', json_pod)
- try:
- resp = self._client.create_namespaced_pod(
- body=sanitized_pod, namespace=pod.metadata.namespace, **kwargs
- )
- self.log.debug('Pod Creation Response: %s', resp)
- except Exception as e:
- self.log.exception('Exception when attempting to create Namespaced Pod: %s', json_pod)
- raise e
- return resp
-
- def delete_pod(self, pod: V1Pod):
- """Deletes POD"""
- try:
- self._client.delete_namespaced_pod(
- pod.metadata.name, pod.metadata.namespace, body=client.V1DeleteOptions()
- )
- except ApiException as e:
- # If the pod is already deleted
- if e.status != 404:
- raise
-
- def start_pod(self, pod: V1Pod, startup_timeout: int = 120):
- """
- Launches the pod synchronously and waits for completion.
-
- :param pod:
- :param startup_timeout: Timeout for startup of the pod (if pod is pending for too long, fails task)
- :return:
- """
- resp = self.run_pod_async(pod)
- curr_time = dt.now()
- if resp.status.start_time is None:
- while self.pod_not_started(pod):
- self.log.warning("Pod not yet started: %s", pod.metadata.name)
- delta = dt.now() - curr_time
- if delta.total_seconds() >= startup_timeout:
- raise AirflowException("Pod took too long to start")
- time.sleep(1)
-
- def monitor_pod(self, pod: V1Pod, get_logs: bool) -> Tuple[State, Optional[str]]:
- """
- Monitors a pod and returns the final state
-
- :param pod: pod spec that will be monitored
- :type pod : V1Pod
- :param get_logs: whether to read the logs locally
- :return: Tuple[State, Optional[str]]
- """
- if get_logs:
- read_logs_since_sec = None
- last_log_time = None
- while True:
- logs = self.read_pod_logs(pod, timestamps=True, since_seconds=read_logs_since_sec)
- for line in logs:
- timestamp, message = self.parse_log_line(line.decode('utf-8'))
- last_log_time = pendulum.parse(timestamp)
- self.log.info(message)
- time.sleep(1)
-
- if not self.base_container_is_running(pod):
- break
-
- self.log.warning('Pod %s log read interrupted', pod.metadata.name)
- if last_log_time:
- delta = pendulum.now() - last_log_time
- # Prefer logs duplication rather than loss
- read_logs_since_sec = math.ceil(delta.total_seconds())
- result = None
- if self.extract_xcom:
- while self.base_container_is_running(pod):
- self.log.info('Container %s has state %s', pod.metadata.name, State.RUNNING)
- time.sleep(2)
- result = self._extract_xcom(pod)
- self.log.info(result)
- result = json.loads(result)
- while self.pod_is_running(pod):
- self.log.info('Pod %s has state %s', pod.metadata.name, State.RUNNING)
- time.sleep(2)
- return self._task_status(self.read_pod(pod)), result
-
- def parse_log_line(self, line: str) -> Tuple[str, str]:
- """
- Parse K8s log line and returns the final state
-
- :param line: k8s log line
- :type line: str
- :return: timestamp and log message
- :rtype: Tuple[str, str]
- """
- split_at = line.find(' ')
- if split_at == -1:
- raise Exception(f'Log not in "{{timestamp}} {{log}}" format. Got: {line}')
- timestamp = line[:split_at]
- message = line[split_at + 1 :].rstrip()
- return timestamp, message
-
- def _task_status(self, event):
- self.log.info('Event: %s had an event of type %s', event.metadata.name, event.status.phase)
- status = self.process_status(event.metadata.name, event.status.phase)
- return status
-
- def pod_not_started(self, pod: V1Pod):
- """Tests if pod has not started"""
- state = self._task_status(self.read_pod(pod))
- return state == State.QUEUED
-
- def pod_is_running(self, pod: V1Pod):
- """Tests if pod is running"""
- state = self._task_status(self.read_pod(pod))
- return state not in (State.SUCCESS, State.FAILED)
-
- def base_container_is_running(self, pod: V1Pod):
- """Tests if base container is running"""
- event = self.read_pod(pod)
- status = next(iter(filter(lambda s: s.name == 'base', event.status.container_statuses)), None)
- if not status:
- return False
- return status.state.running is not None
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod_logs(
- self,
- pod: V1Pod,
- tail_lines: Optional[int] = None,
- timestamps: bool = False,
- since_seconds: Optional[int] = None,
- ):
- """Reads log from the POD"""
- additional_kwargs = {}
- if since_seconds:
- additional_kwargs['since_seconds'] = since_seconds
-
- if tail_lines:
- additional_kwargs['tail_lines'] = tail_lines
-
- try:
- return self._client.read_namespaced_pod_log(
- name=pod.metadata.name,
- namespace=pod.metadata.namespace,
- container='base',
- follow=True,
- timestamps=timestamps,
- _preload_content=False,
- **additional_kwargs,
- )
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod_events(self, pod):
- """Reads events from the POD"""
- try:
- return self._client.list_namespaced_event(
- namespace=pod.metadata.namespace, field_selector=f"involvedObject.name={pod.metadata.name}"
- )
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod(self, pod: V1Pod):
- """Read POD information"""
- try:
- return self._client.read_namespaced_pod(pod.metadata.name, pod.metadata.namespace)
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- def _extract_xcom(self, pod: V1Pod):
- resp = kubernetes_stream(
- self._client.connect_get_namespaced_pod_exec,
- pod.metadata.name,
- pod.metadata.namespace,
- container=PodDefaults.SIDECAR_CONTAINER_NAME,
- command=['/bin/sh'],
- stdin=True,
- stdout=True,
- stderr=True,
- tty=False,
- _preload_content=False,
- )
- try:
- result = self._exec_pod_command(resp, f'cat {PodDefaults.XCOM_MOUNT_PATH}/return.json')
- self._exec_pod_command(resp, 'kill -s SIGINT 1')
- finally:
- resp.close()
- if result is None:
- raise AirflowException(f'Failed to extract xcom from pod: {pod.metadata.name}')
- return result
-
- def _exec_pod_command(self, resp, command):
- if resp.is_open():
- self.log.info('Running command... %s\n', command)
- resp.write_stdin(command + '\n')
- while resp.is_open():
- resp.update(timeout=1)
- if resp.peek_stdout():
- return resp.read_stdout()
- if resp.peek_stderr():
- self.log.info(resp.read_stderr())
- break
- return None
-
- def process_status(self, job_id, status):
- """Process status information for the JOB"""
- status = status.lower()
- if status == PodStatus.PENDING:
- return State.QUEUED
- elif status == PodStatus.FAILED:
- self.log.error('Event with job id %s Failed', job_id)
- return State.FAILED
- elif status == PodStatus.SUCCEEDED:
- self.log.info('Event with job id %s Succeeded', job_id)
- return State.SUCCESS
- elif status == PodStatus.RUNNING:
- return State.RUNNING
- else:
- self.log.error('Event: Invalid state %s on job %s', status, job_id)
- return State.FAILED
+"""This module is deprecated. Please use `kubernetes.client.models for V1ResourceRequirements and Port."""
+# flake8: noqa
+# pylint: disable=unused-import
+import warnings
+
+from airflow.kubernetes import pod_launcher_deprecated # pylint: disable=unused-import
Review comment:
This does not take care of the case if someone would have been importing class from `airflow/kubernetes/pod_launcher.py`
example:
```python
>>> from airflow.kubernetes.pod_launcher import PodStatus, PodLauncher
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
ImportError: cannot import name 'PodStatus'
>>> from airflow.kubernetes.pod_launcher import PodLauncher
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
ImportError: cannot import name 'PodLauncher'
```
This will now raise AttributeNotFoundError
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on a change in pull request #15165: Separate pod_launcher from core airflow
Posted by GitBox <gi...@apache.org>.
potiuk commented on a change in pull request #15165:
URL: https://github.com/apache/airflow/pull/15165#discussion_r607212246
##########
File path: airflow/kubernetes/pod_launcher.py
##########
@@ -14,283 +15,15 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
-"""Launches PODs"""
-import json
-import math
-import time
-from datetime import datetime as dt
-from typing import Optional, Tuple
-
-import pendulum
-import tenacity
-from kubernetes import client, watch
-from kubernetes.client.models.v1_pod import V1Pod
-from kubernetes.client.rest import ApiException
-from kubernetes.stream import stream as kubernetes_stream
-from requests.exceptions import BaseHTTPError
-
-from airflow.exceptions import AirflowException
-from airflow.kubernetes.kube_client import get_kube_client
-from airflow.kubernetes.pod_generator import PodDefaults
-from airflow.settings import pod_mutation_hook
-from airflow.utils.log.logging_mixin import LoggingMixin
-from airflow.utils.state import State
-
-
-class PodStatus:
- """Status of the PODs"""
-
- PENDING = 'pending'
- RUNNING = 'running'
- FAILED = 'failed'
- SUCCEEDED = 'succeeded'
-
-
-class PodLauncher(LoggingMixin):
- """Launches PODS"""
-
- def __init__(
- self,
- kube_client: client.CoreV1Api = None,
- in_cluster: bool = True,
- cluster_context: Optional[str] = None,
- extract_xcom: bool = False,
- ):
- """
- Creates the launcher.
-
- :param kube_client: kubernetes client
- :param in_cluster: whether we are in cluster
- :param cluster_context: context of the cluster
- :param extract_xcom: whether we should extract xcom
- """
- super().__init__()
- self._client = kube_client or get_kube_client(in_cluster=in_cluster, cluster_context=cluster_context)
- self._watch = watch.Watch()
- self.extract_xcom = extract_xcom
-
- def run_pod_async(self, pod: V1Pod, **kwargs):
- """Runs POD asynchronously"""
- pod_mutation_hook(pod)
-
- sanitized_pod = self._client.api_client.sanitize_for_serialization(pod)
- json_pod = json.dumps(sanitized_pod, indent=2)
-
- self.log.debug('Pod Creation Request: \n%s', json_pod)
- try:
- resp = self._client.create_namespaced_pod(
- body=sanitized_pod, namespace=pod.metadata.namespace, **kwargs
- )
- self.log.debug('Pod Creation Response: %s', resp)
- except Exception as e:
- self.log.exception('Exception when attempting to create Namespaced Pod: %s', json_pod)
- raise e
- return resp
-
- def delete_pod(self, pod: V1Pod):
- """Deletes POD"""
- try:
- self._client.delete_namespaced_pod(
- pod.metadata.name, pod.metadata.namespace, body=client.V1DeleteOptions()
- )
- except ApiException as e:
- # If the pod is already deleted
- if e.status != 404:
- raise
-
- def start_pod(self, pod: V1Pod, startup_timeout: int = 120):
- """
- Launches the pod synchronously and waits for completion.
-
- :param pod:
- :param startup_timeout: Timeout for startup of the pod (if pod is pending for too long, fails task)
- :return:
- """
- resp = self.run_pod_async(pod)
- curr_time = dt.now()
- if resp.status.start_time is None:
- while self.pod_not_started(pod):
- self.log.warning("Pod not yet started: %s", pod.metadata.name)
- delta = dt.now() - curr_time
- if delta.total_seconds() >= startup_timeout:
- raise AirflowException("Pod took too long to start")
- time.sleep(1)
-
- def monitor_pod(self, pod: V1Pod, get_logs: bool) -> Tuple[State, Optional[str]]:
- """
- Monitors a pod and returns the final state
-
- :param pod: pod spec that will be monitored
- :type pod : V1Pod
- :param get_logs: whether to read the logs locally
- :return: Tuple[State, Optional[str]]
- """
- if get_logs:
- read_logs_since_sec = None
- last_log_time = None
- while True:
- logs = self.read_pod_logs(pod, timestamps=True, since_seconds=read_logs_since_sec)
- for line in logs:
- timestamp, message = self.parse_log_line(line.decode('utf-8'))
- last_log_time = pendulum.parse(timestamp)
- self.log.info(message)
- time.sleep(1)
-
- if not self.base_container_is_running(pod):
- break
-
- self.log.warning('Pod %s log read interrupted', pod.metadata.name)
- if last_log_time:
- delta = pendulum.now() - last_log_time
- # Prefer logs duplication rather than loss
- read_logs_since_sec = math.ceil(delta.total_seconds())
- result = None
- if self.extract_xcom:
- while self.base_container_is_running(pod):
- self.log.info('Container %s has state %s', pod.metadata.name, State.RUNNING)
- time.sleep(2)
- result = self._extract_xcom(pod)
- self.log.info(result)
- result = json.loads(result)
- while self.pod_is_running(pod):
- self.log.info('Pod %s has state %s', pod.metadata.name, State.RUNNING)
- time.sleep(2)
- return self._task_status(self.read_pod(pod)), result
-
- def parse_log_line(self, line: str) -> Tuple[str, str]:
- """
- Parse K8s log line and returns the final state
-
- :param line: k8s log line
- :type line: str
- :return: timestamp and log message
- :rtype: Tuple[str, str]
- """
- split_at = line.find(' ')
- if split_at == -1:
- raise Exception(f'Log not in "{{timestamp}} {{log}}" format. Got: {line}')
- timestamp = line[:split_at]
- message = line[split_at + 1 :].rstrip()
- return timestamp, message
-
- def _task_status(self, event):
- self.log.info('Event: %s had an event of type %s', event.metadata.name, event.status.phase)
- status = self.process_status(event.metadata.name, event.status.phase)
- return status
-
- def pod_not_started(self, pod: V1Pod):
- """Tests if pod has not started"""
- state = self._task_status(self.read_pod(pod))
- return state == State.QUEUED
-
- def pod_is_running(self, pod: V1Pod):
- """Tests if pod is running"""
- state = self._task_status(self.read_pod(pod))
- return state not in (State.SUCCESS, State.FAILED)
-
- def base_container_is_running(self, pod: V1Pod):
- """Tests if base container is running"""
- event = self.read_pod(pod)
- status = next(iter(filter(lambda s: s.name == 'base', event.status.container_statuses)), None)
- if not status:
- return False
- return status.state.running is not None
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod_logs(
- self,
- pod: V1Pod,
- tail_lines: Optional[int] = None,
- timestamps: bool = False,
- since_seconds: Optional[int] = None,
- ):
- """Reads log from the POD"""
- additional_kwargs = {}
- if since_seconds:
- additional_kwargs['since_seconds'] = since_seconds
-
- if tail_lines:
- additional_kwargs['tail_lines'] = tail_lines
-
- try:
- return self._client.read_namespaced_pod_log(
- name=pod.metadata.name,
- namespace=pod.metadata.namespace,
- container='base',
- follow=True,
- timestamps=timestamps,
- _preload_content=False,
- **additional_kwargs,
- )
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod_events(self, pod):
- """Reads events from the POD"""
- try:
- return self._client.list_namespaced_event(
- namespace=pod.metadata.namespace, field_selector=f"involvedObject.name={pod.metadata.name}"
- )
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod(self, pod: V1Pod):
- """Read POD information"""
- try:
- return self._client.read_namespaced_pod(pod.metadata.name, pod.metadata.namespace)
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- def _extract_xcom(self, pod: V1Pod):
- resp = kubernetes_stream(
- self._client.connect_get_namespaced_pod_exec,
- pod.metadata.name,
- pod.metadata.namespace,
- container=PodDefaults.SIDECAR_CONTAINER_NAME,
- command=['/bin/sh'],
- stdin=True,
- stdout=True,
- stderr=True,
- tty=False,
- _preload_content=False,
- )
- try:
- result = self._exec_pod_command(resp, f'cat {PodDefaults.XCOM_MOUNT_PATH}/return.json')
- self._exec_pod_command(resp, 'kill -s SIGINT 1')
- finally:
- resp.close()
- if result is None:
- raise AirflowException(f'Failed to extract xcom from pod: {pod.metadata.name}')
- return result
-
- def _exec_pod_command(self, resp, command):
- if resp.is_open():
- self.log.info('Running command... %s\n', command)
- resp.write_stdin(command + '\n')
- while resp.is_open():
- resp.update(timeout=1)
- if resp.peek_stdout():
- return resp.read_stdout()
- if resp.peek_stderr():
- self.log.info(resp.read_stderr())
- break
- return None
-
- def process_status(self, job_id, status):
- """Process status information for the JOB"""
- status = status.lower()
- if status == PodStatus.PENDING:
- return State.QUEUED
- elif status == PodStatus.FAILED:
- self.log.error('Event with job id %s Failed', job_id)
- return State.FAILED
- elif status == PodStatus.SUCCEEDED:
- self.log.info('Event with job id %s Succeeded', job_id)
- return State.SUCCESS
- elif status == PodStatus.RUNNING:
- return State.RUNNING
- else:
- self.log.error('Event: Invalid state %s on job %s', status, job_id)
- return State.FAILED
+"""This module is deprecated. Please use `kubernetes.client.models for V1ResourceRequirements and Port."""
+# flake8: noqa
+# pylint: disable=unused-import
+import warnings
+
+from airflow.kubernetes import pod_launcher_deprecated # pylint: disable=unused-import
Review comment:
Looks good now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] dimberman commented on a change in pull request #15165: Separate pod_launcher from core airflow
Posted by GitBox <gi...@apache.org>.
dimberman commented on a change in pull request #15165:
URL: https://github.com/apache/airflow/pull/15165#discussion_r607147803
##########
File path: airflow/kubernetes/pod_launcher.py
##########
@@ -14,283 +15,15 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
-"""Launches PODs"""
-import json
-import math
-import time
-from datetime import datetime as dt
-from typing import Optional, Tuple
-
-import pendulum
-import tenacity
-from kubernetes import client, watch
-from kubernetes.client.models.v1_pod import V1Pod
-from kubernetes.client.rest import ApiException
-from kubernetes.stream import stream as kubernetes_stream
-from requests.exceptions import BaseHTTPError
-
-from airflow.exceptions import AirflowException
-from airflow.kubernetes.kube_client import get_kube_client
-from airflow.kubernetes.pod_generator import PodDefaults
-from airflow.settings import pod_mutation_hook
-from airflow.utils.log.logging_mixin import LoggingMixin
-from airflow.utils.state import State
-
-
-class PodStatus:
- """Status of the PODs"""
-
- PENDING = 'pending'
- RUNNING = 'running'
- FAILED = 'failed'
- SUCCEEDED = 'succeeded'
-
-
-class PodLauncher(LoggingMixin):
- """Launches PODS"""
-
- def __init__(
- self,
- kube_client: client.CoreV1Api = None,
- in_cluster: bool = True,
- cluster_context: Optional[str] = None,
- extract_xcom: bool = False,
- ):
- """
- Creates the launcher.
-
- :param kube_client: kubernetes client
- :param in_cluster: whether we are in cluster
- :param cluster_context: context of the cluster
- :param extract_xcom: whether we should extract xcom
- """
- super().__init__()
- self._client = kube_client or get_kube_client(in_cluster=in_cluster, cluster_context=cluster_context)
- self._watch = watch.Watch()
- self.extract_xcom = extract_xcom
-
- def run_pod_async(self, pod: V1Pod, **kwargs):
- """Runs POD asynchronously"""
- pod_mutation_hook(pod)
-
- sanitized_pod = self._client.api_client.sanitize_for_serialization(pod)
- json_pod = json.dumps(sanitized_pod, indent=2)
-
- self.log.debug('Pod Creation Request: \n%s', json_pod)
- try:
- resp = self._client.create_namespaced_pod(
- body=sanitized_pod, namespace=pod.metadata.namespace, **kwargs
- )
- self.log.debug('Pod Creation Response: %s', resp)
- except Exception as e:
- self.log.exception('Exception when attempting to create Namespaced Pod: %s', json_pod)
- raise e
- return resp
-
- def delete_pod(self, pod: V1Pod):
- """Deletes POD"""
- try:
- self._client.delete_namespaced_pod(
- pod.metadata.name, pod.metadata.namespace, body=client.V1DeleteOptions()
- )
- except ApiException as e:
- # If the pod is already deleted
- if e.status != 404:
- raise
-
- def start_pod(self, pod: V1Pod, startup_timeout: int = 120):
- """
- Launches the pod synchronously and waits for completion.
-
- :param pod:
- :param startup_timeout: Timeout for startup of the pod (if pod is pending for too long, fails task)
- :return:
- """
- resp = self.run_pod_async(pod)
- curr_time = dt.now()
- if resp.status.start_time is None:
- while self.pod_not_started(pod):
- self.log.warning("Pod not yet started: %s", pod.metadata.name)
- delta = dt.now() - curr_time
- if delta.total_seconds() >= startup_timeout:
- raise AirflowException("Pod took too long to start")
- time.sleep(1)
-
- def monitor_pod(self, pod: V1Pod, get_logs: bool) -> Tuple[State, Optional[str]]:
- """
- Monitors a pod and returns the final state
-
- :param pod: pod spec that will be monitored
- :type pod : V1Pod
- :param get_logs: whether to read the logs locally
- :return: Tuple[State, Optional[str]]
- """
- if get_logs:
- read_logs_since_sec = None
- last_log_time = None
- while True:
- logs = self.read_pod_logs(pod, timestamps=True, since_seconds=read_logs_since_sec)
- for line in logs:
- timestamp, message = self.parse_log_line(line.decode('utf-8'))
- last_log_time = pendulum.parse(timestamp)
- self.log.info(message)
- time.sleep(1)
-
- if not self.base_container_is_running(pod):
- break
-
- self.log.warning('Pod %s log read interrupted', pod.metadata.name)
- if last_log_time:
- delta = pendulum.now() - last_log_time
- # Prefer logs duplication rather than loss
- read_logs_since_sec = math.ceil(delta.total_seconds())
- result = None
- if self.extract_xcom:
- while self.base_container_is_running(pod):
- self.log.info('Container %s has state %s', pod.metadata.name, State.RUNNING)
- time.sleep(2)
- result = self._extract_xcom(pod)
- self.log.info(result)
- result = json.loads(result)
- while self.pod_is_running(pod):
- self.log.info('Pod %s has state %s', pod.metadata.name, State.RUNNING)
- time.sleep(2)
- return self._task_status(self.read_pod(pod)), result
-
- def parse_log_line(self, line: str) -> Tuple[str, str]:
- """
- Parse K8s log line and returns the final state
-
- :param line: k8s log line
- :type line: str
- :return: timestamp and log message
- :rtype: Tuple[str, str]
- """
- split_at = line.find(' ')
- if split_at == -1:
- raise Exception(f'Log not in "{{timestamp}} {{log}}" format. Got: {line}')
- timestamp = line[:split_at]
- message = line[split_at + 1 :].rstrip()
- return timestamp, message
-
- def _task_status(self, event):
- self.log.info('Event: %s had an event of type %s', event.metadata.name, event.status.phase)
- status = self.process_status(event.metadata.name, event.status.phase)
- return status
-
- def pod_not_started(self, pod: V1Pod):
- """Tests if pod has not started"""
- state = self._task_status(self.read_pod(pod))
- return state == State.QUEUED
-
- def pod_is_running(self, pod: V1Pod):
- """Tests if pod is running"""
- state = self._task_status(self.read_pod(pod))
- return state not in (State.SUCCESS, State.FAILED)
-
- def base_container_is_running(self, pod: V1Pod):
- """Tests if base container is running"""
- event = self.read_pod(pod)
- status = next(iter(filter(lambda s: s.name == 'base', event.status.container_statuses)), None)
- if not status:
- return False
- return status.state.running is not None
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod_logs(
- self,
- pod: V1Pod,
- tail_lines: Optional[int] = None,
- timestamps: bool = False,
- since_seconds: Optional[int] = None,
- ):
- """Reads log from the POD"""
- additional_kwargs = {}
- if since_seconds:
- additional_kwargs['since_seconds'] = since_seconds
-
- if tail_lines:
- additional_kwargs['tail_lines'] = tail_lines
-
- try:
- return self._client.read_namespaced_pod_log(
- name=pod.metadata.name,
- namespace=pod.metadata.namespace,
- container='base',
- follow=True,
- timestamps=timestamps,
- _preload_content=False,
- **additional_kwargs,
- )
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod_events(self, pod):
- """Reads events from the POD"""
- try:
- return self._client.list_namespaced_event(
- namespace=pod.metadata.namespace, field_selector=f"involvedObject.name={pod.metadata.name}"
- )
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod(self, pod: V1Pod):
- """Read POD information"""
- try:
- return self._client.read_namespaced_pod(pod.metadata.name, pod.metadata.namespace)
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- def _extract_xcom(self, pod: V1Pod):
- resp = kubernetes_stream(
- self._client.connect_get_namespaced_pod_exec,
- pod.metadata.name,
- pod.metadata.namespace,
- container=PodDefaults.SIDECAR_CONTAINER_NAME,
- command=['/bin/sh'],
- stdin=True,
- stdout=True,
- stderr=True,
- tty=False,
- _preload_content=False,
- )
- try:
- result = self._exec_pod_command(resp, f'cat {PodDefaults.XCOM_MOUNT_PATH}/return.json')
- self._exec_pod_command(resp, 'kill -s SIGINT 1')
- finally:
- resp.close()
- if result is None:
- raise AirflowException(f'Failed to extract xcom from pod: {pod.metadata.name}')
- return result
-
- def _exec_pod_command(self, resp, command):
- if resp.is_open():
- self.log.info('Running command... %s\n', command)
- resp.write_stdin(command + '\n')
- while resp.is_open():
- resp.update(timeout=1)
- if resp.peek_stdout():
- return resp.read_stdout()
- if resp.peek_stderr():
- self.log.info(resp.read_stderr())
- break
- return None
-
- def process_status(self, job_id, status):
- """Process status information for the JOB"""
- status = status.lower()
- if status == PodStatus.PENDING:
- return State.QUEUED
- elif status == PodStatus.FAILED:
- self.log.error('Event with job id %s Failed', job_id)
- return State.FAILED
- elif status == PodStatus.SUCCEEDED:
- self.log.info('Event with job id %s Succeeded', job_id)
- return State.SUCCESS
- elif status == PodStatus.RUNNING:
- return State.RUNNING
- else:
- self.log.error('Event: Invalid state %s on job %s', status, job_id)
- return State.FAILED
+"""This module is deprecated. Please use `kubernetes.client.models for V1ResourceRequirements and Port."""
+# flake8: noqa
+# pylint: disable=unused-import
+import warnings
+
+from airflow.providers.cncf.kubernetes.utils import pod_launcher # pylint: disable=unused-import
Review comment:
@SamWheating with the pod_launcher_deprecated class users won't need to download anything new. That said we will not add any new fixes to this class.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on a change in pull request #15165: Separate pod_launcher from core airflow
Posted by GitBox <gi...@apache.org>.
potiuk commented on a change in pull request #15165:
URL: https://github.com/apache/airflow/pull/15165#discussion_r607226417
##########
File path: airflow/kubernetes/pod_launcher_deprecated.py
##########
@@ -0,0 +1,300 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Launches PODs"""
+import json
+import math
+import time
+from datetime import datetime as dt
+from typing import Optional, Tuple
+
+import pendulum
+import tenacity
+from kubernetes import client, watch
+from kubernetes.client.models.v1_pod import V1Pod
+from kubernetes.client.rest import ApiException
+from kubernetes.stream import stream as kubernetes_stream
+from requests.exceptions import BaseHTTPError
+
+from airflow.exceptions import AirflowException
+from airflow.kubernetes.kube_client import get_kube_client
+from airflow.kubernetes.pod_generator import PodDefaults
+from airflow.settings import pod_mutation_hook
+from airflow.utils.log.logging_mixin import LoggingMixin
+from airflow.utils.state import State
+
+
Review comment:
Yep.. Waiting for push to approve it ;)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on a change in pull request #15165: Separate pod_launcher from core airflow
Posted by GitBox <gi...@apache.org>.
potiuk commented on a change in pull request #15165:
URL: https://github.com/apache/airflow/pull/15165#discussion_r606796281
##########
File path: airflow/kubernetes/pod_launcher.py
##########
@@ -14,283 +15,15 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
-"""Launches PODs"""
-import json
-import math
-import time
-from datetime import datetime as dt
-from typing import Optional, Tuple
-
-import pendulum
-import tenacity
-from kubernetes import client, watch
-from kubernetes.client.models.v1_pod import V1Pod
-from kubernetes.client.rest import ApiException
-from kubernetes.stream import stream as kubernetes_stream
-from requests.exceptions import BaseHTTPError
-
-from airflow.exceptions import AirflowException
-from airflow.kubernetes.kube_client import get_kube_client
-from airflow.kubernetes.pod_generator import PodDefaults
-from airflow.settings import pod_mutation_hook
-from airflow.utils.log.logging_mixin import LoggingMixin
-from airflow.utils.state import State
-
-
-class PodStatus:
- """Status of the PODs"""
-
- PENDING = 'pending'
- RUNNING = 'running'
- FAILED = 'failed'
- SUCCEEDED = 'succeeded'
-
-
-class PodLauncher(LoggingMixin):
- """Launches PODS"""
-
- def __init__(
- self,
- kube_client: client.CoreV1Api = None,
- in_cluster: bool = True,
- cluster_context: Optional[str] = None,
- extract_xcom: bool = False,
- ):
- """
- Creates the launcher.
-
- :param kube_client: kubernetes client
- :param in_cluster: whether we are in cluster
- :param cluster_context: context of the cluster
- :param extract_xcom: whether we should extract xcom
- """
- super().__init__()
- self._client = kube_client or get_kube_client(in_cluster=in_cluster, cluster_context=cluster_context)
- self._watch = watch.Watch()
- self.extract_xcom = extract_xcom
-
- def run_pod_async(self, pod: V1Pod, **kwargs):
- """Runs POD asynchronously"""
- pod_mutation_hook(pod)
-
- sanitized_pod = self._client.api_client.sanitize_for_serialization(pod)
- json_pod = json.dumps(sanitized_pod, indent=2)
-
- self.log.debug('Pod Creation Request: \n%s', json_pod)
- try:
- resp = self._client.create_namespaced_pod(
- body=sanitized_pod, namespace=pod.metadata.namespace, **kwargs
- )
- self.log.debug('Pod Creation Response: %s', resp)
- except Exception as e:
- self.log.exception('Exception when attempting to create Namespaced Pod: %s', json_pod)
- raise e
- return resp
-
- def delete_pod(self, pod: V1Pod):
- """Deletes POD"""
- try:
- self._client.delete_namespaced_pod(
- pod.metadata.name, pod.metadata.namespace, body=client.V1DeleteOptions()
- )
- except ApiException as e:
- # If the pod is already deleted
- if e.status != 404:
- raise
-
- def start_pod(self, pod: V1Pod, startup_timeout: int = 120):
- """
- Launches the pod synchronously and waits for completion.
-
- :param pod:
- :param startup_timeout: Timeout for startup of the pod (if pod is pending for too long, fails task)
- :return:
- """
- resp = self.run_pod_async(pod)
- curr_time = dt.now()
- if resp.status.start_time is None:
- while self.pod_not_started(pod):
- self.log.warning("Pod not yet started: %s", pod.metadata.name)
- delta = dt.now() - curr_time
- if delta.total_seconds() >= startup_timeout:
- raise AirflowException("Pod took too long to start")
- time.sleep(1)
-
- def monitor_pod(self, pod: V1Pod, get_logs: bool) -> Tuple[State, Optional[str]]:
- """
- Monitors a pod and returns the final state
-
- :param pod: pod spec that will be monitored
- :type pod : V1Pod
- :param get_logs: whether to read the logs locally
- :return: Tuple[State, Optional[str]]
- """
- if get_logs:
- read_logs_since_sec = None
- last_log_time = None
- while True:
- logs = self.read_pod_logs(pod, timestamps=True, since_seconds=read_logs_since_sec)
- for line in logs:
- timestamp, message = self.parse_log_line(line.decode('utf-8'))
- last_log_time = pendulum.parse(timestamp)
- self.log.info(message)
- time.sleep(1)
-
- if not self.base_container_is_running(pod):
- break
-
- self.log.warning('Pod %s log read interrupted', pod.metadata.name)
- if last_log_time:
- delta = pendulum.now() - last_log_time
- # Prefer logs duplication rather than loss
- read_logs_since_sec = math.ceil(delta.total_seconds())
- result = None
- if self.extract_xcom:
- while self.base_container_is_running(pod):
- self.log.info('Container %s has state %s', pod.metadata.name, State.RUNNING)
- time.sleep(2)
- result = self._extract_xcom(pod)
- self.log.info(result)
- result = json.loads(result)
- while self.pod_is_running(pod):
- self.log.info('Pod %s has state %s', pod.metadata.name, State.RUNNING)
- time.sleep(2)
- return self._task_status(self.read_pod(pod)), result
-
- def parse_log_line(self, line: str) -> Tuple[str, str]:
- """
- Parse K8s log line and returns the final state
-
- :param line: k8s log line
- :type line: str
- :return: timestamp and log message
- :rtype: Tuple[str, str]
- """
- split_at = line.find(' ')
- if split_at == -1:
- raise Exception(f'Log not in "{{timestamp}} {{log}}" format. Got: {line}')
- timestamp = line[:split_at]
- message = line[split_at + 1 :].rstrip()
- return timestamp, message
-
- def _task_status(self, event):
- self.log.info('Event: %s had an event of type %s', event.metadata.name, event.status.phase)
- status = self.process_status(event.metadata.name, event.status.phase)
- return status
-
- def pod_not_started(self, pod: V1Pod):
- """Tests if pod has not started"""
- state = self._task_status(self.read_pod(pod))
- return state == State.QUEUED
-
- def pod_is_running(self, pod: V1Pod):
- """Tests if pod is running"""
- state = self._task_status(self.read_pod(pod))
- return state not in (State.SUCCESS, State.FAILED)
-
- def base_container_is_running(self, pod: V1Pod):
- """Tests if base container is running"""
- event = self.read_pod(pod)
- status = next(iter(filter(lambda s: s.name == 'base', event.status.container_statuses)), None)
- if not status:
- return False
- return status.state.running is not None
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod_logs(
- self,
- pod: V1Pod,
- tail_lines: Optional[int] = None,
- timestamps: bool = False,
- since_seconds: Optional[int] = None,
- ):
- """Reads log from the POD"""
- additional_kwargs = {}
- if since_seconds:
- additional_kwargs['since_seconds'] = since_seconds
-
- if tail_lines:
- additional_kwargs['tail_lines'] = tail_lines
-
- try:
- return self._client.read_namespaced_pod_log(
- name=pod.metadata.name,
- namespace=pod.metadata.namespace,
- container='base',
- follow=True,
- timestamps=timestamps,
- _preload_content=False,
- **additional_kwargs,
- )
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod_events(self, pod):
- """Reads events from the POD"""
- try:
- return self._client.list_namespaced_event(
- namespace=pod.metadata.namespace, field_selector=f"involvedObject.name={pod.metadata.name}"
- )
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod(self, pod: V1Pod):
- """Read POD information"""
- try:
- return self._client.read_namespaced_pod(pod.metadata.name, pod.metadata.namespace)
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- def _extract_xcom(self, pod: V1Pod):
- resp = kubernetes_stream(
- self._client.connect_get_namespaced_pod_exec,
- pod.metadata.name,
- pod.metadata.namespace,
- container=PodDefaults.SIDECAR_CONTAINER_NAME,
- command=['/bin/sh'],
- stdin=True,
- stdout=True,
- stderr=True,
- tty=False,
- _preload_content=False,
- )
- try:
- result = self._exec_pod_command(resp, f'cat {PodDefaults.XCOM_MOUNT_PATH}/return.json')
- self._exec_pod_command(resp, 'kill -s SIGINT 1')
- finally:
- resp.close()
- if result is None:
- raise AirflowException(f'Failed to extract xcom from pod: {pod.metadata.name}')
- return result
-
- def _exec_pod_command(self, resp, command):
- if resp.is_open():
- self.log.info('Running command... %s\n', command)
- resp.write_stdin(command + '\n')
- while resp.is_open():
- resp.update(timeout=1)
- if resp.peek_stdout():
- return resp.read_stdout()
- if resp.peek_stderr():
- self.log.info(resp.read_stderr())
- break
- return None
-
- def process_status(self, job_id, status):
- """Process status information for the JOB"""
- status = status.lower()
- if status == PodStatus.PENDING:
- return State.QUEUED
- elif status == PodStatus.FAILED:
- self.log.error('Event with job id %s Failed', job_id)
- return State.FAILED
- elif status == PodStatus.SUCCEEDED:
- self.log.info('Event with job id %s Succeeded', job_id)
- return State.SUCCESS
- elif status == PodStatus.RUNNING:
- return State.RUNNING
- else:
- self.log.error('Event: Invalid state %s on job %s', status, job_id)
- return State.FAILED
+"""This module is deprecated. Please use `kubernetes.client.models for V1ResourceRequirements and Port."""
Review comment:
I believe this change is breaking. If someone uses old `cncf.kubernetes` provider and new Airflow and still will want to use the deprecated `airflow.kubernets.pod_launcher` module it will break.
This change brings an implicit dependency to the next version of `cncf.kubernetes` provider.
Any proposal how to fix it?
I see few options:
1) Leave the code for the old pod_launcher in the deprecated module (not ideal because of code duplication)
2) Add minimum `cncf.kubernetes` version requirement to `cncf.kubernetes` extra (this will not work for people who already have `airflow` installed and want to upgrade it.
3) Add `cncf.kubernetes` as "preinstalled" module for Airflow (similarly as ftp, http, imap, sqlite and add minimum version required there to the new version of provider. I like this most - imho K8S provider should be pre-installed.
WDYT? Any other options? Or maybe you do not see that as a problem?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] dimberman commented on a change in pull request #15165: Separate pod_launcher from core airflow
Posted by GitBox <gi...@apache.org>.
dimberman commented on a change in pull request #15165:
URL: https://github.com/apache/airflow/pull/15165#discussion_r607220561
##########
File path: airflow/kubernetes/pod_launcher_deprecated.py
##########
@@ -0,0 +1,300 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Launches PODs"""
+import json
+import math
+import time
+from datetime import datetime as dt
+from typing import Optional, Tuple
+
+import pendulum
+import tenacity
+from kubernetes import client, watch
+from kubernetes.client.models.v1_pod import V1Pod
+from kubernetes.client.rest import ApiException
+from kubernetes.stream import stream as kubernetes_stream
+from requests.exceptions import BaseHTTPError
+
+from airflow.exceptions import AirflowException
+from airflow.kubernetes.kube_client import get_kube_client
+from airflow.kubernetes.pod_generator import PodDefaults
+from airflow.settings import pod_mutation_hook
+from airflow.utils.log.logging_mixin import LoggingMixin
+from airflow.utils.state import State
+
+
Review comment:
@potiuk I added the deprecation to the deprecated class. good catch!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] dimberman commented on a change in pull request #15165: Separate pod_launcher from core airflow
Posted by GitBox <gi...@apache.org>.
dimberman commented on a change in pull request #15165:
URL: https://github.com/apache/airflow/pull/15165#discussion_r607229734
##########
File path: airflow/kubernetes/pod_launcher_deprecated.py
##########
@@ -0,0 +1,300 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Launches PODs"""
+import json
+import math
+import time
+from datetime import datetime as dt
+from typing import Optional, Tuple
+
+import pendulum
+import tenacity
+from kubernetes import client, watch
+from kubernetes.client.models.v1_pod import V1Pod
+from kubernetes.client.rest import ApiException
+from kubernetes.stream import stream as kubernetes_stream
+from requests.exceptions import BaseHTTPError
+
+from airflow.exceptions import AirflowException
+from airflow.kubernetes.kube_client import get_kube_client
+from airflow.kubernetes.pod_generator import PodDefaults
+from airflow.settings import pod_mutation_hook
+from airflow.utils.log.logging_mixin import LoggingMixin
+from airflow.utils.state import State
+
+
Review comment:
@potiuk oh oops 🤦 . Pushed!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] dimberman commented on a change in pull request #15165: Separate pod_launcher from core airflow
Posted by GitBox <gi...@apache.org>.
dimberman commented on a change in pull request #15165:
URL: https://github.com/apache/airflow/pull/15165#discussion_r607204487
##########
File path: airflow/kubernetes/pod_launcher.py
##########
@@ -14,283 +15,15 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
-"""Launches PODs"""
-import json
-import math
-import time
-from datetime import datetime as dt
-from typing import Optional, Tuple
-
-import pendulum
-import tenacity
-from kubernetes import client, watch
-from kubernetes.client.models.v1_pod import V1Pod
-from kubernetes.client.rest import ApiException
-from kubernetes.stream import stream as kubernetes_stream
-from requests.exceptions import BaseHTTPError
-
-from airflow.exceptions import AirflowException
-from airflow.kubernetes.kube_client import get_kube_client
-from airflow.kubernetes.pod_generator import PodDefaults
-from airflow.settings import pod_mutation_hook
-from airflow.utils.log.logging_mixin import LoggingMixin
-from airflow.utils.state import State
-
-
-class PodStatus:
- """Status of the PODs"""
-
- PENDING = 'pending'
- RUNNING = 'running'
- FAILED = 'failed'
- SUCCEEDED = 'succeeded'
-
-
-class PodLauncher(LoggingMixin):
- """Launches PODS"""
-
- def __init__(
- self,
- kube_client: client.CoreV1Api = None,
- in_cluster: bool = True,
- cluster_context: Optional[str] = None,
- extract_xcom: bool = False,
- ):
- """
- Creates the launcher.
-
- :param kube_client: kubernetes client
- :param in_cluster: whether we are in cluster
- :param cluster_context: context of the cluster
- :param extract_xcom: whether we should extract xcom
- """
- super().__init__()
- self._client = kube_client or get_kube_client(in_cluster=in_cluster, cluster_context=cluster_context)
- self._watch = watch.Watch()
- self.extract_xcom = extract_xcom
-
- def run_pod_async(self, pod: V1Pod, **kwargs):
- """Runs POD asynchronously"""
- pod_mutation_hook(pod)
-
- sanitized_pod = self._client.api_client.sanitize_for_serialization(pod)
- json_pod = json.dumps(sanitized_pod, indent=2)
-
- self.log.debug('Pod Creation Request: \n%s', json_pod)
- try:
- resp = self._client.create_namespaced_pod(
- body=sanitized_pod, namespace=pod.metadata.namespace, **kwargs
- )
- self.log.debug('Pod Creation Response: %s', resp)
- except Exception as e:
- self.log.exception('Exception when attempting to create Namespaced Pod: %s', json_pod)
- raise e
- return resp
-
- def delete_pod(self, pod: V1Pod):
- """Deletes POD"""
- try:
- self._client.delete_namespaced_pod(
- pod.metadata.name, pod.metadata.namespace, body=client.V1DeleteOptions()
- )
- except ApiException as e:
- # If the pod is already deleted
- if e.status != 404:
- raise
-
- def start_pod(self, pod: V1Pod, startup_timeout: int = 120):
- """
- Launches the pod synchronously and waits for completion.
-
- :param pod:
- :param startup_timeout: Timeout for startup of the pod (if pod is pending for too long, fails task)
- :return:
- """
- resp = self.run_pod_async(pod)
- curr_time = dt.now()
- if resp.status.start_time is None:
- while self.pod_not_started(pod):
- self.log.warning("Pod not yet started: %s", pod.metadata.name)
- delta = dt.now() - curr_time
- if delta.total_seconds() >= startup_timeout:
- raise AirflowException("Pod took too long to start")
- time.sleep(1)
-
- def monitor_pod(self, pod: V1Pod, get_logs: bool) -> Tuple[State, Optional[str]]:
- """
- Monitors a pod and returns the final state
-
- :param pod: pod spec that will be monitored
- :type pod : V1Pod
- :param get_logs: whether to read the logs locally
- :return: Tuple[State, Optional[str]]
- """
- if get_logs:
- read_logs_since_sec = None
- last_log_time = None
- while True:
- logs = self.read_pod_logs(pod, timestamps=True, since_seconds=read_logs_since_sec)
- for line in logs:
- timestamp, message = self.parse_log_line(line.decode('utf-8'))
- last_log_time = pendulum.parse(timestamp)
- self.log.info(message)
- time.sleep(1)
-
- if not self.base_container_is_running(pod):
- break
-
- self.log.warning('Pod %s log read interrupted', pod.metadata.name)
- if last_log_time:
- delta = pendulum.now() - last_log_time
- # Prefer logs duplication rather than loss
- read_logs_since_sec = math.ceil(delta.total_seconds())
- result = None
- if self.extract_xcom:
- while self.base_container_is_running(pod):
- self.log.info('Container %s has state %s', pod.metadata.name, State.RUNNING)
- time.sleep(2)
- result = self._extract_xcom(pod)
- self.log.info(result)
- result = json.loads(result)
- while self.pod_is_running(pod):
- self.log.info('Pod %s has state %s', pod.metadata.name, State.RUNNING)
- time.sleep(2)
- return self._task_status(self.read_pod(pod)), result
-
- def parse_log_line(self, line: str) -> Tuple[str, str]:
- """
- Parse K8s log line and returns the final state
-
- :param line: k8s log line
- :type line: str
- :return: timestamp and log message
- :rtype: Tuple[str, str]
- """
- split_at = line.find(' ')
- if split_at == -1:
- raise Exception(f'Log not in "{{timestamp}} {{log}}" format. Got: {line}')
- timestamp = line[:split_at]
- message = line[split_at + 1 :].rstrip()
- return timestamp, message
-
- def _task_status(self, event):
- self.log.info('Event: %s had an event of type %s', event.metadata.name, event.status.phase)
- status = self.process_status(event.metadata.name, event.status.phase)
- return status
-
- def pod_not_started(self, pod: V1Pod):
- """Tests if pod has not started"""
- state = self._task_status(self.read_pod(pod))
- return state == State.QUEUED
-
- def pod_is_running(self, pod: V1Pod):
- """Tests if pod is running"""
- state = self._task_status(self.read_pod(pod))
- return state not in (State.SUCCESS, State.FAILED)
-
- def base_container_is_running(self, pod: V1Pod):
- """Tests if base container is running"""
- event = self.read_pod(pod)
- status = next(iter(filter(lambda s: s.name == 'base', event.status.container_statuses)), None)
- if not status:
- return False
- return status.state.running is not None
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod_logs(
- self,
- pod: V1Pod,
- tail_lines: Optional[int] = None,
- timestamps: bool = False,
- since_seconds: Optional[int] = None,
- ):
- """Reads log from the POD"""
- additional_kwargs = {}
- if since_seconds:
- additional_kwargs['since_seconds'] = since_seconds
-
- if tail_lines:
- additional_kwargs['tail_lines'] = tail_lines
-
- try:
- return self._client.read_namespaced_pod_log(
- name=pod.metadata.name,
- namespace=pod.metadata.namespace,
- container='base',
- follow=True,
- timestamps=timestamps,
- _preload_content=False,
- **additional_kwargs,
- )
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod_events(self, pod):
- """Reads events from the POD"""
- try:
- return self._client.list_namespaced_event(
- namespace=pod.metadata.namespace, field_selector=f"involvedObject.name={pod.metadata.name}"
- )
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod(self, pod: V1Pod):
- """Read POD information"""
- try:
- return self._client.read_namespaced_pod(pod.metadata.name, pod.metadata.namespace)
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- def _extract_xcom(self, pod: V1Pod):
- resp = kubernetes_stream(
- self._client.connect_get_namespaced_pod_exec,
- pod.metadata.name,
- pod.metadata.namespace,
- container=PodDefaults.SIDECAR_CONTAINER_NAME,
- command=['/bin/sh'],
- stdin=True,
- stdout=True,
- stderr=True,
- tty=False,
- _preload_content=False,
- )
- try:
- result = self._exec_pod_command(resp, f'cat {PodDefaults.XCOM_MOUNT_PATH}/return.json')
- self._exec_pod_command(resp, 'kill -s SIGINT 1')
- finally:
- resp.close()
- if result is None:
- raise AirflowException(f'Failed to extract xcom from pod: {pod.metadata.name}')
- return result
-
- def _exec_pod_command(self, resp, command):
- if resp.is_open():
- self.log.info('Running command... %s\n', command)
- resp.write_stdin(command + '\n')
- while resp.is_open():
- resp.update(timeout=1)
- if resp.peek_stdout():
- return resp.read_stdout()
- if resp.peek_stderr():
- self.log.info(resp.read_stderr())
- break
- return None
-
- def process_status(self, job_id, status):
- """Process status information for the JOB"""
- status = status.lower()
- if status == PodStatus.PENDING:
- return State.QUEUED
- elif status == PodStatus.FAILED:
- self.log.error('Event with job id %s Failed', job_id)
- return State.FAILED
- elif status == PodStatus.SUCCEEDED:
- self.log.info('Event with job id %s Succeeded', job_id)
- return State.SUCCESS
- elif status == PodStatus.RUNNING:
- return State.RUNNING
- else:
- self.log.error('Event: Invalid state %s on job %s', status, job_id)
- return State.FAILED
+"""This module is deprecated. Please use `kubernetes.client.models for V1ResourceRequirements and Port."""
+# flake8: noqa
+# pylint: disable=unused-import
+import warnings
+
+from airflow.kubernetes import pod_launcher_deprecated # pylint: disable=unused-import
Review comment:
@kaxil I changed the import so this should be fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on a change in pull request #15165: Separate pod_launcher from core airflow
Posted by GitBox <gi...@apache.org>.
potiuk commented on a change in pull request #15165:
URL: https://github.com/apache/airflow/pull/15165#discussion_r606796281
##########
File path: airflow/kubernetes/pod_launcher.py
##########
@@ -14,283 +15,15 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
-"""Launches PODs"""
-import json
-import math
-import time
-from datetime import datetime as dt
-from typing import Optional, Tuple
-
-import pendulum
-import tenacity
-from kubernetes import client, watch
-from kubernetes.client.models.v1_pod import V1Pod
-from kubernetes.client.rest import ApiException
-from kubernetes.stream import stream as kubernetes_stream
-from requests.exceptions import BaseHTTPError
-
-from airflow.exceptions import AirflowException
-from airflow.kubernetes.kube_client import get_kube_client
-from airflow.kubernetes.pod_generator import PodDefaults
-from airflow.settings import pod_mutation_hook
-from airflow.utils.log.logging_mixin import LoggingMixin
-from airflow.utils.state import State
-
-
-class PodStatus:
- """Status of the PODs"""
-
- PENDING = 'pending'
- RUNNING = 'running'
- FAILED = 'failed'
- SUCCEEDED = 'succeeded'
-
-
-class PodLauncher(LoggingMixin):
- """Launches PODS"""
-
- def __init__(
- self,
- kube_client: client.CoreV1Api = None,
- in_cluster: bool = True,
- cluster_context: Optional[str] = None,
- extract_xcom: bool = False,
- ):
- """
- Creates the launcher.
-
- :param kube_client: kubernetes client
- :param in_cluster: whether we are in cluster
- :param cluster_context: context of the cluster
- :param extract_xcom: whether we should extract xcom
- """
- super().__init__()
- self._client = kube_client or get_kube_client(in_cluster=in_cluster, cluster_context=cluster_context)
- self._watch = watch.Watch()
- self.extract_xcom = extract_xcom
-
- def run_pod_async(self, pod: V1Pod, **kwargs):
- """Runs POD asynchronously"""
- pod_mutation_hook(pod)
-
- sanitized_pod = self._client.api_client.sanitize_for_serialization(pod)
- json_pod = json.dumps(sanitized_pod, indent=2)
-
- self.log.debug('Pod Creation Request: \n%s', json_pod)
- try:
- resp = self._client.create_namespaced_pod(
- body=sanitized_pod, namespace=pod.metadata.namespace, **kwargs
- )
- self.log.debug('Pod Creation Response: %s', resp)
- except Exception as e:
- self.log.exception('Exception when attempting to create Namespaced Pod: %s', json_pod)
- raise e
- return resp
-
- def delete_pod(self, pod: V1Pod):
- """Deletes POD"""
- try:
- self._client.delete_namespaced_pod(
- pod.metadata.name, pod.metadata.namespace, body=client.V1DeleteOptions()
- )
- except ApiException as e:
- # If the pod is already deleted
- if e.status != 404:
- raise
-
- def start_pod(self, pod: V1Pod, startup_timeout: int = 120):
- """
- Launches the pod synchronously and waits for completion.
-
- :param pod:
- :param startup_timeout: Timeout for startup of the pod (if pod is pending for too long, fails task)
- :return:
- """
- resp = self.run_pod_async(pod)
- curr_time = dt.now()
- if resp.status.start_time is None:
- while self.pod_not_started(pod):
- self.log.warning("Pod not yet started: %s", pod.metadata.name)
- delta = dt.now() - curr_time
- if delta.total_seconds() >= startup_timeout:
- raise AirflowException("Pod took too long to start")
- time.sleep(1)
-
- def monitor_pod(self, pod: V1Pod, get_logs: bool) -> Tuple[State, Optional[str]]:
- """
- Monitors a pod and returns the final state
-
- :param pod: pod spec that will be monitored
- :type pod : V1Pod
- :param get_logs: whether to read the logs locally
- :return: Tuple[State, Optional[str]]
- """
- if get_logs:
- read_logs_since_sec = None
- last_log_time = None
- while True:
- logs = self.read_pod_logs(pod, timestamps=True, since_seconds=read_logs_since_sec)
- for line in logs:
- timestamp, message = self.parse_log_line(line.decode('utf-8'))
- last_log_time = pendulum.parse(timestamp)
- self.log.info(message)
- time.sleep(1)
-
- if not self.base_container_is_running(pod):
- break
-
- self.log.warning('Pod %s log read interrupted', pod.metadata.name)
- if last_log_time:
- delta = pendulum.now() - last_log_time
- # Prefer logs duplication rather than loss
- read_logs_since_sec = math.ceil(delta.total_seconds())
- result = None
- if self.extract_xcom:
- while self.base_container_is_running(pod):
- self.log.info('Container %s has state %s', pod.metadata.name, State.RUNNING)
- time.sleep(2)
- result = self._extract_xcom(pod)
- self.log.info(result)
- result = json.loads(result)
- while self.pod_is_running(pod):
- self.log.info('Pod %s has state %s', pod.metadata.name, State.RUNNING)
- time.sleep(2)
- return self._task_status(self.read_pod(pod)), result
-
- def parse_log_line(self, line: str) -> Tuple[str, str]:
- """
- Parse K8s log line and returns the final state
-
- :param line: k8s log line
- :type line: str
- :return: timestamp and log message
- :rtype: Tuple[str, str]
- """
- split_at = line.find(' ')
- if split_at == -1:
- raise Exception(f'Log not in "{{timestamp}} {{log}}" format. Got: {line}')
- timestamp = line[:split_at]
- message = line[split_at + 1 :].rstrip()
- return timestamp, message
-
- def _task_status(self, event):
- self.log.info('Event: %s had an event of type %s', event.metadata.name, event.status.phase)
- status = self.process_status(event.metadata.name, event.status.phase)
- return status
-
- def pod_not_started(self, pod: V1Pod):
- """Tests if pod has not started"""
- state = self._task_status(self.read_pod(pod))
- return state == State.QUEUED
-
- def pod_is_running(self, pod: V1Pod):
- """Tests if pod is running"""
- state = self._task_status(self.read_pod(pod))
- return state not in (State.SUCCESS, State.FAILED)
-
- def base_container_is_running(self, pod: V1Pod):
- """Tests if base container is running"""
- event = self.read_pod(pod)
- status = next(iter(filter(lambda s: s.name == 'base', event.status.container_statuses)), None)
- if not status:
- return False
- return status.state.running is not None
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod_logs(
- self,
- pod: V1Pod,
- tail_lines: Optional[int] = None,
- timestamps: bool = False,
- since_seconds: Optional[int] = None,
- ):
- """Reads log from the POD"""
- additional_kwargs = {}
- if since_seconds:
- additional_kwargs['since_seconds'] = since_seconds
-
- if tail_lines:
- additional_kwargs['tail_lines'] = tail_lines
-
- try:
- return self._client.read_namespaced_pod_log(
- name=pod.metadata.name,
- namespace=pod.metadata.namespace,
- container='base',
- follow=True,
- timestamps=timestamps,
- _preload_content=False,
- **additional_kwargs,
- )
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod_events(self, pod):
- """Reads events from the POD"""
- try:
- return self._client.list_namespaced_event(
- namespace=pod.metadata.namespace, field_selector=f"involvedObject.name={pod.metadata.name}"
- )
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_exponential(), reraise=True)
- def read_pod(self, pod: V1Pod):
- """Read POD information"""
- try:
- return self._client.read_namespaced_pod(pod.metadata.name, pod.metadata.namespace)
- except BaseHTTPError as e:
- raise AirflowException(f'There was an error reading the kubernetes API: {e}')
-
- def _extract_xcom(self, pod: V1Pod):
- resp = kubernetes_stream(
- self._client.connect_get_namespaced_pod_exec,
- pod.metadata.name,
- pod.metadata.namespace,
- container=PodDefaults.SIDECAR_CONTAINER_NAME,
- command=['/bin/sh'],
- stdin=True,
- stdout=True,
- stderr=True,
- tty=False,
- _preload_content=False,
- )
- try:
- result = self._exec_pod_command(resp, f'cat {PodDefaults.XCOM_MOUNT_PATH}/return.json')
- self._exec_pod_command(resp, 'kill -s SIGINT 1')
- finally:
- resp.close()
- if result is None:
- raise AirflowException(f'Failed to extract xcom from pod: {pod.metadata.name}')
- return result
-
- def _exec_pod_command(self, resp, command):
- if resp.is_open():
- self.log.info('Running command... %s\n', command)
- resp.write_stdin(command + '\n')
- while resp.is_open():
- resp.update(timeout=1)
- if resp.peek_stdout():
- return resp.read_stdout()
- if resp.peek_stderr():
- self.log.info(resp.read_stderr())
- break
- return None
-
- def process_status(self, job_id, status):
- """Process status information for the JOB"""
- status = status.lower()
- if status == PodStatus.PENDING:
- return State.QUEUED
- elif status == PodStatus.FAILED:
- self.log.error('Event with job id %s Failed', job_id)
- return State.FAILED
- elif status == PodStatus.SUCCEEDED:
- self.log.info('Event with job id %s Succeeded', job_id)
- return State.SUCCESS
- elif status == PodStatus.RUNNING:
- return State.RUNNING
- else:
- self.log.error('Event: Invalid state %s on job %s', status, job_id)
- return State.FAILED
+"""This module is deprecated. Please use `kubernetes.client.models for V1ResourceRequirements and Port."""
Review comment:
I believe this change is breaking. If someone uses old `cncf.kubernetes` provider and new Airflow and still will want to use the deprecated `airflow.kubernets.pod_launcher` module. This change brings an implicit dependency to the next version of `cncf.kubernetes` provider.
Any proposal how to fix it?
I see few options:
1) Leave the code for the old pod_launcher in the deprecated module (not ideal because of code duplication)
2) Add minimum `cncf.kubernetes` version requirement to `cncf.kubernetes` extra (this will not work for people who already have `airflow` installed and want to upgrade it.
3) Add `cncf.kubernetes` as "preinstalled" module for Airflow (similarly as ftp, http, imap, sqlite and add minimum version required there to the new version of provider. I like this most - imho K8S provider should be pre-installed.
WDYT? Any other options? Or maybe you do not see that as a problem?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org