feat: 重构接口日志页面与接口监控看板,支持HTTP方法过滤和实时轮询
This commit is contained in:
@@ -0,0 +1,74 @@
|
||||
"""Log external API calls to the database and console."""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from datetime import datetime, timezone
|
||||
from typing import Optional
|
||||
|
||||
from app.database import SessionLocal
|
||||
from app.models.external_api_log import ExternalApiLog
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def log_external_api_call(
|
||||
direction: str,
|
||||
target_type: str,
|
||||
method: str,
|
||||
path: str,
|
||||
url_host: str,
|
||||
duration_ms: int,
|
||||
status_code: Optional[int] = None,
|
||||
success: bool = True,
|
||||
target_id: Optional[int] = None,
|
||||
target_name: Optional[str] = None,
|
||||
error_type: Optional[str] = None,
|
||||
error_message: Optional[str] = None,
|
||||
) -> None:
|
||||
"""Persist an external API call log entry and write a structured console log.
|
||||
|
||||
No request/response bodies, tokens, cookies, or API keys are recorded.
|
||||
"""
|
||||
# Console log (structured, for Docker log aggregation)
|
||||
logger.info(
|
||||
"api_log direction=%s target_type=%s target_id=%s target_name=%s "
|
||||
"method=%s path=%s url_host=%s status_code=%s success=%s "
|
||||
"duration_ms=%d error_type=%s error_message=%s",
|
||||
direction,
|
||||
target_type,
|
||||
target_id,
|
||||
target_name or "",
|
||||
method,
|
||||
path,
|
||||
url_host,
|
||||
status_code or "",
|
||||
success,
|
||||
duration_ms,
|
||||
error_type or "",
|
||||
(error_message or "")[:200],
|
||||
)
|
||||
|
||||
# DB persistence
|
||||
db = SessionLocal()
|
||||
try:
|
||||
log = ExternalApiLog(
|
||||
direction=direction,
|
||||
target_type=target_type,
|
||||
target_id=target_id,
|
||||
target_name=target_name,
|
||||
method=method,
|
||||
path=path,
|
||||
url_host=url_host,
|
||||
status_code=status_code,
|
||||
success=success,
|
||||
duration_ms=duration_ms,
|
||||
error_type=error_type,
|
||||
error_message=error_message,
|
||||
created_at=datetime.now(timezone.utc),
|
||||
)
|
||||
db.add(log)
|
||||
db.commit()
|
||||
except Exception:
|
||||
logger.exception("failed to persist external API log entry")
|
||||
finally:
|
||||
db.close()
|
||||
@@ -19,6 +19,7 @@ from app.services.snapshot_service import diff_snapshots, prune_snapshots
|
||||
from app.services import webhook_service
|
||||
from app.services import website_sync
|
||||
from app.config import get_settings
|
||||
from app.routers.external_api_logs import clean_expired_logs
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -77,6 +78,8 @@ def _check_upstream(upstream_id: int) -> None:
|
||||
auth_config=auth_config,
|
||||
timeout=float(upstream.timeout_seconds),
|
||||
on_auth_config_update=lambda updated: _persist_auth_config_update(upstream, updated),
|
||||
target_id=upstream.id,
|
||||
target_name=upstream.name,
|
||||
) as client:
|
||||
try:
|
||||
client.login()
|
||||
@@ -258,6 +261,8 @@ def _sync_upstream_keys(upstream_id: int, snapshot: dict[str, Any], captured_at:
|
||||
auth_config=auth_config,
|
||||
timeout=float(upstream.timeout_seconds),
|
||||
on_auth_config_update=lambda updated: _persist_auth_config_update(upstream, updated),
|
||||
target_id=upstream.id,
|
||||
target_name=upstream.name,
|
||||
) as client:
|
||||
client.login()
|
||||
remote_key_ids = website_sync._fetch_remote_managed_key_ids(db, client, upstream_id)
|
||||
@@ -328,11 +333,35 @@ def start_scheduler() -> None:
|
||||
upstreams = db.query(Upstream).filter(Upstream.enabled == True).all()
|
||||
for u in upstreams:
|
||||
refresh_upstream(u.id, u.check_interval_seconds, u.enabled)
|
||||
# Daily cleanup of expired external API logs
|
||||
_scheduler.add_job(
|
||||
_cleanup_external_logs,
|
||||
"interval",
|
||||
hours=24,
|
||||
id="cleanup_external_api_logs",
|
||||
replace_existing=True,
|
||||
coalesce=True,
|
||||
max_instances=1,
|
||||
misfire_grace_time=3600,
|
||||
)
|
||||
logger.info("scheduler started with %d upstream job(s)", len(upstreams))
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
def _cleanup_external_logs() -> None:
|
||||
"""Delete external API logs older than the configured retention period."""
|
||||
db = SessionLocal()
|
||||
try:
|
||||
deleted = clean_expired_logs(db)
|
||||
if deleted:
|
||||
logger.info("cleaned %d expired external API log(s)", deleted)
|
||||
except Exception:
|
||||
logger.exception("failed to clean expired external API logs")
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
def stop_scheduler() -> None:
|
||||
if _scheduler.running:
|
||||
_scheduler.shutdown(wait=True)
|
||||
|
||||
@@ -3,13 +3,18 @@ from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import logging
|
||||
import re
|
||||
import time
|
||||
from typing import Any, Callable, Optional
|
||||
from urllib.parse import urljoin
|
||||
from urllib.parse import urljoin, urlparse
|
||||
|
||||
import httpx
|
||||
|
||||
from app.utils.number import decimal_string
|
||||
from app.services.external_api_logger import log_external_api_call
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class UpstreamError(RuntimeError):
|
||||
@@ -429,6 +434,8 @@ class UpstreamClient:
|
||||
auth_config: dict[str, Any],
|
||||
timeout: float = 30.0,
|
||||
on_auth_config_update: Callable[[dict[str, Any]], None] | None = None,
|
||||
target_id: int | None = None,
|
||||
target_name: str | None = None,
|
||||
) -> None:
|
||||
self.base_url = base_url.rstrip("/")
|
||||
self.api_prefix = api_prefix.strip("/")
|
||||
@@ -436,6 +443,8 @@ class UpstreamClient:
|
||||
self.auth_config = auth_config
|
||||
self.timeout = timeout
|
||||
self.on_auth_config_update = on_auth_config_update
|
||||
self.target_id = target_id
|
||||
self.target_name = target_name
|
||||
self._token: str = ""
|
||||
self._cookies: dict[str, str] = {}
|
||||
self._new_api_user: str = ""
|
||||
@@ -549,6 +558,51 @@ class UpstreamClient:
|
||||
if changed and self.on_auth_config_update:
|
||||
self.on_auth_config_update(dict(self.auth_config))
|
||||
|
||||
def _do_request(self, method: str, url: str, **kwargs: Any) -> httpx.Response:
|
||||
"""Wrapped self._client.request() with timing and external-API logging.
|
||||
|
||||
Every HTTP call (initial, retry, refresh token) goes through here so
|
||||
the log is accurate and complete.
|
||||
|
||||
A response with status_code >= 400 is logged as a failure even though
|
||||
no exception was raised, so callers can check success/failure rates
|
||||
accurately. Callers still use raise_for_status() for flow control.
|
||||
"""
|
||||
started = time.monotonic()
|
||||
status_code: int | None = None
|
||||
error_type: str | None = None
|
||||
error_msg: str | None = None
|
||||
try:
|
||||
resp = self._client.request(method, url, **kwargs)
|
||||
status_code = getattr(resp, "status_code", None)
|
||||
if status_code is not None and status_code >= 400:
|
||||
error_type = "HTTPStatus"
|
||||
error_msg = f"HTTP {status_code}"
|
||||
return resp
|
||||
except Exception as exc:
|
||||
error_type = type(exc).__name__
|
||||
error_msg = str(exc)[:500]
|
||||
if hasattr(exc, "response") and hasattr(exc.response, "status_code"):
|
||||
status_code = exc.response.status_code
|
||||
raise
|
||||
finally:
|
||||
elapsed = int((time.monotonic() - started) * 1000)
|
||||
parsed = urlparse(url)
|
||||
log_external_api_call(
|
||||
direction="upstream",
|
||||
target_type="upstream",
|
||||
method=method,
|
||||
path=parsed.path or "",
|
||||
url_host=parsed.hostname or "",
|
||||
duration_ms=elapsed,
|
||||
status_code=status_code,
|
||||
success=error_type is None,
|
||||
target_id=self.target_id,
|
||||
target_name=self.target_name,
|
||||
error_type=error_type,
|
||||
error_message=error_msg,
|
||||
)
|
||||
|
||||
def _refresh_sub2api_bearer_token(self) -> bool:
|
||||
if not self._is_sub2api_bearer():
|
||||
return False
|
||||
@@ -556,7 +610,7 @@ class UpstreamClient:
|
||||
if not refresh_token:
|
||||
return False
|
||||
try:
|
||||
resp = self._client.request(
|
||||
resp = self._do_request(
|
||||
"POST",
|
||||
self._url("/auth/refresh"),
|
||||
json={"refresh_token": refresh_token},
|
||||
@@ -587,7 +641,7 @@ class UpstreamClient:
|
||||
allow_refresh: bool = True,
|
||||
**kwargs: Any,
|
||||
) -> httpx.Response:
|
||||
resp = self._client.request(
|
||||
resp = self._do_request(
|
||||
method,
|
||||
url,
|
||||
headers=self._headers(auth),
|
||||
@@ -602,7 +656,7 @@ class UpstreamClient:
|
||||
and self._is_sub2api_bearer()
|
||||
and self._refresh_sub2api_bearer_token()
|
||||
):
|
||||
resp = self._client.request(
|
||||
resp = self._do_request(
|
||||
method,
|
||||
url,
|
||||
headers=self._headers(auth),
|
||||
@@ -621,18 +675,9 @@ class UpstreamClient:
|
||||
raise UpstreamError("Nox-API endpoint requires Nox-Api-User; please fill user id and retry")
|
||||
url = self._url(path)
|
||||
if body is not None:
|
||||
resp = self._send_request(
|
||||
method,
|
||||
url,
|
||||
json=body,
|
||||
auth=auth,
|
||||
)
|
||||
resp = self._send_request(method, url, json=body, auth=auth)
|
||||
else:
|
||||
resp = self._send_request(
|
||||
method,
|
||||
url,
|
||||
auth=auth,
|
||||
)
|
||||
resp = self._send_request(method, url, auth=auth)
|
||||
resp.raise_for_status()
|
||||
ct = resp.headers.get("content-type", "")
|
||||
if not resp.content:
|
||||
@@ -689,11 +734,7 @@ class UpstreamClient:
|
||||
return items, meta
|
||||
|
||||
def _request_new_api_token_list(self, path: str, params: dict[str, Any]) -> tuple[list[dict[str, Any]], dict[str, Any]]:
|
||||
resp = self._send_request(
|
||||
"GET",
|
||||
self._url(path),
|
||||
params=params,
|
||||
)
|
||||
resp = self._send_request("GET", self._url(path), params=params)
|
||||
resp.raise_for_status()
|
||||
data = resp.json()
|
||||
self._ensure_api_success(data, "list New-API tokens")
|
||||
|
||||
@@ -1,13 +1,16 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import sys
|
||||
import time
|
||||
from decimal import Decimal, InvalidOperation, ROUND_HALF_UP
|
||||
from typing import Any
|
||||
from urllib.parse import quote
|
||||
from urllib.parse import quote, urlparse
|
||||
|
||||
import httpx
|
||||
|
||||
from app.utils.number import fixed_decimal_number, fixed_decimal_string
|
||||
from app.services.external_api_logger import log_external_api_call
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -136,12 +139,16 @@ class Sub2ApiWebsiteClient:
|
||||
auth_type: str,
|
||||
auth_config: dict[str, Any],
|
||||
timeout: float = 30.0,
|
||||
target_id: int | None = None,
|
||||
target_name: str | None = None,
|
||||
) -> None:
|
||||
self.base_url = base_url.rstrip("/")
|
||||
self.api_prefix = api_prefix.strip("/")
|
||||
self.auth_type = auth_type
|
||||
self.auth_config = auth_config
|
||||
self.timeout = timeout
|
||||
self.target_id = target_id
|
||||
self.target_name = target_name
|
||||
self._client = httpx.Client(timeout=timeout)
|
||||
|
||||
def close(self) -> None:
|
||||
@@ -171,21 +178,63 @@ class Sub2ApiWebsiteClient:
|
||||
return headers
|
||||
|
||||
def _request(self, method: str, path: str, body: Any = None) -> Any:
|
||||
url = self._url(path)
|
||||
started = time.monotonic()
|
||||
status_code: int | None = None
|
||||
error_type: str | None = None
|
||||
error_msg: str | None = None
|
||||
try:
|
||||
resp = self._client.request(method, self._url(path), json=body, headers=self._headers())
|
||||
resp.raise_for_status()
|
||||
except httpx.HTTPStatusError as exc:
|
||||
raise WebsiteError(_friendly_http_error(exc)) from exc
|
||||
except httpx.TimeoutException as exc:
|
||||
raise WebsiteError(_friendly_connection_error(exc)) from exc
|
||||
except httpx.ConnectError as exc:
|
||||
raise WebsiteError(_friendly_connection_error(exc)) from exc
|
||||
if not resp.content:
|
||||
return None
|
||||
text = resp.text
|
||||
if "application/json" not in resp.headers.get("content-type", "") and text.lstrip().startswith("<"):
|
||||
raise WebsiteError(f"{method} {path} 返回了 HTML,请检查接口地址是否正确")
|
||||
return resp.json()
|
||||
try:
|
||||
resp = self._client.request(method, url, json=body, headers=self._headers())
|
||||
except httpx.TimeoutException as exc:
|
||||
error_type, error_msg = type(exc).__name__, str(exc)[:500]
|
||||
raise WebsiteError(_friendly_connection_error(exc)) from exc
|
||||
except httpx.ConnectError as exc:
|
||||
error_type, error_msg = type(exc).__name__, str(exc)[:500]
|
||||
raise WebsiteError(_friendly_connection_error(exc)) from exc
|
||||
except httpx.HTTPStatusError as exc:
|
||||
error_type, error_msg = type(exc).__name__, str(exc)[:500]
|
||||
status_code = exc.response.status_code
|
||||
raise WebsiteError(_friendly_http_error(exc)) from exc
|
||||
status_code = getattr(resp, "status_code", None)
|
||||
try:
|
||||
resp.raise_for_status()
|
||||
except httpx.HTTPStatusError as exc:
|
||||
error_type, error_msg = type(exc).__name__, str(exc)[:500]
|
||||
status_code = exc.response.status_code
|
||||
raise WebsiteError(_friendly_http_error(exc)) from exc
|
||||
if not resp.content:
|
||||
return None
|
||||
text = resp.text
|
||||
if "application/json" not in resp.headers.get("content-type", "") and text.lstrip().startswith("<"):
|
||||
error_type = "WebsiteError"
|
||||
error_msg = f"{method} {path} returned HTML"
|
||||
raise WebsiteError(f"{method} {path} 返回了 HTML,请检查接口地址是否正确")
|
||||
return resp.json()
|
||||
except WebsiteError:
|
||||
if error_type is None:
|
||||
error_type = "WebsiteError"
|
||||
e = sys.exc_info()[1]
|
||||
if e:
|
||||
error_msg = str(e)[:500]
|
||||
raise
|
||||
finally:
|
||||
elapsed = int((time.monotonic() - started) * 1000)
|
||||
parsed = urlparse(url)
|
||||
log_external_api_call(
|
||||
direction="website",
|
||||
target_type="website",
|
||||
method=method,
|
||||
path=parsed.path or path,
|
||||
url_host=parsed.hostname or "",
|
||||
duration_ms=elapsed,
|
||||
status_code=status_code,
|
||||
success=error_type is None,
|
||||
target_id=self.target_id,
|
||||
target_name=self.target_name,
|
||||
error_type=error_type,
|
||||
error_message=error_msg,
|
||||
)
|
||||
|
||||
def get_groups(self, endpoint: str = "/groups") -> list[dict[str, Any]]:
|
||||
"""拉取分组列表,尝试 endpoint 和 fallback /groups/all。"""
|
||||
|
||||
@@ -93,6 +93,8 @@ def _client_for(website: Website) -> Sub2ApiWebsiteClient:
|
||||
auth_type=website.auth_type,
|
||||
auth_config=json.loads(website.auth_config_json or "{}"),
|
||||
timeout=float(website.timeout_seconds),
|
||||
target_id=website.id,
|
||||
target_name=website.name,
|
||||
)
|
||||
|
||||
|
||||
@@ -325,6 +327,8 @@ def _backfill_missing_target_groups_from_remote_accounts(
|
||||
auth_type=website.auth_type,
|
||||
auth_config=json.loads(website.auth_config_json or "{}"),
|
||||
timeout=float(website.timeout_seconds),
|
||||
target_id=website.id,
|
||||
target_name=website.name,
|
||||
) as client:
|
||||
accounts = client.list_accounts() or []
|
||||
except Exception as exc:
|
||||
@@ -838,6 +842,8 @@ def reconcile_upstream_keys_full(db: Session, upstream_id: int) -> bool:
|
||||
auth_config=auth_config,
|
||||
timeout=float(upstream.timeout_seconds),
|
||||
on_auth_config_update=lambda updated: _persist_upstream_auth_config(upstream, updated),
|
||||
target_id=upstream.id,
|
||||
target_name=upstream.name,
|
||||
) as client:
|
||||
client.login()
|
||||
# 获取远端 Key 列表(支持自定义 managed_prefix)
|
||||
|
||||
Reference in New Issue
Block a user