Refactor serve_logs with FastAPI

airflow-core/pyproject.toml

@@ -80,10 +80,6 @@ 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",
-    # 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"',
     'importlib_metadata>=7.0;python_version>="3.12"',

airflow-core/src/airflow/utils/serve_logs/__init__.py

@@ -0,0 +1,22 @@
+# 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.utils.serve_logs.core import serve_logs
+from airflow.utils.serve_logs.log_server import create_app
+
+__all__ = ["serve_logs", "create_app"]

airflow-core/src/airflow/utils/serve_logs/core.py

@@ -0,0 +1,65 @@
+# 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.
+"""Serve logs process."""
+
+from __future__ import annotations
+
+import logging
+import socket
+import sys
+
+import uvicorn
+
+from airflow.configuration import conf
+
+logger = logging.getLogger(__name__)
+
+
+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")
+
+    port = port or conf.getint("logging", "WORKER_LOG_SERVER_PORT")
+
+    # If dual stack is available and IPV6_V6ONLY is not enabled on the socket
+    # then when IPV6 is bound to it will also bind to IPV4 automatically
+    if getattr(socket, "has_dualstack_ipv6", lambda: False)():
+        host = "::"  # ASGI uses `::` syntax for IPv6 binding instead of the `[::]` notation used in WSGI, while preserving the `[::]` format in logs
+        serve_log_uri = f"http://[::]:{port}"
+    else:
+        host = "0.0.0.0"
+        serve_log_uri = f"http://{host}:{port}"
+
+    logger.info("Starting log server on %s", serve_log_uri)
+
+    # Use uvicorn directly for ASGI applications
+    uvicorn.run("airflow.utils.serve_logs.log_server:app", host=host, port=port, workers=2, log_level="info")
+    # Note: if we want to use more than 1 workers, we **can't** use the instance of FastAPI directly
+    # This is way we split the instantiation of log server to a separate module
+    #
+    # https://github.com/encode/uvicorn/blob/374bb6764e8d7f34abab0746857db5e3d68ecfdd/docs/deployment/index.md?plain=1#L50-L63
+
+
+if __name__ == "__main__":
+    serve_logs()

airflow-core/src/airflow/utils/serve_logs.py → airflow-core/src/airflow/utils/serve_logs/log_server.py

@@ -14,26 +14,23 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
-"""Serve logs process."""
+"""Log server written in FastAPI."""
 
 from __future__ import annotations
 
 import logging
 import os
-import socket
-import sys
-from collections import namedtuple
+from typing import cast
 
-import gunicorn.app.base
-from flask import Flask, abort, request, send_from_directory
+from fastapi import FastAPI, HTTPException, Request, status
+from fastapi.staticfiles import StaticFiles
 from jwt.exceptions import (
     ExpiredSignatureError,
     ImmatureSignatureError,
     InvalidAudienceError,
     InvalidIssuedAtError,
     InvalidSignatureError,
 )
-from werkzeug.exceptions import HTTPException
 
 from airflow.api_fastapi.auth.tokens import JWTValidator, get_signing_key
 from airflow.configuration import conf
@@ -43,74 +40,55 @@
 logger = logging.getLogger(__name__)
 
 
-def create_app():
-    flask_app = Flask(__name__, static_folder=None)
-    leeway = conf.getint("webserver", "log_request_clock_grace", fallback=30)
-    log_directory = os.path.expanduser(conf.get("logging", "BASE_LOG_FOLDER"))
-    log_config_class = conf.get("logging", "logging_config_class")
-    if log_config_class:
-        logger.info("Detected user-defined logging config. Attempting to load %s", log_config_class)
-        try:
-            logging_config = import_string(log_config_class)
-            try:
-                base_log_folder = logging_config["handlers"]["task"]["base_log_folder"]
-            except KeyError:
-                base_log_folder = None
-            if base_log_folder is not None:
-                log_directory = base_log_folder
-                logger.info(
-                    "Successfully imported user-defined logging config. Flask App will serve log from %s",
-                    log_directory,
-                )
-            else:
-                logger.warning(
-                    "User-defined logging config does not specify 'base_log_folder'. "
-                    "Flask App will use default log directory %s",
-                    base_log_folder,
-                )
-        except Exception as e:
-            raise ImportError(f"Unable to load {log_config_class} due to error: {e}")
-    signer = JWTValidator(
-        issuer=None,
-        secret_key=get_signing_key("api", "secret_key"),
-        algorithm="HS512",
-        leeway=leeway,
-        audience="task-instance-logs",
-    )
+class JWTAuthStaticFiles(StaticFiles):
+    """StaticFiles with JWT authentication."""
 
-    # Prevent direct access to the logs port
-    @flask_app.before_request
-    def validate_pre_signed_url():
+    # reference from https://github.com/fastapi/fastapi/issues/858#issuecomment-876564020
+
+    def __init__(self, *args, **kwargs) -> None:
+        super().__init__(*args, **kwargs)
+
+    async def __call__(self, scope, receive, send) -> None:
+        request = Request(scope, receive)
+        await self.validate_jwt_token(request)
+        await super().__call__(scope, receive, send)
+
+    async def validate_jwt_token(self, request: Request):
+        # we get the signer from the app state instead of creating a new instance for each request
+        signer = cast("JWTValidator", request.app.state.signer)
         try:
             auth = request.headers.get("Authorization")
             if auth is None:
                 logger.warning("The Authorization header is missing: %s.", request.headers)
