Airflow: KubernetesExecutor single task run error: Only works with the Celery or Kubernetes executors, sorry

[mpolatcan]

Apache Airflow version: 1.10.12

Kubernetes version (if you are using kubernetes) (use kubectl version): v1.16.13-gke.401

Environment: AWS EKS

• Cloud provider or hardware configuration:
• OS (e.g. from /etc/os-release):
• Kernel (e.g. uname -a):
• Install tools:
• Others:**

What happened:

Hi Airflow community ✋ I have detected bug on Airflow webserver. I have deployed Airflow cluster that uses Kubernetes executor, runs on production AWS EKS cluster. All configurations are correct and runs successfully tasks on DAGs. But when I want to run specific task on DAG, that throws error like that:

But, I have checked my KubernetesExecutor configurations all of them are correct. So, I have digged error on source code of Airflow webserver at tag 1.10.12 which I have deployed version of Airflow. And found that some import errors are ignored for KubernetesExecutor on file https://github.com/apache/airflow/blob/1.10.12/airflow/www/views.py at line 1152

executor = ExecutorLoader.get_default_executor()
valid_celery_config = False
valid_kubernetes_config = False

try:
   from airflow.executors.celery_executor import CeleryExecutor  # noqa

   valid_celery_config = isinstance(executor, CeleryExecutor)
except ImportError:
    pass

try:
  from airflow.contrib.executors.kubernetes_executor import KubernetesExecutor  # noqa

   valid_kubernetes_config = isinstance(executor, KubernetesExecutor)
except ImportError:
   pass

if not valid_celery_config and not valid_kubernetes_config:
   flash("Only works with the Celery or Kubernetes executors, sorry", "error")
   return redirect(origin)

In there, bug in this code snippet

try:
  from airflow.contrib.executors.kubernetes_executor import KubernetesExecutor  # BUG IS HERE !!!

   valid_kubernetes_config = isinstance(executor, KubernetesExecutor)
except ImportError:
   pass

Import statement raises error KubernetesExecutor not found in this module because when you look that module at tag 1.10.12 https://github.com/apache/airflow/blob/1.10.12/airflow/contrib/executors/kubernetes_executor.py, there is no KubernetesExecutor class, only includes kubernetes_executor module which it is KubernetesExecutor's location. But when you use that you need to change the code like that:

try:
  from airflow.contrib.executors.kubernetes_executor import kubernetes_executor

  valid_kubernetes_config = isinstance(executor, kubernetes_executor.KubernetesExecutor)
except ImportError:
   pass

or:

try:
  from airflow.executors.kubernetes_executor import KubernetesExecutor

  valid_kubernetes_config = isinstance(executor, KubernetesExecutor)
except ImportError:
   pass

What you expected:

Runs successfully single task with KubernetesExecutor. This error block user to run single task with KubernetesExecutor without run whole DAG.

Thanks, have a nice day 👍 😄

[boring-cyborg[bot]]

Thanks for opening your first issue here! Be sure to follow the issue template!

[gfeldman]

Confirming @mpolatcan diagnosis and that changing the import from airflow.contrib.executors.kubernetes_executor to airflow.executors.kubernetes_executor in airflow/www/views.py and airflow/www_rbac/views.py resolves the issue.

[turbaszek]

CC @dimberman @kaxil , should we release this in 1.10.13?

[kaxil]

This is already fixed in 1.10.13rc1 : https://github.com/apache/airflow/commit/25c16e87536b4c12acf1ae19bfa3ce1e3f5d6171

[mpolatcan]

Thanks @kaxil, I am awaiting Airflow version 1.10.13 😄 👍