This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a commit to branch v3-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-4-test by this push:
new 63f9ab0e144 [v3-4-test] Read a team's running task logs through the
team's executor (#75255) (#75408)
63f9ab0e144 is described below
commit 63f9ab0e14452865da7501eb78cc3e4aac958c2f
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Thu Oct 22 23:48:43 2026 +0530
[v3-4-test] Read a team's running task logs through the team's executor
(#75255) (#75408)
In multi-team mode, the API server reads a running task's logs through the
global default executor, or looks up the task's named executor in the
global scope only. A team's executor has its own config, such as the
cluster its KubernetesExecutor pods run on, so a team task's live logs came
from the wrong executor: nothing from a global LocalExecutor, or "Cannot
find pod" from a global KubernetesExecutor on another cluster.
Resolve the executor the same way the scheduler does: the team's executor,
and a global one if the team doesn't have it.
(cherry picked from commit 1e3a23293b834db9c362c123bf9580ddef81800a)
Co-authored-by: Sean Ghaeli <[email protected]>
---
.../src/airflow/executors/executor_loader.py | 36 ++++++++++++++++
.../src/airflow/utils/log/file_task_handler.py | 17 ++++++--
.../tests/unit/executors/test_executor_loader.py | 48 +++++++++++++++++++++-
airflow-core/tests/unit/utils/test_log_handlers.py | 30 ++++++++++++++
4 files changed, 127 insertions(+), 4 deletions(-)
diff --git a/airflow-core/src/airflow/executors/executor_loader.py
b/airflow-core/src/airflow/executors/executor_loader.py
index 79555d49daf..101a5fc48e6 100644
--- a/airflow-core/src/airflow/executors/executor_loader.py
+++ b/airflow-core/src/airflow/executors/executor_loader.py
@@ -38,6 +38,8 @@ from airflow.models.team import Team
log = structlog.get_logger(__name__)
if TYPE_CHECKING:
+ from collections.abc import Iterable
+
from airflow.executors.base_executor import BaseExecutor
@@ -332,6 +334,40 @@ class ExecutorLoader:
return executor_name
raise UnknownExecutorException(f"Unknown executor being loaded:
{executor_name_str}")
+ @classmethod
+ def resolve_executor_name(
+ cls,
+ executor: str | None,
+ team_name: str | None = None,
+ *,
+ executor_names: Iterable[ExecutorName] | None = None,
+ ) -> ExecutorName:
+ """
+ Return the executor name for a task's ``executor`` and team.
+
+ The team's own executor wins over a global one, and a named executor
that doesn't match raises
+ instead of falling back to the default.
+
+ :param executor: alias, module path or class name, or None for the
default
+ :param team_name: the task's team
+ :param executor_names: names to search, the configured ones by default
+ """
+ global_match = None
+ for name in cls._get_executor_names() if executor_names is None else
executor_names:
+ if executor is not None and executor not in (
+ name.alias,
+ name.module_path,
+ name.module_path.split(".")[-1],
+ ):
+ continue
+ if team_name and name.team_name == team_name:
+ return name
+ if name.team_name is None and global_match is None:
+ global_match = name
+ if global_match is None:
+ raise UnknownExecutorException(f"Unknown executor {executor!r} for
team {team_name!r}")
+ return global_match
+
@classmethod
def load_executor(cls, executor_name: ExecutorName | str | None) ->
BaseExecutor:
"""
diff --git a/airflow-core/src/airflow/utils/log/file_task_handler.py
b/airflow-core/src/airflow/utils/log/file_task_handler.py
index a2cd4c29322..f20db27722b 100644
--- a/airflow-core/src/airflow/utils/log/file_task_handler.py
+++ b/airflow-core/src/airflow/utils/log/file_task_handler.py
@@ -531,13 +531,24 @@ class FileTaskHandler(logging.Handler):
:return: executor of the task instance
"""
executor_name = ti.executor or self.DEFAULT_EXECUTOR_KEY
- executor = self.executor_instances.get(executor_name)
+ team_name = None
+ if conf.getboolean("core", "multi_team"):
+ from airflow.models.dag import DagModel
+
+ team_name = DagModel.get_team_name(ti.dag_id)
+ # A team's executor has its own config, such as the cluster its pods
run on.
+ cache_key = f"{team_name}:{executor_name}" if team_name else
executor_name
+ executor = self.executor_instances.get(cache_key)
if executor is None:
- if executor_name == self.DEFAULT_EXECUTOR_KEY:
+ if team_name:
+ executor = ExecutorLoader.load_executor(
+ ExecutorLoader.resolve_executor_name(ti.executor,
team_name)
+ )
+ elif executor_name == self.DEFAULT_EXECUTOR_KEY:
executor = ExecutorLoader.get_default_executor()
else:
executor = ExecutorLoader.load_executor(executor_name)
- self.executor_instances[executor_name] = executor
+ self.executor_instances[cache_key] = executor
return executor
diff --git a/airflow-core/tests/unit/executors/test_executor_loader.py
b/airflow-core/tests/unit/executors/test_executor_loader.py
index c0c0a2f6331..550e0fa708f 100644
--- a/airflow-core/tests/unit/executors/test_executor_loader.py
+++ b/airflow-core/tests/unit/executors/test_executor_loader.py
@@ -21,7 +21,7 @@ from unittest.mock import patch
import pytest
-from airflow.exceptions import AirflowConfigException
+from airflow.exceptions import AirflowConfigException, UnknownExecutorException
from airflow.executors import executor_loader
from airflow.executors.executor_loader import ConnectorSource, ExecutorName
from airflow.executors.local_executor import LocalExecutor
@@ -37,6 +37,13 @@ celery_executor =
pytest.importorskip("airflow.providers.celery.executors.celery
ecs_executor =
pytest.importorskip("airflow.providers.amazon.aws.executors.ecs.ecs_executor")
+TEAM_EXECUTORS_CONFIG = {
+ ("core", "multi_team"): "True",
+ ("core", "executor"):
"global_celery_exec:CeleryExecutor,LocalExecutor,pkg_a.mod.MyExec,pkg_b.mod.MyExec;"
+ "team_a=team_celery_exec:CeleryExecutor;team_b=KubernetesExecutor",
+}
+
+
@pytest.mark.usefixtures("clean_executor_loader")
class TestExecutorLoader:
def test_no_executor_configured(self):
@@ -749,6 +756,45 @@ class TestExecutorLoader:
)
assert result.alias == "LocalExecutor"
+ @pytest.mark.parametrize(
+ ("executor", "team_name", "expected"),
+ [
+ pytest.param("CeleryExecutor", "team_a", ("team_celery_exec",
"team_a"), id="classname-to-team"),
+ pytest.param(
+
"airflow.providers.celery.executors.celery_executor.CeleryExecutor",
+ "team_a",
+ ("team_celery_exec", "team_a"),
+ id="module-path-to-team",
+ ),
+ pytest.param("team_celery_exec", "team_a", ("team_celery_exec",
"team_a"), id="team-alias"),
+ pytest.param("global_celery_exec", "team_a",
("global_celery_exec", None), id="global-alias"),
+ pytest.param("LocalExecutor", "team_a", ("LocalExecutor", None),
id="global-fallback"),
+ pytest.param(None, "team_a", ("team_celery_exec", "team_a"),
id="team-default"),
+ pytest.param(None, "team_c", ("global_celery_exec", None),
id="team-without-executors-default"),
+ pytest.param(None, None, ("global_celery_exec", None),
id="global-default"),
+ pytest.param("MyExec", None, ("pkg_a.mod.MyExec", None),
id="first-of-several-global-matches"),
+ ],
+ )
+ @conf_vars(TEAM_EXECUTORS_CONFIG)
+ @patch.object(executor_loader.ExecutorLoader,
"_validate_teams_exist_in_database")
+ def test_resolve_executor_name(self, _, executor, team_name, expected):
+ name = executor_loader.ExecutorLoader.resolve_executor_name(executor,
team_name)
+ assert (name.alias or name.module_path, name.team_name) == expected
+
+ @pytest.mark.parametrize(
+ ("executor", "team_name"),
+ [
+ pytest.param("team_celery_exec", None,
id="team-alias-without-team"),
+ pytest.param("KubernetesExecutor", "team_a",
id="other-teams-executor"),
+ pytest.param("EcsExecutor", "team_a",
id="unknown-does-not-fall-back-to-default"),
+ ],
+ )
+ @conf_vars(TEAM_EXECUTORS_CONFIG)
+ @patch.object(executor_loader.ExecutorLoader,
"_validate_teams_exist_in_database")
+ def test_resolve_executor_name_no_match(self, _, executor, team_name):
+ with pytest.raises(UnknownExecutorException):
+ executor_loader.ExecutorLoader.resolve_executor_name(executor,
team_name)
+
def test_get_executor_names_default_validates_teams(self):
"""Test that get_executor_names validates teams by default."""
with patch.object(executor_loader.Team, "get_all_team_names") as
mock_get_team_names:
diff --git a/airflow-core/tests/unit/utils/test_log_handlers.py
b/airflow-core/tests/unit/utils/test_log_handlers.py
index a29e73b0ec0..f779ac8b44b 100644
--- a/airflow-core/tests/unit/utils/test_log_handlers.py
+++ b/airflow-core/tests/unit/utils/test_log_handlers.py
@@ -336,6 +336,36 @@ class TestFileTaskLogHandler:
mock_get_default_executor.assert_not_called()
mock_load_executor.assert_called_once_with(executor_name)
+ @pytest.mark.parametrize(
+ ("dag_id", "ti_executor", "expected"),
+ [
+ pytest.param("team_a_dag", None, ("LocalExecutor", "team_a"),
id="team-default"),
+ pytest.param("team_b_dag", None, ("CeleryExecutor", "team_b"),
id="other-team-default"),
+ pytest.param("team_a_dag", "LocalExecutor", ("LocalExecutor",
"team_a"), id="team-named"),
+ pytest.param("team_a_dag", "CeleryExecutor", ("CeleryExecutor",
None), id="global-named"),
+ pytest.param("global_dag", None, ("CeleryExecutor", None),
id="no-team"),
+ ],
+ )
+ @conf_vars(
+ {
+ ("core", "multi_team"): "True",
+ ("core", "executor"):
"CeleryExecutor;team_a=LocalExecutor;team_b=CeleryExecutor",
+ }
+ )
+ def test_get_executor_uses_the_teams_executor(self, dag_id, ti_executor,
expected, clean_executor_loader):
+ teams = {"team_a_dag": "team_a", "team_b_dag": "team_b"}
+ ti = mock.MagicMock(dag_id=dag_id, executor=ti_executor)
+ handler = FileTaskHandler("")
+ handler.executor_instances = {}
+ with (
+ mock.patch.object(executor_loader.ExecutorLoader,
"_validate_teams_exist_in_database"),
+ mock.patch("airflow.models.dag.DagModel.get_team_name",
side_effect=teams.get),
+ mock.patch.object(executor_loader.ExecutorLoader, "load_executor")
as load_executor,
+ ):
+ handler._get_executor(ti)
+ (executor_name,), _ = load_executor.call_args
+ assert (executor_name.alias, executor_name.team_name) == expected
+
def test_file_task_handler_running(self, dag_maker):
def task_callable(ti):
ti.log.info("test")
