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 2022/11/11 00:00:52 UTC
[GitHub] [airflow] XD-DENG opened a new pull request, #27611: Make Kubernetes Executor & Scheduler Resilient to error during PMH execution
XD-DENG opened a new pull request, #27611:
URL: https://github.com/apache/airflow/pull/27611
Currently there is no try-catch logic when `pod_mutation_hook` is called in the Kubernetes Executor. So if there is any error/exception during the execution of the `pod_mutation_hook` user created, the whole executor/scheduler will crash.
Such error may happen:
- user who authored the `pod_mutation_hook` made an error inside.
- if there is any code inside the `pod_mutation_hook` encounters transient error (e.g. it queries DB and the DB gives a timeout error).
- ... you name it. It can simply go wrong.
This PR aims to make Kubernetes Executor & Scheduler more resilient to such error during PMH execution (current behavior, directly crashing, is a bit scary)
---
**^ Add meaningful description above**
Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst#pull-request-guidelines)** for more information.
In case of fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+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 a newsfragment file, named `{pr_number}.significant.rst` or `{issue_number}.significant.rst`, in [newsfragments](https://github.com/apache/airflow/tree/main/newsfragments).
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] XD-DENG commented on a diff in pull request #27611: Make Kubernetes Executor & Scheduler Resilient to error during PMH execution
Posted by GitBox <gi...@apache.org>.
XD-DENG commented on code in PR #27611:
URL: https://github.com/apache/airflow/pull/27611#discussion_r1019723290
##########
airflow/executors/kubernetes_executor.py:
##########
@@ -646,6 +649,14 @@ def sync(self) -> None:
json.loads(e.body)["message"],
)
self.task_queue.put(task)
+ except PodMutationHookException as e:
+ key, _, _, _ = task
+ self.log.warning(
+ 'Pod Mutation Hook failed for the task %s. Re-queueing. Details: %s',
Review Comment:
The reason for what the `pod_mutation_hook` failed may be different and can vary case by case. That's why I chose to log & re-queue it, just similar to how the non-`400`/`422` `ApiException` are handled. Not sure if it's sensible to you? @jedcunningham
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] XD-DENG merged pull request #27611: Make Kubernetes Executor & Scheduler Resilient to error during PMH execution
Posted by GitBox <gi...@apache.org>.
XD-DENG merged PR #27611:
URL: https://github.com/apache/airflow/pull/27611
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] XD-DENG commented on a diff in pull request #27611: Make Kubernetes Executor & Scheduler Resilient to error during PMH execution
Posted by GitBox <gi...@apache.org>.
XD-DENG commented on code in PR #27611:
URL: https://github.com/apache/airflow/pull/27611#discussion_r1019753709
##########
tests/executors/test_kubernetes_executor.py:
##########
@@ -293,6 +293,41 @@ def test_run_next_exception_requeue(
assert kubernetes_executor.task_queue.empty()
assert kubernetes_executor.event_buffer[task_instance_key][0] == State.FAILED
+ @pytest.mark.skipif(
+ AirflowKubernetesScheduler is None, reason="kubernetes python package is not installed"
+ )
+ @mock.patch("airflow.executors.kubernetes_executor.pod_mutation_hook")
+ @mock.patch("airflow.executors.kubernetes_executor.get_kube_client")
+ def test_run_next_pmh_error_requeue(self, mock_get_kube_client, mock_pmh):
+ """
+ Exception during Pod Mutation Hook execution should be handled gracefully.
+ The job should also be re-queued in case the Pod Mutation Hook failed due to transient reason.
+
+ """
+ mock_pmh.side_effect = Exception("Purposely generate error for test")
+
+ mock_kube_client = mock.patch("kubernetes.client.CoreV1Api", autospec=True)
+ mock_kube_client.create_namespaced_pod = mock.MagicMock()
+ mock_get_kube_client.return_value = mock_kube_client
+
+ kubernetes_executor = self.kubernetes_executor
+ kubernetes_executor.start()
+ try_number = 1
+ task_instance_key = TaskInstanceKey("dag", "task", "run_id", try_number)
+ kubernetes_executor.execute_async(
+ key=task_instance_key,
+ queue=None,
+ command=["airflow", "tasks", "run", "true", "some_parameter"],
+ )
+ kubernetes_executor.sync()
+
+ # The pod_mutation_hook should have been called once.
+ assert mock_pmh.call_count == 1
+ # There should be no pod creation request sent
+ assert mock_kube_client.create_namespaced_pod.call_count == 0
+ # The task is not re-queued
+ assert kubernetes_executor.task_queue.empty()
Review Comment:
Thanks @jedcunningham .
Added more comprehensive test updates at https://github.com/apache/airflow/pull/27611/commits/0b86fb3693540e3ee3fba434ceaaf9c8934cb85c.
Will merge after the CI passed.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] XD-DENG commented on a diff in pull request #27611: Make Kubernetes Executor & Scheduler Resilient to error during PMH execution
Posted by GitBox <gi...@apache.org>.
XD-DENG commented on code in PR #27611:
URL: https://github.com/apache/airflow/pull/27611#discussion_r1038790399
##########
airflow/executors/kubernetes_executor.py:
##########
@@ -256,7 +256,10 @@ def __init__(
def run_pod_async(self, pod: k8s.V1Pod, **kwargs):
"""Runs POD asynchronously"""
- pod_mutation_hook(pod)
+ try:
+ pod_mutation_hook(pod)
+ except Exception as e:
+ raise PodMutationHookException(e)
Review Comment:
Hi @hterik , thanks for the feedback!
I tested the two different ways below. I find the difference is minor, and I personally prefer the current way where the detailed exception reason appears in the same line as the `PodMutationHookException`, so it's clearer to the log reader (this is also the original reason for what I wrote it this way). WDYT?
## Case -1
```python
from airflow.exceptions import PodMutationHookException
try:
int("a")
except Exception as e:
raise PodMutationHookException(e)
```
We get
```
Traceback (most recent call last):
File "<stdin>", line 2, in <module>
ValueError: invalid literal for int() with base 10: 'a'
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "<stdin>", line 4, in <module>
airflow.exceptions.PodMutationHookException: invalid literal for int() with base 10: 'a'
```
## Case -2
If we use the way you suggested
```python
from airflow.exceptions import PodMutationHookException
try:
int("a")
except Exception as e:
raise PodMutationHookException() from e
```
We get
```
Traceback (most recent call last):
File "<stdin>", line 2, in <module>
ValueError: invalid literal for int() with base 10: 'a'
The above exception was the direct cause of the following exception:
Traceback (most recent call last):
File "<stdin>", line 4, in <module>
airflow.exceptions.PodMutationHookException
```
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] XD-DENG commented on a diff in pull request #27611: Make Kubernetes Executor & Scheduler Resilient to error during PMH execution
Posted by GitBox <gi...@apache.org>.
XD-DENG commented on code in PR #27611:
URL: https://github.com/apache/airflow/pull/27611#discussion_r1019730009
##########
airflow/executors/kubernetes_executor.py:
##########
@@ -646,6 +649,14 @@ def sync(self) -> None:
json.loads(e.body)["message"],
)
self.task_queue.put(task)
+ except PodMutationHookException as e:
+ key, _, _, _ = task
+ self.log.warning(
+ 'Pod Mutation Hook failed for the task %s. Re-queueing. Details: %s',
Review Comment:
I already thought of this: it can literally be any type of error, so unfortunately there is no way for us to differentiate more granularly.
But yes I'm sold that we may not want to re-queue here. Let me update accordingly.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] XD-DENG commented on a diff in pull request #27611: Make Kubernetes Executor & Scheduler Resilient to error during PMH execution
Posted by GitBox <gi...@apache.org>.
XD-DENG commented on code in PR #27611:
URL: https://github.com/apache/airflow/pull/27611#discussion_r1038790399
##########
airflow/executors/kubernetes_executor.py:
##########
@@ -256,7 +256,10 @@ def __init__(
def run_pod_async(self, pod: k8s.V1Pod, **kwargs):
"""Runs POD asynchronously"""
- pod_mutation_hook(pod)
+ try:
+ pod_mutation_hook(pod)
+ except Exception as e:
+ raise PodMutationHookException(e)
Review Comment:
Hi @hterik , I tested the two different ways below. I find the difference is minor, and I personally prefer the current way where the detailed exception reason appears in the same line as the `PodMutationHookException`, so it's clearer to the log reader (this is also the original reason for what I wrote it this way). WDYT?
## Case -1
```python
from airflow.exceptions import PodMutationHookException
try:
int("a")
except Exception as e:
raise PodMutationHookException(e)
```
We get
```
Traceback (most recent call last):
File "<stdin>", line 2, in <module>
ValueError: invalid literal for int() with base 10: 'a'
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "<stdin>", line 4, in <module>
airflow.exceptions.PodMutationHookException: invalid literal for int() with base 10: 'a'
```
## Case -2
If we use the way you suggested
```python
from airflow.exceptions import PodMutationHookException
try:
int("a")
except Exception as e:
raise PodMutationHookException() from e
```
We get
```
Traceback (most recent call last):
File "<stdin>", line 2, in <module>
ValueError: invalid literal for int() with base 10: 'a'
The above exception was the direct cause of the following exception:
Traceback (most recent call last):
File "<stdin>", line 4, in <module>
airflow.exceptions.PodMutationHookException
```
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] hterik commented on a diff in pull request #27611: Make Kubernetes Executor & Scheduler Resilient to error during PMH execution
Posted by GitBox <gi...@apache.org>.
hterik commented on code in PR #27611:
URL: https://github.com/apache/airflow/pull/27611#discussion_r1038788637
##########
airflow/executors/kubernetes_executor.py:
##########
@@ -256,7 +256,10 @@ def __init__(
def run_pod_async(self, pod: k8s.V1Pod, **kwargs):
"""Runs POD asynchronously"""
- pod_mutation_hook(pod)
+ try:
+ pod_mutation_hook(pod)
+ except Exception as e:
+ raise PodMutationHookException(e)
Review Comment:
I believe this should be `raise PodMutationHookException() from e` ? Otherwise it will be logged as an exception happening while handling another exception.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] jedcunningham commented on a diff in pull request #27611: Make Kubernetes Executor & Scheduler Resilient to error during PMH execution
Posted by GitBox <gi...@apache.org>.
jedcunningham commented on code in PR #27611:
URL: https://github.com/apache/airflow/pull/27611#discussion_r1019728052
##########
airflow/executors/kubernetes_executor.py:
##########
@@ -646,6 +649,14 @@ def sync(self) -> None:
json.loads(e.body)["message"],
)
self.task_queue.put(task)
+ except PodMutationHookException as e:
+ key, _, _, _ = task
+ self.log.warning(
+ 'Pod Mutation Hook failed for the task %s. Re-queueing. Details: %s',
Review Comment:
The k8s logic at least tries to differentiate between "probably always fatal" and "probably not always fatal" errors. Not sure we can easily do that with the mutation hook though, so taking the "safe" path of failing makes the most sense to me.
Said another way, I have no problem requeuing if we only try it when we think it might succeed the next time instead of having blind confidence it'll eventually work.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] XD-DENG commented on a diff in pull request #27611: Make Kubernetes Executor & Scheduler Resilient to error during PMH execution
Posted by GitBox <gi...@apache.org>.
XD-DENG commented on code in PR #27611:
URL: https://github.com/apache/airflow/pull/27611#discussion_r1019735200
##########
airflow/executors/kubernetes_executor.py:
##########
@@ -646,6 +649,14 @@ def sync(self) -> None:
json.loads(e.body)["message"],
)
self.task_queue.put(task)
+ except PodMutationHookException as e:
+ key, _, _, _ = task
+ self.log.warning(
+ 'Pod Mutation Hook failed for the task %s. Re-queueing. Details: %s',
Review Comment:
Hi @jedcunningham , addressed with https://github.com/apache/airflow/pull/27611/commits/022a93fb4b499df7ed76a19fe159ece21eb045c2
Mind helping take another look? Cheers!
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] jedcunningham commented on a diff in pull request #27611: Make Kubernetes Executor & Scheduler Resilient to error during PMH execution
Posted by GitBox <gi...@apache.org>.
jedcunningham commented on code in PR #27611:
URL: https://github.com/apache/airflow/pull/27611#discussion_r1019740790
##########
tests/executors/test_kubernetes_executor.py:
##########
@@ -293,6 +293,41 @@ def test_run_next_exception_requeue(
assert kubernetes_executor.task_queue.empty()
assert kubernetes_executor.event_buffer[task_instance_key][0] == State.FAILED
+ @pytest.mark.skipif(
+ AirflowKubernetesScheduler is None, reason="kubernetes python package is not installed"
+ )
+ @mock.patch("airflow.executors.kubernetes_executor.pod_mutation_hook")
+ @mock.patch("airflow.executors.kubernetes_executor.get_kube_client")
+ def test_run_next_pmh_error_requeue(self, mock_get_kube_client, mock_pmh):
+ """
+ Exception during Pod Mutation Hook execution should be handled gracefully.
+ The job should also be re-queued in case the Pod Mutation Hook failed due to transient reason.
+
+ """
+ mock_pmh.side_effect = Exception("Purposely generate error for test")
+
+ mock_kube_client = mock.patch("kubernetes.client.CoreV1Api", autospec=True)
+ mock_kube_client.create_namespaced_pod = mock.MagicMock()
+ mock_get_kube_client.return_value = mock_kube_client
+
+ kubernetes_executor = self.kubernetes_executor
+ kubernetes_executor.start()
+ try_number = 1
+ task_instance_key = TaskInstanceKey("dag", "task", "run_id", try_number)
+ kubernetes_executor.execute_async(
+ key=task_instance_key,
+ queue=None,
+ command=["airflow", "tasks", "run", "true", "some_parameter"],
+ )
+ kubernetes_executor.sync()
+
+ # The pod_mutation_hook should have been called once.
+ assert mock_pmh.call_count == 1
+ # There should be no pod creation request sent
+ assert mock_kube_client.create_namespaced_pod.call_count == 0
+ # The task is not re-queued
+ assert kubernetes_executor.task_queue.empty()
Review Comment:
```suggestion
# The task is failed
assert kubernetes_executor.task_queue.empty()
assert kubernetes_executor.event_buffer[task_instance_key][0] == State.FAILED
```
Lets check both the task_queue and event_buffer.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] jedcunningham commented on a diff in pull request #27611: Make Kubernetes Executor & Scheduler Resilient to error during PMH execution
Posted by GitBox <gi...@apache.org>.
jedcunningham commented on code in PR #27611:
URL: https://github.com/apache/airflow/pull/27611#discussion_r1019721454
##########
airflow/executors/kubernetes_executor.py:
##########
@@ -646,6 +649,14 @@ def sync(self) -> None:
json.loads(e.body)["message"],
)
self.task_queue.put(task)
+ except PodMutationHookException as e:
+ key, _, _, _ = task
+ self.log.warning(
+ 'Pod Mutation Hook failed for the task %s. Re-queueing. Details: %s',
Review Comment:
I feel like we should fail it instead. It could just as easily always fail and end up stuck, probably better to rely on retries to give it another chance.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] potiuk commented on a diff in pull request #27611: Make Kubernetes Executor & Scheduler Resilient to error during PMH execution
Posted by GitBox <gi...@apache.org>.
potiuk commented on code in PR #27611:
URL: https://github.com/apache/airflow/pull/27611#discussion_r1039099044
##########
airflow/executors/kubernetes_executor.py:
##########
@@ -256,7 +256,10 @@ def __init__(
def run_pod_async(self, pod: k8s.V1Pod, **kwargs):
"""Runs POD asynchronously"""
- pod_mutation_hook(pod)
+ try:
+ pod_mutation_hook(pod)
+ except Exception as e:
+ raise PodMutationHookException(e)
Review Comment:
Better indeed.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] hterik commented on a diff in pull request #27611: Make Kubernetes Executor & Scheduler Resilient to error during PMH execution
Posted by GitBox <gi...@apache.org>.
hterik commented on code in PR #27611:
URL: https://github.com/apache/airflow/pull/27611#discussion_r1039269266
##########
airflow/executors/kubernetes_executor.py:
##########
@@ -256,7 +256,10 @@ def __init__(
def run_pod_async(self, pod: k8s.V1Pod, **kwargs):
"""Runs POD asynchronously"""
- pod_mutation_hook(pod)
+ try:
+ pod_mutation_hook(pod)
+ except Exception as e:
+ raise PodMutationHookException(e)
Review Comment:
Oh sorry, maybe `raise PodMutationHookException(e) from e` in that case ;) That will have best of both worlds. But yeah, i think the difference is minor and i don't think it has any logical difference other than the log message in between the two tracebacks.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org