diff --git a/airflow/cli/commands/scheduler_command.py b/airflow/cli/commands/scheduler_command.py index fccb65210e4db..3b06c8510b90e 100644 --- a/airflow/cli/commands/scheduler_command.py +++ b/airflow/cli/commands/scheduler_command.py @@ -81,7 +81,7 @@ def _serve_logs(skip_serve_logs: bool = False): sub_proc = None executor_class, _ = ExecutorLoader.import_default_executor_cls() - if executor_class.is_local: + if executor_class.serve_logs: if skip_serve_logs is False: sub_proc = Process(target=serve_logs) sub_proc.start() diff --git a/airflow/executors/base_executor.py b/airflow/executors/base_executor.py index a56b20ec840ad..02715a84ad93b 100644 --- a/airflow/executors/base_executor.py +++ b/airflow/executors/base_executor.py @@ -114,6 +114,8 @@ class BaseExecutor(LoggingMixin): is_local: bool = False + serve_logs: bool = False + def __init__(self, parallelism: int = PARALLELISM): super().__init__() self.parallelism: int = parallelism diff --git a/airflow/executors/celery_kubernetes_executor.py b/airflow/executors/celery_kubernetes_executor.py index 474c6d4a8be04..af3d42a816718 100644 --- a/airflow/executors/celery_kubernetes_executor.py +++ b/airflow/executors/celery_kubernetes_executor.py @@ -48,6 +48,8 @@ class CeleryKubernetesExecutor(LoggingMixin): is_local: bool = False + serve_logs: bool = False + def __init__(self, celery_executor: CeleryExecutor, kubernetes_executor: KubernetesExecutor): super().__init__() self._job_id: int | None = None diff --git a/airflow/executors/local_executor.py b/airflow/executors/local_executor.py index 79f75475672e3..fc6eaf221be27 100644 --- a/airflow/executors/local_executor.py +++ b/airflow/executors/local_executor.py @@ -208,6 +208,8 @@ class LocalExecutor(BaseExecutor): is_local: bool = True supports_pickling: bool = False + serve_logs: bool = True + def __init__(self, parallelism: int = PARALLELISM): super().__init__(parallelism=parallelism) if self.parallelism < 0: diff --git a/airflow/executors/local_kubernetes_executor.py b/airflow/executors/local_kubernetes_executor.py index b582b37e8a9c8..c935a13af7ef0 100644 --- a/airflow/executors/local_kubernetes_executor.py +++ b/airflow/executors/local_kubernetes_executor.py @@ -48,6 +48,8 @@ class LocalKubernetesExecutor(LoggingMixin): is_local: bool = False + serve_logs: bool = True + def __init__(self, local_executor: LocalExecutor, kubernetes_executor: KubernetesExecutor): super().__init__() self._job_id: str | None = None diff --git a/airflow/executors/sequential_executor.py b/airflow/executors/sequential_executor.py index 466c9ea4aa844..d96c380243eba 100644 --- a/airflow/executors/sequential_executor.py +++ b/airflow/executors/sequential_executor.py @@ -47,6 +47,8 @@ class SequentialExecutor(BaseExecutor): supports_pickling: bool = False is_local: bool = True + serve_logs: bool = True + def __init__(self): super().__init__() self.commands_to_run = [] diff --git a/tests/cli/commands/test_scheduler_command.py b/tests/cli/commands/test_scheduler_command.py index 3779bdfd9a142..2ae3b4e9002c1 100644 --- a/tests/cli/commands/test_scheduler_command.py +++ b/tests/cli/commands/test_scheduler_command.py @@ -42,6 +42,7 @@ def setup_class(cls): ("LocalExecutor", True), ("SequentialExecutor", True), ("KubernetesExecutor", False), + ("LocalKubernetesExecutor", True), ], ) @mock.patch("airflow.cli.commands.scheduler_command.SchedulerJob") diff --git a/tests/executors/test_base_executor.py b/tests/executors/test_base_executor.py index af2894f891dcd..80650c83d41a3 100644 --- a/tests/executors/test_base_executor.py +++ b/tests/executors/test_base_executor.py @@ -44,6 +44,10 @@ def test_is_local_default_value(): assert not BaseExecutor.is_local +def test_serve_logs_default_value(): + assert not BaseExecutor.serve_logs + + def test_get_event_buffer(): executor = BaseExecutor() diff --git a/tests/executors/test_celery_kubernetes_executor.py b/tests/executors/test_celery_kubernetes_executor.py index d8568d93d10e4..361481417f843 100644 --- a/tests/executors/test_celery_kubernetes_executor.py +++ b/tests/executors/test_celery_kubernetes_executor.py @@ -40,6 +40,9 @@ def test_supports_sentry(self): def test_is_local_default_value(self): assert not CeleryKubernetesExecutor.is_local + def test_serve_logs_default_value(self): + assert not CeleryKubernetesExecutor.serve_logs + def test_queued_tasks(self): celery_executor_mock = mock.MagicMock() k8s_executor_mock = mock.MagicMock() diff --git a/tests/executors/test_local_executor.py b/tests/executors/test_local_executor.py index 4de9250f10c52..d292d20154b65 100644 --- a/tests/executors/test_local_executor.py +++ b/tests/executors/test_local_executor.py @@ -40,6 +40,9 @@ def test_supports_sentry(self): def test_is_local_default_value(self): assert LocalExecutor.is_local + def test_serve_logs_default_value(self): + assert LocalExecutor.serve_logs + @mock.patch("airflow.executors.local_executor.subprocess.check_call") def execution_parallelism_subprocess(self, mock_check_call, parallelism=0): success_command = ["airflow", "tasks", "run", "true", "some_parameter", "2020-10-07"] diff --git a/tests/executors/test_local_kubernetes_executor.py b/tests/executors/test_local_kubernetes_executor.py index 0d3d86bad052b..809b0277df575 100644 --- a/tests/executors/test_local_kubernetes_executor.py +++ b/tests/executors/test_local_kubernetes_executor.py @@ -35,6 +35,9 @@ def test_supports_sentry(self): def test_is_local_default_value(self): assert not LocalKubernetesExecutor.is_local + def test_serve_logs_default_value(self): + assert LocalKubernetesExecutor.serve_logs + def test_queued_tasks(self): local_executor_mock = mock.MagicMock() k8s_executor_mock = mock.MagicMock() diff --git a/tests/executors/test_sequential_executor.py b/tests/executors/test_sequential_executor.py index 61cd3ef238b80..d4d29c76f38fc 100644 --- a/tests/executors/test_sequential_executor.py +++ b/tests/executors/test_sequential_executor.py @@ -32,6 +32,9 @@ def test_supports_sentry(self): def test_is_local_default_value(self): assert SequentialExecutor.is_local + def test_serve_logs_default_value(self): + assert SequentialExecutor.serve_logs + @mock.patch("airflow.executors.sequential_executor.SequentialExecutor.sync") @mock.patch("airflow.executors.base_executor.BaseExecutor.trigger_tasks") @mock.patch("airflow.executors.base_executor.Stats.gauge")