Minor updates
This commit is contained in:
@@ -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,
|
||||||
|
)
|
||||||
@@ -14,6 +14,8 @@ from fastapi.responses import JSONResponse, Response
|
|||||||
from starlette.middleware.base import BaseHTTPMiddleware
|
from starlette.middleware.base import BaseHTTPMiddleware
|
||||||
from pydantic import BaseModel, Field, ValidationError, field_validator
|
from pydantic import BaseModel, Field, ValidationError, field_validator
|
||||||
|
|
||||||
|
from common.request_logging import install_request_logging
|
||||||
|
|
||||||
logger = logging.getLogger("gateway")
|
logger = logging.getLogger("gateway")
|
||||||
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s %(message)s")
|
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 = FastAPI(title="Skinbase Vision Gateway", version="1.0.0", lifespan=lifespan)
|
||||||
app.add_middleware(APIKeyMiddleware)
|
app.add_middleware(APIKeyMiddleware)
|
||||||
|
install_request_logging(app, logger, "gateway")
|
||||||
|
|
||||||
|
|
||||||
@app.exception_handler(LLMGatewayError)
|
@app.exception_handler(LLMGatewayError)
|
||||||
|
|||||||
@@ -14,6 +14,8 @@ import httpx
|
|||||||
from fastapi import FastAPI, HTTPException, Request
|
from fastapi import FastAPI, HTTPException, Request
|
||||||
from fastapi.responses import JSONResponse
|
from fastapi.responses import JSONResponse
|
||||||
|
|
||||||
|
from common.request_logging import install_request_logging
|
||||||
|
|
||||||
logger = logging.getLogger("llm")
|
logger = logging.getLogger("llm")
|
||||||
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s %(message)s")
|
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)
|
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]:
|
def _health_payload(status: str) -> Dict[str, Any]:
|
||||||
|
|||||||
@@ -85,15 +85,27 @@ class GatewayLLMTests(unittest.IsolatedAsyncioTestCase):
|
|||||||
async def test_llm_endpoint_requires_api_key(self):
|
async def test_llm_endpoint_requires_api_key(self):
|
||||||
module = load_gateway_module(llm_enabled=True)
|
module = load_gateway_module(llm_enabled=True)
|
||||||
|
|
||||||
response = await self._request(
|
with self.assertLogs("gateway", level="INFO") as captured:
|
||||||
module,
|
response = await self._request(
|
||||||
"POST",
|
module,
|
||||||
"/ai/chat",
|
"POST",
|
||||||
json_payload={"messages": [{"role": "user", "content": "hello"}]},
|
"/ai/chat",
|
||||||
)
|
json_payload={"messages": [{"role": "user", "content": "hello"}]},
|
||||||
|
)
|
||||||
|
|
||||||
self.assertEqual(response.status_code, 401)
|
self.assertEqual(response.status_code, 401)
|
||||||
self.assertEqual(response.json()["error"]["code"], "unauthorized")
|
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):
|
async def test_llm_disabled_returns_503(self):
|
||||||
module = load_gateway_module(llm_enabled=False)
|
module = load_gateway_module(llm_enabled=False)
|
||||||
|
|||||||
@@ -44,8 +44,9 @@ class LLMServiceTests(unittest.IsolatedAsyncioTestCase):
|
|||||||
)
|
)
|
||||||
|
|
||||||
transport = httpx.ASGITransport(app=module.app)
|
transport = httpx.ASGITransport(app=module.app)
|
||||||
async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
|
with self.assertLogs("llm", level="INFO") as captured:
|
||||||
response = await client.get("/health")
|
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(response.status_code, 200)
|
||||||
self.assertEqual(
|
self.assertEqual(
|
||||||
@@ -59,6 +60,17 @@ class LLMServiceTests(unittest.IsolatedAsyncioTestCase):
|
|||||||
"gpu_layers": 0,
|
"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):
|
async def test_health_reports_unavailable_when_process_is_down(self):
|
||||||
module = load_llm_module()
|
module = load_llm_module()
|
||||||
|
|||||||
Reference in New Issue
Block a user