Fix for LocalKubernetesExecutor scheduler is not serving logs

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()

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

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

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:

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

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 = []

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")

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()
 

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()

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"]

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()

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")