Develop to Main

README.md

@@ -27,6 +27,7 @@ Run a local demo instance in seconds.
 
 ```bash
 # Clone and launch instantly
+cp .env.example .env
 docker compose -f docker-compose.yml -f docker-compose.demo.yml up --build
 ```
 

backend-services/doorman.py

@@ -96,6 +96,7 @@
 from routes.grpc_routes import grpc_router
 from utils.audit_util import audit
 from middleware.security_audit_middleware import SecurityAuditMiddleware
+from middleware.logging_middleware import GlobalLoggingMiddleware
 from middleware.latency_injection_middleware import LatencyInjectionMiddleware
 from middleware.websocket_reject_middleware import WebSocketRejectMiddleware
 from utils.auth_blacklist import purge_expired_tokens
@@ -111,6 +112,7 @@
     restore_memory_from_file,
 )
 from utils.metrics_util import metrics_store
+from utils.enhanced_metrics_util import enhanced_metrics_store
 from utils.response_util import process_response
 from utils.security_settings_util import (
     get_cached_settings,
@@ -368,11 +370,18 @@ async def app_lifespan(app: FastAPI):
     except Exception as e:
         gateway_logger.debug(f'Metrics restore skipped: {e}')
 
+    ENHANCED_METRICS_FILE = os.path.join(LOGS_DIR, 'enhanced_metrics.json')
+    try:
+        enhanced_metrics_store.load_from_file(ENHANCED_METRICS_FILE)
+    except Exception as e:
+        gateway_logger.debug(f'Enhanced metrics restore skipped: {e}')
+
     async def _metrics_autosave(interval_s: int = 60):
         while True:
             try:
                 await asyncio.sleep(interval_s)
                 metrics_store.save_to_file(METRICS_FILE)
+                enhanced_metrics_store.save_to_file(ENHANCED_METRICS_FILE)
             except asyncio.CancelledError:
                 break
             except Exception:
@@ -647,6 +656,8 @@ async def _sighup_reload():
         try:
             METRICS_FILE = os.path.join(LOGS_DIR, 'metrics.json')
             metrics_store.save_to_file(METRICS_FILE)
+            ENHANCED_METRICS_FILE = os.path.join(LOGS_DIR, 'enhanced_metrics.json')
+            enhanced_metrics_store.save_to_file(ENHANCED_METRICS_FILE)
         except Exception:
             pass
 
@@ -1484,6 +1495,7 @@ async def send_wrapper(message):
 
 
 doorman.add_middleware(PlatformCORSMiddleware)
+doorman.add_middleware(GlobalLoggingMiddleware)
 
 # Add tier-based rate limiting middleware (skip in live/test to avoid 429 floods)
 try:
@@ -1514,41 +1526,45 @@ async def send_wrapper(message):
 @doorman.middleware('http')
 async def request_id_middleware(request: Request, call_next):
     try:
+        from utils.correlation_util import get_correlation_id, set_correlation_id
+
         rid = (
-            request.headers.get('x-request-id')
+            getattr(request.state, 'request_id', None)
+            or get_correlation_id()
+            or request.headers.get('x-request-id')
             or request.headers.get('request-id')
-            or request.headers.get('x-request-id'.title())
+            or request.headers.get('X-Request-ID')
         )
         if not rid:
             rid = str(uuid.uuid4())
 
-        try:
+        # Store in state
+        if not hasattr(request.state, 'request_id'):
             request.state.request_id = rid
-        except Exception:
-            pass
 
-        try:
-            from utils.correlation_util import set_correlation_id
+        # Ensure correlation ID is set
+        set_correlation_id(rid)
 
-            set_correlation_id(rid)
-        except Exception:
-            pass
+        # Optional logging
         try:
             settings = get_cached_settings()
-            trust_xff = bool(settings.get('trust_x_forwarded_for'))
+            trust_xff = bool(settings.get('trust_x_forwarded_for', False))
             direct_ip = getattr(getattr(request, 'client', None), 'host', None)
             effective_ip = _policy_get_client_ip(request, trust_xff)
             gateway_logger.info(
-                f'{rid} | Entry: client_ip={direct_ip} effective_ip={effective_ip} method={request.method} path={str(request.url.path)}'
+                f'Entry: client_ip={direct_ip} effective_ip={effective_ip} method={request.method} path={str(request.url.path)}'
             )
         except Exception:
             pass
+
         response = await call_next(request)
+
         try:
             response.headers['X-Request-ID'] = rid
             response.headers['request_id'] = rid
         except Exception as e:
             gateway_logger.warning(f'Failed to set response headers: {str(e)}')
+
         return response
     except Exception as e:
         gateway_logger.error(f'Request ID middleware error: {str(e)}', exc_info=True)
@@ -1629,10 +1645,12 @@ async def security_headers(request: Request, call_next):
 # Build formatters
 class JSONFormatter(logging.Formatter):
     def format(self, record: logging.LogRecord) -> str:
+        request_id = getattr(record, 'request_id', 'no-request-id')
         payload = {
             'time': self.formatTime(record, '%Y-%m-%dT%H:%M:%S'),
             'name': record.name,
             'level': record.levelname,
+            'request_id': request_id,
             'message': record.getMessage(),
         }
         try:
@@ -1654,7 +1672,7 @@ def format(self, record: logging.LogRecord) -> str:
     _file_handler.setFormatter(
         JSONFormatter()
         if _fmt_is_json
-        else logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+        else logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(request_id)s | %(message)s')
     )
 except Exception as _e:
     # Attempt fallback to /tmp which is commonly writable on containers
@@ -1670,7 +1688,7 @@ def format(self, record: logging.LogRecord) -> str:
         _file_handler.setFormatter(
             JSONFormatter()
             if _fmt_is_json
-            else logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+            else logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(request_id)s | %(message)s')
         )
         logging.getLogger('doorman.gateway').warning(
             f'Primary LOGS_DIR={LOGS_DIR} not writable ({_e}); falling back to {_tmp_dir}'
@@ -1685,12 +1703,16 @@ def format(self, record: logging.LogRecord) -> str:
 # Configure all doorman loggers to use the same handler and prevent propagation
 def configure_logger(logger_name: str) -> logging.Logger:
     logger = logging.getLogger(logger_name)
-    logger.setLevel(logging.INFO)
+    log_level = os.getenv('LOG_LEVEL', 'INFO').upper()
+    numeric_level = getattr(logging, log_level, logging.INFO)
+    logger.setLevel(numeric_level)
     logger.propagate = False
 
     for handler in logger.handlers[:]:
         logger.removeHandler(handler)
 
+    from utils.correlation_util import correlation_id, RequestIdFilter
+
     class RedactFilter(logging.Filter):
         """Comprehensive logging redaction filter for sensitive data.
 
@@ -1767,19 +1789,21 @@ def filter(self, record: logging.LogRecord) -> bool:
             return True
 
     console = logging.StreamHandler(stream=sys.stdout)
-    console.setLevel(logging.INFO)
+    console.setLevel(numeric_level)
     console.setFormatter(
         JSONFormatter()
         if _fmt_is_json
-        else logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+        else logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(request_id)s | %(message)s')
     )
+    console.addFilter(RequestIdFilter())
     console.addFilter(RedactFilter())
     logger.addHandler(console)
 
     if _file_handler is not None:
         if not any(
             isinstance(f, logging.Filter) and hasattr(f, 'PATTERNS') for f in _file_handler.filters
         ):
+            _file_handler.addFilter(RequestIdFilter())
             _file_handler.addFilter(RedactFilter())
         logger.addHandler(_file_handler)
     return logger
@@ -1795,7 +1819,9 @@ def filter(self, record: logging.LogRecord) -> bool:
 
     _mem_enabled = os.getenv('MEMORY_LOG_ENABLED', 'true').lower() != 'false'
     if _mem_enabled:
-        _mem_handler = MemoryLogHandler()
+        _lvl = os.getenv('LOG_LEVEL', 'INFO').upper()
+        _num_lvl = getattr(logging, _lvl, logging.INFO)
+        _mem_handler = MemoryLogHandler(level=_num_lvl)
         for _name in (
             'doorman.gateway',
             'doorman.logging',

backend-services/middleware/analytics_middleware.py

@@ -48,14 +48,27 @@ async def dispatch(self, request: Request, call_next: Callable) -> Response:
         method = request.method
         path = str(request.url.path)
 
-        # Estimate request size (headers + body)
+        # Estimate request size (headers + body + protocol overhead)
         request_size = 0
         try:
-            # Headers size
-            request_size += sum(len(k) + len(v) for k, v in request.headers.items())
-            # Body size (if available)
+            # Request line: "METHOD /path?query HTTP/1.1\r\n"
+            request_line = f"{request.method} {request.url.path}"
+            if request.url.query:
+                request_line += f"?{request.url.query}"
+            request_line += " HTTP/1.1\r\n"
+            request_size += len(request_line)
+
+            # Headers with proper separators (key: value\r\n)
+            for k, v in request.headers.items():
+                request_size += len(f"{k}: {v}\r\n")
+            request_size += 2  # Final \r\n separator
+
+            # Body size
             if 'content-length' in request.headers:
                 request_size += int(request.headers['content-length'])
+            elif hasattr(request, '_body') and request._body:
+                # Fallback: use actual body if available
+                request_size += len(request._body)
         except Exception:
             pass
 
@@ -83,14 +96,29 @@ async def dispatch(self, request: Request, call_next: Callable) -> Response:
         # Extract response metadata
         status_code = response.status_code
 
-        # Estimate response size
+        # Estimate response size (headers + body + protocol overhead)
         response_size = 0
         try:
-            # Headers size
-            response_size += sum(len(k) + len(v) for k, v in response.headers.items())
-            # Body size (if available)
+            # Status line: "HTTP/1.1 200 OK\r\n"
+            status_text = response.status_code
+            try:
+                from http import HTTPStatus
+                status_text = f"{response.status_code} {HTTPStatus(response.status_code).phrase}"
+            except Exception:
+                status_text = str(response.status_code)
+            response_size += len(f"HTTP/1.1 {status_text}\r\n")
+
+            # Headers with proper separators (key: value\r\n)
+            for k, v in response.headers.items():
+                response_size += len(f"{k}: {v}\r\n")
+            response_size += 2  # Final \r\n separator
+
+            # Body size
             if 'content-length' in response.headers:
                 response_size += int(response.headers['content-length'])
+            # Note: For streaming/chunked responses without Content-Length,
+            # we can't accurately measure body size in middleware without wrapping.
+            # This is acceptable as most responses have Content-Length.
         except Exception:
             pass
 
@@ -104,6 +132,9 @@ async def dispatch(self, request: Request, call_next: Callable) -> Response:
         except Exception:
             pass
 
+        # Extract request_id for logging
+        request_id = getattr(request.state, 'request_id', None) or request.headers.get('X-Request-ID') or 'unknown'
+
         # Parse API and endpoint from path
         api_key, endpoint_uri = self._parse_api_endpoint(path)
 

backend-services/middleware/logging_middleware.py

@@ -0,0 +1,68 @@
+import logging
+import time
+import uuid
+from starlette.middleware.base import BaseHTTPMiddleware
+from fastapi import Request
+
+logger = logging.getLogger('doorman.gateway')
+
+class GlobalLoggingMiddleware(BaseHTTPMiddleware):
+    async def dispatch(self, request: Request, call_next):
+        # 1. Generate or extract Request ID
+        from utils.correlation_util import correlation_id
+
+        request_id = (
+            getattr(request.state, 'request_id', None)
+            or correlation_id.get()
+            or request.headers.get('X-Request-ID')
+            or request.headers.get('request-id')
+        )
+        if not request_id:
+            request_id = str(uuid.uuid4())
+
+        correlation_id.set(request_id)
+
+        # Store in state if not already present
+        if not hasattr(request.state, 'request_id'):
+            request.state.request_id = request_id
+
+        # 2. Start Timer
+        start_time = time.time()
+
+        # 3. Log Request Entry (Optional - helpful for incomplete requests)
+        # We can default to 'platform' or 'gateway' in type inference if needed.
+        # Using a consistent prefix helps.
+        # logger.info(f"Request started: {request.method} {request.url.path}")
+
+        try:
+            # 4. Process Request
+            response = await call_next(request)
+
+            # 5. Calculate Duration
+            duration = (time.time() - start_time) * 1000
+
+            # 6. Log Response
+            # Format matches LoggingService extraction regexes:
+            # Endpoint: {method} {path}
+            # status_code: {code}
+            # Total time: {ms}ms
+            logger.info(
+                f"Endpoint: {request.method} {request.url.path} "
+                f"| status_code: {response.status_code} "
+                f"| Total time: {duration:.2f}ms"
+            )
+
+            # Ensure Request ID is returned in headers for debugging
+            response.headers['X-Request-ID'] = request_id
+
+            return response
+
+        except Exception as e:
+            # 7. Log Unhandled Exceptions
+            duration = (time.time() - start_time) * 1000
+            logger.error(
+                f"Request failed: {request.method} {request.url.path} "
+                f"| Error: {str(e)} | Time: {duration:.2f}ms",
+                exc_info=True
+            )
+            raise