diff --git a/routstr/core/admin.py b/routstr/core/admin.py index 3baad78b..032b33ae 100644 --- a/routstr/core/admin.py +++ b/routstr/core/admin.py @@ -3,7 +3,7 @@ import secrets from datetime import datetime, timezone from pathlib import Path -from fastapi import APIRouter, Depends, HTTPException, Request +from fastapi import APIRouter, Depends, HTTPException, Query, Request from fastapi.responses import HTMLResponse, RedirectResponse from pydantic import BaseModel from sqlmodel import select @@ -20,6 +20,7 @@ from ..wallet import ( from .db import ApiKey, ModelRow, UpstreamProviderRow, create_session from .logging import get_logger from .settings import SettingsService, settings +from .usage_metrics import UsageMetricsService, list_metric_definitions logger = get_logger(__name__) @@ -168,6 +169,38 @@ async def get_balances_api(request: Request) -> list[dict[str, object]]: return [dict(d) for d in balance_details] +@admin_router.get( + "/api/usage-metrics/definitions", dependencies=[Depends(require_admin_api)] +) +async def get_usage_metric_definitions() -> list[dict[str, str]]: + return list_metric_definitions() + + +@admin_router.get("/api/usage-metrics", dependencies=[Depends(require_admin_api)]) +async def get_usage_metrics( + metrics: str | None = Query( + default=None, + description="Comma-separated list of usage metric identifiers to include", + ), + bucket_minutes: int = Query(default=15, ge=1, le=24 * 60), + hours: int = Query(default=24, ge=1, le=24 * 7), +) -> dict[str, object]: + metric_list = ( + [item.strip() for item in metrics.split(",") if item.strip()] + if metrics + else [] + ) + try: + result = await UsageMetricsService.collect( + metrics=metric_list, + bucket_minutes=bucket_minutes, + hours=hours, + ) + except ValueError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + return result.to_dict() + + @admin_router.get("/api/settings", dependencies=[Depends(require_admin_api)]) async def get_settings(request: Request) -> dict: data = settings.dict() diff --git a/routstr/core/usage_metrics.py b/routstr/core/usage_metrics.py new file mode 100644 index 00000000..6ab243b5 --- /dev/null +++ b/routstr/core/usage_metrics.py @@ -0,0 +1,264 @@ +from __future__ import annotations + +import asyncio +import json +import math +from dataclasses import asdict, dataclass +from datetime import datetime, timedelta +from pathlib import Path +from typing import Callable, Literal, cast + +UsageMetricName = Literal["errors", "chat_completions_success"] + + +@dataclass(frozen=True) +class MetricDefinition: + name: UsageMetricName + label: str + description: str + matcher: Callable[[dict[str, object]], bool] + + +@dataclass(frozen=True) +class UsageMetricPoint: + bucket_start: datetime + count: int + + +@dataclass(frozen=True) +class UsageMetricSeries: + name: UsageMetricName + label: str + description: str + total: int + points: list[UsageMetricPoint] + + +@dataclass(frozen=True) +class UsageMetricsComputation: + bucket_minutes: int + bucket_count: int + start: datetime + end: datetime + series: list[UsageMetricSeries] + + def to_dict(self) -> dict[str, object]: + return asdict(self) + + +def _is_error(record: dict[str, object]) -> bool: + level = record.get("levelname") + if not isinstance(level, str): + return False + return level.upper() in {"ERROR", "CRITICAL"} + + +def _coerce_int(value: object) -> int | None: + if isinstance(value, int): + return value + if isinstance(value, str): + try: + return int(value.strip()) + except ValueError: + return None + return None + + +def _is_successful_chat_completion(record: dict[str, object]) -> bool: + message = record.get("message") + if not isinstance(message, str) or message != "Received upstream response": + return False + status_code = _coerce_int(record.get("status_code")) + if status_code != 200: + return False + path = record.get("path") + return isinstance(path, str) and path.endswith("chat/completions") + + +METRIC_DEFINITIONS: dict[UsageMetricName, MetricDefinition] = { + "errors": MetricDefinition( + name="errors", + label="Errors", + description="Log entries emitted at ERROR or CRITICAL level", + matcher=_is_error, + ), + "chat_completions_success": MetricDefinition( + name="chat_completions_success", + label="200 chat/completions", + description="Successful upstream responses for chat/completions", + matcher=_is_successful_chat_completion, + ), +} + + +def list_metric_definitions() -> list[dict[str, str]]: + return [ + { + "name": definition.name, + "label": definition.label, + "description": definition.description, + } + for definition in METRIC_DEFINITIONS.values() + ] + + +def default_metric_names() -> list[UsageMetricName]: + return list(METRIC_DEFINITIONS.keys()) + + +class UsageMetricsService: + @classmethod + async def collect( + cls, + metrics: list[str], + bucket_minutes: int, + hours: int, + log_dir: Path | None = None, + now: datetime | None = None, + ) -> UsageMetricsComputation: + return await asyncio.to_thread( + cls._collect_sync, metrics, bucket_minutes, hours, log_dir, now + ) + + @classmethod + def _collect_sync( + cls, + metrics: list[str], + bucket_minutes: int, + hours: int, + log_dir: Path | None, + now: datetime | None, + ) -> UsageMetricsComputation: + metric_list = cls._sanitize_metrics(metrics) + bucket_minutes = cls._clamp(bucket_minutes, minimum=1, maximum=24 * 60) + hours = cls._clamp(hours, minimum=1, maximum=24 * 7) + now_dt = now or datetime.now() + start = now_dt - timedelta(hours=hours) + bucket_seconds = bucket_minutes * 60 + bucket_count = max(1, math.ceil((hours * 3600) / bucket_seconds)) + bucket_starts = [ + start + timedelta(seconds=bucket_seconds * index) + for index in range(bucket_count) + ] + counts: dict[UsageMetricName, list[int]] = { + name: [0] * bucket_count for name in metric_list + } + + for log_file in cls._iter_log_files(log_dir): + try: + with log_file.open("r", encoding="utf-8") as handle: + for line in handle: + record = cls._parse_record(line) + if not record: + continue + timestamp = cls._parse_timestamp(record.get("asctime")) + if timestamp is None or timestamp < start or timestamp > now_dt: + continue + bucket_index = cls._bucket_index( + timestamp, start, bucket_seconds, bucket_count + ) + if bucket_index is None: + continue + for metric_name in metric_list: + definition = METRIC_DEFINITIONS[metric_name] + if definition.matcher(record): + counts[metric_name][bucket_index] += 1 + except (OSError, UnicodeDecodeError): + continue + + series = [ + UsageMetricSeries( + name=metric_name, + label=METRIC_DEFINITIONS[metric_name].label, + description=METRIC_DEFINITIONS[metric_name].description, + total=sum(counts[metric_name]), + points=[ + UsageMetricPoint(bucket_start=bucket_starts[index], count=count) + for index, count in enumerate(counts[metric_name]) + ], + ) + for metric_name in metric_list + ] + + return UsageMetricsComputation( + bucket_minutes=bucket_minutes, + bucket_count=bucket_count, + start=start, + end=now_dt, + series=series, + ) + + @staticmethod + def _sanitize_metrics(metrics: list[str]) -> list[UsageMetricName]: + requested = [metric.strip() for metric in metrics if metric.strip()] + unique: list[UsageMetricName] = [] + for normalized in requested: + if normalized not in METRIC_DEFINITIONS: + continue + metric_name = cast(UsageMetricName, normalized) + if metric_name not in unique: + unique.append(metric_name) + if unique: + return unique + if requested: + raise ValueError("No valid usage metrics requested") + return default_metric_names() + + @staticmethod + def _clamp(value: int, *, minimum: int, maximum: int) -> int: + return max(minimum, min(maximum, value)) + + @staticmethod + def _iter_log_files(log_dir: Path | None) -> list[Path]: + directory = log_dir or Path("logs") + if not directory.exists(): + return [] + log_files = sorted( + directory.glob("*.log"), + key=UsageMetricsService._safe_mtime, + reverse=True, + ) + return log_files + + @staticmethod + def _safe_mtime(path: Path) -> float: + try: + return path.stat().st_mtime + except OSError: + return 0.0 + + @staticmethod + def _parse_record(line: str) -> dict[str, object] | None: + stripped = line.strip() + if not stripped: + return None + try: + data = json.loads(stripped) + except json.JSONDecodeError: + return None + return data if isinstance(data, dict) else None + + @staticmethod + def _parse_timestamp(value: object) -> datetime | None: + if not isinstance(value, str): + return None + try: + return datetime.strptime(value, "%Y-%m-%d %H:%M:%S") + except ValueError: + return None + + @staticmethod + def _bucket_index( + timestamp: datetime, + start: datetime, + bucket_seconds: int, + bucket_count: int, + ) -> int | None: + delta_seconds = (timestamp - start).total_seconds() + if delta_seconds < 0: + return None + index = int(delta_seconds // bucket_seconds) + if index >= bucket_count: + index = bucket_count - 1 + return index + diff --git a/tests/unit/test_usage_metrics.py b/tests/unit/test_usage_metrics.py new file mode 100644 index 00000000..68f6afde --- /dev/null +++ b/tests/unit/test_usage_metrics.py @@ -0,0 +1,79 @@ +from __future__ import annotations + +import json +from datetime import datetime, timedelta +from pathlib import Path + +import pytest + +from routstr.core.usage_metrics import UsageMetricsService + + +def _write_records(path: Path, records: list[dict[str, object]]) -> None: + with path.open("w", encoding="utf-8") as handle: + for record in records: + handle.write(json.dumps(record) + "\n") + + +@pytest.mark.asyncio +async def test_usage_metrics_counts(tmp_path: Path) -> None: + log_dir = tmp_path / "logs" + log_dir.mkdir() + now = datetime(2025, 1, 1, 12, 0, 0) + records = [ + { + "asctime": (now - timedelta(minutes=5)).strftime("%Y-%m-%d %H:%M:%S"), + "levelname": "ERROR", + "message": "Proxy failure", + }, + { + "asctime": (now - timedelta(minutes=10)).strftime("%Y-%m-%d %H:%M:%S"), + "levelname": "INFO", + "message": "Received upstream response", + "status_code": 200, + "path": "chat/completions", + }, + { + "asctime": (now - timedelta(minutes=20)).strftime("%Y-%m-%d %H:%M:%S"), + "levelname": "INFO", + "message": "Received upstream response", + "status_code": 500, + "path": "chat/completions", + }, + ] + _write_records(log_dir / "app_2025-01-01.log", records) + + result = await UsageMetricsService.collect( + metrics=["errors", "chat_completions_success"], + bucket_minutes=15, + hours=1, + log_dir=log_dir, + now=now, + ) + + errors_series = next(series for series in result.series if series.name == "errors") + completions_series = next( + series + for series in result.series + if series.name == "chat_completions_success" + ) + + assert errors_series.total == 1 + assert sum(point.count for point in errors_series.points) == 1 + assert completions_series.total == 1 + assert any(point.count == 1 for point in completions_series.points) + + +@pytest.mark.asyncio +async def test_usage_metrics_invalid_metric(tmp_path: Path) -> None: + log_dir = tmp_path / "logs" + log_dir.mkdir() + now = datetime(2025, 1, 1, 12, 0, 0) + with pytest.raises(ValueError): + await UsageMetricsService.collect( + metrics=["unknown"], + bucket_minutes=15, + hours=1, + log_dir=log_dir, + now=now, + ) diff --git a/ui/app/page.tsx b/ui/app/page.tsx index da8ffba3..5e7d6514 100644 --- a/ui/app/page.tsx +++ b/ui/app/page.tsx @@ -10,6 +10,7 @@ import { TemporaryBalances } from '@/components/temporary-balances'; import { ToggleGroup, ToggleGroupItem } from '@/components/ui/toggle-group'; import type { DisplayUnit } from '@/lib/types/units'; import { fetchBtcUsdPrice, btcToSatsRate } from '@/lib/exchange-rate'; +import { UsageTracking } from '@/components/usage-tracking'; export default function Page() { const [displayUnit, setDisplayUnit] = useState('sat'); @@ -65,22 +66,25 @@ export default function Page() { -
-
- +
+
+ +
+
+ +
+
+ +
-
- -
-
diff --git a/ui/components/usage-tracking.tsx b/ui/components/usage-tracking.tsx new file mode 100644 index 00000000..3be3902f --- /dev/null +++ b/ui/components/usage-tracking.tsx @@ -0,0 +1,417 @@ +'use client'; + +import { useEffect, useMemo, useState } from 'react'; +import { useQuery } from '@tanstack/react-query'; +import { + Activity, + AlertCircle, + BarChart3, + RefreshCw, + TrendingDown, + TrendingUp, +} from 'lucide-react'; +import { + AdminService, + UsageMetricDefinition, + UsageMetricName, +} from '@/lib/api/services/admin'; +import { + Card, + CardContent, + CardDescription, + CardHeader, + CardTitle, +} from '@/components/ui/card'; +import { Button } from '@/components/ui/button'; +import { Badge } from '@/components/ui/badge'; +import { + Select, + SelectContent, + SelectItem, + SelectTrigger, + SelectValue, +} from '@/components/ui/select'; +import { + ChartConfig, + ChartContainer, + ChartTooltip, + ChartTooltipContent, +} from '@/components/ui/chart'; +import { Area, AreaChart, CartesianGrid, XAxis } from 'recharts'; +import { Skeleton } from '@/components/ui/skeleton'; +import { cn } from '@/lib/utils'; + +const bucketOptions = [ + { label: '15 minutes', value: 15 }, + { label: '1 hour', value: 60 }, +]; + +const rangeOptions = [ + { label: 'Last 6 hours', value: 6 }, + { label: 'Last 24 hours', value: 24 }, + { label: 'Last 72 hours', value: 72 }, +]; + +const palette = [ + 'hsl(var(--chart-1))', + 'hsl(var(--chart-2))', + 'hsl(var(--chart-3))', + 'hsl(var(--chart-4))', + 'hsl(var(--chart-5))', +]; + +const defaultMetrics: UsageMetricName[] = [ + 'errors', + 'chat_completions_success', +]; + +export function UsageTracking() { + const [bucketMinutes, setBucketMinutes] = useState(15); + const [hours, setHours] = useState(24); + const [selectedMetrics, setSelectedMetrics] = + useState(defaultMetrics); + + const { + data: metricDefinitions, + isLoading: definitionsLoading, + isError: definitionsError, + } = useQuery({ + queryKey: ['usage-metric-definitions'], + queryFn: async () => AdminService.getUsageMetricDefinitions(), + staleTime: 10 * 60 * 1000, + }); + + useEffect(() => { + if (!metricDefinitions || !metricDefinitions.length) { + return; + } + setSelectedMetrics((current) => { + const filtered = current.filter((metric) => + metricDefinitions.some((definition) => definition.name === metric) + ) as UsageMetricName[]; + if (filtered.length) { + return filtered; + } + return metricDefinitions.map( + (definition) => definition.name as UsageMetricName + ); + }); + }, [metricDefinitions]); + + const metricsKey = useMemo( + () => [...selectedMetrics].sort().join(','), + [selectedMetrics] + ); + + const { + data: metricsData, + isLoading: metricsLoading, + isFetching: metricsFetching, + isError: metricsError, + error: metricsErrorObject, + refetch: refetchMetrics, + } = useQuery({ + queryKey: ['usage-metrics', metricsKey, bucketMinutes, hours], + queryFn: async () => + AdminService.getUsageMetrics({ + metrics: selectedMetrics, + bucket_minutes: bucketMinutes, + hours, + }), + enabled: selectedMetrics.length > 0, + refetchInterval: 60_000, + keepPreviousData: true, + }); + + const colorMap = useMemo(() => { + const map: Partial> = {}; + metricDefinitions?.forEach((definition, index) => { + map[definition.name] = palette[index % palette.length]; + }); + return map; + }, [metricDefinitions]); + + const chartConfig = useMemo(() => { + const config: ChartConfig = {}; + metricDefinitions?.forEach((definition) => { + const color = colorMap[definition.name]; + config[definition.name] = { + label: definition.label, + color, + }; + }); + return config; + }, [metricDefinitions, colorMap]); + + const chartData = useMemo(() => { + if (!metricsData || !metricsData.series.length) { + return []; + } + const referenceSeries = metricsData.series[0]; + return referenceSeries.points.map((point, index) => { + const base: Record = { + bucketStart: point.bucket_start, + label: formatBucketLabel(point.bucket_start, hours), + }; + metricsData.series.forEach((series) => { + base[series.name] = series.points[index]?.count ?? 0; + }); + return base; + }); + }, [metricsData, hours]); + + const summary = useMemo(() => { + if (!metricsData) { + return []; + } + return metricsData.series.map((series) => ({ + name: series.name, + label: series.label, + total: series.total, + latest: series.points.at(-1)?.count ?? 0, + averagePerHour: series.total / Math.max(1, hours), + })); + }, [metricsData, hours]); + + const handleMetricToggle = (metric: UsageMetricName) => { + setSelectedMetrics((current) => { + if (current.includes(metric)) { + if (current.length === 1) { + return current; + } + return current.filter((item) => item !== metric); + } + return [...current, metric]; + }); + }; + + const renderSummaryCard = (definition: UsageMetricDefinition) => { + const stat = summary.find((item) => item.name === definition.name); + const Icon = definition.name === 'errors' ? AlertCircle : Activity; + return ( +
+
+
+ + {definition.label} +
+ + {bucketMinutes >= 60 + ? `${bucketMinutes / 60}h buckets` + : `${bucketMinutes}m buckets`} + +
+
+
+
+ {stat ? stat.total.toLocaleString() : '—'} +
+

+ total events in range +

+
+ {stat && ( +
+
+ + {stat.latest} latest bucket +
+
+ + {stat.averagePerHour.toFixed(2)} avg/hr +
+
+ )} +
+
+ ); + }; + + const isLoading = + definitionsLoading || + metricsLoading || + (selectedMetrics.length > 0 && !metricsData); + + return ( + + +
+
+ + + Usage Tracking + + + Monitor errors and successful upstream traffic over time + +
+
+ + + +
+
+ + + {definitionsError ? ( +
+ + Failed to load metric definitions. +
+ ) : ( +
+ {metricDefinitions?.map((definition) => ( + + ))} +
+ )} + + {metricsError && ( +
+ + {metricsErrorObject instanceof Error + ? metricsErrorObject.message + : 'Failed to load usage metrics.'} +
+ )} + + {isLoading ? ( + + ) : ( + metricsData && + summary.length > 0 && ( +
+ {metricDefinitions + ?.filter((definition) => + selectedMetrics.includes(definition.name as UsageMetricName) + ) + .map((definition) => renderSummaryCard(definition))} +
+ ) + )} + + {isLoading ? ( + + ) : chartData.length > 0 ? ( + + + + + value as string} + /> + } + /> + {metricsData?.series.map((series) => ( + + ))} + + + ) : ( +
+ +

No data available for the selected window.

+
+ )} + + + ); +} + +function formatBucketLabel(value: string | undefined, rangeHours: number) { + if (!value) { + return ''; + } + const date = new Date(value); + if (Number.isNaN(date.getTime())) { + return value; + } + if (rangeHours <= 24) { + return date.toLocaleTimeString([], { hour: '2-digit', minute: '2-digit' }); + } + return date.toLocaleString([], { + month: 'short', + day: 'numeric', + hour: '2-digit', + }); +} diff --git a/ui/lib/api/services/admin.ts b/ui/lib/api/services/admin.ts index 9645a6a1..cbf7571f 100644 --- a/ui/lib/api/services/admin.ts +++ b/ui/lib/api/services/admin.ts @@ -802,6 +802,30 @@ export class AdminService { '/admin/api/temporary-balances' ); } + + static async getUsageMetricDefinitions(): Promise { + return await apiClient.get( + '/admin/api/usage-metrics/definitions' + ); + } + + static async getUsageMetrics(params: { + metrics: UsageMetricName[]; + bucket_minutes: number; + hours: number; + }): Promise { + const query: Record = { + bucket_minutes: params.bucket_minutes, + hours: params.hours, + }; + if (params.metrics.length > 0) { + query.metrics = params.metrics.join(','); + } + return await apiClient.get( + '/admin/api/usage-metrics', + query + ); + } } export const TemporaryBalanceSchema = z.object({ @@ -814,3 +838,32 @@ export const TemporaryBalanceSchema = z.object({ }); export type TemporaryBalance = z.infer; + +export type UsageMetricName = 'errors' | 'chat_completions_success'; + +export interface UsageMetricDefinition { + name: UsageMetricName; + label: string; + description: string; +} + +export interface UsageMetricPoint { + bucket_start: string; + count: number; +} + +export interface UsageMetricSeries { + name: UsageMetricName; + label: string; + description: string; + total: number; + points: UsageMetricPoint[]; +} + +export interface UsageMetricsResponse { + bucket_minutes: number; + bucket_count: number; + start: string; + end: string; + series: UsageMetricSeries[]; +}