chore: improve API logs (#1227)

Author: davidgamez

.gitignore

@@ -61,6 +61,7 @@ vars-*.env
 config/.env.*
 venv
 cypress.env.json
+.deploy-env-dev
 
 __pycache__
 /functions-python/*/.env.local

api/Dockerfile

@@ -15,7 +15,7 @@
 #
 
 # Builder Stage
-FROM --platform=$BUILDPLATFORM python:3.11 AS builder
+FROM --platform=$TARGETPLATFORM python:3.11 AS builder
 
 WORKDIR /usr/src/app
 

api/src/feeds/impl/error_handling.py

@@ -1,3 +1,5 @@
+import logging
+
 from fastapi import HTTPException
 
 from shared.common.error_handling import InternalHTTPException
@@ -21,10 +23,12 @@ def raise_http_error(status_code: int, error: str):
         "detail": "Invalid date format for 'field_name'. Expected ISO 8601 format, example: '2021-01-01T00:00:00Z'"
     }
     """
-    raise HTTPException(
+    exception = HTTPException(
         status_code=status_code,
         detail=error,
     )
+    logging.error(exception)
+    raise exception
 
 
 def raise_http_validation_error(error: str):

api/src/main.py

@@ -15,7 +15,6 @@
 #
 # This files allows to add extra application decorators aside from the generated code.
 # The app created here is intended to replace the generated feeds_gen.main:app variable.
-import logging
 import os
 
 import uvicorn
@@ -31,7 +30,8 @@
 from starlette.middleware.cors import CORSMiddleware
 
 from middleware.request_context_middleware import RequestContextMiddleware
-from utils.logger import API_ACCESS_LOG, GCPLogHandler
+from utils.logger import global_logging_setup
+
 
 app = FastAPI(
     title="Mobility Data Catalog API",
@@ -58,11 +58,7 @@
 
 @app.on_event("startup")
 async def startup_event():
-    # Add the GCP log handler to the logger.
-    # This is required to log the API access logs to the GCP logging service.
-    logger = logging.getLogger(API_ACCESS_LOG)
-    handler = GCPLogHandler()
-    logger.handlers.append(handler)
+    global_logging_setup()
 
 
 if __name__ == "__main__":

api/src/middleware/request_context.py

@@ -1,3 +1,4 @@
+import logging
 from contextvars import ContextVar
 
 import requests
@@ -49,7 +50,7 @@ def decode_jwt(self, token: str):
                 return None
             return jwt.decode(token, self.google_public_keys, audience=get_config(PROJECT_ID))
         except Exception as e:
-            print(f"Error decoding JWT: {e}")
+            logging.warning("Error decoding JWT: %s", e)
             return None
 
     def _extract_from_headers(self, headers: dict, scope: Scope) -> None:

api/src/middleware/request_context_middleware.py

@@ -38,17 +38,21 @@ def create_http_request(
         """
         request_method = scope["method"]
         request_path = scope["path"]
+        query_string = scope.get("query_string", b"").decode("utf-8")
+        full_path = f"{request_path}?{query_string}" if query_string else request_path
+
         protocol = scope["scheme"].upper() + "/" + str(scope["http_version"])
+
         return HttpRequest(
             requestMethod=request_method,
-            requestUrl=f"{request_context.protocol}://{request_context.host}{request_path}",
+            requestUrl=f"{request_context.protocol}://{request_context.host}{full_path}",
             remoteIp=request_context.client_host,
             protocol=protocol,
             status=status_code,
             responseSize=content_length,
             userAgent=request_context.client_user_agent,
             serverIp=request_context.server_ip,
-            latency=latency,
+            latency=f"{latency:.9f}s",
         )
 
     def log_api_access(

api/src/utils/logger.py

@@ -1,13 +1,17 @@
-import asyncio
-import json
 import logging
-from dataclasses import dataclass
+import os
+import sys
+from dataclasses import dataclass, asdict
 from typing import Final, Optional
 
 from middleware.request_context import RequestContext, get_request_context
+from shared.common.logging_utils import get_env_logging_level
 from utils.config import get_config, PROJECT_ID
 from utils.dict_utils import get_safe_value
 
+import google.cloud.logging
+from google.cloud.logging_v2.handlers import CloudLoggingFilter, CloudLoggingHandler
+
 API_ACCESS_LOG: Final[str] = "api-access-log"
 CLOUD_RUN_SERVICE_ID: Final[str] = "K_SERVICE"
 CLOUD_RUN_REVISION_ID: Final[str] = "K_REVISION"
@@ -27,101 +31,125 @@ class HttpRequest:
     userAgent: str
     remoteIp: str
     serverIp: str
-    latency: float
+    latency: str
     protocol: str
 
 
-@dataclass
-class LogRecord:
+def get_trace(request_context: RequestContext):
     """
-    Data class for Log Record
+    Get the trace id from the log record
     """
+    trace = ""
+    trace_id = get_safe_value(request_context, "trace_id")
+    if trace_id:
+        trace = f"projects/{get_config(PROJECT_ID, '')}/traces/{trace_id}"
+    return trace
 
-    user_id: str
-    httpRequest: dict
-    trace: str
-    spanId: str
-    traceSampled: bool
-    textPayload: Optional[str]
-    jsonPayload: Optional[dict]
+
+def get_http_request(record) -> HttpRequest | None:
+    """
+    Get the http request from the log record
+    If the http request is not found, return None
+    """
+    context = record.__getattribute__("context") if hasattr(record, "context") else None
+    return context.get("http_request") if context else {}
 
 
-class AsyncStreamHandler(logging.StreamHandler):
+class GoogleCloudLogFilter(CloudLoggingFilter):
     """
-    Async Stream Handler
+    Log filter for Google Cloud Logging.
+    This filter adds the trace, span and http_request fields to the log record.
     """
 
-    def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
-        self.loop = asyncio.get_event_loop()
+    def filter(self, record: logging.LogRecord) -> bool:
+        request_context = get_request_context()
+        http_request = get_http_request(record)
+        if http_request:
+            record.http_request = asdict(http_request)
+        span_id = request_context.get("span_id")
+        trace = get_trace(request_context)
+        record.trace = trace
+        record.span_id = span_id
+
+        record._log_fields = {
+            "logging.googleapis.com/trace": trace,
+            "logging.googleapis.com/spanId": span_id,
+            "logging.googleapis.com/httpRequest": asdict(http_request) if http_request else None,
+            "logging.googleapis.com/trace_sampled": True,
+        }
+        super().filter(record)
+
+        return True
+
+
+class StderrToLog:
+    """
+    Redirect stderr to log
+    """
 
-    def emit(self, record):
-        """
-        Emit the log record
-        """
-        asyncio.ensure_future(self.async_emit(record))
+    def __init__(self, logger):
+        self.logger = logger
 
-    async def async_emit(self, record):
-        """
-        Async emit the log record
-        """
-        msg = self.format(record)
-        stream = self.stream
-        await self.loop.run_in_executor(None, stream.write, msg)
-        await self.loop.run_in_executor(None, stream.flush)
+    def write(self, message):
+        message = message.strip()
+        if message:
+            self.logger.error(message)
+
+    def flush(self):
+        pass
 
 
-class GCPLogHandler(AsyncStreamHandler):
+def get_logger(name: Optional[str]):
     """
-    GCP Log Handler
+    Returns a logger with the name making sure the propagate flag is set to True.
     """
+    logger = logging.getLogger(name)
+    logger.propagate = True
+    return logger
 
-    def __init__(self):
-        console_handler = logging.StreamHandler()
-        self.logger = logging.getLogger()
-        self.logger.addHandler(console_handler)
-        self.logger.setLevel(logging.DEBUG)
-        super().__init__()
-
-    @staticmethod
-    def get_trace(request_context: RequestContext):
-        """
-        Get the trace id from the log record
-        """
-        trace = ""
-        trace_id = get_safe_value(request_context, "trace_id")
-        if trace_id:
-            trace = f"projects/{get_config(PROJECT_ID, '')}/traces/{trace_id}"
-        return trace
-
-    @staticmethod
-    def get_http_request(record) -> HttpRequest:
-        context = record.__getattribute__("context") if hasattr(record, "context") else None
-        return context.get("http_request") if context else {}
-
-    async def async_emit(self, record):
-        """
-        Emit the GCP log record
-        """
-        http_request = self.get_http_request(record)
-        request_context = get_request_context()
-        text_payload = None
-        json_payload = None
-        message = record.msg if hasattr(record, "msg") else None
-        message = record.getMessage() if message is None and hasattr(record, "getMessage") else message
-        if message:
-            if type(message) is dict:
-                json_payload = message
-            else:
-                text_payload = str(message)
-
-        log_record: LogRecord = LogRecord(
-            httpRequest=http_request.__dict__ if not isinstance(http_request, dict) else {},
-            trace=self.get_trace(request_context),
-            spanId=request_context.get("span_id"),
-            traceSampled=request_context.get("trace_sampled"),
-            user_id=request_context.get("user_id"),
-            textPayload=text_payload,
-            jsonPayload=json_payload,
-        )
-        self.logger.info(json.dumps(log_record.__dict__))
+
+def is_local_env():
+    """
+    Returns: True if the environment is local, False otherwise
+    """
+    return os.getenv("K_SERVICE") is None
+
+
+def global_logging_setup():
+    if is_local_env():
+        logging.basicConfig(level=get_env_logging_level())
+        return
+
+    # Send warnings through logging
+    logging.captureWarnings(True)
+    # Replace sys.stderr
+    sys.stderr = StderrToLog(logging.getLogger("stderr"))
+    try:
+        client = google.cloud.logging.Client()
+        handler = CloudLoggingHandler(client, structured=True)
+        handler.setLevel(get_env_logging_level())
+        handler.addFilter(GoogleCloudLogFilter(project=client.project))
+    except Exception as e:
+        logging.error("Error initializing cloud logging: %s", e)
+        logging.basicConfig(level=get_env_logging_level())
+        return
+
+    # Configure root logger
+    root_logger = logging.getLogger()
+    root_logger.setLevel(get_env_logging_level())
+    root_logger.handlers.clear()
+    root_logger.addHandler(handler)
+
+    # This overrides individual logs essential for debugging purposes.
+    for name in [
+        "sqlalchemy",
+        "uvicorn",
+        "uvicorn.error",
+        "uvicorn.access",
+        "sqlalchemy.exc",
+        "feed-api",
+        "sqlalchemy.engine",
+    ]:
+        get_logger(name)
+
+    logging.info("Setting cloud up logging completed")