-                abort(403)
-            payload = signer.validated_claims(auth)
+                raise HTTPException(
+                    status_code=status.HTTP_403_FORBIDDEN, detail="Authorization header missing"
+                )
+            payload = await signer.avalidated_claims(auth)
             token_filename = payload.get("filename")
-            request_filename = request.view_args["filename"]
+            # Extract filename from url path
+            request_filename = request.url.path.lstrip("/log/")
             if token_filename is None:
                 logger.warning("The payload does not contain 'filename' key: %s.", payload)
-                abort(403)
+                raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
             if token_filename != request_filename:
                 logger.warning(
                     "The payload log_relative_path key is different than the one in token:"
                     "Request path: %s. Token path: %s.",
                     request_filename,
                     token_filename,
                 )
-                abort(403)
+                raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
         except HTTPException:
             raise
         except InvalidAudienceError:
             logger.warning("Invalid audience for the request", exc_info=True)
-            abort(403)
+            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
         except InvalidSignatureError:
             logger.warning("The signature of the request was wrong", exc_info=True)
-            abort(403)
+            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
         except ImmatureSignatureError:
             logger.warning("The signature of the request was sent from the future", exc_info=True)
-            abort(403)
+            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
         except ExpiredSignatureError:
             logger.warning(
                 "The signature of the request has expired. Make sure that all components "
@@ -119,78 +97,64 @@ def validate_pre_signed_url():
                 get_docs_url("configurations-ref.html#secret-key"),
                 exc_info=True,
             )
-            abort(403)
+            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
         except InvalidIssuedAtError:
             logger.warning(
-                "The request was issues in the future. Make sure that all components "
+                "The request was issued in the future. Make sure that all components "
                 "in your system have synchronized clocks. "
                 "See more at %s",
                 get_docs_url("configurations-ref.html#secret-key"),
                 exc_info=True,
             )
-            abort(403)
+            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
         except Exception:
             logger.warning("Unknown error", exc_info=True)
-            abort(403)
-
-    @flask_app.route("/log/<path:filename>")
-    def serve_logs_view(filename):
-        return send_from_directory(log_directory, filename, mimetype="application/json", as_attachment=False)
-
-    return flask_app
-
-
-GunicornOption = namedtuple("GunicornOption", ["key", "value"])
-
+            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
 
-class StandaloneGunicornApplication(gunicorn.app.base.BaseApplication):
-    """
-    Standalone Gunicorn application/serve for usage with any WSGI-application.
 
-    Code inspired by an example from the Gunicorn documentation.
-    https://github.com/benoitc/gunicorn/blob/cf55d2cec277f220ebd605989ce78ad1bb553c46/examples/standalone_app.py
-
-    For details, about standalone gunicorn application, see:
-    https://docs.gunicorn.org/en/stable/custom.html
-    """
-
-    def __init__(self, app, options=None):
-        self.options = options or []
-        self.application = app
-        super().__init__()
-
-    def load_config(self):
-        for option in self.options:
-            self.cfg.set(option.key.lower(), option.value)
-
-    def load(self):
-        return self.application
-
-
-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()
+def create_app():
+    leeway = conf.getint("webserver", "log_request_clock_grace", fallback=30)
+    log_directory = os.path.expanduser(conf.get("logging", "BASE_LOG_FOLDER"))
+    log_config_class = conf.get("logging", "logging_config_class")
+    if log_config_class:
+        logger.info("Detected user-defined logging config. Attempting to load %s", log_config_class)
+        try:
+            logging_config = import_string(log_config_class)
+            try:
+                base_log_folder = logging_config["handlers"]["task"]["base_log_folder"]
+            except KeyError:
+                base_log_folder = None
+            if base_log_folder is not None:
+                log_directory = base_log_folder
+                logger.info(
+                    "Successfully imported user-defined logging config. FastAPI App will serve log from %s",
+                    log_directory,
+                )
+            else:
+                logger.warning(
+                    "User-defined logging config does not specify 'base_log_folder'. "
+                    "FastAPI App will use default log directory %s",
+                    base_log_folder,
+                )
+        except Exception as e:
+            raise ImportError(f"Unable to load {log_config_class} due to error: {e}")
 
-    port = port or conf.getint("logging", "WORKER_LOG_SERVER_PORT")
+    fastapi_app = FastAPI()
+    fastapi_app.state.signer = JWTValidator(
+        issuer=None,
+        secret_key=get_signing_key("api", "secret_key"),
+        algorithm="HS512",
+        leeway=leeway,
+        audience="task-instance-logs",
+    )
 
-    # If dual stack is available and IPV6_V6ONLY is not enabled on the socket
-    # then when IPV6 is bound to it will also bind to IPV4 automatically
-    if getattr(socket, "has_dualstack_ipv6", lambda: False)():
-        bind_option = GunicornOption("bind", f"[::]:{port}")
-    else:
-        bind_option = GunicornOption("bind", f"0.0.0.0:{port}")
+    fastapi_app.mount(
+        "/log",
+        JWTAuthStaticFiles(directory=log_directory, html=False),
+        name="serve_logs",
+    )
 
-    options = [bind_option, GunicornOption("workers", 2)]
-    StandaloneGunicornApplication(wsgi_app, options).run()
+    return fastapi_app
 
 
-if __name__ == "__main__":
-    serve_logs()
+app = create_app()