feat(serve): add --quiet mode and developer dashboard API

Headless serve with plain stderr logging; /api/dashboard/* (status, sessions, jobs, logs, SSE/poll events) behind a separate serve.dashboard.key.
This commit is contained in:
imSp4rky
2026-08-28 13:18:41 -06:00
parent c49fccf3b5
commit e3f4e52444
13 changed files with 831 additions and 26 deletions
+82
View File
@@ -0,0 +1,82 @@
# Developer Dashboard
The dashboard endpoints show the operator of a `serve` instance what the server is doing:
version and uptime, request counts, live remote sessions, download jobs, and the recent
log lines. They exist so you can watch a headless server (`serve --quiet`) from a web page
instead of a terminal.
The dashboard endpoints are for the person who runs the server, not for API users. They
have their own key: a user key or a tier key never opens them.
## Enable
Set a dashboard key under `serve.dashboard` in `unshackle.yaml`:
```yaml
serve:
api_secret: ...
dashboard:
key: a-long-random-string
```
Without `dashboard.key` the server does not register the routes, and they answer `404`.
`serve --no-key` turns off user authentication only; the dashboard key stays required.
Once the dashboard key is set, every `/api/dashboard/` request must carry it in
`X-Secret-Key`.
Every other endpoint refuses the dashboard key, and the dashboard refuses user keys. The
routes are available in `--remote-only` mode too.
Run the server headless with `--quiet`: no banner, no Rich output, one plain log line per
event on stderr, and `api.*` loggers stay at `INFO` so remote session activity reaches the
log ring. Add `--debug` to raise the level to `DEBUG` and also write the full log to
`directories.logs` as `unshackle_serve_<time>.log`. Nothing else the dashboard shows reaches
the disk: stats, the log ring and the event stream live in memory and reset on restart, so a
dashboard UI that wants history must keep its own copy.
```
unshackle serve --remote-only --quiet
```
## Endpoints
All are `GET`.
| 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), loaded service count, session count, `max_sessions`, `session_ttl`, job counts by status. |
| `/api/dashboard/sessions` | Every live remote session: `id`, `owner` (username or masked key), `creator_ip`, `service`, `title_id`, `titles`, `tracks`, `title` (the first resolved title's display name), `auth_status`, `auth_error`, `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`. `client` is whatever the remote client sent as `client` in its session create request (the current CLI sends nothing, so it is `{}`). `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/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/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. |
`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
characters of the API key.
## Event stream
`/api/dashboard/events` sends a `stats` event (the same body as `/status`) on connect and
then every 5 seconds, so the connection stays alive and the counters stay current. Between
those it sends:
| Event | Data |
|-------|------|
| `log` | One log record: `seq`, `ts`, `level`, `logger`, `msg`. |
| `session` | `action` (`create`, `update` on every request the session makes and when titles or tracks load, or `delete`) plus the full remote session summary, so each event is a complete snapshot. A `delete` also carries `reason`: `closed` (the client ended it), `expired` (TTL sweep) or `evicted` (`max_sessions` reached). A session opened by `--skip-dl` can live only a few seconds, so a UI that wants history must keep its own record of these events. |
| `job` | `event` (`queued`, `progress`, `status`, or the terminal status) plus the full job object. |
Every frame carries an `id:` (the event seq). After a dropped connection an `EventSource`
reconnects with `Last-Event-ID` and the server replays the events it missed. The connection
never closes on its own.
A browser `EventSource` cannot set headers, so this route also accepts the dashboard key as
`?secret_key=`. The dashboard key then appears in proxy and access logs and in browser history, so
use the query parameter only for `/events` and send the header everywhere else.
```js
const es = new EventSource(`${server}/api/dashboard/events?secret_key=${key}`);
es.addEventListener("log", (e) => console.log(JSON.parse(e.data).msg));
```
CORS is open (`Access-Control-Allow-Origin: *`), so a dashboard page served from another
origin works without a proxy.
+1
View File
@@ -119,6 +119,7 @@ is the [REST API](../../dev/rest-api/index.md) section. These are the config key
| `users` | dict | `{}` | Per-user API keys and their allowlists (see below). | | `users` | dict | `{}` | Per-user API keys and their allowlists (see below). |
| `services` | list | *(unset)* | Global service allowlist. Omit to allow all. | | `services` | list | *(unset)* | Global service allowlist. Omit to allow all. |
| `remote_only` | bool | `false` | Expose only the remote service session endpoints (health, services, search, session) and disable the rest of the REST API. | | `remote_only` | bool | `false` | Expose only the remote service session endpoints (health, services, search, session) and disable the rest of the REST API. |
| `dashboard` | dict | *(unset)* | Developer dashboard: `key` is the API key for the read-only `/api/dashboard/` endpoints (status, sessions, jobs, logs, SSE events). Unset leaves those routes unregistered. See [dashboard endpoints](../../dev/rest-api/dashboard.md). |
| `session_ttl` | int (s) | `300` | Lifetime of an interactive auth session. | | `session_ttl` | int (s) | `300` | Lifetime of an interactive auth session. |
| `max_sessions` | int | `100` | Maximum concurrent sessions. | | `max_sessions` | int | `100` | Maximum concurrent sessions. |
| `history_limit` | int | `100` | How many finished jobs to retain in history. | | `history_limit` | int | `100` | How many finished jobs to retain in history. |
+1
View File
@@ -140,6 +140,7 @@ nav:
- Authentication: dev/rest-api/authentication.md - Authentication: dev/rest-api/authentication.md
- Endpoints: dev/rest-api/endpoints.md - Endpoints: dev/rest-api/endpoints.md
- Job Lifecycle: dev/rest-api/job-lifecycle.md - Job Lifecycle: dev/rest-api/job-lifecycle.md
- Dashboard: dev/rest-api/dashboard.md
- Remote Sessions: dev/rest-api/remote-sessions.md - Remote Sessions: dev/rest-api/remote-sessions.md
- Error Responses: dev/rest-api/errors.md - Error Responses: dev/rest-api/errors.md
- API Reference: - API Reference:
+110
View File
@@ -0,0 +1,110 @@
"""Dashboard auth split, stats middleware, log ring buffer and session events."""
from __future__ import annotations
import asyncio
import logging
import pytest
from aiohttp import web
from unshackle.core.api.events import bus
from unshackle.core.api.handlers import dashboard_authentication
from unshackle.core.api.routes import cors_middleware, setup_routes
from unshackle.core.api.session_store import SessionStore
from unshackle.core.api.stats import RingLogHandler, stats, stats_middleware
from unshackle.core.config import config
pytestmark = pytest.mark.unit
@pytest.fixture
def dashboard_cfg(monkeypatch):
monkeypatch.setitem(config.serve, "dashboard", {"key": "dash-secret"})
monkeypatch.setitem(config.serve, "users", {"tier-key": {"username": "tier1"}})
stats.requests_total = 0
stats.requests_by_key.clear()
def make_app(dashboard: bool = True) -> web.Application:
app = web.Application(middlewares=[cors_middleware, stats_middleware, dashboard_authentication])
app["config"] = {"users": {}}
app["debug_api"] = False
setup_routes(app, remote_only=True, dashboard=dashboard)
return app
async def test_dashboard_key_split(aiohttp_client, dashboard_cfg) -> None:
client = await aiohttp_client(make_app())
assert (await client.get("/api/dashboard/status", headers={"X-Secret-Key": "tier-key"})).status == 401
assert (await client.get("/api/dashboard/status")).status == 401
resp = await client.get("/api/dashboard/status", headers={"X-Secret-Key": "dash-secret"})
assert resp.status == 200
body = await resp.json()
assert body["mode"] and "uptime_seconds" in body and body["sessions"] == 0
# rejected requests count only as rejected; accepted ones are keyed by masked label
assert stats.requests_total == 3 and stats.requests_rejected == 2 and stats.requests_by_key == {"dash…": 1}
await client.get("/api/health", headers={"X-Secret-Key": "tier-key"})
assert stats.requests_by_key["tier1"] == 1
async def test_dashboard_routes_absent_without_key(aiohttp_client, dashboard_cfg) -> None:
client = await aiohttp_client(make_app(dashboard=False))
assert (await client.get("/api/dashboard/status", headers={"X-Secret-Key": "dash-secret"})).status == 404
def test_ring_since_and_level() -> None:
ring = RingLogHandler(maxlen=3)
ring.setFormatter(logging.Formatter("%(message)s"))
logger = logging.getLogger("test.ring")
logger.propagate = False
logger.addHandler(ring)
logger.setLevel(logging.DEBUG)
for i in range(5):
(logger.warning if i % 2 else logger.info)(f"m{i}")
assert [r["msg"] for r in ring.records] == ["m2", "m3", "m4"]
assert ring.seq == 5
assert [r["msg"] for r in ring.since(3)] == ["m3", "m4"]
assert [r["msg"] for r in ring.since(0, "warning")] == ["m3"]
assert ring.since(0, logger="test") and not ring.since(0, logger="other")
async def test_session_events() -> None:
queue = bus.subscribe()
try:
store = SessionStore()
entry = await store.create("EXAMPLE", object())
assert [e.session_id for e in store.list()] == [entry.session_id]
await store.delete(entry.session_id)
create, delete = queue.get_nowait(), queue.get_nowait()
assert create["event"] == "session" and create["data"]["action"] == "create"
assert delete["data"]["action"] == "delete" and delete["data"]["id"] == entry.session_id
assert "owner_key" not in create["data"]
finally:
bus.unsubscribe(queue)
await asyncio.sleep(0)
async def test_events_burst_mode(aiohttp_client, dashboard_cfg) -> None:
client = await aiohttp_client(make_app())
mark = bus.seq
bus.publish("session", {"action": "create", "id": "s1"})
bus.publish("session", {"action": "delete", "id": "s1"})
resp = await client.get(f"/api/dashboard/events?since={mark}", headers={"X-Secret-Key": "dash-secret"})
assert resp.status == 200
body = await resp.json()
assert body["seq"] == bus.seq and "stats" in body
assert [e["data"]["action"] for e in body["events"] if e["event"] == "session"] == ["create", "delete"]
async def test_session_action_log() -> None:
store = SessionStore()
entry = await store.create("EXAMPLE", object(), owner_key="k")
queue = bus.subscribe()
try:
store.record_action(entry.session_id, {"action": "titles", "status": 200})
store.record_action("missing", {"action": "titles", "status": 200})
assert entry.summary()["actions"] == [{"action": "titles", "status": 200}]
assert queue.get_nowait()["data"]["action"] == "update" and queue.empty()
finally:
bus.unsubscribe(queue)
+1
View File
@@ -32,6 +32,7 @@ def test_serve_help_lists_documented_flags(runner: CliRunner) -> None:
"--debug-api", "--debug-api",
"--debug", "--debug",
"--remote-only", "--remote-only",
"--quiet",
): ):
assert flag in out, f"missing flag in --help: {flag}" assert flag in out, f"missing flag in --help: {flag}"
+81 -12
View File
@@ -2,7 +2,9 @@ import asyncio
import hmac import hmac
import logging import logging
import subprocess import subprocess
import sys
from contextlib import suppress from contextlib import suppress
from datetime import datetime
import click import click
from aiohttp import web from aiohttp import web
@@ -10,8 +12,15 @@ from aiohttp import web
from unshackle.core import binaries from unshackle.core import binaries
from unshackle.core.api import cors_middleware, setup_routes, setup_swagger from unshackle.core.api import cors_middleware, setup_routes, setup_swagger
from unshackle.core.api.compression import compression_middleware from unshackle.core.api.compression import compression_middleware
from unshackle.core.api.handlers import request_secret_key from unshackle.core.api.handlers import (
DASHBOARD_PREFIX,
dashboard_authentication,
dashboard_key,
request_secret_key,
)
from unshackle.core.api.stats import ring, stats, stats_middleware
from unshackle.core.config import config from unshackle.core.config import config
from unshackle.core.console import console
from unshackle.core.constants import context_settings from unshackle.core.constants import context_settings
from unshackle.core.downloaders import format_speed, parse_speed_limit, set_speed_limit from unshackle.core.downloaders import format_speed, parse_speed_limit, set_speed_limit
@@ -92,6 +101,13 @@ def _install_service_refresh(app: web.Application) -> None:
default=False, default=False,
help="Only expose remote service session endpoints (health, services, search, session).", help="Only expose remote service session endpoints (health, services, search, session).",
) )
@click.option(
"-q",
"--quiet",
is_flag=True,
default=False,
help="Headless mode: no banner and no Rich output, plain log lines on stderr.",
)
def serve( def serve(
host: str, host: str,
port: int, port: int,
@@ -103,6 +119,7 @@ def serve(
debug_api: bool, debug_api: bool,
debug: bool, debug: bool,
remote_only: bool, remote_only: bool,
quiet: bool,
) -> None: ) -> None:
""" """
Serve your Local Widevine and PlayReady Devices and REST API for Remote Access. Serve your Local Widevine and PlayReady Devices and REST API for Remote Access.
@@ -136,12 +153,35 @@ def serve(
log = logging.getLogger("serve") log = logging.getLogger("serve")
if quiet:
logging.basicConfig(
force=True,
level=logging.DEBUG if debug else logging.INFO,
format="%(asctime)s %(name)s %(levelname)s %(message)s",
stream=sys.stderr,
)
console.quiet = True
if not debug:
for handler in logging.getLogger().handlers:
handler.addFilter(lambda record: record.name != "aiohttp.access")
if debug:
log_path = config.directories.logs / config.filenames.log.format(
name="serve", time=datetime.now().strftime("%Y%m%d-%H%M%S")
)
log_path.parent.mkdir(parents=True, exist_ok=True)
file_handler = logging.FileHandler(log_path, encoding="utf-8")
file_handler.setFormatter(logging.Formatter("%(asctime)s %(name)s %(levelname)s %(message)s"))
logging.getLogger().addHandler(file_handler)
log.info(f"Writing log to {log_path}")
elif debug:
logging.getLogger().setLevel(logging.DEBUG)
if debug: if debug:
logging.basicConfig(level=logging.DEBUG, format="%(name)s - %(levelname)s - %(message)s")
log.info("Debug logging enabled for API operations") log.info("Debug logging enabled for API operations")
else: elif not quiet:
logging.getLogger("api").setLevel(logging.WARNING) logging.getLogger("api").setLevel(logging.WARNING)
logging.getLogger("api.remote").setLevel(logging.WARNING) logging.getLogger("api.remote").setLevel(logging.WARNING)
ring.setFormatter(logging.Formatter("%(message)s"))
logging.getLogger().addHandler(ring)
if not no_key: if not no_key:
api_secret = config.serve.get("api_secret") api_secret = config.serve.get("api_secret")
@@ -180,7 +220,7 @@ def serve(
@web.middleware @web.middleware
async def api_key_authentication(request: web.Request, handler) -> web.StreamResponse: async def api_key_authentication(request: web.Request, handler) -> web.StreamResponse:
"""Authenticate API requests using X-Secret-Key header.""" """Authenticate API requests using X-Secret-Key header."""
if request.path == "/api/health": if request.path == "/api/health" or request.path.startswith(DASHBOARD_PREFIX):
return await handler(request) return await handler(request)
secret_key = request_secret_key(request) secret_key = request_secret_key(request)
if not secret_key: if not secret_key:
@@ -194,6 +234,13 @@ def serve(
remote_only = remote_only or config.serve.get("remote_only", False) remote_only = remote_only or config.serve.get("remote_only", False)
if remote_only: if remote_only:
api_only = True api_only = True
stats.host, stats.port = host, port
stats.mode = "remote_only" if remote_only else "api_only" if api_only else "full"
dashboard = bool(dashboard_key())
if dashboard:
log.info(f"Developer dashboard endpoints available at http://{host}:{port}/api/dashboard/")
elif quiet:
log.info("Developer dashboard disabled: set serve.dashboard.key in unshackle.yaml to enable it")
try: try:
global_speed_limit = parse_speed_limit(config.serve.get("global_speed_limit")) global_speed_limit = parse_speed_limit(config.serve.get("global_speed_limit"))
@@ -220,10 +267,20 @@ def serve(
if api_only: if api_only:
log.info("Starting REST API server (pywidevine/pyplayready CDM disabled)") log.info("Starting REST API server (pywidevine/pyplayready CDM disabled)")
if no_key: if no_key:
app = web.Application(middlewares=[cors_middleware, compression_middleware]) app = web.Application(
middlewares=[cors_middleware, stats_middleware, dashboard_authentication, compression_middleware]
)
app["config"] = {"users": {}} app["config"] = {"users": {}}
else: else:
app = web.Application(middlewares=[cors_middleware, api_key_authentication, compression_middleware]) app = web.Application(
middlewares=[
cors_middleware,
stats_middleware,
dashboard_authentication,
api_key_authentication,
compression_middleware,
]
)
api_users: dict = {api_secret: {"devices": [], "username": "api_user"}} api_users: dict = {api_secret: {"devices": [], "username": "api_user"}}
if isinstance(users, dict): if isinstance(users, dict):
api_users.update(users) api_users.update(users)
@@ -245,9 +302,9 @@ def serve(
app.on_cleanup.append(stop_session_cleanup) app.on_cleanup.append(stop_session_cleanup)
_install_service_refresh(app) _install_service_refresh(app)
setup_routes(app, remote_only=remote_only) setup_routes(app, remote_only=remote_only, dashboard=dashboard)
if not remote_only: if not remote_only:
setup_swagger(app) setup_swagger(app, dashboard=dashboard)
log.info(f"REST API endpoints available at http://{host}:{port}/api/") log.info(f"REST API endpoints available at http://{host}:{port}/api/")
log.info(f"Swagger UI available at http://{host}:{port}/api/docs/") log.info(f"Swagger UI available at http://{host}:{port}/api/docs/")
else: else:
@@ -295,6 +352,8 @@ def serve(
@web.middleware @web.middleware
async def serve_authentication(request: web.Request, handler) -> web.StreamResponse: async def serve_authentication(request: web.Request, handler) -> web.StreamResponse:
secret_key = request_secret_key(request) secret_key = request_secret_key(request)
if request.path.startswith(DASHBOARD_PREFIX):
return await handler(request)
if serve_playready_flag and request.path in ("/playready", "/playready/"): if serve_playready_flag and request.path in ("/playready", "/playready/"):
response = await handler(request) response = await handler(request)
elif secret_key and not request.headers.get("X-Secret-Key"): elif secret_key and not request.headers.get("X-Secret-Key"):
@@ -316,10 +375,20 @@ def serve(
return serve_authentication return serve_authentication
if no_key: if no_key:
app = web.Application(middlewares=[cors_middleware, compression_middleware]) app = web.Application(
middlewares=[cors_middleware, stats_middleware, dashboard_authentication, compression_middleware]
)
else: else:
serve_auth = create_serve_authentication(serve_playready and bool(prd_devices)) serve_auth = create_serve_authentication(serve_playready and bool(prd_devices))
app = web.Application(middlewares=[cors_middleware, serve_auth, compression_middleware]) app = web.Application(
middlewares=[
cors_middleware,
dashboard_authentication,
serve_auth,
stats_middleware,
compression_middleware,
]
)
app["config"] = serve_config app["config"] = serve_config
app["debug_api"] = debug_api app["debug_api"] = debug_api
@@ -382,14 +451,14 @@ def serve(
elif serve_playready: elif serve_playready:
log.info("No PlayReady devices found, skipping PlayReady CDM endpoints") log.info("No PlayReady devices found, skipping PlayReady CDM endpoints")
setup_routes(app, remote_only=remote_only) setup_routes(app, remote_only=remote_only, dashboard=dashboard)
if serve_widevine: if serve_widevine:
log.info(f"Widevine CDM endpoints available at http://{host}:{port}/{{device}}/open") log.info(f"Widevine CDM endpoints available at http://{host}:{port}/{{device}}/open")
if remote_only: if remote_only:
log.info(f"Remote service endpoints available at http://{host}:{port}/api/session/") log.info(f"Remote service endpoints available at http://{host}:{port}/api/session/")
else: else:
setup_swagger(app) setup_swagger(app, dashboard=dashboard)
log.info(f"REST API endpoints available at http://{host}:{port}/api/") log.info(f"REST API endpoints available at http://{host}:{port}/api/")
log.info(f"Swagger UI available at http://{host}:{port}/api/docs/") log.info(f"Swagger UI available at http://{host}:{port}/api/docs/")
log.info("(Press CTRL+C to quit)") log.info("(Press CTRL+C to quit)")
+4
View File
@@ -1,5 +1,6 @@
import atexit import atexit
import logging import logging
import sys
from datetime import datetime from datetime import datetime
import click import click
@@ -52,6 +53,9 @@ def main(version: bool, debug: bool) -> None:
traceback.install(console=console, width=80, suppress=[click]) traceback.install(console=console, width=80, suppress=[click])
if "--quiet" in sys.argv[1:] or "-q" in sys.argv[1:]:
return
console.print( console.print(
Padding( Padding(
Group( Group(
+3
View File
@@ -15,6 +15,7 @@ from enum import Enum
from pathlib import Path from pathlib import Path
from typing import Any, Callable, Dict, List, Optional from typing import Any, Callable, Dict, List, Optional
from unshackle.core.api.events import bus
from unshackle.core.api.sanitize import sanitize_log from unshackle.core.api.sanitize import sanitize_log
from unshackle.core.utils.redact import REDACTED, URL_USERINFO_RE, redact_text from unshackle.core.utils.redact import REDACTED, URL_USERINFO_RE, redact_text
@@ -675,6 +676,7 @@ class DownloadQueueManager:
self._job_queue.put_nowait(job) self._job_queue.put_nowait(job)
log.info(f"Created download job {job_id} for {sanitize_log(service)}:{sanitize_log(title_id)}") log.info(f"Created download job {job_id} for {sanitize_log(service)}:{sanitize_log(title_id)}")
bus.publish("job", {"event": "queued", **job.to_dict(include_full_details=True)})
return job return job
def get_job(self, job_id: str) -> Optional[DownloadJob]: def get_job(self, job_id: str) -> Optional[DownloadJob]:
@@ -703,6 +705,7 @@ class DownloadQueueManager:
def publish(self, job: DownloadJob, event: str) -> None: def publish(self, job: DownloadJob, event: str) -> None:
"""Fan an event carrying the job's full detail dict out to every listener.""" """Fan an event carrying the job's full detail dict out to every listener."""
bus.publish("job", {"event": event, **job.to_dict(include_full_details=True)})
queues = self._subscribers.get(job.job_id) queues = self._subscribers.get(job.job_id)
if not queues: if not queues:
return return
+56
View File
@@ -0,0 +1,56 @@
"""Server-wide event broadcaster feeding the dashboard SSE event stream."""
from __future__ import annotations
import asyncio
import threading
from collections import deque
from contextlib import suppress
from typing import Any, Deque, Dict, List, Optional
class EventBus:
"""Fan events out to every subscriber queue; drop when a subscriber falls behind."""
def __init__(self, history: int = 20000) -> None:
self._subs: List[asyncio.Queue] = []
self.seq = 0
self.history: Deque[Dict[str, Any]] = deque(maxlen=history)
self._loop: Optional[asyncio.AbstractEventLoop] = None
self._loop_thread: Optional[threading.Thread] = None
def subscribe(self) -> asyncio.Queue:
self._loop = asyncio.get_running_loop()
self._loop_thread = threading.current_thread()
queue: asyncio.Queue = asyncio.Queue(maxsize=200)
self._subs.append(queue)
return queue
def unsubscribe(self, queue: asyncio.Queue) -> None:
with suppress(ValueError):
self._subs.remove(queue)
def since(self, seq: int) -> List[Dict[str, Any]]:
return [item for item in self.history if item["seq"] > seq]
def publish(self, event: str, data: Dict[str, Any]) -> None:
self.seq += 1
item = {"seq": self.seq, "event": event, "data": data}
self.history.append(item)
if not self._subs:
return
on_loop = self._loop is None or threading.current_thread() is self._loop_thread
for queue in list(self._subs):
if on_loop:
_put_drop(queue, item)
else:
assert self._loop is not None
self._loop.call_soon_threadsafe(_put_drop, queue, item)
def _put_drop(queue: asyncio.Queue, item: Dict[str, Any]) -> None:
with suppress(asyncio.QueueFull):
queue.put_nowait(item)
bus = EventBus()
+159 -6
View File
@@ -15,6 +15,7 @@ from unshackle.core.api.compression import safe_inflate
from unshackle.core.api.errors import APIError, APIErrorCode, handle_api_exception from unshackle.core.api.errors import APIError, APIErrorCode, handle_api_exception
from unshackle.core.api.input_bridge import AuthStatus, InputBridge from unshackle.core.api.input_bridge import AuthStatus, InputBridge
from unshackle.core.api.sanitize import safe_cache_key, sanitize_log from unshackle.core.api.sanitize import safe_cache_key, sanitize_log
from unshackle.core.api.session_store import SessionStore
from unshackle.core.config import config from unshackle.core.config import config
from unshackle.core.constants import AUDIO_CODEC_MAP, DYNAMIC_RANGE_MAP, VIDEO_CODEC_MAP from unshackle.core.constants import AUDIO_CODEC_MAP, DYNAMIC_RANGE_MAP, VIDEO_CODEC_MAP
from unshackle.core.providers.anilist import parse_anilist_ref from unshackle.core.providers.anilist import parse_anilist_ref
@@ -325,6 +326,9 @@ def server_proxy_allowed(request: Optional[web.Request] = None) -> bool:
JOB_EVENTS_ROUTE = "/api/download/jobs/{job_id}/events" JOB_EVENTS_ROUTE = "/api/download/jobs/{job_id}/events"
DASHBOARD_PREFIX = "/api/dashboard/"
DASHBOARD_EVENTS_ROUTE = DASHBOARD_PREFIX + "events"
SSE_QUERY_KEY_ROUTES = {JOB_EVENTS_ROUTE, DASHBOARD_EVENTS_ROUTE}
CORS_HEADERS = { CORS_HEADERS = {
"Access-Control-Allow-Origin": "*", "Access-Control-Allow-Origin": "*",
@@ -335,16 +339,42 @@ CORS_HEADERS = {
def request_secret_key(request: web.Request) -> Optional[str]: def request_secret_key(request: web.Request) -> Optional[str]:
"""The caller's API key: the X-Secret-Key header, or on the events route the secret_key """The caller's API key: the X-Secret-Key header, or on SSE routes the secret_key query param."""
query param (EventSource cannot send headers)."""
key = request.headers.get("X-Secret-Key") key = request.headers.get("X-Secret-Key")
if not key: if not key:
resource = request.match_info.route.resource resource = request.match_info.route.resource
if resource is not None and resource.canonical == JOB_EVENTS_ROUTE: if resource is not None and resource.canonical in SSE_QUERY_KEY_ROUTES:
key = request.query.get("secret_key") key = request.query.get("secret_key")
return key return key
def dashboard_key() -> Optional[str]:
"""The developer dashboard key from serve.dashboard.key, or None when the dashboard is off."""
dashboard = (config.serve or {}).get("dashboard")
key = dashboard.get("key") if isinstance(dashboard, dict) else None
return str(key) if key else None
@web.middleware
async def dashboard_authentication(request: web.Request, handler: Any) -> web.StreamResponse:
"""Gate /api/dashboard/ behind serve.dashboard.key only; user and tier keys never pass.
This middleware runs before the user-key middleware, so it answers the dashboard routes
first and no other route ever accepts the dashboard key.
"""
import hmac
if not request.path.startswith(DASHBOARD_PREFIX):
return await handler(request)
expected = dashboard_key()
if not expected:
return await handler(request) # routes are unregistered, so this 404s
provided = request_secret_key(request)
if not provided or not hmac.compare_digest(provided, expected):
return web.json_response({"status": 401, "message": "Dashboard key is invalid."}, status=401)
return await handler(request)
def caller_key(request: Optional[web.Request] = None) -> str: def caller_key(request: Optional[web.Request] = None) -> str:
"""The authenticating API key for a request, or 'anonymous' when unauthenticated.""" """The authenticating API key for a request, or 'anonymous' when unauthenticated."""
if not request: if not request:
@@ -1422,7 +1452,7 @@ async def download_job_events_handler(job_id: str, request: web.Request) -> web.
response = web.StreamResponse( response = web.StreamResponse(
headers={ headers={
"Content-Type": "text/event-stream", "Content-Type": "text/event-stream",
"Cache-Control": "no-cache", "Cache-Control": "no-cache, no-transform",
# Stops nginx from buffering the stream. # Stops nginx from buffering the stream.
"X-Accel-Buffering": "no", "X-Accel-Buffering": "no",
**CORS_HEADERS, **CORS_HEADERS,
@@ -1468,6 +1498,124 @@ async def download_job_events_handler(job_id: str, request: web.Request) -> web.
return response return response
async def dashboard_status_handler(request: web.Request) -> web.Response:
"""Server-wide stats for the developer dashboard."""
from collections import Counter
from unshackle.core.api.download_manager import get_download_manager
from unshackle.core.api.session_store import get_session_store
from unshackle.core.api.stats import stats
store = get_session_store()
jobs = Counter(job.status.value for job in get_download_manager().list_jobs())
return web.json_response(
{
**stats.to_dict(),
"services": len(Services.get_tags()),
"sessions": store.session_count,
"max_sessions": store.max_sessions,
"session_ttl": store.ttl,
"jobs": dict(jobs),
}
)
async def dashboard_sessions_handler(request: web.Request) -> web.Response:
"""Every live remote session, not owner-scoped."""
from unshackle.core.api.session_store import get_session_store
return web.json_response([entry.summary() for entry in get_session_store().list()])
async def dashboard_jobs_handler(request: web.Request) -> web.Response:
"""Every download job, not owner-scoped."""
from unshackle.core.api.download_manager import get_download_manager
return web.json_response([job.to_dict(include_full_details=True) for job in get_download_manager().list_jobs()])
async def dashboard_logs_handler(request: web.Request) -> web.Response:
"""Recent log records; `since` (seq) and `level` filter the ring buffer."""
from unshackle.core.api.stats import ring
try:
since = int(request.query.get("since", 0))
except ValueError:
since = 0
records = ring.since(since, request.query.get("level"), request.query.get("logger"))
return web.json_response({"seq": ring.seq, "records": records})
async def dashboard_events_handler(request: web.Request) -> web.StreamResponse:
"""Server-wide SSE event stream: `log`, `session` and `job` events, plus `stats` every 5 s (also the keep-alive).
Every frame carries `id: <seq>`; a reconnecting EventSource sends Last-Event-ID, and the
server replays the missed events from the bus history. With `?since=<seq>` the same missed
events come back as one JSON burst instead of an event stream, for clients that poll.
"""
from unshackle.core.api.events import bus
if "since" in request.query:
try:
since = int(request.query["since"])
except ValueError:
since = 0
status = await dashboard_status_handler(request)
return web.json_response({"seq": bus.seq, "stats": json.loads(status.text or "{}"), "events": bus.since(since)})
response = web.StreamResponse(
headers={
"Content-Type": "text/event-stream",
# no-transform stops a CDN from compressing or buffering the event stream.
"Cache-Control": "no-cache, no-transform",
"X-Accel-Buffering": "no",
**CORS_HEADERS,
}
)
await response.prepare(request)
async def send(event: str, data: Dict[str, Any], event_id: Optional[int] = None) -> None:
payload = json.dumps(data, separators=(",", ":"), default=str)
head = f"id: {event_id}\n" if event_id is not None else ""
await response.write(f"{head}event: {event}\ndata: {payload}\n\n".encode())
async def send_stats() -> None:
status = await dashboard_status_handler(request)
await send("stats", json.loads(status.text or "{}"))
queue = bus.subscribe()
get_task: Optional[asyncio.Task] = None
loop = asyncio.get_running_loop()
try:
await send_stats()
next_stats = loop.time() + 5
try:
last_id = int(request.headers.get("Last-Event-ID", 0))
except ValueError:
last_id = 0
for item in bus.since(last_id) if last_id else []:
await send(item["event"], item["data"], item["seq"])
while True:
if get_task is None:
get_task = asyncio.create_task(queue.get())
done, _ = await asyncio.wait({get_task}, timeout=max(0.0, next_stats - loop.time()))
if not done:
await send_stats()
next_stats = loop.time() + 5
continue
item = get_task.result()
get_task = None
await send(item["event"], item["data"], item["seq"])
except (ConnectionResetError, asyncio.CancelledError):
log.debug("SSE dashboard client disconnected")
finally:
if get_task is not None:
get_task.cancel()
bus.unsubscribe(queue)
return response
async def cancel_download_job_handler(job_id: str, request: Optional[web.Request] = None) -> web.Response: async def cancel_download_job_handler(job_id: str, request: Optional[web.Request] = None) -> web.Response:
"""Answer the cancel/remove download job request.""" """Answer the cancel/remove download job request."""
from unshackle.core.api.download_manager import TERMINAL_STATUSES, get_download_manager from unshackle.core.api.download_manager import TERMINAL_STATUSES, get_download_manager
@@ -2080,10 +2228,13 @@ async def session_create_handler(data: Dict[str, Any], request: Optional[web.Req
normalized_service, normalized_service,
service_instance, service_instance,
session_id=session_id, session_id=session_id,
owner_key=api_key,
creator_ip=request.remote if request else None,
) )
session.creator_ip = request.remote if request else None
session.owner_key = api_key
session.cache_tag = session_cache_tag session.cache_tag = session_cache_tag
# Echoed to every dashboard viewer, so cap what an arbitrary client can push into it.
if isinstance(data.get("client"), dict) and len(json.dumps(data["client"], default=str)) <= 4096:
session.client = data["client"]
session.input_bridge = bridge session.input_bridge = bridge
session.auth_status = AuthStatus.AUTHENTICATING session.auth_status = AuthStatus.AUTHENTICATING
@@ -2145,6 +2296,7 @@ async def session_titles_handler(session_id: str, request: Optional[web.Request]
tid = str(t.id) if hasattr(t, "id") else str(id(t)) tid = str(t.id) if hasattr(t, "id") else str(id(t))
session.title_map[tid] = t session.title_map[tid] = t
serialized_titles.append(stamp_service_flags(serialize_title(t), service_instance)) serialized_titles.append(stamp_service_flags(serialize_title(t), service_instance))
SessionStore.publish_update(session)
return web.json_response( return web.json_response(
{ {
@@ -2202,6 +2354,7 @@ async def session_tracks_handler(
title_tracks[str(track.id)] = track title_tracks[str(track.id)] = track
session.tracks[str(track.id)] = track session.tracks[str(track.id)] = track
session.tracks_by_title[str(title_id)] = title_tracks session.tracks_by_title[str(title_id)] = title_tracks
SessionStore.publish_update(session)
try: try:
chapters = service_instance.get_chapters(title) chapters = service_instance.get_chapters(title)
+149 -5
View File
@@ -13,11 +13,18 @@ from unshackle.core import services as services_module
from unshackle.core.api.errors import APIError, APIErrorCode, build_error_response, handle_api_exception from unshackle.core.api.errors import APIError, APIErrorCode, build_error_response, handle_api_exception
from unshackle.core.api.handlers import ( from unshackle.core.api.handlers import (
CORS_HEADERS, CORS_HEADERS,
DASHBOARD_EVENTS_ROUTE,
DASHBOARD_PREFIX,
JOB_EVENTS_ROUTE, JOB_EVENTS_ROUTE,
cancel_download_job_handler, cancel_download_job_handler,
clear_cache_handler, clear_cache_handler,
clear_finished_download_jobs_handler, clear_finished_download_jobs_handler,
clear_temp_handler, clear_temp_handler,
dashboard_events_handler,
dashboard_jobs_handler,
dashboard_logs_handler,
dashboard_sessions_handler,
dashboard_status_handler,
delete_history_handler, delete_history_handler,
download_handler, download_handler,
download_history_handler, download_history_handler,
@@ -1930,6 +1937,132 @@ async def session_prompt_submit(request: web.Request) -> web.Response:
) )
@api_handler
async def dashboard_status(request: web.Request) -> web.Response:
"""
Dashboard: server status.
---
summary: Dashboard server status
description: >
Version, uptime, bind address, mode, request counters per API key, 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
exist without a dashboard key.
tags: [Dashboard]
responses:
'200':
description: Server status
'401':
description: Dashboard key missing or invalid
"""
return await dashboard_status_handler(request)
@api_handler
async def dashboard_sessions(request: web.Request) -> web.Response:
"""
Dashboard: active remote sessions.
---
summary: Dashboard remote sessions
description: Every live remote session on the server, with the owner key masked.
tags: [Dashboard]
responses:
'200':
description: Session list
'401':
description: Dashboard key missing or invalid
"""
return await dashboard_sessions_handler(request)
@api_handler
async def dashboard_jobs(request: web.Request) -> web.Response:
"""
Dashboard: download jobs.
---
summary: Dashboard download jobs
description: Every download job on the server regardless of owner. Empty in --remote-only mode.
tags: [Dashboard]
responses:
'200':
description: Job list
'401':
description: Dashboard key missing or invalid
"""
return await dashboard_jobs_handler(request)
@api_handler
async def dashboard_logs(request: web.Request) -> web.Response:
"""
Dashboard: recent log records.
---
summary: Dashboard log ring buffer
description: The last 1000 log records. Poll with `since` set to the `seq` of the last record seen.
tags: [Dashboard]
parameters:
- name: since
in: query
required: false
schema:
type: integer
description: Return only records with seq greater than this value
- name: level
in: query
required: false
schema:
type: string
description: Minimum level (DEBUG, INFO, WARNING, ERROR)
- name: logger
in: query
required: false
schema:
type: string
description: Keep only this logger and its children, e.g. serve or aiohttp.access
responses:
'200':
description: Records plus the current seq
'401':
description: Dashboard key missing or invalid
"""
return await dashboard_logs_handler(request)
@api_handler
async def dashboard_events(request: web.Request) -> web.StreamResponse:
"""
Dashboard: server-wide event stream.
---
summary: Dashboard SSE stream
description: >
Server-Sent Events: a `stats` event on connect and every 5 seconds (also the keep-alive),
plus `log`, `session` (action create/update/delete) and `job` events as they happen. A browser
EventSource can authenticate with the `secret_key` query parameter.
tags: [Dashboard]
parameters:
- name: secret_key
in: query
required: false
schema:
type: string
description: Dashboard key, for clients that cannot send headers
- name: since
in: query
required: false
schema:
type: integer
description: >
Poll mode. Return the events after this seq as one JSON burst
({seq, stats, events}) instead of opening an event stream. Use seq 0 on first call.
responses:
'200':
description: text/event-stream, or JSON for a request that carries `since`
'401':
description: Dashboard key missing or invalid
"""
return await dashboard_events_handler(request)
# Single source of truth for all API routes. `remote` marks endpoints exposed in # Single source of truth for all API routes. `remote` marks endpoints exposed in
# --remote-only server mode; full mode registers every route. Both setup_routes and # --remote-only server mode; full mode registers every route. Both setup_routes and
# setup_swagger derive from this table so the live routes and the docs cannot drift. # setup_swagger derive from this table so the live routes and the docs cannot drift.
@@ -1967,21 +2100,30 @@ ROUTES: list[tuple[str, str, Handler, bool]] = [
("DELETE", "/api/session/{session_id}", session_delete, True), ("DELETE", "/api/session/{session_id}", session_delete, True),
] ]
DASHBOARD_ROUTES: list[tuple[str, str, Handler, bool]] = [
("GET", DASHBOARD_PREFIX + "status", dashboard_status, True),
("GET", DASHBOARD_PREFIX + "sessions", dashboard_sessions, True),
("GET", DASHBOARD_PREFIX + "jobs", dashboard_jobs, True),
("GET", DASHBOARD_PREFIX + "logs", dashboard_logs, True),
("GET", DASHBOARD_EVENTS_ROUTE, dashboard_events, True),
]
def setup_routes(app: web.Application, remote_only: bool = False) -> None:
"""Setup API routes. When remote_only=True, only the remote session endpoints operate.""" def setup_routes(app: web.Application, remote_only: bool = False, dashboard: bool = False) -> None:
"""Setup API routes. When remote_only=True, only the remote session endpoints operate.
When dashboard=True, this also registers the /api/dashboard/ routes."""
add: dict[str, Callable[..., Any]] = { add: dict[str, Callable[..., Any]] = {
"GET": app.router.add_get, "GET": app.router.add_get,
"POST": app.router.add_post, "POST": app.router.add_post,
"DELETE": app.router.add_delete, "DELETE": app.router.add_delete,
} }
for method, path, handler, remote in ROUTES: for method, path, handler, remote in ROUTES + (DASHBOARD_ROUTES if dashboard else []):
if remote_only and not remote: if remote_only and not remote:
continue continue
add[method](path, handler) add[method](path, handler)
def setup_swagger(app: web.Application) -> None: def setup_swagger(app: web.Application, dashboard: bool = False) -> None:
"""Setup Swagger UI documentation.""" """Setup Swagger UI documentation."""
swagger = SwaggerDocs( swagger = SwaggerDocs(
app, app,
@@ -1994,4 +2136,6 @@ def setup_swagger(app: web.Application) -> None:
) )
route: dict[str, Callable[..., Any]] = {"GET": web.get, "POST": web.post, "DELETE": web.delete} route: dict[str, Callable[..., Any]] = {"GET": web.get, "POST": web.post, "DELETE": web.delete}
swagger.add_routes([route[method](path, handler) for method, path, handler, _ in ROUTES]) swagger.add_routes(
[route[method](path, handler) for method, path, handler, _ in ROUTES + (DASHBOARD_ROUTES if dashboard else [])]
)
+64 -3
View File
@@ -11,10 +11,12 @@ from __future__ import annotations
import asyncio import asyncio
import logging import logging
import uuid import uuid
from collections import deque
from dataclasses import dataclass, field from dataclasses import dataclass, field
from datetime import datetime, timezone from datetime import datetime, timezone
from typing import Any, Dict, List, Optional from typing import Any, Deque, Dict, List, Optional
from unshackle.core.api.events import bus
from unshackle.core.api.input_bridge import AUTH_INPUT_TIMEOUT, AuthStatus, InputBridge from unshackle.core.api.input_bridge import AUTH_INPUT_TIMEOUT, AuthStatus, InputBridge
from unshackle.core.api.sanitize import sanitize_log from unshackle.core.api.sanitize import sanitize_log
from unshackle.core.config import config from unshackle.core.config import config
@@ -41,6 +43,8 @@ class SessionEntry:
input_bridge: Optional[InputBridge] = None input_bridge: Optional[InputBridge] = None
auth_status: AuthStatus = AuthStatus.AUTHENTICATED auth_status: AuthStatus = AuthStatus.AUTHENTICATED
auth_error: Optional[str] = None auth_error: Optional[str] = None
client: Dict[str, Any] = field(default_factory=dict) # version/argv/platform the client reported
actions: Deque[Dict[str, Any]] = field(default_factory=lambda: deque(maxlen=500))
created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
last_accessed: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) last_accessed: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
@@ -48,6 +52,39 @@ class SessionEntry:
"""Update last_accessed timestamp.""" """Update last_accessed timestamp."""
self.last_accessed = datetime.now(timezone.utc) self.last_accessed = datetime.now(timezone.utc)
def summary(self) -> Dict[str, Any]:
"""Dashboard view of the session: no service instance, no raw owner key."""
from unshackle.core.api.stats import mask_key
now = datetime.now(timezone.utc)
return {
"id": self.session_id,
"owner": mask_key(self.owner_key),
"creator_ip": self.creator_ip,
"service": self.service_tag,
"title_id": next(iter(self.title_map), None),
"title": str(next(iter(self.title_map.values()), "")) or None,
"titles": len(self.title_map),
"tracks": len(self.tracks),
"auth_status": self.auth_status.value,
"auth_error": self.auth_error,
"client": self.client,
"actions": list(self.actions),
"created_at": self.created_at.isoformat(),
"last_accessed": self.last_accessed.isoformat(),
"created_ts": self.created_at.timestamp(),
"last_accessed_ts": self.last_accessed.timestamp(),
"age_seconds": int((now - self.created_at).total_seconds()),
"idle_seconds": int((now - self.last_accessed).total_seconds()),
}
def _publish(action: str, entry: SessionEntry, reason: Optional[str] = None) -> None:
data = {"action": action, **entry.summary()}
if reason:
data["reason"] = reason
bus.publish("session", data)
class SessionStore: class SessionStore:
"""Thread-safe remote session store with TTL-based expiration.""" """Thread-safe remote session store with TTL-based expiration."""
@@ -72,21 +109,26 @@ class SessionStore:
service_tag: str, service_tag: str,
service_instance: Any, service_instance: Any,
session_id: Optional[str] = None, session_id: Optional[str] = None,
owner_key: Optional[str] = None,
creator_ip: Optional[str] = None,
) -> SessionEntry: ) -> SessionEntry:
"""Make a new remote session with an authenticated service instance.""" """Make a new remote session with an authenticated service instance."""
async with self._lock: async with self._lock:
if len(self._sessions) >= self.max_sessions: if len(self._sessions) >= self.max_sessions:
oldest_id = min(self._sessions, key=lambda k: self._sessions[k].last_accessed) oldest_id = min(self._sessions, key=lambda k: self._sessions[k].last_accessed)
log.warning(f"Max sessions reached ({self.max_sessions}), evicting oldest: {oldest_id}") log.warning(f"Max sessions reached ({self.max_sessions}), evicting oldest: {oldest_id}")
del self._sessions[oldest_id] _publish("delete", self._sessions.pop(oldest_id), "evicted")
session_id = session_id or str(uuid.uuid4()) session_id = session_id or str(uuid.uuid4())
entry = SessionEntry( entry = SessionEntry(
session_id=session_id, session_id=session_id,
service_tag=service_tag, service_tag=service_tag,
service_instance=service_instance, service_instance=service_instance,
owner_key=owner_key,
creator_ip=creator_ip,
) )
self._sessions[session_id] = entry self._sessions[session_id] = entry
_publish("create", entry)
log.info(f"Created session {sanitize_log(session_id)} for service {sanitize_log(service_tag)}") log.info(f"Created session {sanitize_log(session_id)} for service {sanitize_log(service_tag)}")
return entry return entry
@@ -101,7 +143,7 @@ class SessionStore:
elapsed = (datetime.now(timezone.utc) - entry.last_accessed).total_seconds() elapsed = (datetime.now(timezone.utc) - entry.last_accessed).total_seconds()
if elapsed > self.ttl: if elapsed > self.ttl:
log.info(f"Session {sanitize_log(session_id)} expired (elapsed={elapsed:.0f}s, ttl={self.ttl}s)") log.info(f"Session {sanitize_log(session_id)} expired (elapsed={elapsed:.0f}s, ttl={self.ttl}s)")
del self._sessions[session_id] _publish("delete", self._sessions.pop(session_id), "expired")
return None return None
entry.touch() entry.touch()
@@ -115,6 +157,7 @@ class SessionStore:
if entry.input_bridge: if entry.input_bridge:
entry.input_bridge.cancel() entry.input_bridge.cancel()
self.cleanup_cache_dir(entry.cache_tag) self.cleanup_cache_dir(entry.cache_tag)
_publish("delete", entry, "closed")
log.info(f"Deleted session {sanitize_log(session_id)}") log.info(f"Deleted session {sanitize_log(session_id)}")
return True return True
return False return False
@@ -136,6 +179,7 @@ class SessionStore:
if entry.input_bridge: if entry.input_bridge:
entry.input_bridge.cancel() entry.input_bridge.cancel()
self.cleanup_cache_dir(entry.cache_tag) self.cleanup_cache_dir(entry.cache_tag)
_publish("delete", entry, "expired")
if expired: if expired:
log.info(f"Cleaned up {len(expired)} expired sessions") log.info(f"Cleaned up {len(expired)} expired sessions")
return len(expired) return len(expired)
@@ -194,6 +238,23 @@ class SessionStore:
except OSError: except OSError:
break break
@staticmethod
def publish_update(entry: SessionEntry) -> None:
"""Tell dashboard listeners that a session's summary changed (titles or tracks loaded, auth settled)."""
_publish("update", entry)
def record_action(self, session_id: str, action: Dict[str, Any]) -> None:
"""Append one request to the session's action log and tell dashboard listeners."""
entry = self._sessions.get(session_id)
if entry is None:
return
entry.actions.append(action)
_publish("update", entry)
def list(self) -> List[SessionEntry]:
"""Snapshot of every live session."""
return list(self._sessions.values())
@property @property
def session_count(self) -> int: def session_count(self) -> int:
"""Number of active sessions.""" """Number of active sessions."""
+120
View File
@@ -0,0 +1,120 @@
"""Request counters and a log ring buffer for the developer dashboard."""
from __future__ import annotations
import logging
import time
from collections import Counter, deque
from dataclasses import dataclass, field
from typing import Any, Deque, Dict, List, Optional
from aiohttp import web
from unshackle.core import __code_hash__, __version__
from unshackle.core.api.events import bus
from unshackle.core.config import config
def mask_key(key: Optional[str]) -> str:
"""A display label for an API key: its configured username, else a truncated prefix."""
if not key:
return "anonymous"
user = (config.serve or {}).get("users", {}).get(key)
if isinstance(user, dict) and user.get("username"):
return str(user["username"])
return key[:4] + "…"
@dataclass
class ServerStats:
started_at: float = field(default_factory=time.time)
host: str = ""
port: int = 0
mode: str = "full"
requests_total: int = 0
requests_rejected: int = 0
requests_by_key: Counter = field(default_factory=Counter)
def to_dict(self) -> Dict[str, Any]:
return {
"version": __version__,
"code_hash": __code_hash__,
"host": self.host,
"port": self.port,
"mode": self.mode,
"started_at": self.started_at,
"uptime_seconds": int(time.time() - self.started_at),
"requests_total": self.requests_total,
"requests_rejected": self.requests_rejected,
"requests_by_key": dict(self.requests_by_key),
}
stats = ServerStats()
@web.middleware
async def stats_middleware(request: web.Request, handler: Any) -> web.StreamResponse:
from unshackle.core.api.handlers import request_secret_key
stats.requests_total += 1
started = time.perf_counter()
response = await handler(request)
if response.status == 401:
stats.requests_rejected += 1
else:
stats.requests_by_key[mask_key(request_secret_key(request))] += 1
session_id = request.match_info.get("session_id")
if session_id and request.method != "OPTIONS":
from unshackle.core.api.session_store import get_session_store
get_session_store().record_action(
session_id,
{
"ts": time.time(),
"method": request.method,
"action": request.path.split(f"/{session_id}", 1)[-1].strip("/") or "info",
"query": dict(request.query),
"status": response.status,
"ms": round((time.perf_counter() - started) * 1000, 1),
"bytes_in": request.content_length or 0,
"bytes_out": getattr(response, "content_length", None) or 0,
},
)
return response
class RingLogHandler(logging.Handler):
"""Keep the last N log records in memory and publish each one on the event bus."""
def __init__(self, maxlen: int = 1000) -> None:
super().__init__()
self.records: Deque[Dict[str, Any]] = deque(maxlen=maxlen)
self.seq = 0
def emit(self, record: logging.LogRecord) -> None:
self.seq += 1
item = {
"seq": self.seq,
"ts": record.created,
"level": record.levelname,
"logger": record.name,
"msg": self.format(record),
}
self.records.append(item)
bus.publish("log", item)
def since(self, seq: int = 0, level: Optional[str] = None, logger: Optional[str] = None) -> List[Dict[str, Any]]:
min_level = logging.getLevelName(level.upper()) if level else 0
if not isinstance(min_level, int):
min_level = 0
return [
r
for r in self.records
if r["seq"] > seq
and logging.getLevelName(r["level"]) >= min_level
and (not logger or r["logger"] == logger or r["logger"].startswith(logger + "."))
]
ring = RingLogHandler()