100 lines
2.7 KiB
Python
100 lines
2.7 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
from time import perf_counter
|
|
from typing import Any, Mapping
|
|
|
|
|
|
logger = logging.getLogger("cmhub.api.generation_usage")
|
|
|
|
EVENT_NAME = "generation_route_usage"
|
|
MAX_CLIENT_VERSION_LENGTH = 64
|
|
MAX_ALIAS_LENGTH = 64
|
|
|
|
|
|
def telemetry_start_time() -> float:
|
|
return perf_counter()
|
|
|
|
|
|
def telemetry_elapsed_ms(started: float) -> int:
|
|
return max(0, int((perf_counter() - started) * 1000))
|
|
|
|
|
|
def request_alias(data: Mapping[str, Any]) -> str:
|
|
try:
|
|
return normalize_text(data.get("model", ""), MAX_ALIAS_LENGTH)
|
|
except AttributeError:
|
|
return ""
|
|
|
|
|
|
def log_generation_route_usage(
|
|
*,
|
|
route_type: str,
|
|
request,
|
|
alias: str,
|
|
status: str,
|
|
latency_ms: int,
|
|
error_code: str = "",
|
|
http_status: int | None = None,
|
|
) -> dict[str, Any]:
|
|
event = build_generation_route_usage_event(
|
|
route_type=route_type,
|
|
request=request,
|
|
alias=alias,
|
|
status=status,
|
|
latency_ms=latency_ms,
|
|
error_code=error_code,
|
|
http_status=http_status,
|
|
)
|
|
logger.info(
|
|
"%s %s",
|
|
EVENT_NAME,
|
|
json.dumps(event, ensure_ascii=False, sort_keys=True),
|
|
extra={"generation_route_usage": event},
|
|
)
|
|
return event
|
|
|
|
|
|
def build_generation_route_usage_event(
|
|
*,
|
|
route_type: str,
|
|
request,
|
|
alias: str,
|
|
status: str,
|
|
latency_ms: int,
|
|
error_code: str = "",
|
|
http_status: int | None = None,
|
|
) -> dict[str, Any]:
|
|
api_key = getattr(request, "auth", None)
|
|
user = getattr(request, "user", None)
|
|
client_device = getattr(request, "client_device", None)
|
|
return {
|
|
"event": EVENT_NAME,
|
|
"route_type": normalize_text(route_type, 16),
|
|
"api_key_id": getattr(api_key, "id", None),
|
|
"api_key_prefix": normalize_text(getattr(api_key, "key_prefix", ""), 32),
|
|
"user_id": getattr(user, "id", None),
|
|
"product_code": normalize_text(
|
|
getattr(client_device, "product_code", ""),
|
|
32,
|
|
),
|
|
"client_device_id": getattr(client_device, "id", None),
|
|
"device_session_present": bool(
|
|
getattr(request, "device_session_present", False)
|
|
),
|
|
"client_version": normalize_text(
|
|
request.headers.get("X-Client-Version", ""),
|
|
MAX_CLIENT_VERSION_LENGTH,
|
|
),
|
|
"alias": normalize_text(alias, MAX_ALIAS_LENGTH),
|
|
"status": normalize_text(status, 32),
|
|
"latency_ms": max(0, int(latency_ms)),
|
|
"error_code": normalize_text(error_code, 64),
|
|
"http_status": http_status,
|
|
}
|
|
|
|
|
|
def normalize_text(value: Any, max_length: int) -> str:
|
|
return str(value or "").strip()[:max_length]
|