You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "o-nikolas (via GitHub)" <gi...@apache.org> on 2024/02/22 21:00:23 UTC
[PR] Introduce mechanism to support multiple executor configuration [airflow]
o-nikolas opened a new pull request, #37635:
URL: https://github.com/apache/airflow/pull/37635
This commit delivers an isolated component of AIP-61: https://cwiki.apache.org/confluence/display/AIRFLOW/AIP-61+Hybrid+Execution
It updates the executor loader logic to allow configuring multiple executors (and aliases per executor) in a list form (comma delimited) from airflow configuration; as well as associated methods for core Airflow code to load the default executor (first executor in the list) or any other configured executor.
Testing is included with the changes but not documentation, because AIP-61 is not yet complete. The code delivered is behind gating logic that will prohibit users from accidentally using it. User facing documentation will be delivered when the entire AIP-61 feature is fully completed.
<!--
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.
-->
<!--
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 an 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/
-->
<!-- Please keep an empty line above the dashes. -->
---
**^ Add meaningful description above**
Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.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
Re: [PR] Introduce mechanism to support multiple executor configuration [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on code in PR #37635:
URL: https://github.com/apache/airflow/pull/37635#discussion_r1510022587
##########
airflow/executors/executor_loader.py:
##########
@@ -69,56 +71,181 @@ class ExecutorLoader:
}
@classmethod
- def get_default_executor_name(cls) -> str:
- """Return the default executor name from Airflow configuration.
+ def block_use_of_hybrid_exec(cls, executor_config: list):
+ """Raise an exception if the user tries to use multiple executors before the feature is complete.
- :return: executor name from Airflow configuration
+ This check is built into a method so that it can be easily mocked in unit tests.
+
+ :param executor_config: core.executor configuration value.
+ """
+ if len(executor_config) > 1 or ":" in "".join(executor_config):
+ raise AirflowConfigException(
+ "Configuring multiple executors and executor aliases are not yet supported!: "
+ f"{executor_config}"
+ )
+
+ @classmethod
+ def _get_executor_names(cls) -> list[ExecutorName]:
+ """Return the executor names from Airflow configuration.
+
+ :return: List of executor names from Airflow configuration
"""
from airflow.configuration import conf
- return conf.get_mandatory_value("core", "EXECUTOR")
+ if _executor_names:
+ return _executor_names
+
+ executor_names_raw = conf.get_mandatory_list_value("core", "EXECUTOR")
+
+ # AIP-61 is WIP. Unblock configuring multiple executors when the feature is ready to launch
+ cls.block_use_of_hybrid_exec(executor_names_raw)
+
+ executor_names = []
+ for name in executor_names_raw:
+ if len(split_name := name.split(":")) == 1:
+ name = split_name[0]
+ # Check if this is an alias for a core airflow executor, module
+ # paths won't be provided by the user in that case.
+ if core_executor_module := cls.executors.get(name):
+ executor_names.append(ExecutorName(alias=name, module_path=core_executor_module))
+ # Only a module path or plugin name was provided
+ else:
+ executor_names.append(ExecutorName(alias=None, module_path=name))
+ # An alias was provided with the module path
+ elif len(split_name) == 2:
+ # Ensure the user is not trying to override the existing aliases of any of the core
+ # executors by providing an alias along with the existing core airflow executor alias
+ # (e.g. my_local_exec_alias:LocalExecutor). Allowing this makes things unnecessarily
+ # complicated. Multiple Executors of the same type will be supported by a future multitenancy
+ # AIP.
+ # The module component should always be a module or plugin path.
+ if not split_name[1] or split_name[1] in CORE_EXECUTOR_NAMES:
+ raise AirflowConfigException(
+ f"Incorrectly formatted executor configuration: {name}\n"
+ "second portion of an executor configuration must be a module path"
+ )
+ else:
+ executor_names.append(ExecutorName(alias=split_name[0], module_path=split_name[1]))
+ else:
+ raise AirflowConfigException(f"Incorrectly formatted executor configuration: {name}")
+
+ # As of now, we do not allow duplicate executors.
+ # Add all module paths/plugin names to a set, since the actual code is what is unique
+ unique_modules = set([exec_name.module_path for exec_name in executor_names])
+ if len(unique_modules) < len(executor_names):
+ msg = (
+ "At least one executor was configured twice. Duplicate executors is not yet supported. "
Review Comment:
```suggestion
"At least one executor was configured twice. Duplicate executors are not yet supported. "
```
--
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
Re: [PR] Introduce mechanism to support multiple executor configuration [airflow]
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas commented on code in PR #37635:
URL: https://github.com/apache/airflow/pull/37635#discussion_r1503293491
##########
airflow/executors/executor_constants.py:
##########
@@ -16,6 +16,18 @@
# under the License.
from __future__ import annotations
+from enum import Enum, unique
+
+
+@unique
+class ConnectorSource(Enum):
Review Comment:
Moved here since it is now used in a couple modules and was causing import cycles. I'd consider an Enum a constant, curious what others think.
--
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
Re: [PR] Introduce mechanism to support multiple executor configuration [airflow]
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas commented on code in PR #37635:
URL: https://github.com/apache/airflow/pull/37635#discussion_r1503295194
##########
airflow/executors/executor_utils.py:
##########
@@ -0,0 +1,66 @@
+# 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.
+from __future__ import annotations
+
+from airflow.executors.executor_constants import CORE_EXECUTOR_NAMES, ConnectorSource
+from airflow.utils.log.logging_mixin import LoggingMixin
+
+
+class ExecutorName(LoggingMixin):
+ """Representation of an executor config/name."""
+
+ def __init__(self, module_path, alias=None):
+ self.module_path = module_path
+ self.alias = alias
+ self.set_connector_source()
+
+ def set_connector_source(self):
+ if self.alias in CORE_EXECUTOR_NAMES:
+ self.connector_source = ConnectorSource.CORE
+ # If there is only one dot, then this is likely a plugin. This is the best we can do
+ # to determine.
+ elif self.module_path.count(".") == 1:
+ self.log.debug(
+ "The executor name looks like the plugin path (executor_name=%s) due to having "
+ "just two period delimited parts. Treating executor as a plugin",
+ self.module_path,
+ )
+ self.connector_source = ConnectorSource.PLUGIN
Review Comment:
This isn't new code, this logic came from another place in the executor loader. I don't love it, but I also don't want to re-invent that wheel at the moment.
--
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
Re: [PR] Introduce mechanism to support multiple executor configuration [airflow]
Posted by "syedahsn (via GitHub)" <gi...@apache.org>.
syedahsn commented on code in PR #37635:
URL: https://github.com/apache/airflow/pull/37635#discussion_r1511880593
##########
airflow/executors/executor_loader.py:
##########
@@ -69,56 +71,181 @@ class ExecutorLoader:
}
@classmethod
- def get_default_executor_name(cls) -> str:
- """Return the default executor name from Airflow configuration.
+ def block_use_of_hybrid_exec(cls, executor_config: list):
+ """Raise an exception if the user tries to use multiple executors before the feature is complete.
- :return: executor name from Airflow configuration
+ This check is built into a method so that it can be easily mocked in unit tests.
+
+ :param executor_config: core.executor configuration value.
+ """
+ if len(executor_config) > 1 or ":" in "".join(executor_config):
+ raise AirflowConfigException(
+ "Configuring multiple executors and executor aliases are not yet supported!: "
+ f"{executor_config}"
+ )
+
+ @classmethod
+ def _get_executor_names(cls) -> list[ExecutorName]:
+ """Return the executor names from Airflow configuration.
+
+ :return: List of executor names from Airflow configuration
"""
from airflow.configuration import conf
- return conf.get_mandatory_value("core", "EXECUTOR")
+ if _executor_names:
+ return _executor_names
+
+ executor_names_raw = conf.get_mandatory_list_value("core", "EXECUTOR")
+
+ # AIP-61 is WIP. Unblock configuring multiple executors when the feature is ready to launch
+ cls.block_use_of_hybrid_exec(executor_names_raw)
+
+ executor_names = []
+ for name in executor_names_raw:
+ if len(split_name := name.split(":")) == 1:
+ name = split_name[0]
+ # Check if this is an alias for a core airflow executor, module
+ # paths won't be provided by the user in that case.
+ if core_executor_module := cls.executors.get(name):
+ executor_names.append(ExecutorName(alias=name, module_path=core_executor_module))
+ # Only a module path or plugin name was provided
+ else:
+ executor_names.append(ExecutorName(alias=None, module_path=name))
+ # An alias was provided with the module path
+ elif len(split_name) == 2:
+ # Ensure the user is not trying to override the existing aliases of any of the core
+ # executors by providing an alias along with the existing core airflow executor alias
+ # (e.g. my_local_exec_alias:LocalExecutor). Allowing this makes things unnecessarily
+ # complicated. Multiple Executors of the same type will be supported by a future multitenancy
+ # AIP.
+ # The module component should always be a module or plugin path.
+ if not split_name[1] or split_name[1] in CORE_EXECUTOR_NAMES:
+ raise AirflowConfigException(
+ f"Incorrectly formatted executor configuration: {name}\n"
+ "second portion of an executor configuration must be a module path"
+ )
+ else:
+ executor_names.append(ExecutorName(alias=split_name[0], module_path=split_name[1]))
+ else:
+ raise AirflowConfigException(f"Incorrectly formatted executor configuration: {name}")
+
+ # As of now, we do not allow duplicate executors.
+ # Add all module paths/plugin names to a set, since the actual code is what is unique
+ unique_modules = set([exec_name.module_path for exec_name in executor_names])
+ if len(unique_modules) < len(executor_names):
+ msg = (
+ "At least one executor was configured twice. Duplicate executors is not yet supported. "
+ "Please check your configuration again to correct the issue."
+ )
+ raise AirflowConfigException(msg)
+
+ # Populate some mappings for fast future lookups
+ for executor_name in executor_names:
+ # Executors will not always have aliases
+ if executor_name.alias:
+ _alias_to_executors[executor_name.alias] = executor_name
+ # All executors will have a module path
+ _module_to_executors[executor_name.module_path] = executor_name
+ # Cache the executor names, so the logic of this method only runs once
+ _executor_names.append(executor_name)
+
+ return executor_names
+
+ @classmethod
+ def get_default_executor_name(cls) -> ExecutorName:
+ """Return the default executor name from Airflow configuration.
+
+ :return: executor name from Airflow configuration
+ """
+ # The default executor is the first configured executor in the list
+ return cls._get_executor_names()[0]
@classmethod
def get_default_executor(cls) -> BaseExecutor:
"""Create a new instance of the configured executor if none exists and returns it."""
- if cls._default_executor is not None:
- return cls._default_executor
+ default_executor = cls.load_executor(cls.get_default_executor_name())
+
+ return default_executor
+
+ @classmethod
+ def init_executors(cls) -> list[BaseExecutor]:
+ """Create a new instance of all configured executors if not cached already."""
+ executor_names = cls._get_executor_names()
+ loaded_executors = []
+ for executor_name in executor_names:
+ loaded_executor = cls.load_executor(executor_name.module_path)
+ if executor_name.alias:
+ cls.executors[executor_name.alias] = executor_name.module_path
+ else:
+ cls.executors[loaded_executor.__class__.__name__] = executor_name.module_path
- return cls.load_executor(cls.get_default_executor_name())
+ loaded_executors.append(loaded_executor)
+
+ return loaded_executors
+
+ @classmethod
+ def lookup_executor_name_by_str(cls, executor_name_str: str) -> ExecutorName:
+ # lookup the executor by alias first, if not check if we're given a module path
+ if executor_name := _alias_to_executors.get(executor_name_str):
+ return executor_name
+ elif executor_name := _module_to_executors.get(executor_name_str):
+ return executor_name
+ else:
+ raise AirflowException(f"Unknown executor being loaded: {executor_name}")
@classmethod
- def load_executor(cls, executor_name: str) -> BaseExecutor:
+ def load_executor(cls, executor_name: ExecutorName | str) -> BaseExecutor:
"""
Load the executor.
This supports the following formats:
* by executor name for core executor
* by ``{plugin_name}.{class_name}`` for executor from plugins
* by import path.
+ * by ExecutorName object specification
:return: an instance of executor class via executor_name
"""
- if executor_name == CELERY_KUBERNETES_EXECUTOR:
- return cls.__load_celery_kubernetes_executor()
- elif executor_name == LOCAL_KUBERNETES_EXECUTOR:
- return cls.__load_local_kubernetes_executor()
+ if isinstance(executor_name, str):
+ _executor_name = cls.lookup_executor_name_by_str(executor_name)
+ else:
+ _executor_name = executor_name
+
+ # Check if the executor has been previously loaded. Avoid constructing a new object
+ if _executor_name in _loaded_executors:
+ return _loaded_executors[_executor_name]
try:
- executor_cls, import_source = cls.import_executor_cls(executor_name)
- log.debug("Loading executor %s from %s", executor_name, import_source.value)
+ if _executor_name.alias == CELERY_KUBERNETES_EXECUTOR:
Review Comment:
Just wondering why these executors are mentioned here by name. Is this introducing coupling?
--
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
Re: [PR] Introduce mechanism to support multiple executor configuration [airflow]
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas commented on PR #37635:
URL: https://github.com/apache/airflow/pull/37635#issuecomment-1960350732
Converting to draft while I work on test failures :heart:
--
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
Re: [PR] Introduce mechanism to support multiple executor configuration [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on code in PR #37635:
URL: https://github.com/apache/airflow/pull/37635#discussion_r1510022352
##########
airflow/executors/executor_constants.py:
##########
@@ -16,6 +16,18 @@
# under the License.
from __future__ import annotations
+from enum import Enum, unique
+
+
+@unique
+class ConnectorSource(Enum):
Review Comment:
Yes. Same thinking, and yes - extracting to shared module is a good idea.
--
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
Re: [PR] Introduce mechanism to support multiple executor configuration [airflow]
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas commented on code in PR #37635:
URL: https://github.com/apache/airflow/pull/37635#discussion_r1503295868
##########
tests/cli/commands/test_celery_command.py:
##########
@@ -69,6 +70,7 @@ class TestCeleryStopCommand:
@classmethod
def setup_class(cls):
with conf_vars({("core", "executor"): "CeleryExecutor"}):
+ importlib.reload(executor_loader)
Review Comment:
We do more caching in the executor loader now, use `reload` to get fresh module state. Similar to cli_parser.
--
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
Re: [PR] Introduce mechanism to support multiple executor configuration [airflow]
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas commented on code in PR #37635:
URL: https://github.com/apache/airflow/pull/37635#discussion_r1513596587
##########
airflow/executors/executor_loader.py:
##########
@@ -69,56 +71,181 @@ class ExecutorLoader:
}
@classmethod
- def get_default_executor_name(cls) -> str:
- """Return the default executor name from Airflow configuration.
+ def block_use_of_hybrid_exec(cls, executor_config: list):
+ """Raise an exception if the user tries to use multiple executors before the feature is complete.
- :return: executor name from Airflow configuration
+ This check is built into a method so that it can be easily mocked in unit tests.
+
+ :param executor_config: core.executor configuration value.
+ """
+ if len(executor_config) > 1 or ":" in "".join(executor_config):
+ raise AirflowConfigException(
+ "Configuring multiple executors and executor aliases are not yet supported!: "
+ f"{executor_config}"
+ )
+
+ @classmethod
+ def _get_executor_names(cls) -> list[ExecutorName]:
+ """Return the executor names from Airflow configuration.
+
+ :return: List of executor names from Airflow configuration
"""
from airflow.configuration import conf
- return conf.get_mandatory_value("core", "EXECUTOR")
+ if _executor_names:
+ return _executor_names
+
+ executor_names_raw = conf.get_mandatory_list_value("core", "EXECUTOR")
+
+ # AIP-61 is WIP. Unblock configuring multiple executors when the feature is ready to launch
+ cls.block_use_of_hybrid_exec(executor_names_raw)
+
+ executor_names = []
+ for name in executor_names_raw:
+ if len(split_name := name.split(":")) == 1:
+ name = split_name[0]
+ # Check if this is an alias for a core airflow executor, module
+ # paths won't be provided by the user in that case.
+ if core_executor_module := cls.executors.get(name):
+ executor_names.append(ExecutorName(alias=name, module_path=core_executor_module))
+ # Only a module path or plugin name was provided
+ else:
+ executor_names.append(ExecutorName(alias=None, module_path=name))
+ # An alias was provided with the module path
+ elif len(split_name) == 2:
+ # Ensure the user is not trying to override the existing aliases of any of the core
+ # executors by providing an alias along with the existing core airflow executor alias
+ # (e.g. my_local_exec_alias:LocalExecutor). Allowing this makes things unnecessarily
+ # complicated. Multiple Executors of the same type will be supported by a future multitenancy
+ # AIP.
+ # The module component should always be a module or plugin path.
+ if not split_name[1] or split_name[1] in CORE_EXECUTOR_NAMES:
+ raise AirflowConfigException(
+ f"Incorrectly formatted executor configuration: {name}\n"
+ "second portion of an executor configuration must be a module path"
+ )
+ else:
+ executor_names.append(ExecutorName(alias=split_name[0], module_path=split_name[1]))
+ else:
+ raise AirflowConfigException(f"Incorrectly formatted executor configuration: {name}")
+
+ # As of now, we do not allow duplicate executors.
+ # Add all module paths/plugin names to a set, since the actual code is what is unique
+ unique_modules = set([exec_name.module_path for exec_name in executor_names])
+ if len(unique_modules) < len(executor_names):
+ msg = (
+ "At least one executor was configured twice. Duplicate executors is not yet supported. "
+ "Please check your configuration again to correct the issue."
+ )
+ raise AirflowConfigException(msg)
+
+ # Populate some mappings for fast future lookups
+ for executor_name in executor_names:
+ # Executors will not always have aliases
+ if executor_name.alias:
+ _alias_to_executors[executor_name.alias] = executor_name
+ # All executors will have a module path
+ _module_to_executors[executor_name.module_path] = executor_name
+ # Cache the executor names, so the logic of this method only runs once
+ _executor_names.append(executor_name)
+
+ return executor_names
+
+ @classmethod
+ def get_default_executor_name(cls) -> ExecutorName:
+ """Return the default executor name from Airflow configuration.
+
+ :return: executor name from Airflow configuration
+ """
+ # The default executor is the first configured executor in the list
+ return cls._get_executor_names()[0]
@classmethod
def get_default_executor(cls) -> BaseExecutor:
"""Create a new instance of the configured executor if none exists and returns it."""
- if cls._default_executor is not None:
- return cls._default_executor
+ default_executor = cls.load_executor(cls.get_default_executor_name())
+
+ return default_executor
+
+ @classmethod
+ def init_executors(cls) -> list[BaseExecutor]:
+ """Create a new instance of all configured executors if not cached already."""
+ executor_names = cls._get_executor_names()
+ loaded_executors = []
+ for executor_name in executor_names:
+ loaded_executor = cls.load_executor(executor_name.module_path)
+ if executor_name.alias:
+ cls.executors[executor_name.alias] = executor_name.module_path
+ else:
+ cls.executors[loaded_executor.__class__.__name__] = executor_name.module_path
- return cls.load_executor(cls.get_default_executor_name())
+ loaded_executors.append(loaded_executor)
+
+ return loaded_executors
+
+ @classmethod
+ def lookup_executor_name_by_str(cls, executor_name_str: str) -> ExecutorName:
+ # lookup the executor by alias first, if not check if we're given a module path
+ if executor_name := _alias_to_executors.get(executor_name_str):
+ return executor_name
+ elif executor_name := _module_to_executors.get(executor_name_str):
+ return executor_name
+ else:
+ raise AirflowException(f"Unknown executor being loaded: {executor_name}")
@classmethod
- def load_executor(cls, executor_name: str) -> BaseExecutor:
+ def load_executor(cls, executor_name: ExecutorName | str) -> BaseExecutor:
"""
Load the executor.
This supports the following formats:
* by executor name for core executor
* by ``{plugin_name}.{class_name}`` for executor from plugins
* by import path.
+ * by ExecutorName object specification
:return: an instance of executor class via executor_name
"""
- if executor_name == CELERY_KUBERNETES_EXECUTOR:
- return cls.__load_celery_kubernetes_executor()
- elif executor_name == LOCAL_KUBERNETES_EXECUTOR:
- return cls.__load_local_kubernetes_executor()
+ if isinstance(executor_name, str):
+ _executor_name = cls.lookup_executor_name_by_str(executor_name)
+ else:
+ _executor_name = executor_name
+
+ # Check if the executor has been previously loaded. Avoid constructing a new object
+ if _executor_name in _loaded_executors:
+ return _loaded_executors[_executor_name]
try:
- executor_cls, import_source = cls.import_executor_cls(executor_name)
- log.debug("Loading executor %s from %s", executor_name, import_source.value)
+ if _executor_name.alias == CELERY_KUBERNETES_EXECUTOR:
Review Comment:
Great question! These are the old style hybrid executors, concrete classes that combine two executors. They need to be treated differently because they're not natively supported.
This aip-61 project itself will allow us to deprecate these!
--
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
Re: [PR] Introduce mechanism to support multiple executor configuration [airflow]
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas commented on PR #37635:
URL: https://github.com/apache/airflow/pull/37635#issuecomment-1986300785
@potiuk
Do you have time to have another look and approve? I've made your suggestion and rebased and tests are green
--
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
Re: [PR] Introduce mechanism to support multiple executor configuration [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk merged PR #37635:
URL: https://github.com/apache/airflow/pull/37635
--
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
Re: [PR] Introduce mechanism to support multiple executor configuration [airflow]
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas commented on code in PR #37635:
URL: https://github.com/apache/airflow/pull/37635#discussion_r1503294232
##########
airflow/executors/executor_loader.py:
##########
@@ -69,56 +71,181 @@ class ExecutorLoader:
}
@classmethod
- def get_default_executor_name(cls) -> str:
- """Return the default executor name from Airflow configuration.
+ def block_use_of_hybrid_exec(cls, executor_config: list):
+ """Raise an exception if the user tries to use multiple executors before the feature is complete.
- :return: executor name from Airflow configuration
+ This check is built into a method so that it can be easily mocked in unit tests.
+
+ :param executor_config: core.executor configuration value.
+ """
+ if len(executor_config) > 1 or ":" in "".join(executor_config):
+ raise AirflowConfigException(
+ "Configuring multiple executors and executor aliases are not yet supported!: "
+ f"{executor_config}"
+ )
+
+ @classmethod
+ def _get_executor_names(cls) -> list[ExecutorName]:
+ """Return the executor names from Airflow configuration.
+
+ :return: List of executor names from Airflow configuration
"""
from airflow.configuration import conf
- return conf.get_mandatory_value("core", "EXECUTOR")
+ if _executor_names:
+ return _executor_names
+
+ executor_names_raw = conf.get_mandatory_list_value("core", "EXECUTOR")
+
+ # AIP-61 is WIP. Unblock configuring multiple executors when the feature is ready to launch
+ cls.block_use_of_hybrid_exec(executor_names_raw)
+
+ executor_names = []
+ for name in executor_names_raw:
+ if len(split_name := name.split(":")) == 1:
+ name = split_name[0]
+ # Check if this is an alias for a core airflow executor, module
+ # paths won't be provided by the user in that case.
+ if core_executor_module := cls.executors.get(name):
+ executor_names.append(ExecutorName(alias=name, module_path=core_executor_module))
+ # Only a module path or plugin name was provided
+ else:
+ executor_names.append(ExecutorName(alias=None, module_path=name))
+ # An alias was provided with the module path
+ elif len(split_name) == 2:
+ # Ensure the user is not trying to override the existing aliases of any of the core
+ # executors by providing an alias along with the existing core airflow executor alias
+ # (e.g. my_local_exec_alias:LocalExecutor). Allowing this makes things unnecessarily
+ # complicated. Multiple Executors of the same type will be supported by a future multitenancy
+ # AIP.
+ # The module component should always be a module or plugin path.
+ if not split_name[1] or split_name[1] in CORE_EXECUTOR_NAMES:
+ raise AirflowConfigException(
+ f"Incorrectly formatted executor configuration: {name}\n"
+ "second portion of an executor configuration must be a module path"
+ )
+ else:
+ executor_names.append(ExecutorName(alias=split_name[0], module_path=split_name[1]))
+ else:
+ raise AirflowConfigException(f"Incorrectly formatted executor configuration: {name}")
+
+ # As of now, we do not allow duplicate executors.
+ # Add all module paths/plugin names to a set, since the actual code is what is unique
+ unique_modules = set([exec_name.module_path for exec_name in executor_names])
+ if len(unique_modules) < len(executor_names):
+ msg = (
+ "At least one executor was configured twice. Duplicate executors is not yet supported. "
+ "Please check your configuration again to correct the issue."
+ )
+ raise AirflowConfigException(msg)
+
+ # Populate some mappings for fast future lookups
+ for executor_name in executor_names:
+ # Executors will not always have aliases
+ if executor_name.alias:
+ _alias_to_executors[executor_name.alias] = executor_name
+ # All executors will have a module path
+ _module_to_executors[executor_name.module_path] = executor_name
+ # Cache the executor names, so the logic of this method only runs once
+ _executor_names.append(executor_name)
+
+ return executor_names
+
+ @classmethod
+ def get_default_executor_name(cls) -> ExecutorName:
+ """Return the default executor name from Airflow configuration.
+
+ :return: executor name from Airflow configuration
+ """
+ # The default executor is the first configured executor in the list
+ return cls._get_executor_names()[0]
@classmethod
def get_default_executor(cls) -> BaseExecutor:
"""Create a new instance of the configured executor if none exists and returns it."""
- if cls._default_executor is not None:
- return cls._default_executor
+ default_executor = cls.load_executor(cls.get_default_executor_name())
+
+ return default_executor
+
+ @classmethod
+ def init_executors(cls) -> list[BaseExecutor]:
Review Comment:
Not used by any code here. Will be used by the scheduler and backfill job runners (or more precisely the jobs within).
--
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