From bd0abab7599bd52bda326a53332b40b085889c98 Mon Sep 17 00:00:00 2001 From: Gregor Klevze Date: Tue, 25 Aug 2026 07:11:22 +0200 Subject: [PATCH] Minor updates --- common/request_logging.py | 135 ++++++++++++++++++++++++++++++++++++++ gateway/main.py | 3 + llm/main.py | 3 + tests/test_gateway_llm.py | 24 +++++-- tests/test_llm_service.py | 16 ++++- 5 files changed, 173 insertions(+), 8 deletions(-) create mode 100644 common/request_logging.py diff --git a/common/request_logging.py b/common/request_logging.py new file mode 100644 index 0000000..5da4783 --- /dev/null +++ b/common/request_logging.py @@ -0,0 +1,135 @@ +from __future__ import annotations + +import ctypes +import os +import time +from typing import Callable + +from fastapi import FastAPI, Request + + +class _ProcessMemoryCounters(ctypes.Structure): + _fields_ = [ + ("cb", ctypes.c_ulong), + ("PageFaultCount", ctypes.c_ulong), + ("PeakWorkingSetSize", ctypes.c_size_t), + ("WorkingSetSize", ctypes.c_size_t), + ("QuotaPeakPagedPoolUsage", ctypes.c_size_t), + ("QuotaPagedPoolUsage", ctypes.c_size_t), + ("QuotaPeakNonPagedPoolUsage", ctypes.c_size_t), + ("QuotaNonPagedPoolUsage", ctypes.c_size_t), + ("PagefileUsage", ctypes.c_size_t), + ("PeakPagefileUsage", ctypes.c_size_t), + ] + + +def _get_process_cpu_seconds() -> float: + cpu_times = os.times() + return cpu_times.user + cpu_times.system + + +def _get_process_rss_bytes() -> int | None: + if os.name == "nt": + try: + kernel32 = ctypes.WinDLL("kernel32", use_last_error=True) + psapi = ctypes.WinDLL("psapi", use_last_error=True) + kernel32.GetCurrentProcess.restype = ctypes.c_void_p + psapi.GetProcessMemoryInfo.argtypes = [ctypes.c_void_p, ctypes.c_void_p, ctypes.c_ulong] + psapi.GetProcessMemoryInfo.restype = ctypes.c_int + + counters = _ProcessMemoryCounters() + counters.cb = ctypes.sizeof(_ProcessMemoryCounters) + ok = psapi.GetProcessMemoryInfo( + kernel32.GetCurrentProcess(), + ctypes.byref(counters), + counters.cb, + ) + if ok: + return int(counters.WorkingSetSize) + except Exception: + return None + return None + + try: + with open("/proc/self/statm", "r", encoding="ascii") as handle: + resident_pages = int(handle.read().split()[1]) + return resident_pages * os.sysconf("SC_PAGE_SIZE") + except Exception: + return None + + +def _format_megabytes(value: int | None) -> str: + if value is None: + return "unknown" + return f"{value / (1024 * 1024):.1f}" + + +def _format_megabyte_delta(start: int | None, end: int | None) -> str: + if start is None or end is None: + return "unknown" + return f"{(end - start) / (1024 * 1024):.1f}" + + +def _get_caller(request: Request) -> str: + forwarded_for = request.headers.get("x-forwarded-for") + if forwarded_for: + return forwarded_for.split(",", 1)[0].strip() + if request.client is not None: + return request.client.host + return "unknown" + + +def install_request_logging(app: FastAPI, logger, service_name: str) -> None: + @app.middleware("http") + async def log_request(request: Request, call_next: Callable): + started_at = time.perf_counter() + cpu_started = _get_process_cpu_seconds() + rss_started = _get_process_rss_bytes() + caller = _get_caller(request) + forwarded_for = request.headers.get("x-forwarded-for", "-") + user_agent = request.headers.get("user-agent", "-") + request_id = request.headers.get("x-request-id") or request.headers.get("x-correlation-id") or "-" + content_length = request.headers.get("content-length", "0") + error_type = "-" + status_code = 500 + + logger.info( + "incoming_request service=%s caller=%s forwarded_for=%s method=%s path=%s query=%s user_agent=%r request_id=%s content_length=%s", + service_name, + caller, + forwarded_for, + request.method, + request.url.path, + request.url.query or "-", + user_agent, + request_id, + content_length, + ) + + try: + response = await call_next(request) + status_code = response.status_code + return response + except Exception as exc: + error_type = type(exc).__name__ + raise + finally: + elapsed_ms = (time.perf_counter() - started_at) * 1000 + cpu_time_ms = max(0.0, (_get_process_cpu_seconds() - cpu_started) * 1000) + cpu_percent = 0.0 if elapsed_ms <= 0 else (cpu_time_ms / elapsed_ms) * 100 + rss_finished = _get_process_rss_bytes() + logger.info( + "completed_request service=%s caller=%s method=%s path=%s status_code=%s response_time_ms=%.1f cpu_time_ms=%.1f cpu_percent=%.1f rss_mb=%s rss_delta_mb=%s request_id=%s error_type=%s", + service_name, + caller, + request.method, + request.url.path, + status_code, + elapsed_ms, + cpu_time_ms, + cpu_percent, + _format_megabytes(rss_finished), + _format_megabyte_delta(rss_started, rss_finished), + request_id, + error_type, + ) \ No newline at end of file diff --git a/gateway/main.py b/gateway/main.py index 3e2212f..42e0e49 100644 --- a/gateway/main.py +++ b/gateway/main.py @@ -14,6 +14,8 @@ from fastapi.responses import JSONResponse, Response from starlette.middleware.base import BaseHTTPMiddleware from pydantic import BaseModel, Field, ValidationError, field_validator +from common.request_logging import install_request_logging + logger = logging.getLogger("gateway") logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s %(message)s") @@ -142,6 +144,7 @@ def _is_llm_path(path: str) -> bool: app = FastAPI(title="Skinbase Vision Gateway", version="1.0.0", lifespan=lifespan) app.add_middleware(APIKeyMiddleware) +install_request_logging(app, logger, "gateway") @app.exception_handler(LLMGatewayError) diff --git a/llm/main.py b/llm/main.py index e63498a..4675018 100644 --- a/llm/main.py +++ b/llm/main.py @@ -14,6 +14,8 @@ import httpx from fastapi import FastAPI, HTTPException, Request from fastapi.responses import JSONResponse +from common.request_logging import install_request_logging + logger = logging.getLogger("llm") logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s %(message)s") @@ -135,6 +137,7 @@ async def lifespan(app: FastAPI): app = FastAPI(title="Skinbase LLM Service", version="1.0.0", lifespan=lifespan) +install_request_logging(app, logger, "llm") def _health_payload(status: str) -> Dict[str, Any]: diff --git a/tests/test_gateway_llm.py b/tests/test_gateway_llm.py index 5452edb..0dcbfba 100644 --- a/tests/test_gateway_llm.py +++ b/tests/test_gateway_llm.py @@ -85,15 +85,27 @@ class GatewayLLMTests(unittest.IsolatedAsyncioTestCase): async def test_llm_endpoint_requires_api_key(self): module = load_gateway_module(llm_enabled=True) - response = await self._request( - module, - "POST", - "/ai/chat", - json_payload={"messages": [{"role": "user", "content": "hello"}]}, - ) + with self.assertLogs("gateway", level="INFO") as captured: + response = await self._request( + module, + "POST", + "/ai/chat", + json_payload={"messages": [{"role": "user", "content": "hello"}]}, + ) self.assertEqual(response.status_code, 401) self.assertEqual(response.json()["error"]["code"], "unauthorized") + joined_logs = "\n".join(captured.output) + self.assertIn("incoming_request service=gateway", joined_logs) + self.assertIn("completed_request service=gateway", joined_logs) + self.assertIn("caller=127.0.0.1", joined_logs) + self.assertIn("method=POST", joined_logs) + self.assertIn("path=/ai/chat", joined_logs) + self.assertIn("status_code=401", joined_logs) + self.assertIn("response_time_ms=", joined_logs) + self.assertIn("cpu_time_ms=", joined_logs) + self.assertIn("cpu_percent=", joined_logs) + self.assertIn("rss_mb=", joined_logs) async def test_llm_disabled_returns_503(self): module = load_gateway_module(llm_enabled=False) diff --git a/tests/test_llm_service.py b/tests/test_llm_service.py index e7c906b..22e5a8b 100644 --- a/tests/test_llm_service.py +++ b/tests/test_llm_service.py @@ -44,8 +44,9 @@ class LLMServiceTests(unittest.IsolatedAsyncioTestCase): ) transport = httpx.ASGITransport(app=module.app) - async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client: - response = await client.get("/health") + with self.assertLogs("llm", level="INFO") as captured: + async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client: + response = await client.get("/health") self.assertEqual(response.status_code, 200) self.assertEqual( @@ -59,6 +60,17 @@ class LLMServiceTests(unittest.IsolatedAsyncioTestCase): "gpu_layers": 0, }, ) + joined_logs = "\n".join(captured.output) + self.assertIn("incoming_request service=llm", joined_logs) + self.assertIn("completed_request service=llm", joined_logs) + self.assertIn("caller=127.0.0.1", joined_logs) + self.assertIn("method=GET", joined_logs) + self.assertIn("path=/health", joined_logs) + self.assertIn("status_code=200", joined_logs) + self.assertIn("response_time_ms=", joined_logs) + self.assertIn("cpu_time_ms=", joined_logs) + self.assertIn("cpu_percent=", joined_logs) + self.assertIn("rss_mb=", joined_logs) async def test_health_reports_unavailable_when_process_is_down(self): module = load_llm_module()