mirror of
https://github.com/unshackle-dl/unshackle.git
synced 2026-11-12 20:15:10 +00:00
feat(api): tag each error with a ref and count errors by service
Every structured error carries a short ref, appended to its message and logged with the request; /api/dashboard/status gains per-service error counts. Gzip now compresses in place.
This commit is contained in:
@@ -43,7 +43,7 @@ All are `GET`.
|
|||||||
|
|
||||||
| Path | Returns |
|
| Path | Returns |
|
||||||
|------|---------|
|
|------|---------|
|
||||||
| `/api/dashboard/status` | Version, `code_hash`, bind host and port, `mode` (`full`, `api_only`, `remote_only`), uptime, `requests_total`, `requests_rejected` (401s), `requests_by_key` (label → count), configured service count (the same services as `/api/dashboard/services`), session count, `max_sessions` (`null` when the operator configured no limit), `session_ttl`, job counts by status. The service count includes services that failed to import - use `/api/dashboard/services` for the number that loaded. |
|
| `/api/dashboard/status` | Version, `code_hash`, bind host and port, `mode` (`full`, `api_only`, `remote_only`), uptime, `requests_total`, `requests_rejected` (401s), `requests_by_key` (label → count), `errors` (error counts by service tag, then by `error_code`), configured service count (the same services as `/api/dashboard/services`), session count, `max_sessions` (`null` when the operator configured no limit), `session_ttl`, job counts by status. The service count includes services that failed to import - use `/api/dashboard/services` for the number that loaded. |
|
||||||
| `/api/dashboard/sessions` | Every live remote session: `id`, `owner` (username or masked key), `creator_ip`, `service`, `title_id`, `titles`, `tracks`, `title` (the display name of the title the client is on: the last one it asked tracks for, else the first resolved title), `auth_status`, `auth_error`, `server_account` (the server profile lent to the remote session, else `null`), `server_cdm` (`false` when the client's own device licenses the remote session), `server_cdm_max_height` (the tallest video the server CDM licenses live, else `null`), `client`, `actions`, `created_at` and `last_accessed` (ISO 8601) with `created_ts` and `last_accessed_ts` (Unix epoch, the same clock as log `ts`), `age_seconds`, `idle_seconds`, `log_seq`. `client` is whatever the remote client sent as `client` in its session create request. The CLI sends `version`, `code_hash` (the commit the client runs, `null` when its source cannot be read), `platform` and `argv`. `argv` is the command line the user ran: proxy and URL userinfo, secret query parameters and credential values become `***`, home and install paths shorten as in the logs, and the line is cut at 3000 characters. An older client sends less or nothing and the field is `{}`. `log_seq` is the last sequence number in the remote session's service log - poll `/api/dashboard/sessions/{id}/logs` when it changes. `actions` is the session's request log, newest last, capped at 500: `ts`, `method`, `action` (`titles`, `tracks`, `segments`, `license`, `prompt`, …), `query`, `status`, `ms`, `bytes_in`, `bytes_out`. |
|
| `/api/dashboard/sessions` | Every live remote session: `id`, `owner` (username or masked key), `creator_ip`, `service`, `title_id`, `titles`, `tracks`, `title` (the display name of the title the client is on: the last one it asked tracks for, else the first resolved title), `auth_status`, `auth_error`, `server_account` (the server profile lent to the remote session, else `null`), `server_cdm` (`false` when the client's own device licenses the remote session), `server_cdm_max_height` (the tallest video the server CDM licenses live, else `null`), `client`, `actions`, `created_at` and `last_accessed` (ISO 8601) with `created_ts` and `last_accessed_ts` (Unix epoch, the same clock as log `ts`), `age_seconds`, `idle_seconds`, `log_seq`. `client` is whatever the remote client sent as `client` in its session create request. The CLI sends `version`, `code_hash` (the commit the client runs, `null` when its source cannot be read), `platform` and `argv`. `argv` is the command line the user ran: proxy and URL userinfo, secret query parameters and credential values become `***`, home and install paths shorten as in the logs, and the line is cut at 3000 characters. An older client sends less or nothing and the field is `{}`. `log_seq` is the last sequence number in the remote session's service log - poll `/api/dashboard/sessions/{id}/logs` when it changes. `actions` is the session's request log, newest last, capped at 500: `ts`, `method`, `action` (`titles`, `tracks`, `segments`, `license`, `prompt`, …), `query`, `status`, `ms`, `bytes_in`, `bytes_out`. |
|
||||||
| `/api/dashboard/jobs` | Every download job with full detail, regardless of owner. Empty in `--remote-only` mode. |
|
| `/api/dashboard/jobs` | Every download job with full detail, regardless of owner. Empty in `--remote-only` mode. |
|
||||||
| `/api/dashboard/logs` | `{"seq": N, "records": [...]}`: the last 1000 log records. `?since=<seq>` returns only newer records; `?level=WARNING` sets the minimum level; `?logger=serve` keeps one logger and its children (`aiohttp.access` is the noisiest). `ts` is a Unix epoch in seconds. |
|
| `/api/dashboard/logs` | `{"seq": N, "records": [...]}`: the last 1000 log records. `?since=<seq>` returns only newer records; `?level=WARNING` sets the minimum level; `?logger=serve` keeps one logger and its children (`aiohttp.access` is the noisiest). `ts` is a Unix epoch in seconds. |
|
||||||
@@ -54,6 +54,11 @@ All are `GET`.
|
|||||||
| `/api/dashboard/health` | Preflight: whether this instance could finish a download. Cached 30 seconds. |
|
| `/api/dashboard/health` | Preflight: whether this instance could finish a download. Cached 30 seconds. |
|
||||||
| `/api/dashboard/events` | Server-Sent Events stream. With `?since=<seq>` it returns `{"seq", "stats", "events": [...]}` instead: every event after `seq` (the server keeps the last 20000, log records included, so a poll every few seconds is safe even under a busy access log) as one JSON burst, so a UI that polls every few seconds still sees a remote session that opened and closed in between. |
|
| `/api/dashboard/events` | Server-Sent Events stream. With `?since=<seq>` it returns `{"seq", "stats", "events": [...]}` instead: every event after `seq` (the server keeps the last 20000, log records included, so a poll every few seconds is safe even under a busy access log) as one JSON burst, so a UI that polls every few seconds still sees a remote session that opened and closed in between. |
|
||||||
|
|
||||||
|
`errors` counts every structured error response since the server started, for example
|
||||||
|
`{"EXAMPLE": {"SERVICE_ERROR": 4}, "-": {"SESSION_NOT_FOUND": 1}}`. The service is the one of
|
||||||
|
the remote session, else the `service` field of the JSON body. `-` holds an error with no
|
||||||
|
installed service. Each error also logs its [`ref`](errors.md#find-an-error-from-its-ref).
|
||||||
|
|
||||||
`requests_rejected` counts 401 responses; those never appear in `requests_by_key`. The
|
`requests_rejected` counts 401 responses; those never appear in `requests_by_key`. The
|
||||||
label is the `username` from the `users` entry when one is set, else the first four
|
label is the `username` from the `users` entry when one is set, else the first four
|
||||||
characters of the API key.
|
characters of the API key.
|
||||||
|
|||||||
@@ -25,7 +25,8 @@ A structured error is always a JSON object with `status` set to the literal stri
|
|||||||
{
|
{
|
||||||
"status": "error",
|
"status": "error",
|
||||||
"error_code": "INVALID_PARAMETERS",
|
"error_code": "INVALID_PARAMETERS",
|
||||||
"message": "vcodec must be one of: H264, H265, ...",
|
"message": "vcodec must be one of: H264, H265, ... (ref 7f3a2c)",
|
||||||
|
"ref": "7f3a2c",
|
||||||
"timestamp": "2026-07-03T12:34:56.789012+00:00",
|
"timestamp": "2026-07-03T12:34:56.789012+00:00",
|
||||||
"details": { "field": "vcodec" },
|
"details": { "field": "vcodec" },
|
||||||
"retryable": true,
|
"retryable": true,
|
||||||
@@ -40,7 +41,8 @@ A structured error is always a JSON object with `status` set to the literal stri
|
|||||||
|---|---|---|---|
|
|---|---|---|---|
|
||||||
| `status` | string | Yes | Always the literal `"error"`. Lets a client distinguish an error body from a success body at a glance. |
|
| `status` | string | Yes | Always the literal `"error"`. Lets a client distinguish an error body from a success body at a glance. |
|
||||||
| `error_code` | string | Yes | A stable, machine-readable code from the [catalogue below](#error-code-catalogue). Match on this, not on `message`. |
|
| `error_code` | string | Yes | A stable, machine-readable code from the [catalogue below](#error-code-catalogue). Match on this, not on `message`. |
|
||||||
| `message` | string | Yes | A human-readable description of what went wrong. Wording may change between versions, so do not parse it. |
|
| `message` | string | Yes | A human-readable description of what went wrong, followed by `(ref <ref>)`. Wording may change between versions, so do not parse it. |
|
||||||
|
| `ref` | string | Yes | A short random reference for this one error. The server logs it with the request, so an operator can find the record from a user's report. |
|
||||||
| `timestamp` | string | Yes | ISO 8601 timestamp (UTC) of when the error response was built. |
|
| `timestamp` | string | Yes | ISO 8601 timestamp (UTC) of when the error response was built. |
|
||||||
| `details` | object | Only when non-empty | Structured, error-specific context (for example the invalid field, a list of available titles, or the set of currently active jobs). |
|
| `details` | object | Only when non-empty | Structured, error-specific context (for example the invalid field, a list of available titles, or the set of currently active jobs). |
|
||||||
| `retryable` | boolean | Only when `true` | Present and set to `true` when retrying the same request may succeed later (network blips, rate limiting, transient outages). Absent otherwise. |
|
| `retryable` | boolean | Only when `true` | Present and set to `true` when retrying the same request may succeed later (network blips, rate limiting, transient outages). Absent otherwise. |
|
||||||
@@ -52,6 +54,17 @@ A structured error is always a JSON object with `status` set to the literal stri
|
|||||||
every code maps to a default status (see the catalogue), and the pairing is
|
every code maps to a default status (see the catalogue), and the pairing is
|
||||||
stable across releases.
|
stable across releases.
|
||||||
|
|
||||||
|
### Find an error from its `ref`
|
||||||
|
|
||||||
|
For each error response, the server logs one `WARNING` line on the `api` logger:
|
||||||
|
|
||||||
|
```text
|
||||||
|
Error ref 7f3a2c: SERVICE_ERROR 422 POST /api/session/<id>/license service=EXAMPLE user=alice: Licence request failed: ...
|
||||||
|
```
|
||||||
|
|
||||||
|
Search the server log for `Error ref <ref>`. The handler's traceback, when there is one,
|
||||||
|
is directly above that line. The [developer dashboard](dashboard.md) log view also shows it.
|
||||||
|
|
||||||
### The `details` object
|
### The `details` object
|
||||||
|
|
||||||
When present, `details` carries extra context specific to the error. It is only
|
When present, `details` carries extra context specific to the error. It is only
|
||||||
|
|||||||
@@ -61,4 +61,4 @@ def test_compresses_large_json_when_accepted() -> None:
|
|||||||
assert resp.headers.get("Content-Encoding") == "gzip"
|
assert resp.headers.get("Content-Encoding") == "gzip"
|
||||||
decompressed = gzip.decompress(resp.body)
|
decompressed = gzip.decompress(resp.body)
|
||||||
assert json.loads(decompressed) == big
|
assert json.loads(decompressed) == big
|
||||||
assert resp.headers["Content-Length"] == str(len(resp.body))
|
assert resp.content_length == len(resp.body)
|
||||||
|
|||||||
@@ -73,7 +73,7 @@ def test_build_error_response_from_api_error() -> None:
|
|||||||
body = _body(resp)
|
body = _body(resp)
|
||||||
assert body["status"] == "error"
|
assert body["status"] == "error"
|
||||||
assert body["error_code"] == "SESSION_NOT_FOUND"
|
assert body["error_code"] == "SESSION_NOT_FOUND"
|
||||||
assert body["message"] == "no such session"
|
assert body["message"] == f"no such session (ref {body['ref']})"
|
||||||
assert body["details"] == {"session_id": "abc"}
|
assert body["details"] == {"session_id": "abc"}
|
||||||
assert "retryable" not in body
|
assert "retryable" not in body
|
||||||
assert "debug_info" not in body
|
assert "debug_info" not in body
|
||||||
@@ -91,7 +91,7 @@ def test_build_error_response_from_generic_exception() -> None:
|
|||||||
assert resp.status == 500
|
assert resp.status == 500
|
||||||
body = _body(resp)
|
body = _body(resp)
|
||||||
assert body["error_code"] == "INTERNAL_ERROR"
|
assert body["error_code"] == "INTERNAL_ERROR"
|
||||||
assert body["message"] == "oops"
|
assert body["message"] == f"oops (ref {body['ref']})"
|
||||||
|
|
||||||
|
|
||||||
def test_build_error_response_debug_mode_includes_traceback() -> None:
|
def test_build_error_response_debug_mode_includes_traceback() -> None:
|
||||||
@@ -188,6 +188,6 @@ def test_error_contract_apierror_passthrough_full_shape() -> None:
|
|||||||
assert resp.status == 429
|
assert resp.status == 429
|
||||||
body = _body(resp)
|
body = _body(resp)
|
||||||
assert body["error_code"] == "RATE_LIMITED"
|
assert body["error_code"] == "RATE_LIMITED"
|
||||||
assert body["message"] == "slow down"
|
assert body["message"] == f"slow down (ref {body['ref']})"
|
||||||
assert body["details"] == {"retry_after": 5}
|
assert body["details"] == {"retry_after": 5}
|
||||||
assert body["retryable"] is True
|
assert body["retryable"] is True
|
||||||
|
|||||||
@@ -9,6 +9,7 @@ from __future__ import annotations
|
|||||||
|
|
||||||
import asyncio
|
import asyncio
|
||||||
import json
|
import json
|
||||||
|
from collections import Counter, defaultdict
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
from aiohttp import web
|
from aiohttp import web
|
||||||
@@ -132,7 +133,7 @@ async def test_api_handler_maps_apierror(aiohttp_client) -> None:
|
|||||||
body = await resp.json()
|
body = await resp.json()
|
||||||
assert body["status"] == "error"
|
assert body["status"] == "error"
|
||||||
assert body["error_code"] == "NOT_FOUND"
|
assert body["error_code"] == "NOT_FOUND"
|
||||||
assert body["message"] == "nope"
|
assert body["message"] == f"nope (ref {body['ref']})"
|
||||||
assert body["details"] == {"x": 1}
|
assert body["details"] == {"x": 1}
|
||||||
|
|
||||||
|
|
||||||
@@ -180,6 +181,7 @@ async def test_heartbeat_late_error_counts_as_rejected(aiohttp_client, monkeypat
|
|||||||
|
|
||||||
monkeypatch.setattr(routes, "HEARTBEAT_INTERVAL", 0.05)
|
monkeypatch.setattr(routes, "HEARTBEAT_INTERVAL", 0.05)
|
||||||
monkeypatch.setattr(stats, "keys", {})
|
monkeypatch.setattr(stats, "keys", {})
|
||||||
|
monkeypatch.setattr(stats, "errors", defaultdict(Counter))
|
||||||
|
|
||||||
@heartbeat
|
@heartbeat
|
||||||
async def slow_fail(request: web.Request) -> web.Response:
|
async def slow_fail(request: web.Request) -> web.Response:
|
||||||
@@ -195,6 +197,33 @@ async def test_heartbeat_late_error_counts_as_rejected(aiohttp_client, monkeypat
|
|||||||
sent = len(await resp.read())
|
sent = len(await resp.read())
|
||||||
(entry,) = stats.keys.values()
|
(entry,) = stats.keys.values()
|
||||||
assert resp.status == 200 and entry.rejected == 1 and entry.bytes_out >= sent > 0
|
assert resp.status == 200 and entry.rejected == 1 and entry.bytes_out >= sent > 0
|
||||||
|
assert stats.errors == {"-": {"SERVICE_ERROR": 1}}
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize(("service", "counted_as"), [("EXAMPLE", "EXAMPLE"), ("NOT_INSTALLED", "-")])
|
||||||
|
async def test_error_is_counted_by_service_and_logged_with_its_ref(
|
||||||
|
aiohttp_client, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture, service: str, counted_as: str
|
||||||
|
) -> None:
|
||||||
|
"""A gzipped error still reaches the stats: counted under an installed service only, and logged with its ref."""
|
||||||
|
from unshackle.core.api.stats import stats, stats_middleware
|
||||||
|
|
||||||
|
monkeypatch.setattr(stats, "errors", defaultdict(Counter))
|
||||||
|
|
||||||
|
@api_handler
|
||||||
|
async def fail(request: web.Request) -> web.Response:
|
||||||
|
await request.json()
|
||||||
|
raise APIError(APIErrorCode.SERVICE_ERROR, "denied", details={"pad": "x" * 300})
|
||||||
|
|
||||||
|
app = web.Application(middlewares=[stats_middleware, compression_middleware])
|
||||||
|
app["debug_api"] = False
|
||||||
|
app.router.add_post("/fail", fail)
|
||||||
|
client = await aiohttp_client(app)
|
||||||
|
|
||||||
|
resp = await client.post("/fail", json={"service": service}, headers={"Accept-Encoding": "gzip"})
|
||||||
|
body = await resp.json()
|
||||||
|
assert resp.headers["Content-Encoding"] == "gzip"
|
||||||
|
assert stats.errors == {counted_as: {"SERVICE_ERROR": 1}}
|
||||||
|
assert f"Error ref {body['ref']}: SERVICE_ERROR 422 POST /fail service={counted_as}" in caplog.text
|
||||||
|
|
||||||
|
|
||||||
async def test_heartbeat_client_disconnect_is_quiet(
|
async def test_heartbeat_client_disconnect_is_quiet(
|
||||||
|
|||||||
@@ -49,12 +49,6 @@ async def compression_middleware(request: web.Request, handler) -> web.StreamRes
|
|||||||
if not level:
|
if not level:
|
||||||
return response
|
return response
|
||||||
|
|
||||||
compressed = gzip.compress(body, compresslevel=level)
|
response.body = gzip.compress(body, compresslevel=level)
|
||||||
headers = dict(response.headers)
|
response.headers["Content-Encoding"] = "gzip"
|
||||||
headers["Content-Encoding"] = "gzip"
|
return response
|
||||||
headers["Content-Length"] = str(len(compressed))
|
|
||||||
return web.Response(
|
|
||||||
body=compressed,
|
|
||||||
status=response.status,
|
|
||||||
headers=headers,
|
|
||||||
)
|
|
||||||
|
|||||||
@@ -7,6 +7,7 @@ and optional debug information for the unshackle REST API.
|
|||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import secrets
|
||||||
import traceback
|
import traceback
|
||||||
from datetime import datetime, timezone
|
from datetime import datetime, timezone
|
||||||
from enum import Enum
|
from enum import Enum
|
||||||
@@ -114,6 +115,9 @@ def build_error_response(
|
|||||||
"""
|
"""
|
||||||
Assemble a structured JSON error response.
|
Assemble a structured JSON error response.
|
||||||
|
|
||||||
|
Each error gets a short ``ref``, also appended to ``message`` so that every client prints it.
|
||||||
|
The stats middleware logs the ref with the request, so a reported ref finds the server-side record.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
error: APIError or generic Exception to convert to response
|
error: APIError or generic Exception to convert to response
|
||||||
debug_mode: Whether to include technical debug information
|
debug_mode: Whether to include technical debug information
|
||||||
@@ -135,10 +139,12 @@ def build_error_response(
|
|||||||
http_status = 500
|
http_status = 500
|
||||||
retryable = False
|
retryable = False
|
||||||
|
|
||||||
|
ref = secrets.token_hex(3)
|
||||||
response_data: dict[str, Any] = {
|
response_data: dict[str, Any] = {
|
||||||
"status": "error",
|
"status": "error",
|
||||||
"error_code": error_code,
|
"error_code": error_code,
|
||||||
"message": message,
|
"message": f"{message} (ref {ref})",
|
||||||
|
"ref": ref,
|
||||||
"timestamp": datetime.now(timezone.utc).isoformat(),
|
"timestamp": datetime.now(timezone.utc).isoformat(),
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -161,7 +167,9 @@ def build_error_response(
|
|||||||
|
|
||||||
response_data["debug_info"] = debug_info
|
response_data["debug_info"] = debug_info
|
||||||
|
|
||||||
return web.json_response(response_data, status=http_status)
|
response = web.json_response(response_data, status=http_status)
|
||||||
|
response["error"] = {"code": error_code, "ref": ref, "message": message}
|
||||||
|
return response
|
||||||
|
|
||||||
|
|
||||||
def categorize_exception(
|
def categorize_exception(
|
||||||
|
|||||||
@@ -147,6 +147,7 @@ def heartbeat(handler: Handler) -> Handler:
|
|||||||
log.exception(f"Error in {request.path}")
|
log.exception(f"Error in {request.path}")
|
||||||
result = handle_api_exception(e, debug_mode=debug_mode)
|
result = handle_api_exception(e, debug_mode=debug_mode)
|
||||||
response["late_status"] = result.status
|
response["late_status"] = result.status
|
||||||
|
response["error"] = result.get("error")
|
||||||
body = result.body if isinstance(result, web.Response) else None
|
body = result.body if isinstance(result, web.Response) else None
|
||||||
await response.write(bytes(body) if isinstance(body, (bytes, bytearray)) else b"{}")
|
await response.write(bytes(body) if isinstance(body, (bytes, bytearray)) else b"{}")
|
||||||
await response.write_eof()
|
await response.write_eof()
|
||||||
@@ -2581,8 +2582,8 @@ async def dashboard_status(request: web.Request) -> web.Response:
|
|||||||
---
|
---
|
||||||
summary: Dashboard server status
|
summary: Dashboard server status
|
||||||
description: >
|
description: >
|
||||||
Version, uptime, bind address, mode, request counters per API key, and counts of
|
Version, uptime, bind address, mode, request counters per API key, error counts by
|
||||||
services, remote sessions and jobs. Every /api/dashboard/ route needs
|
service and error code, and counts of services, remote sessions and jobs. Every /api/dashboard/ route needs
|
||||||
`serve.dashboard.key` in X-Secret-Key; a user key never passes, and the routes do not
|
`serve.dashboard.key` in X-Secret-Key; a user key never passes, and the routes do not
|
||||||
exist without a dashboard key.
|
exist without a dashboard key.
|
||||||
tags: [Dashboard]
|
tags: [Dashboard]
|
||||||
|
|||||||
@@ -7,7 +7,7 @@ import hashlib
|
|||||||
import json
|
import json
|
||||||
import logging
|
import logging
|
||||||
import time
|
import time
|
||||||
from collections import Counter, deque
|
from collections import Counter, defaultdict, deque
|
||||||
from dataclasses import dataclass, field
|
from dataclasses import dataclass, field
|
||||||
from functools import lru_cache
|
from functools import lru_cache
|
||||||
from typing import Any, Deque, Dict, List, Optional, Tuple
|
from typing import Any, Deque, Dict, List, Optional, Tuple
|
||||||
@@ -143,6 +143,7 @@ class ServerStats:
|
|||||||
requests_total: int = 0
|
requests_total: int = 0
|
||||||
requests_rejected: int = 0
|
requests_rejected: int = 0
|
||||||
keys: Dict[str, KeyStats] = field(default_factory=dict)
|
keys: Dict[str, KeyStats] = field(default_factory=dict)
|
||||||
|
errors: Dict[str, Counter] = field(default_factory=lambda: defaultdict(Counter)) # service -> error_code -> count
|
||||||
|
|
||||||
def key_stats(self, key: Optional[str]) -> KeyStats:
|
def key_stats(self, key: Optional[str]) -> KeyStats:
|
||||||
"""The counters for an API key, created on first sight."""
|
"""The counters for an API key, created on first sight."""
|
||||||
@@ -194,6 +195,7 @@ class ServerStats:
|
|||||||
"requests_total": self.requests_total,
|
"requests_total": self.requests_total,
|
||||||
"requests_rejected": self.requests_rejected,
|
"requests_rejected": self.requests_rejected,
|
||||||
"requests_by_key": dict(self.requests_by_key),
|
"requests_by_key": dict(self.requests_by_key),
|
||||||
|
"errors": {service: dict(codes) for service, codes in self.errors.items()},
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
@@ -203,16 +205,18 @@ stats = ServerStats()
|
|||||||
@web.middleware
|
@web.middleware
|
||||||
async def stats_middleware(request: web.Request, handler: Any) -> web.StreamResponse:
|
async def stats_middleware(request: web.Request, handler: Any) -> web.StreamResponse:
|
||||||
from unshackle.core.api.handlers import request_secret_key
|
from unshackle.core.api.handlers import request_secret_key
|
||||||
|
from unshackle.core.api.sanitize import sanitize_log
|
||||||
|
|
||||||
stats.requests_total += 1
|
stats.requests_total += 1
|
||||||
started = time.perf_counter()
|
started = time.perf_counter()
|
||||||
response = await handler(request)
|
response = await handler(request)
|
||||||
status = response.get("late_status", response.status)
|
status = response.get("late_status", response.status)
|
||||||
bytes_out = getattr(response, "content_length", None) or response.body_length
|
bytes_out = getattr(response, "content_length", None) or response.body_length
|
||||||
|
key = None if status == 401 else configured_key(request_secret_key(request))
|
||||||
if status == 401:
|
if status == 401:
|
||||||
stats.requests_rejected += 1
|
stats.requests_rejected += 1
|
||||||
else:
|
else:
|
||||||
entry = stats.key_stats(configured_key(request_secret_key(request)))
|
entry = stats.key_stats(key)
|
||||||
entry.requests += 1
|
entry.requests += 1
|
||||||
entry.last_seen = time.time()
|
entry.last_seen = time.time()
|
||||||
entry.bytes_out += bytes_out
|
entry.bytes_out += bytes_out
|
||||||
@@ -235,9 +239,33 @@ async def stats_middleware(request: web.Request, handler: Any) -> web.StreamResp
|
|||||||
"bytes_out": bytes_out,
|
"bytes_out": bytes_out,
|
||||||
},
|
},
|
||||||
)
|
)
|
||||||
|
error = response.get("error")
|
||||||
|
if error:
|
||||||
|
service = await _request_service(request, session_id)
|
||||||
|
stats.errors[service][error["code"]] += 1
|
||||||
|
logging.getLogger("api").warning(
|
||||||
|
f"Error ref {error['ref']}: {error['code']} {status} {request.method} {sanitize_log(request.path)} "
|
||||||
|
f"service={service} user={sanitize_log(mask_key(key))}: {sanitize_log(error['message'])}"
|
||||||
|
)
|
||||||
return response
|
return response
|
||||||
|
|
||||||
|
|
||||||
|
async def _request_service(request: web.Request, session_id: Optional[str]) -> str:
|
||||||
|
"""The installed service tag a request is for, from its remote session or its JSON body; ``-`` if none.
|
||||||
|
|
||||||
|
Only an installed tag counts, so a caller cannot grow ``stats.errors`` with made-up names.
|
||||||
|
"""
|
||||||
|
from unshackle.core.api.session_store import get_session_store
|
||||||
|
from unshackle.core.services import Services
|
||||||
|
|
||||||
|
entry = get_session_store().peek(session_id) if session_id else None
|
||||||
|
service = entry.service_tag if entry is not None else (await _consumed_json(request)).get("service")
|
||||||
|
if not isinstance(service, str):
|
||||||
|
return "-"
|
||||||
|
tag = Services.get_tag(service)
|
||||||
|
return tag if tag in Services.get_tags() else "-"
|
||||||
|
|
||||||
|
|
||||||
class SeqRing:
|
class SeqRing:
|
||||||
"""Keep the last N records in memory, each with a ``seq`` one above the record before it."""
|
"""Keep the last N records in memory, each with a ``seq`` one above the record before it."""
|
||||||
|
|
||||||
@@ -347,20 +375,20 @@ def _reply_json(response: Any) -> Dict[str, Any]:
|
|||||||
return data if isinstance(data, dict) else {}
|
return data if isinstance(data, dict) else {}
|
||||||
|
|
||||||
|
|
||||||
async def _request_session_id(request: web.Request) -> Any:
|
async def _consumed_json(request: web.Request) -> Dict[str, Any]:
|
||||||
"""The ``session_id`` of the JSON request body, when the handler consumed it: aiohttp caches that read.
|
"""The JSON object of the request body when the handler consumed it (aiohttp caches that read), else ``{}``.
|
||||||
|
|
||||||
A body nobody read belongs to a call refused before the handler, a 401 or a 429. It stays
|
A body nobody read belongs to a call refused before the handler, a 401 or a 429. It stays
|
||||||
unread: reading it would spend time and memory on an unauthenticated caller, and wait on a
|
unread: reading it would spend time and memory on an unauthenticated caller, and wait on a
|
||||||
body that never finishes arriving.
|
body that never finishes arriving.
|
||||||
"""
|
"""
|
||||||
if not request.content.at_eof():
|
if not request.content.at_eof():
|
||||||
return None
|
return {}
|
||||||
try:
|
try:
|
||||||
body = await request.json()
|
body = await request.json()
|
||||||
except Exception:
|
except Exception:
|
||||||
return None
|
return {}
|
||||||
return body.get("session_id") if isinstance(body, dict) else None
|
return body if isinstance(body, dict) else {}
|
||||||
|
|
||||||
|
|
||||||
async def _cdm_record(request: web.Request, drm: str, op: str, outcome: Any) -> Dict[str, Any]:
|
async def _cdm_record(request: web.Request, drm: str, op: str, outcome: Any) -> Dict[str, Any]:
|
||||||
@@ -388,7 +416,7 @@ async def _cdm_record(request: web.Request, drm: str, op: str, outcome: Any) ->
|
|||||||
elif op == "close":
|
elif op == "close":
|
||||||
session_id = request.match_info.get("session_id")
|
session_id = request.match_info.get("session_id")
|
||||||
else:
|
else:
|
||||||
session_id = await _request_session_id(request)
|
session_id = (await _consumed_json(request)).get("session_id")
|
||||||
return {
|
return {
|
||||||
"drm": drm,
|
"drm": drm,
|
||||||
"op": op,
|
"op": op,
|
||||||
|
|||||||
Reference in New Issue
Block a user