Log request times and content lengths (#570)

Author: markwaddle

libraries/python/semantic-workbench-assistant/semantic_workbench_assistant/assistant_service.py

@@ -40,6 +40,7 @@
 from starlette.exceptions import HTTPException as StarletteHTTPException
 
 from . import auth, settings
+from .logging_config import log_request_middleware
 
 logger = logging.getLogger(__name__)
 
@@ -129,7 +130,7 @@ async def _ping_semantic_workbench(self, client: workbench_service_client.Assist
                     name=self.service_name,
                     description=self.service_description,
                     url=HttpUrl(settings.callback_url),
-                    online_expires_in_seconds=settings.workbench_service_ping_interval_seconds * 2.5,
+                    online_expires_in_seconds=settings.workbench_service_ping_interval_seconds * 3.5,
                 ),
             )
 
@@ -292,6 +293,7 @@ def _assistant_service_api(
             exclude_paths=set(settings.anonymous_paths),
         )
     app.add_middleware(asgi_correlation_id.CorrelationIdMiddleware)
+    app.middleware("http")(log_request_middleware())
 
     @app.exception_handler(StarletteHTTPException)
     async def custom_http_exception_handler(request: Request, exc: StarletteHTTPException) -> Response:

libraries/python/semantic-workbench-assistant/semantic_workbench_assistant/logging_config.py

@@ -1,8 +1,11 @@
 import logging
 import logging.config
 import re
+from time import perf_counter
+from typing import Awaitable, Callable
 
 import asgi_correlation_id
+from fastapi import Request, Response
 from pydantic_settings import BaseSettings
 from pythonjsonlogger import json as jsonlogger
 
@@ -141,3 +144,31 @@ def config(settings: LoggingSettings):
             },
         },
     })
+
+
+def log_request_middleware(
+    logger: logging.Logger | None = None,
+) -> Callable[[Request, Callable[[Request], Awaitable[Response]]], Awaitable[Response]]:
+    access_logger = logger or logging.getLogger("access_log")
+
+    async def middleware(request: Request, call_next: Callable[[Request], Awaitable[Response]]) -> Response:
+        """
+        This middleware will log all requests and their processing time.
+        E.g. log:
+        0.0.0.0:1234 - "GET /ping HTTP/1.1" 200 OK 1.00ms 0b
+        """
+        url = f"{request.url.path}?{request.query_params}" if request.query_params else request.url.path
+        start_time = perf_counter()
+        response = await call_next(request)
+        process_time = (perf_counter() - start_time) * 1000
+        formatted_process_time = "{0:.2f}".format(process_time)
+        host = getattr(getattr(request, "client", None), "host", None)
+        port = getattr(getattr(request, "client", None), "port", None)
+        http_version = f"HTTP/{request.scope.get('http_version', '1.1')}"
+        content_length = response.headers.get("content-length", 0)
+        access_logger.info(
+            f'{host}:{port} - "{request.method} {url} {http_version}" {response.status_code} {formatted_process_time}ms {content_length}b',
+        )
+        return response
+
+    return middleware

libraries/python/semantic-workbench-assistant/semantic_workbench_assistant/settings.py

@@ -16,7 +16,7 @@ class Settings(BaseSettings):
 
     workbench_service_url: HttpUrl = HttpUrl("http://127.0.0.1:3000")
     workbench_service_api_key: str = ""
-    workbench_service_ping_interval_seconds: float = 20.0
+    workbench_service_ping_interval_seconds: float = 30.0
 
     assistant_service_id: str | None = None
     assistant_service_name: str | None = None

libraries/python/semantic-workbench-assistant/semantic_workbench_assistant/start.py

@@ -95,6 +95,7 @@ def main():
         host=settings.host,
         port=settings.port,
         reload=reload,
+        access_log=False,
         log_config={"version": 1, "disable_existing_loggers": False},
     )
 

workbench-service/semantic_workbench_service/logging_config.py

@@ -1,7 +1,10 @@
 import logging
 import re
+from time import perf_counter
+from typing import Awaitable, Callable
 
 import asgi_correlation_id
+from fastapi import Request, Response
 from pydantic_settings import BaseSettings
 from pythonjsonlogger import json as jsonlogger
 from rich.logging import RichHandler
@@ -65,3 +68,31 @@ def config(settings: LoggingSettings):
         datefmt="[%X]",
         handlers=[handler],
     )
+
+
+def log_request_middleware(
+    logger: logging.Logger | None = None,
+) -> Callable[[Request, Callable[[Request], Awaitable[Response]]], Awaitable[Response]]:
+    access_logger = logger or logging.getLogger("access_log")
+
+    async def middleware(request: Request, call_next: Callable[[Request], Awaitable[Response]]) -> Response:
+        """
+        This middleware will log all requests and their processing time.
+        E.g. log:
+        0.0.0.0:1234 - "GET /ping HTTP/1.1" 200 OK 1.00ms 0b
+        """
+        url = f"{request.url.path}?{request.query_params}" if request.query_params else request.url.path
+        start_time = perf_counter()
+        response = await call_next(request)
+        process_time = (perf_counter() - start_time) * 1000
+        formatted_process_time = "{0:.2f}".format(process_time)
+        host = getattr(getattr(request, "client", None), "host", None)
+        port = getattr(getattr(request, "client", None), "port", None)
+        http_version = f"HTTP/{request.scope.get('http_version', '1.1')}"
+        content_length = response.headers.get("content-length", 0)
+        access_logger.info(
+            f'{host}:{port} - "{request.method} {url} {http_version}" {response.status_code} {formatted_process_time}ms {content_length}b',
+        )
+        return response
+
+    return middleware

workbench-service/semantic_workbench_service/service.py

@@ -86,6 +86,7 @@
 from sse_starlette import EventSourceResponse, ServerSentEvent
 
 from semantic_workbench_service import azure_speech
+from semantic_workbench_service.logging_config import log_request_middleware
 
 from . import assistant_api_key, auth, controller, db, files, middleware, settings
 from .event import ConversationEventQueueItem
@@ -160,11 +161,15 @@ async def _notify_event(queue_item: ConversationEventQueueItem) -> None:
         )
 
         if "user" in queue_item.event_audience:
+            enqueued_count = 0
             async with conversation_sse_queues_lock:
                 for queue in conversation_sse_queues.get(queue_item.event.conversation_id, {}):
+                    enqueued_count += 1
                     await queue.put(queue_item.event)
+
             logger.debug(
-                "enqueued event for SSE; conversation_id: %s, event: %s, event_id: %s",
+                "enqueued event for SSE; count: %d, conversation_id: %s, event: %s, event_id: %s",
+                enqueued_count,
                 queue_item.event.conversation_id,
                 queue_item.event.event,
                 queue_item.event.id,
@@ -265,6 +270,8 @@ async def _notify_user_event(event: ConversationEvent) -> None:
     )
     app.add_middleware(CorrelationIdMiddleware)
 
+    app.middleware("http")(log_request_middleware())
+
     user_controller = controller.UserController(get_session=_controller_get_session)
     assistant_controller = controller.AssistantController(
         get_session=_controller_get_session,

workbench-service/semantic_workbench_service/start.py

@@ -46,6 +46,7 @@ def main():
         app,
         host=settings.service.host,
         port=settings.service.port,
+        access_log=False,
         log_config={"version": 1, "disable_existing_loggers": False},
     )