tech: improve grafana logging

Author: stephenprn

aigle/settings.py

@@ -16,6 +16,7 @@
 
 from core.utils.parsing import strtobool
 
+
 DEPLOYMENT_DATETIME = datetime.now()
 
 # Build paths inside the project like this: BASE_DIR / 'subdir'.
@@ -129,6 +130,22 @@
     },
 }
 
+if ENVIRONMENT in ["production", "preprod"]:
+    LOGGING["handlers"]["scaleway_loki"] = {
+        "class": "logging_loki.LokiHandler",
+        "url": os.environ.get("SCW_COCKPIT_URL"),
+        "tags": {
+            "job": "django_api",
+            "environment": ENVIRONMENT,
+        },
+        "auth": (
+            os.environ.get("SCW_SECRET_KEY"),
+            os.environ.get("SCW_COCKPIT_TOKEN_SECRET_KEY"),
+        ),
+        "version": "1",
+    }
+
+
 ALLOWED_HOSTS = []
 
 if os.environ.get("ALLOWED_HOSTS"):

core/middlewares/logs.py

@@ -1,8 +1,7 @@
-import logging
 import time
 import uuid
 
-logger = logging.getLogger("aigle")  # Replace 'myapp' with your app name
+from core.utils.logs_helpers import log_api_call
 
 
 class RequestLoggingMiddleware:
@@ -14,29 +13,20 @@ def __call__(self, request):
         request.id = str(uuid.uuid4())
         start_time = time.time()
 
-        # Log incoming request
-        logger.info(
-            f"Request {request.method} {request.path}",
-            extra={
-                "request_id": request.id,
-                "method": request.method,
-                "path": request.path,
-                "user": str(request.user) if hasattr(request, "user") else "Anonymous",
-                "ip": self.get_client_ip(request),
-            },
-        )
-
         response = self.get_response(request)
 
         # Log response
         duration = time.time() - start_time
-        logger.info(
-            f"Response {response.status_code}",
-            extra={
-                "request_id": request.id,
-                "status_code": response.status_code,
-                "duration_ms": round(duration * 1000, 2),
-            },
+
+        # Log incoming request
+        log_api_call(
+            endpoint=request.path,
+            method=request.method,
+            user=str(request.user) if hasattr(request, "user") else "Anonymous",
+            ip=str(self.get_client_ip(request)),
+            request_id=request.id,
+            duration_ms=round(duration * 1000, 2),
+            status_code=response.status_code,
         )
 
         return response

core/utils/logs.py

@@ -12,11 +12,11 @@ def setup_scaleway_logger():
     # Only use Scaleway logging in production/staging
     if environment in ["production", "staging"]:
         handler = logging_loki.LokiHandler(
-            url="https://ad795441-15db-4609-a53f-30ddf578f82d.logs.cockpit.fr-par.scw.cloud",
+            url=os.environ.get("SCW_COCKPIT_URL"),
             tags={"job": "django_api", "environment": environment},
             auth=(
                 os.environ.get("SCW_SECRET_KEY"),
-                os.environ.get("COCKPIT_TOKEN_SECRET_KEY"),
+                os.environ.get("SCW_COCKPIT_TOKEN_SECRET_KEY"),
             ),
             version="1",
         )

core/utils/logs_helpers.py

@@ -1,39 +1,39 @@
 import logging
+from typing import Optional
 
 
 def get_app_logger():
     return logging.getLogger("aigle")
 
 
-def log_user_action(user, action, **kwargs):
-    logger = get_app_logger()
-    logger.info(
-        f"User action: {action}",
-        extra={
-            "user_id": user.id if hasattr(user, "id") else None,
-            "username": str(user),
-            "action": action,
-            **kwargs,
-        },
-    )
-
-
-def log_api_call(endpoint, method, status_code, duration=None, **kwargs):
+def log_api_call(
+    endpoint: str,
+    method: str,
+    user: str,
+    ip: str,
+    request_id: str,
+    status_code: str,
+    duration_ms: Optional[str] = None,
+    **kwargs,
+):
     """Log API calls"""
     logger = get_app_logger()
     logger.info(
         f"API call: {method} {endpoint}",
         extra={
             "endpoint": endpoint,
             "method": method,
+            "user": user,
+            "ip": ip,
+            "request_id": request_id,
             "status_code": status_code,
-            "duration_ms": duration,
+            "duration_ms": duration_ms,
             **kwargs,
         },
     )
 
 
-def log_command_event(command_name, info, **kwargs):
+def log_command_event(command_name: str, info: str, **kwargs):
     logger = get_app_logger()
     logger.info(
         f"Command event: {command_name}",