[AIRFLOW-6004] Untangle Executors class to avoid cyclic imports

[potiuk]

There are cyclic imports detected seemingly randomly by pylint checks when some
of the PRs are run in CI:

************* Module airflow.utils.log.json_formatter
airflow/utils/log/json_formatter.py:1:0: R0401: Cyclic import
(airflow.executors -> airflow.executors.kubernetes_executor ->
airflow.kubernetes.pod_generator) (cyclic-import)
airflow/utils/log/json_formatter.py:1:0: R0401: Cyclic import (airflow ->
airflow.executors -> airflow.executors.kubernetes_executor ->
airflow.kubernetes.pod_launcher) (cyclic-import)
airflow/utils/log/json_formatter.py:1:0: R0401: Cyclic import
(airflow.executors -> airflow.executors.kubernetes_executor ->
airflow.kubernetes.worker_configuration -> airflow.kubernetes.pod_generator)
(cyclic-import)

The problem is that airflow's init contains a few convenience imports
(AirflowException, Executors etc.) but it also imports a number of packages
(for example kubernetes_executor) that in turn import the airflow package
objects - for example airflow.Executor. This leads to cyclic imports if you
import first the executors before airflow. Similar problem happens with
executor.init.py containing class "Executors" imported by all executors but
at the same time some of the executors (for example KubernetesExecutor) import
the very same Executor class.   

This might happen in pylint checks in pre-commit because they split a number of
files they process between the multiple threads you have at your machine and
sometimes it might happen that the files are imported in different order.  

As a solution, the executors "list" should be moved to a separate module.
Also the name of constants was changed to not to be confused with class
names and Executors class was renamed to AvailableExecutors.

Make sure you have checked all steps below.

Jira

• My PR addresses the following Airflow Jira issues and references them in the PR title. For example, "[AIRFLOW-XXX] My Airflow PR"
  • https://issues.apache.org/jira/browse/AIRFLOW-6004

Description

• Here are some details about my PR, including screenshots of any UI changes:

Tests

• My PR adds the following unit tests OR does not need testing for this extremely good reason:

Commits

• My commits all reference Jira issues in their subject lines, and I have squashed multiple commits if they address the same issue. In addition, my commits follow the guidelines from "How to write a good git commit message":
  1. Subject is separated from body by a blank line
  2. Subject is limited to 50 characters (not including Jira issue reference)
  3. Subject does not end with a period
  4. Subject uses the imperative mood ("add", not "adding")
  5. Body wraps at 72 characters
  6. Body explains "what" and "why", not "how"

Documentation

• In case of new functionality, my PR adds documentation that describes how to use it.
  • All the public functions and the classes in the PR contain docstrings that explain what it does
  • If you implement backwards incompatible changes, please leave a note in the Updating.md so we can assign it to a appropriate release

[potiuk]

Still a few unrelated files changed.

Also we've got inconsistent imports airflow.configuration import conf vs airflow import conf and airflow import AirflowException vs airflow.exceptions import AirflowException.

I fixed few of those in the fixup that's coming but I left the gist of it for a follow-up commit where I implemented also validation in pre-commit.

[potiuk]

I think NOW I resolved everything that needs to be resolved in this commit (some more commits are following), I let Travis CI to take a look now. I also made the commit message much shorter - we do not need the full context there, I left more of it in JIRA. @ashb -> let me know if you think it's OK.

[kaxil]

I would probably leave the if conditions instead of asserts

[potiuk]

I am happy to discuss it and see other's opinion there. I was myself doubtful about asserts but they are really nicer and in many cases they simply make sense (see above) my comments. And if we discourage asserts we should remove them everywhere (they were used in other parts of the code).

[ashb]

I think this is a good case for asserts.

Under normal circumstances they should pass, and it's only if you're developing really that you might get this wrong, so having this be asserts that could be disabled by python -O seems like the right thing to do.

[uncletoxa]

I'd rather check our options first. How we may avoid assertions.

1. If we need a oneliner: if not self.futures: raise Error(NOT_STARTED_MESSAGE). It is not very elegant but easy to understand.

2. Another option to consider is writing a single checker and call it wether we need it:

   def _check_started(self):
           if None in (self.client, self.futures):
               raise AirflowError(NOT_STARTED_MESSAGE)

3. Or go more pythonic and introduce decorator.

   def check_futures(fn):
       def wrapped(self):
           if not self.futures:
               raise AirflowError(NOT_STARTED_MESSAGE)```
       return wrapped
   
   @check_futures
   def execute_async()
       <...>

4. If we expect that some methods need some time to wait for executor start, tenacity could be useful:

   from tenacity import retry
   
   @retry(stop=stop_after_attempt(3))
   def execute_async()
       try:
           <...>
       except Exception:
           raise AirflowError(NOT_STARTED_MESSAGE)

My vote goes to decorator solution -- either 3 or 4. Anyway, I think assertions could be occasionally useful for internal checks only. Like you checking all good and raise your exception with meaningful message to end user. Otherwise it looks like a dirty hack. Just a simple AssertionError would confuse average pythonist as developer expects (and official docs clearly states that) assertion usage in test environment only. Confusion harm would outweight readability benefit.

[mik-laj]

solution no.3 is not mypy compatibilie. It is not too smart to understand this. Most often, we want to have assertions to make pylint happy.

[ashb]

Style point that w should decide upon: In cases like this where th function arg is typed we don't also need to specify the type on the instance attribute - mypy will do that automatically.

So this could be written as:

    def __init__(self:
                 dag_directory: str,
                 file_paths: List[str],
                 max_runs: int,
                 processor_factory: Callable[[str, List[Any]], AbstractDagFileProcessor],
                 processor_timeout: timedelta,
                 signal_conn: Connection,
                 async_mode: bool = True):
        self._file_paths file_paths
        self._file_path_queue: List[str] = []
        self._dag_directory = dag_directory
        self._max_runs = max_runs
        self._processor_factory = processor_factory
        self._signal_conn = signal_conn
        self._async_mode = async_mode

[potiuk]

Yeah. I can fix it here. And I agree it's not needed.

[ashb]

Pre-emptively approving once the three changes (UPDATING, remove two instances of load_test_config) are done.

[potiuk]

Rebased with latest master and pushed all changes.

[potiuk]

And @ashb -> the "UPDATING.md" I think there was no warning as it was not possible to detect this way of importing. So I will leave it in UPDATING just in case someone has a problem with this way of import not working.

[potiuk]

Thanks @kaxil @ashb and others :). Merged finally.