Bring back serve_logs to be in the core (#49031)

potiuk · kaxil · web-flow · commit 00aec39a4c49 · 2025-04-10T14:35:54.000+05:30
The `serve_logs` has been moved in #48457 to fab provider by mistake. It should remain in the airflow-core. We still depend on flask in the core (but as a next step we can replace it by starlette and unicorn). Fixes: #49028 Co-authored-by: Kaxil Naik <kaxilnaik@gmail.com>
diff --git a/airflow-core/pyproject.toml b/airflow-core/pyproject.toml
@@ -84,7 +84,10 @@ dependencies = [
     # 0.115.10 fastapi was a bad release that broke our API's and static checks.
     # Related fastapi issue here: https://github.com/fastapi/fastapi/discussions/13431
     "fastapi[standard]>=0.115.0,!=0.115.10",
+    # We could get rid of flask and gunicorn if we replace serve_logs with a starlette + unicorn
+    "flask>=2.1.1",
     "gitpython>=3.1.40",
+    # We could get rid of flask and gunicorn if we replace serve_logs with a starlette + unicorn
     "gunicorn>=20.1.0",
     "httpx>=0.25.0",
     'importlib_metadata>=6.5;python_version<"3.12"',
diff --git a/airflow-core/src/airflow/cli/commands/scheduler_command.py b/airflow-core/src/airflow/cli/commands/scheduler_command.py
@@ -29,7 +29,6 @@
 from airflow.executors.executor_loader import ExecutorLoader
 from airflow.jobs.job import Job, run_job
 from airflow.jobs.scheduler_job_runner import SchedulerJobRunner
-from airflow.providers.celery.version_compat import AIRFLOW_V_3_0_PLUS
 from airflow.utils import cli as cli_utils
 from airflow.utils.providers_configuration_loader import providers_configuration_loaded
 from airflow.utils.scheduler_health import serve_health_check
@@ -60,18 +59,7 @@ def scheduler(args: Namespace):
 
 @contextmanager
 def _serve_logs(skip_serve_logs: bool = False):
-    """Start serve_logs sub-process."""
-    if AIRFLOW_V_3_0_PLUS:
-        try:
-            from airflow.providers.fab.www.serve_logs import serve_logs
-        except ImportError:
-            raise ImportError(
-                "Celery requires FAB provider to be installed in order to run this command. "
-                "Please install the FAB provider by running: "
-                "pip install apache-airflow-providers-celery[fab]"
-            )
-    else:
-        from airflow.utils.serve_logs import serve_logs  # type: ignore[no-redef]
+    from airflow.utils.serve_logs import serve_logs
 
     sub_proc = None
     executor_class, _ = ExecutorLoader.import_default_executor_cls()
diff --git a/airflow-core/src/airflow/cli/commands/triggerer_command.py b/airflow-core/src/airflow/cli/commands/triggerer_command.py
@@ -28,25 +28,14 @@
 from airflow.configuration import conf
 from airflow.jobs.job import Job, run_job
 from airflow.jobs.triggerer_job_runner import TriggererJobRunner
-from airflow.providers.celery.version_compat import AIRFLOW_V_3_0_PLUS
 from airflow.utils import cli as cli_utils
 from airflow.utils.providers_configuration_loader import providers_configuration_loaded
 
 
 @contextmanager
 def _serve_logs(skip_serve_logs: bool = False) -> Generator[None, None, None]:
     """Start serve_logs sub-process."""
-    if AIRFLOW_V_3_0_PLUS:
-        try:
-            from airflow.providers.fab.www.serve_logs import serve_logs
-        except ImportError:
-            raise ImportError(
-                "Celery requires FAB provider to be installed in order to run this command. "
-                "Please install the FAB provider by running: "
-                "pip install apache-airflow-providers-celery[fab]"
-            )
-    else:
-        from airflow.utils.serve_logs import serve_logs  # type: ignore[no-redef]
+    from airflow.utils.serve_logs import serve_logs
 
     sub_proc = None
     if skip_serve_logs is False:
diff --git a/airflow-core/src/airflow/utils/serve_logs.py b/airflow-core/src/airflow/utils/serve_logs.py
@@ -21,6 +21,7 @@
 import logging
 import os
 import socket
+import sys
 from collections import namedtuple
 
 import gunicorn.app.base
@@ -32,7 +33,6 @@
     InvalidIssuedAtError,
     InvalidSignatureError,
 )
-from setproctitle import setproctitle
 from werkzeug.exceptions import HTTPException
 
 from airflow.api_fastapi.auth.tokens import JWTValidator, get_signing_key
@@ -169,6 +169,13 @@ def load(self):
 
 def serve_logs(port=None):
     """Serve logs generated by Worker."""
+    # setproctitle causes issue on Mac OS: https://github.com/benoitc/gunicorn/issues/3021
+    os_type = sys.platform
+    if os_type == "darwin":
+        logger.debug("Mac OS detected, skipping setproctitle")
+    else:
+        from setproctitle import setproctitle
+
     setproctitle("airflow serve-logs")
     wsgi_app = create_app()
 
diff --git a/airflow-core/tests/unit/cli/commands/test_scheduler_command.py b/airflow-core/tests/unit/cli/commands/test_scheduler_command.py
@@ -25,8 +25,8 @@
 from airflow.cli import cli_parser
 from airflow.cli.commands import scheduler_command
 from airflow.executors import executor_loader
-from airflow.providers.fab.www.serve_logs import serve_logs
 from airflow.utils.scheduler_health import serve_health_check
+from airflow.utils.serve_logs import serve_logs
 
 from tests_common.test_utils.config import conf_vars
 
diff --git a/airflow-core/tests/unit/utils/test_serve_logs.py b/airflow-core/tests/unit/utils/test_serve_logs.py
@@ -26,8 +26,8 @@
 
 from airflow.api_fastapi.auth.tokens import JWTGenerator
 from airflow.config_templates.airflow_local_settings import DEFAULT_LOGGING_CONFIG
-from airflow.providers.fab.www.serve_logs import create_app
 from airflow.utils import timezone
+from airflow.utils.serve_logs import create_app
 
 from tests_common.test_utils.config import conf_vars
 
@@ -81,7 +81,7 @@ def sample_log(request, tmp_path):
         base_log_dir = Path(DEFAULT_LOGGING_CONFIG["handlers"]["task"]["base_log_folder"])
     else:
         raise ValueError(f"Unknown client fixture: {client}")
-
+    base_log_dir.mkdir(exist_ok=True, parents=True)
     f = base_log_dir.joinpath("sample.log")
     f.write_text(LOG_DATA)
     return f
diff --git a/dev/breeze/tests/test_selective_checks.py b/dev/breeze/tests/test_selective_checks.py
@@ -2157,7 +2157,7 @@ def test_upgrade_to_newer_dependencies(
         ),
         pytest.param(
             ("providers/celery/src/airflow/providers/celery/file.py",),
-            {"docs-list-as-string": "celery cncf.kubernetes fab"},
+            {"docs-list-as-string": "celery cncf.kubernetes"},
             id="Celery python files changed",
         ),
         pytest.param(
diff --git a/providers/celery/README.rst b/providers/celery/README.rst
@@ -75,7 +75,6 @@ You can install such cross-provider dependencies when installing from PyPI. For
 Dependent package                                                                                                       Extra
 ======================================================================================================================  ===================
 `apache-airflow-providers-cncf-kubernetes <https://airflow.apache.org/docs/apache-airflow-providers-cncf-kubernetes>`_  ``cncf.kubernetes``
-`apache-airflow-providers-fab <https://airflow.apache.org/docs/apache-airflow-providers-fab>`_                          ``fab``
 ======================================================================================================================  ===================
 
 The changelog for the provider package can be found in the
diff --git a/providers/celery/pyproject.toml b/providers/celery/pyproject.toml
@@ -72,17 +72,13 @@ dependencies = [
 "cncf.kubernetes" = [
     "apache-airflow-providers-cncf-kubernetes>=7.4.0",
 ]
-"fab" = [
-    "apache-airflow-providers-fab>=2.0.0",
-]
 
 [dependency-groups]
 dev = [
     "apache-airflow",
     "apache-airflow-task-sdk",
     "apache-airflow-devel-common",
     "apache-airflow-providers-cncf-kubernetes",
-    "apache-airflow-providers-fab",
     # Additional devel dependencies (do not remove this line and add extra development dependencies)
 ]
 
diff --git a/providers/celery/src/airflow/providers/celery/cli/celery_command.py b/providers/celery/src/airflow/providers/celery/cli/celery_command.py
@@ -107,17 +107,7 @@ def flower(args):
 @contextmanager
 def _serve_logs(skip_serve_logs: bool = False):
     """Start serve_logs sub-process."""
-    if AIRFLOW_V_3_0_PLUS:
-        try:
-            from airflow.providers.fab.www.serve_logs import serve_logs
-        except ImportError:
-            raise ImportError(
-                "Celery requires FAB provider to be installed in order to run this command. "
-                "Please install the FAB provider by running: "
-                "pip install apache-airflow-providers-celery[fab]"
-            )
-    else:
-        from airflow.utils.serve_logs import serve_logs  # type: ignore[no-redef]
+    from airflow.utils.serve_logs import serve_logs
 
     sub_proc = None
     if skip_serve_logs is False:
