Repository: apache/airflow
Pull request: Scheduler should not fail when invalid executor_config is passed
closes https://github.com/apache/airflow/issues/14182

<!--
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).


Problem statement:
Scheduler dies if executor_config isnt passed a dict when using K8s executor
**Apache Airflow version**: 2.0.1


**Kubernetes version (if you are using kubernetes)** (use `kubectl version`): 1.15

**Environment**:

- **Cloud provider or hardware configuration**: k8s on bare metal
- **OS** (e.g. from /etc/os-release): 
- **Kernel** (e.g. `uname -a`):
- **Install tools**: pip3
- **Others**:

**What happened**:
Scheduler dies with 
```
[2021-02-10 21:09:27,469] {scheduler_job.py:1298} ERROR - Exception when executing SchedulerJob._run_schedu
ler_loop
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 1280, in _execute
    self._run_scheduler_loop()
  File "/usr/local/lib/python3.8/site-packages/airflow/jobs/scheduler_job.py", line 1384, in _run_scheduler
_loop
    self.executor.heartbeat()
  File "/usr/local/lib/python3.8/site-packages/airflow/executors/base_executor.py", line 158, in heartbeat
    self.trigger_tasks(open_slots)
  File "/usr/local/lib/python3.8/site-packages/airflow/executors/base_executor.py", line 188, in trigger_ta
sks
    self.execute_async(key=key, command=command, queue=None, executor_config=ti.executor_config)
  File "/usr/local/lib/python3.8/site-packages/airflow/executors/kubernetes_executor.py", line 493, in exec
ute_async
    kube_executor_config = PodGenerator.from_obj(executor_config)
  File "/usr/local/lib/python3.8/site-packages/airflow/kubernetes/pod_generator.py", line 175, in from_obj
    k8s_legacy_object = obj.get("KubernetesExecutor", None)
AttributeError: 'V1Pod' object has no attribute 'get'
[2021-02-10 21:09:28,475] {process_utils.py:100} INFO - Sending Signals.SIGTERM to GPID 60
[2021-02-10 21:09:29,222] {process_utils.py:66} INFO - Process psutil.Process(pid=66, status='terminated',
started='21:09:27') (66) terminated with exit code None
[2021-02-10 21:09:29,697] {process_utils.py:206} INFO - Waiting up to 5 seconds for processes to exit...
[2021-02-10 21:09:29,716] {process_utils.py:66} INFO - Process psutil.Process(pid=75, status='terminated',
started='21:09:28') (75) terminated with exit code None
[2021-02-10 21:09:29,717] {process_utils.py:66} INFO - Process psutil.Process(pid=60, status='terminated',
exitcode=0, started='21:09:27') (60) terminated with exit code 0
[2021-02-10 21:09:29,717] {scheduler_job.py:1301} INFO - Exited execute loop
```

**What you expected to happen**:
DAG loading fails, producing an error for just that DAG, instead of crashing the scheduler.

**How to reproduce it**:
Create a task like
```
    test = DummyOperator(task_id="new-pod-spec",
                              executor_config=k8s.V1Pod(
                                      spec=k8s.V1PodSpec(
                                          containers=[
                                              k8s.V1Container(
                                                  name="base",
                                                  image="myimage",
                                                  image_pull_policy="Always"
                                              )
                                          ]
                                      )))
```
or 
```
    test = DummyOperator(task_id="new-pod-spec",
                              executor_config={"KubernetesExecutor": k8s.V1Pod(
                                      spec=k8s.V1PodSpec(
                                          containers=[
                                              k8s.V1Container(
                                                  name="base",
                                                  image="myimage",
                                                  image_pull_policy="Always"
                                              )
                                          ]
                                      ))})
```
essentially anything where it expects a dict but gets something else, and run the scheduler using the kubernetes executor

Hints from issue discussion:
Hi @fjmacagno so if I'm understanding the issue: rather than killing the entire scheduler when this happens, a better result would be for this to return a task failure with the error upon run?

@kaxil @jhtimmins @ephraimbuddy this would be a good start k8sexecutor bug if you're interested. Glad to help you find the solution.
Yeah, in general i want to avoid letting my users break airflow for everyone, and since this is a mis-configuration of a DAG I feel it should not load in the first place.

Maybe Executors need a verifyExecutorConfig method?
@fjmacagno that could work or at the minimum wrapping it in a try/catch block. 

