[WEB-5575]feat: enhance APITokenLogMiddleware to support logging to MongoDB (#8241)

* feat: enhance APITokenLogMiddleware to support logging to MongoDB

- Added functionality to log external API requests to MongoDB, with a fallback to PostgreSQL if MongoDB is unavailable.
- Implemented error handling for MongoDB connection and logging operations.
- Introduced additional fields for MongoDB logs, including timestamps and user identifiers.
- Refactored request logging logic to streamline the process and improve maintainability.

* fix: improve MongoDB availability checks in APITokenLogMiddleware

- Enhanced the logic for determining MongoDB availability by checking if the collection is not None.
- Added a check for MongoDB configuration before attempting to retrieve the collection.
- Updated error handling to ensure the middleware correctly reflects the state of MongoDB connectivity.

* feat: implement logging functionality in logger_task for API activity

- Added a new logger_task module to handle logging of API activity to MongoDB and PostgreSQL.
- Introduced functions for safely decoding request/response bodies and processing logs based on MongoDB availability.
- Refactored APITokenLogMiddleware to utilize the new logging functions, improving code organization and maintainability.

* refactor: simplify MongoDB logging in logger_task and middleware

- Removed direct dependency on MongoDB collection in log_to_mongo function, now retrieving it internally.
- Updated process_logs to check MongoDB configuration before logging, enhancing error handling.
- Cleaned up logger.py by removing unused imports related to MongoDB.

* feat: add Celery task decorator to process_logs function in logger_task

- Introduced the @shared_task decorator to the process_logs function, enabling asynchronous processing of log data.
- Updated function signature to include a return type of None for clarity.

apps/api/plane/bgtasks/logger_task.py

@@ -0,0 +1,96 @@
+# Python imports
+import logging
+from typing import Optional, Dict, Any
+
+# Third party imports
+from pymongo.collection import Collection
+from celery import shared_task
+
+# Django imports
+from plane.settings.mongo import MongoConnection
+from plane.utils.exception_logger import log_exception
+from plane.db.models import APIActivityLog
+
+
+logger = logging.getLogger("plane.worker")
+
+
+def get_mongo_collection() -> Optional[Collection]:
+    """
+    Returns the MongoDB collection for external API activity logs.
+    """
+    if not MongoConnection.is_configured():
+        logger.info("MongoDB not configured")
+        return None
+
+    try:
+        return MongoConnection.get_collection("api_activity_logs")
+    except Exception as e:
+        logger.error(f"Error getting MongoDB collection: {str(e)}")
+        log_exception(e)
+        return None
+
+
+def safe_decode_body(content: bytes) -> Optional[str]:
+    """
+    Safely decodes request/response body content, handling binary data.
+    Returns "[Binary Content]" if the content is binary, or a string representation of the content.
+    Returns None if the content is None or empty.
+    """
+    # If the content is None, return None
+    if content is None:
+        return None
+
+    # If the content is an empty bytes object, return None
+    if content == b"":
+        return None
+
+    # Check if content is binary by looking for common binary file signatures
+    if content.startswith(b"\x89PNG") or content.startswith(b"\xff\xd8\xff") or content.startswith(b"%PDF"):
+        return "[Binary Content]"
+
+    try:
+        return content.decode("utf-8")
+    except UnicodeDecodeError:
+        return "[Could not decode content]"
+
+
+def log_to_mongo(log_document: Dict[str, Any]) -> bool:
+    """
+    Logs the request to MongoDB if available.
+    """
+    mongo_collection = get_mongo_collection()
+    if mongo_collection is None:
+        logger.error("MongoDB not configured")
+        return False
+
+    try:
+        mongo_collection.insert_one(log_document)
+        return True
+    except Exception as e:
+        log_exception(e)
+        return False
+
+
+def log_to_postgres(log_data: Dict[str, Any]) -> bool:
+    """
+    Fallback to logging to PostgreSQL if MongoDB is unavailable.
+    """
+    try:
+        APIActivityLog.objects.create(**log_data)
+        return True
+    except Exception as e:
+        log_exception(e)
+        return False
+
+
+@shared_task
+def process_logs(log_data: Dict[str, Any], mongo_log: Dict[str, Any]) -> None:
+    """
+    Process logs to save to MongoDB or Postgres based on the configuration
+    """
+
+    if MongoConnection.is_configured():
+        log_to_mongo(mongo_log)
+    else:
+        log_to_postgres(log_data)

apps/api/plane/middleware/logger.py

@@ -4,13 +4,15 @@ import time
 
 # Django imports
 from django.http import HttpRequest
+from django.utils import timezone
 
 # Third party imports
 from rest_framework.request import Request
 
 # Module imports
 from plane.utils.ip_address import get_client_ip
-from plane.db.models import APIActivityLog
+from plane.utils.exception_logger import log_exception
+from plane.bgtasks.logger_task import process_logs
 
 api_logger = logging.getLogger("plane.api.request")
 
@@ -70,6 +72,10 @@ class RequestLoggerMiddleware:
 
 
 class APITokenLogMiddleware:
+    """
+    Middleware to log External API requests to MongoDB or PostgreSQL.
+    """
+
     def __init__(self, get_response):
         self.get_response = get_response
 
@@ -104,24 +110,41 @@ class APITokenLogMiddleware:
     def process_request(self, request, response, request_body):
         api_key_header = "X-Api-Key"
         api_key = request.headers.get(api_key_header)
-        # If the API key is present, log the request
-        if api_key:
-            try:
-                APIActivityLog.objects.create(
-                    token_identifier=api_key,
-                    path=request.path,
-                    method=request.method,
-                    query_params=request.META.get("QUERY_STRING", ""),
-                    headers=str(request.headers),
-                    body=(self._safe_decode_body(request_body) if request_body else None),
-                    response_body=(self._safe_decode_body(response.content) if response.content else None),
-                    response_code=response.status_code,
-                    ip_address=get_client_ip(request=request),
-                    user_agent=request.META.get("HTTP_USER_AGENT", None),
-                )
 
-            except Exception as e:
+        # If the API key is not present, return
-                api_logger.exception(e)
+        if not api_key:
-                # If the token does not exist, you can decide whether to log this as an invalid attempt
+            return
+
+        try:
+            log_data = {
+                "token_identifier": api_key,
+                "path": request.path,
+                "method": request.method,
+                "query_params": request.META.get("QUERY_STRING", ""),
+                "headers": str(request.headers),
+                "body": self._safe_decode_body(request_body) if request_body else None,
+                "response_body": self._safe_decode_body(response.content) if response.content else None,
+                "response_code": response.status_code,
+                "ip_address": get_client_ip(request=request),
+                "user_agent": request.META.get("HTTP_USER_AGENT", None),
+            }
+            user_id = (
+                str(request.user.id)
+                if getattr(request, "user") and getattr(request.user, "is_authenticated", False)
+                else None
+            )
+            # Additional fields for MongoDB
+            mongo_log = {
+                **log_data,
+                "created_at": timezone.now(),
+                "updated_at": timezone.now(),
+                "created_by": user_id,
+                "updated_by": user_id,
+            }
+
+            process_logs.delay(log_data=log_data, mongo_log=mongo_log)
+
+        except Exception as e:
+            log_exception(e)
 
         return None