This should be a very easy fix, so you or anyone mentioned should be able to take it on and I'd be glad to help/review (I'm trying to spread k8sexecutor knowledge so we can start answering these tickets more quickly)

Commit message: Scheduler should not fail when invalid executor_config is passed

closes https://github.com/apache/airflow/issues/14182

Patch to review:
diff --git a/airflow/executors/kubernetes_executor.py b/airflow/executors/kubernetes_executor.py
index 88e26be885932..2ae3b9acc6c70 100644
--- a/airflow/executors/kubernetes_executor.py
+++ b/airflow/executors/kubernetes_executor.py
@@ -490,7 +490,13 @@ def execute_async(
     ) -> None:
         """Executes task asynchronously"""
         self.log.info('Add task %s with command %s with executor_config %s', key, command, executor_config)
-        kube_executor_config = PodGenerator.from_obj(executor_config)
+        try:
+            kube_executor_config = PodGenerator.from_obj(executor_config)
+        except Exception:  # pylint: disable=broad-except
+            self.log.error("Invalid executor_config for %s", key)
+            self.change_state(key=key, state=State.FAILED, info="Invalid executor_config passed")
+            return
+
         if executor_config:
             pod_template_file = executor_config.get("pod_template_override", None)
         else:
diff --git a/tests/executors/test_kubernetes_executor.py b/tests/executors/test_kubernetes_executor.py
index 9abb32884310b..dc7cbbbf7b532 100644
--- a/tests/executors/test_kubernetes_executor.py
+++ b/tests/executors/test_kubernetes_executor.py
@@ -194,6 +194,26 @@ def test_gauge_executor_metrics(self, mock_stats_gauge, mock_trigger_tasks, mock
         ]
         mock_stats_gauge.assert_has_calls(calls)
 
+    @mock.patch('airflow.executors.kubernetes_executor.KubernetesJobWatcher')
+    @mock.patch('airflow.executors.kubernetes_executor.get_kube_client')
+    def test_invalid_executor_config(self, mock_get_kube_client, mock_kubernetes_job_watcher):
+        executor = self.kubernetes_executor
+        executor.start()
+
+        assert executor.event_buffer == {}
+        executor.execute_async(
+            key=('dag', 'task', datetime.utcnow(), 1),
+            queue=None,
+            command=['airflow', 'tasks', 'run', 'true', 'some_parameter'],
+            executor_config=k8s.V1Pod(
+                spec=k8s.V1PodSpec(
+                    containers=[k8s.V1Container(name="base", image="myimage", image_pull_policy="Always")]
+                )
+            ),
+        )
+
+        assert list(executor.event_buffer.values())[0][1] == "Invalid executor_config passed"
+
     @mock.patch('airflow.executors.kubernetes_executor.KubernetesJobWatcher')
     @mock.patch('airflow.executors.kubernetes_executor.get_kube_client')
     def test_change_state_running(self, mock_get_kube_client, mock_kubernetes_job_watcher):


Code review comments:
diff_hunk: @@ -490,7 +490,13 @@ def execute_async(
     ) -> None:
         """Executes task asynchronously"""
         self.log.info('Add task %s with command %s with executor_config %s', key, command, executor_config)
-        kube_executor_config = PodGenerator.from_obj(executor_config)
+        try:
+            kube_executor_config = PodGenerator.from_obj(executor_config)
+        except Exception:  # pylint: disable=broad-except
+            self.log.error("Invalid executor_config for %s", key)
+            self.change_state(key=key, state=State.FAILED, info="Invalid executor_config passed")
path: airflow/executors/kubernetes_executor.py
text: @user1:
maybe can use the existing `self.fail()` method?

@author:
Updated

diff_hunk: @@ -490,7 +490,13 @@ def execute_async(
     ) -> None:
         """Executes task asynchronously"""
         self.log.info('Add task %s with command %s with executor_config %s', key, command, executor_config)
-        kube_executor_config = PodGenerator.from_obj(executor_config)
+        try:
+            kube_executor_config = PodGenerator.from_obj(executor_config)
+        except Exception:  # pylint: disable=broad-except
+            self.log.error("Invalid executor_config for %s", key)
+            self.change_state(key=key, state=State.FAILED, info="Invalid executor_config passed")
path: airflow/executors/kubernetes_executor.py
text: @author:
```suggestion
            self.fail(key=key, info="Invalid executor_config passed")
```