mirror of
https://github.com/unshackle-dl/unshackle.git
synced 2026-11-12 20:15:10 +00:00
feat(serve): hot-reload service repos on a timer without a restart
Add serve.services_refresh_interval; refresh pulls the repos and re-imports only the changed services, staging any with active jobs until they finish. Report skipped/broken services in the log and on /api/services.
This commit is contained in:
@@ -62,7 +62,7 @@ If the update check fails (for example, no network), `update_available` and `lat
|
|||||||
|
|
||||||
### `GET /api/services`
|
### `GET /api/services`
|
||||||
|
|
||||||
Show the streaming services available on this server, filtered by your allowlist. Each entry gives the service's tag, matching rules, capability flags, and its CLI parameters (useful when you make a UI that accepts service-specific options).
|
Show the streaming services available on this server, filtered by your allowlist. Each entry gives the service's tag, matching rules, capability flags, and its CLI parameters (useful when you make a UI that accepts service-specific options). `load_errors` lists the services the server skipped because they failed to import, at startup or after a service repository refresh; the server also writes each one to its log. An entry marked `pending_update` has newer code staged: it keeps its current code until its running or queued jobs finish.
|
||||||
|
|
||||||
=== "Request"
|
=== "Request"
|
||||||
|
|
||||||
@@ -100,7 +100,8 @@ Show the streaming services available on this server, filtered by your allowlist
|
|||||||
"has_drm": true,
|
"has_drm": true,
|
||||||
"auth_methods": ["cookies"]
|
"auth_methods": ["cookies"]
|
||||||
}
|
}
|
||||||
]
|
],
|
||||||
|
"load_errors": []
|
||||||
}
|
}
|
||||||
```
|
```
|
||||||
|
|
||||||
@@ -118,6 +119,7 @@ Field notes:
|
|||||||
| `needs_auth` | Whether the service overrides `authenticate()`. |
|
| `needs_auth` | Whether the service overrides `authenticate()`. |
|
||||||
| `has_search` | Whether the service supports `POST /api/search`. |
|
| `has_search` | Whether the service supports `POST /api/search`. |
|
||||||
| `has_drm` | Whether the service overrides a Widevine or PlayReady license hook. |
|
| `has_drm` | Whether the service overrides a Widevine or PlayReady license hook. |
|
||||||
|
| `pending_update` | Set to `true` while a service repository refresh waits for this service's jobs to finish before it swaps in the new code. Absent otherwise. |
|
||||||
| `auth_methods` | Declared or inferred auth methods, e.g. `["cookies"]`, `["credentials"]`. |
|
| `auth_methods` | Declared or inferred auth methods, e.g. `["cookies"]`, `["credentials"]`. |
|
||||||
|
|
||||||
| Status | Meaning |
|
| Status | Meaning |
|
||||||
@@ -773,7 +775,7 @@ The `409` guard is about active file I/O, not an arbitrary lock. An in-flight jo
|
|||||||
|
|
||||||
### `POST /api/maintenance/refresh-services`
|
### `POST /api/maintenance/refresh-services`
|
||||||
|
|
||||||
git-pull every service repository configured under `directories.services`. `refreshed` becomes `true` when all repos updated successfully, or when you configure no repository.
|
git-pull every service repository configured under `directories.services` and re-import the services that changed, without a restart. `refreshed` becomes `true` when all repos updated successfully, or when you configure no repository. `deferred` lists the services that have a running or queued job: they keep their current code until that job finishes, then swap. `load_errors` lists the services whose new code failed to import; the server drops them until the next successful refresh.
|
||||||
|
|
||||||
=== "Response `200`"
|
=== "Response `200`"
|
||||||
|
|
||||||
@@ -781,7 +783,13 @@ git-pull every service repository configured under `directories.services`. `refr
|
|||||||
{
|
{
|
||||||
"refreshed": true,
|
"refreshed": true,
|
||||||
"repos": [
|
"repos": [
|
||||||
{ "spec": "github.com/example/services", "updated": true, "changes": ["~EXAMPLE"] }
|
{
|
||||||
|
"spec": "github.com/example/services",
|
||||||
|
"updated": true,
|
||||||
|
"changes": ["~EXAMPLE", "+EXAMPLE2"],
|
||||||
|
"deferred": ["EXAMPLE"],
|
||||||
|
"load_errors": []
|
||||||
|
}
|
||||||
]
|
]
|
||||||
}
|
}
|
||||||
```
|
```
|
||||||
|
|||||||
@@ -183,7 +183,7 @@ request and response in full.
|
|||||||
|---|---|---|
|
|---|---|---|
|
||||||
| `POST` | `/api/maintenance/clear-cache` | Clear the cache directory. |
|
| `POST` | `/api/maintenance/clear-cache` | Clear the cache directory. |
|
||||||
| `POST` | `/api/maintenance/clear-temp` | Clear the temp directory. |
|
| `POST` | `/api/maintenance/clear-temp` | Clear the temp directory. |
|
||||||
| `POST` | `/api/maintenance/refresh-services` | `git pull` each configured service repository. |
|
| `POST` | `/api/maintenance/refresh-services` | `git pull` each configured service repository and re-import the services that changed. |
|
||||||
|
|
||||||
### Remote-download sessions
|
### Remote-download sessions
|
||||||
|
|
||||||
|
|||||||
@@ -550,7 +550,7 @@ unshackle util COMMAND [ARGS]...
|
|||||||
unshackle util refresh-services
|
unshackle util refresh-services
|
||||||
```
|
```
|
||||||
|
|
||||||
Force a refresh (git pull / hard reset) of all service repos configured under `directories.services`.
|
Force a refresh (git pull / hard reset) of all service repos configured under `directories.services`. This updates the clones on disk. A server that is already running reloads the changed services through `POST /api/maintenance/refresh-services` or `serve.services_refresh_interval` instead.
|
||||||
|
|
||||||
### `util crop PATH ASPECT`
|
### `util crop PATH ASPECT`
|
||||||
|
|
||||||
|
|||||||
@@ -76,6 +76,14 @@ directories:
|
|||||||
services_repo_force: true
|
services_repo_force: true
|
||||||
```
|
```
|
||||||
|
|
||||||
|
!!! info "Refresh inside `serve`"
|
||||||
|
A running server does not restart to pick up a refresh. Both the periodic refresh
|
||||||
|
(`serve.services_refresh_interval`, see [serve](services.md#serve)) and the manual
|
||||||
|
`POST /api/maintenance/refresh-services` endpoint hard-reset the clones and then re-import
|
||||||
|
only the services that changed. A service with a running or queued download job keeps its
|
||||||
|
current code until that job finishes, then swaps; `GET /api/services` marks it with
|
||||||
|
`pending_update` in the meantime.
|
||||||
|
|
||||||
!!! warning "Read-only installs and reinstalls"
|
!!! warning "Read-only installs and reinstalls"
|
||||||
If the services directory lives inside the installed package, a reinstall can delete the
|
If the services directory lives inside the installed package, a reinstall can delete the
|
||||||
`_repos` clones. They are re-cloned on next use. On **read-only installs** you must
|
`_repos` clones. They are re-cloned on next use. On **read-only installs** you must
|
||||||
|
|||||||
@@ -117,6 +117,7 @@ is the [REST API](../../dev/rest-api/index.md) section. These are the config key
|
|||||||
| `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. |
|
||||||
| `compression_level` | int | `1` | gzip level for responses. |
|
| `compression_level` | int | `1` | gzip level for responses. |
|
||||||
|
| `services_refresh_interval` | int (s) | `0` | How often the server pulls the git-backed service repositories in `directories.services` and hot-reloads the services that changed. `0` turns it off. A service with a running or queued job swaps to the new code as soon as its last job finishes. |
|
||||||
| `global_speed_limit` | str | *(unlimited)* | Server-wide download speed cap, e.g. `10M`, `1.5G` or plain bytes/sec (same format as `speed_limit`). One shared budget across all concurrent jobs; per-job speed limits are ignored while it is set. |
|
| `global_speed_limit` | str | *(unlimited)* | Server-wide download speed cap, e.g. `10M`, `1.5G` or plain bytes/sec (same format as `speed_limit`). One shared budget across all concurrent jobs; per-job speed limits are ignored while it is set. |
|
||||||
| `cdm_overrides` | list or bool | *(unset)* | Allowed per-request CDM overrides: a list of permitted device names, or `true` for any. Unset rejects every override. |
|
| `cdm_overrides` | list or bool | *(unset)* | Allowed per-request CDM overrides: a list of permitted device names, or `true` for any. Unset rejects every override. |
|
||||||
| `allow_job_credentials` | bool | `false` | Permit clients to supply credentials per job. |
|
| `allow_job_credentials` | bool | `false` | Permit clients to supply credentials per job. |
|
||||||
@@ -141,11 +142,12 @@ serve:
|
|||||||
api_secret: change-me
|
api_secret: change-me
|
||||||
remote_only: true
|
remote_only: true
|
||||||
services: [EXAMPLE1, EXAMPLE2]
|
services: [EXAMPLE1, EXAMPLE2]
|
||||||
|
services_refresh_interval: 3600 # pull the service repositories hourly and reload what changed
|
||||||
# server-wide download defaults (same keys as the `dl:` block)
|
# server-wide download defaults (same keys as the `dl:` block)
|
||||||
downloads: 3
|
downloads: 3
|
||||||
best_available: true
|
best_available: true
|
||||||
users:
|
users:
|
||||||
a1b2c3d4: # this user's API key
|
a1b2c3d4: # this user's API key
|
||||||
services: [EXAMPLE1] # may only use EXAMPLE1
|
services: [EXAMPLE1] # may only use EXAMPLE1
|
||||||
e5f6a7b8:
|
e5f6a7b8:
|
||||||
server_cdm: true # this key may have the server do the licensing
|
server_cdm: true # this key may have the server do the licensing
|
||||||
|
|||||||
@@ -0,0 +1,74 @@
|
|||||||
|
"""Pins hot-reload of services: re-import on change, drop on removal, and staged swaps for busy services."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import os
|
||||||
|
import shutil
|
||||||
|
import time
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from unshackle.core import services
|
||||||
|
from unshackle.core.config import config
|
||||||
|
|
||||||
|
pytestmark = pytest.mark.unit
|
||||||
|
|
||||||
|
|
||||||
|
def write_service(root: Path, version: int) -> None:
|
||||||
|
pkg = root / "FOO"
|
||||||
|
pkg.mkdir(exist_ok=True)
|
||||||
|
init = pkg / "__init__.py"
|
||||||
|
init.write_text(f"class FOO:\n VERSION = {version}\n ALIASES = ('foo{version}',)\n")
|
||||||
|
# the bytecode cache keys on mtime and size; same-second rewrites of equal length would reuse the stale .pyc
|
||||||
|
os.utime(init, (time.time() + version, time.time() + version))
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def service_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
|
||||||
|
root = tmp_path / "svc"
|
||||||
|
root.mkdir()
|
||||||
|
write_service(root, 1)
|
||||||
|
monkeypatch.setattr(config.directories, "services", [str(root)])
|
||||||
|
monkeypatch.setattr(services, "SERVICES", [])
|
||||||
|
monkeypatch.setattr(services, "MODULES", {})
|
||||||
|
monkeypatch.setattr(services, "ALIASES", {})
|
||||||
|
monkeypatch.setattr(services, "LOAD_ERRORS", [])
|
||||||
|
monkeypatch.setattr(services, "PENDING", set())
|
||||||
|
assert services.reload_services(["FOO"]) == []
|
||||||
|
assert services.MODULES["FOO"].VERSION == 1
|
||||||
|
yield root
|
||||||
|
services.reload_services(["FOO"])
|
||||||
|
|
||||||
|
|
||||||
|
def test_reload_picks_up_new_code_and_removal(service_dir: Path):
|
||||||
|
write_service(service_dir, 2)
|
||||||
|
assert services.reload_services(["FOO"]) == []
|
||||||
|
assert services.MODULES["FOO"].VERSION == 2
|
||||||
|
assert services.ALIASES["FOO"] == ("foo2",)
|
||||||
|
assert services.Services.get_tag("foo2") == "FOO"
|
||||||
|
|
||||||
|
shutil.rmtree(service_dir / "FOO")
|
||||||
|
assert services.reload_services(["FOO"]) == []
|
||||||
|
assert "FOO" not in services.MODULES
|
||||||
|
assert "FOO" not in services.Services.get_tags()
|
||||||
|
|
||||||
|
|
||||||
|
def test_busy_service_is_staged_then_applied(service_dir: Path, monkeypatch: pytest.MonkeyPatch):
|
||||||
|
write_service(service_dir, 2)
|
||||||
|
monkeypatch.setattr(services, "repo_specs", lambda: ["example/repo"])
|
||||||
|
monkeypatch.setattr(services, "refresh_repo", lambda spec: (service_dir, ["~FOO"]))
|
||||||
|
|
||||||
|
repos = services.refresh_and_reload(busy={"FOO"})
|
||||||
|
assert repos == [
|
||||||
|
{"spec": "example/repo", "updated": True, "changes": ["~FOO"], "deferred": ["FOO"], "load_errors": []}
|
||||||
|
]
|
||||||
|
assert services.MODULES["FOO"].VERSION == 1
|
||||||
|
assert "FOO" in services.PENDING
|
||||||
|
|
||||||
|
assert services.apply_pending(busy={"FOO"}) == []
|
||||||
|
assert services.MODULES["FOO"].VERSION == 1
|
||||||
|
|
||||||
|
assert services.apply_pending(busy=set()) == ["FOO"]
|
||||||
|
assert services.MODULES["FOO"].VERSION == 2
|
||||||
|
assert services.PENDING == set()
|
||||||
@@ -1,6 +1,8 @@
|
|||||||
|
import asyncio
|
||||||
import hmac
|
import hmac
|
||||||
import logging
|
import logging
|
||||||
import subprocess
|
import subprocess
|
||||||
|
from contextlib import suppress
|
||||||
|
|
||||||
import click
|
import click
|
||||||
from aiohttp import web
|
from aiohttp import web
|
||||||
@@ -14,6 +16,51 @@ 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
|
||||||
|
|
||||||
|
|
||||||
|
def _install_service_refresh(app: web.Application) -> None:
|
||||||
|
"""Report service load issues, then periodically pull the service repos and hot-reload changed services."""
|
||||||
|
from unshackle.core import services
|
||||||
|
|
||||||
|
log = logging.getLogger("serve")
|
||||||
|
services.log_load_issues()
|
||||||
|
interval = int(config.serve.get("services_refresh_interval", 0) or 0)
|
||||||
|
if interval <= 0 or not services.repo_specs():
|
||||||
|
return
|
||||||
|
|
||||||
|
async def loop() -> None:
|
||||||
|
from unshackle.core.api.download_manager import get_download_manager
|
||||||
|
|
||||||
|
manager = get_download_manager()
|
||||||
|
while True:
|
||||||
|
await asyncio.sleep(interval)
|
||||||
|
try:
|
||||||
|
applied = await asyncio.to_thread(services.apply_pending, manager.busy_services())
|
||||||
|
if applied:
|
||||||
|
log.info(f"Services reloaded: {', '.join(applied)}")
|
||||||
|
for r in await asyncio.to_thread(services.refresh_and_reload, manager.busy_services()):
|
||||||
|
if r["changes"]:
|
||||||
|
log.info(f"Services refreshed {r['spec']}: {', '.join(r['changes'])}")
|
||||||
|
if r["deferred"]:
|
||||||
|
log.info(f"Services staged until their jobs finish: {', '.join(r['deferred'])}")
|
||||||
|
for err in r["load_errors"]:
|
||||||
|
log.error(f"Service reload failed: {err}")
|
||||||
|
except Exception:
|
||||||
|
log.exception("Service refresh failed")
|
||||||
|
|
||||||
|
async def start(_app: web.Application) -> None:
|
||||||
|
_app["service_refresh_task"] = asyncio.create_task(loop())
|
||||||
|
log.info(f"Service repos refresh every {interval}s")
|
||||||
|
|
||||||
|
async def stop(_app: web.Application) -> None:
|
||||||
|
task = _app.get("service_refresh_task")
|
||||||
|
if task:
|
||||||
|
task.cancel()
|
||||||
|
with suppress(asyncio.CancelledError):
|
||||||
|
await task
|
||||||
|
|
||||||
|
app.on_startup.append(start)
|
||||||
|
app.on_cleanup.append(stop)
|
||||||
|
|
||||||
|
|
||||||
@click.command(
|
@click.command(
|
||||||
short_help="Serve your Local Widevine/PlayReady Devices and REST API for Remote Access.",
|
short_help="Serve your Local Widevine/PlayReady Devices and REST API for Remote Access.",
|
||||||
context_settings=context_settings,
|
context_settings=context_settings,
|
||||||
@@ -196,6 +243,7 @@ def serve(
|
|||||||
|
|
||||||
app.on_startup.append(start_session_cleanup)
|
app.on_startup.append(start_session_cleanup)
|
||||||
app.on_cleanup.append(stop_session_cleanup)
|
app.on_cleanup.append(stop_session_cleanup)
|
||||||
|
_install_service_refresh(app)
|
||||||
|
|
||||||
setup_routes(app, remote_only=remote_only)
|
setup_routes(app, remote_only=remote_only)
|
||||||
if not remote_only:
|
if not remote_only:
|
||||||
@@ -289,6 +337,7 @@ def serve(
|
|||||||
|
|
||||||
app.on_startup.append(start_session_cleanup)
|
app.on_startup.append(start_session_cleanup)
|
||||||
app.on_cleanup.append(stop_session_cleanup)
|
app.on_cleanup.append(stop_session_cleanup)
|
||||||
|
_install_service_refresh(app)
|
||||||
|
|
||||||
if serve_widevine:
|
if serve_widevine:
|
||||||
app.on_startup.append(pywidevine_serve._startup)
|
app.on_startup.append(pywidevine_serve._startup)
|
||||||
|
|||||||
@@ -710,10 +710,32 @@ class DownloadQueueManager:
|
|||||||
self.put(queue, payload)
|
self.put(queue, payload)
|
||||||
|
|
||||||
def publish_terminal(self, job: DownloadJob) -> None:
|
def publish_terminal(self, job: DownloadJob) -> None:
|
||||||
"""Publish a job's final state, then close every listener's event stream."""
|
"""Publish a job's final state, close every listener's event stream, then apply staged service reloads."""
|
||||||
self.publish(job, job.status.value)
|
self.publish(job, job.status.value)
|
||||||
for queue in self._subscribers.pop(job.job_id, []):
|
for queue in self._subscribers.pop(job.job_id, []):
|
||||||
self.put(queue, None)
|
self.put(queue, None)
|
||||||
|
self.schedule_pending_reload()
|
||||||
|
|
||||||
|
def busy_services(self) -> set[str]:
|
||||||
|
"""Tags with a job that has not reached a terminal status; queued jobs count so none starts on half-swapped code."""
|
||||||
|
return {job.service for job in self._jobs.values() if job.status not in TERMINAL_STATUSES}
|
||||||
|
|
||||||
|
def schedule_pending_reload(self) -> None:
|
||||||
|
"""Swap in staged service updates whose last job has finished, off the event loop."""
|
||||||
|
from unshackle.core import services
|
||||||
|
|
||||||
|
if not services.PENDING:
|
||||||
|
return
|
||||||
|
|
||||||
|
async def run() -> None:
|
||||||
|
try:
|
||||||
|
applied = await asyncio.to_thread(services.apply_pending, self.busy_services())
|
||||||
|
if applied:
|
||||||
|
log.info(f"Services reloaded after job completion: {', '.join(applied)}")
|
||||||
|
except Exception:
|
||||||
|
log.exception("Applying pending service reloads failed")
|
||||||
|
|
||||||
|
asyncio.get_running_loop().create_task(run())
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def put(queue: asyncio.Queue, item: Optional[Dict[str, Any]]) -> None:
|
def put(queue: asyncio.Queue, item: Optional[Dict[str, Any]]) -> None:
|
||||||
|
|||||||
@@ -1796,20 +1796,13 @@ async def clear_temp_handler(request: Optional[web.Request] = None) -> web.Respo
|
|||||||
|
|
||||||
|
|
||||||
async def refresh_services_handler(request: Optional[web.Request] = None) -> web.Response:
|
async def refresh_services_handler(request: Optional[web.Request] = None) -> web.Response:
|
||||||
"""Answer the request to refresh the service repos configured in directories.services."""
|
"""Refresh the service repos configured in directories.services and reload the changed services."""
|
||||||
from unshackle.core.service_repo import is_repo_spec, refresh_repo
|
from unshackle.core.api.download_manager import get_download_manager
|
||||||
|
from unshackle.core.services import refresh_and_reload
|
||||||
|
|
||||||
try:
|
try:
|
||||||
entries = config.directories.services
|
busy = get_download_manager().busy_services()
|
||||||
if not isinstance(entries, list):
|
repos = await asyncio.to_thread(refresh_and_reload, busy)
|
||||||
entries = [entries]
|
|
||||||
specs = [e for e in entries if isinstance(e, str) and is_repo_spec(e)]
|
|
||||||
|
|
||||||
repos = []
|
|
||||||
for spec in specs:
|
|
||||||
dest, changes = await asyncio.to_thread(refresh_repo, spec)
|
|
||||||
repos.append({"spec": spec, "updated": dest is not None, "changes": list(changes or [])})
|
|
||||||
|
|
||||||
return web.json_response({"refreshed": all(r["updated"] for r in repos), "repos": repos})
|
return web.json_response({"refreshed": all(r["updated"] for r in repos), "repos": repos})
|
||||||
|
|
||||||
except APIError:
|
except APIError:
|
||||||
|
|||||||
@@ -9,6 +9,7 @@ from aiohttp import web
|
|||||||
from aiohttp_swagger3 import SwaggerDocs, SwaggerInfo, SwaggerUiSettings
|
from aiohttp_swagger3 import SwaggerDocs, SwaggerInfo, SwaggerUiSettings
|
||||||
|
|
||||||
from unshackle.core import __code_hash__, __version__
|
from unshackle.core import __code_hash__, __version__
|
||||||
|
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,
|
||||||
@@ -150,6 +151,11 @@ async def services(request: web.Request) -> web.Response:
|
|||||||
schema:
|
schema:
|
||||||
type: object
|
type: object
|
||||||
properties:
|
properties:
|
||||||
|
load_errors:
|
||||||
|
type: array
|
||||||
|
description: Services skipped because they failed to import (startup or repo refresh)
|
||||||
|
items:
|
||||||
|
type: string
|
||||||
services:
|
services:
|
||||||
type: array
|
type: array
|
||||||
items:
|
items:
|
||||||
@@ -157,6 +163,9 @@ async def services(request: web.Request) -> web.Response:
|
|||||||
properties:
|
properties:
|
||||||
tag:
|
tag:
|
||||||
type: string
|
type: string
|
||||||
|
pending_update:
|
||||||
|
type: boolean
|
||||||
|
description: Present (true) while a repo refresh waits for this service's jobs to finish
|
||||||
aliases:
|
aliases:
|
||||||
type: array
|
type: array
|
||||||
items:
|
items:
|
||||||
@@ -221,6 +230,8 @@ async def services(request: web.Request) -> web.Response:
|
|||||||
"url": None,
|
"url": None,
|
||||||
"help": None,
|
"help": None,
|
||||||
}
|
}
|
||||||
|
if tag in services_module.PENDING:
|
||||||
|
service_data["pending_update"] = True
|
||||||
|
|
||||||
try:
|
try:
|
||||||
service_module = Services.load(tag)
|
service_module = Services.load(tag)
|
||||||
@@ -322,7 +333,7 @@ async def services(request: web.Request) -> web.Response:
|
|||||||
|
|
||||||
services_info.append(service_data)
|
services_info.append(service_data)
|
||||||
|
|
||||||
return web.json_response({"services": services_info})
|
return web.json_response({"services": services_info, "load_errors": list(services_module.LOAD_ERRORS)})
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
log.exception("Error listing services")
|
log.exception("Error listing services")
|
||||||
debug_mode = request.app.get("debug_api", False)
|
debug_mode = request.app.get("debug_api", False)
|
||||||
@@ -1397,9 +1408,12 @@ async def maintenance_refresh_services(request: web.Request) -> web.Response:
|
|||||||
---
|
---
|
||||||
summary: Refresh service repos
|
summary: Refresh service repos
|
||||||
description: >
|
description: >
|
||||||
Force-sync (git pull) every service repo configured in directories.services.
|
Force-sync (git pull) every service repo configured in directories.services and
|
||||||
`refreshed` is true when all repos synced, or when you configured no repos. The
|
re-import the services that changed, without a restart. `refreshed` is true when all
|
||||||
`repos` field holds the per-repo results.
|
repos synced, or when you configured no repos. The `repos` field holds the per-repo
|
||||||
|
results: `deferred` names the services that keep their current code until their
|
||||||
|
running or queued jobs finish, and `load_errors` names the services whose new code
|
||||||
|
failed to import.
|
||||||
responses:
|
responses:
|
||||||
'200':
|
'200':
|
||||||
description: Refresh results
|
description: Refresh results
|
||||||
@@ -1423,6 +1437,16 @@ async def maintenance_refresh_services(request: web.Request) -> web.Response:
|
|||||||
type: array
|
type: array
|
||||||
items:
|
items:
|
||||||
type: string
|
type: string
|
||||||
|
deferred:
|
||||||
|
type: array
|
||||||
|
description: Services staged until their running or queued jobs finish
|
||||||
|
items:
|
||||||
|
type: string
|
||||||
|
load_errors:
|
||||||
|
type: array
|
||||||
|
description: Services whose new code failed to import
|
||||||
|
items:
|
||||||
|
type: string
|
||||||
'500':
|
'500':
|
||||||
description: Server error
|
description: Server error
|
||||||
"""
|
"""
|
||||||
|
|||||||
+146
-28
@@ -3,48 +3,66 @@ from __future__ import annotations
|
|||||||
import inspect
|
import inspect
|
||||||
import logging
|
import logging
|
||||||
import re
|
import re
|
||||||
|
import sys
|
||||||
|
import threading
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
from typing import Any, Iterable
|
||||||
|
|
||||||
import click
|
import click
|
||||||
|
|
||||||
from unshackle.core.config import config
|
from unshackle.core.config import config
|
||||||
from unshackle.core.service import Service
|
from unshackle.core.service import Service
|
||||||
from unshackle.core.service_repo import DirtyServiceRepo, is_repo_spec, resolve_service_repo
|
from unshackle.core.service_repo import DirtyServiceRepo, is_repo_spec, refresh_repo, resolve_service_repo
|
||||||
from unshackle.core.utilities import import_module_by_path
|
from unshackle.core.utilities import import_module_by_path
|
||||||
from unshackle.core.utils.redact import redact_path
|
from unshackle.core.utils.redact import redact_path
|
||||||
|
|
||||||
log = logging.getLogger("services")
|
log = logging.getLogger("services")
|
||||||
|
|
||||||
raw = config.directories.services
|
|
||||||
if not isinstance(raw, list):
|
|
||||||
raw = [raw]
|
|
||||||
# resolve in config order - priority IS list order: the first source to define a tag is the source,
|
|
||||||
# later sources (local or repo) with the same tag are shadowed. List local last to make it a fallback.
|
|
||||||
service_dirs: list[Path] = []
|
|
||||||
DIRTY_REPOS: list[str] = []
|
DIRTY_REPOS: list[str] = []
|
||||||
for _entry in raw:
|
|
||||||
if isinstance(_entry, str) and is_repo_spec(_entry):
|
|
||||||
try:
|
|
||||||
_resolved = resolve_service_repo(_entry, force=config.services_repo_force)
|
|
||||||
except DirtyServiceRepo as e:
|
|
||||||
# local edits in the clone - record it and let check_load_errors() exit cleanly
|
|
||||||
DIRTY_REPOS.append(redact_path(str(e.path)))
|
|
||||||
_resolved = None
|
|
||||||
if _resolved:
|
|
||||||
service_dirs.append(_resolved)
|
|
||||||
else:
|
|
||||||
service_dirs.append(Path(_entry))
|
|
||||||
|
|
||||||
seen: dict[str, Path] = {}
|
|
||||||
SHADOWED: list[str] = []
|
SHADOWED: list[str] = []
|
||||||
for _dir in service_dirs:
|
|
||||||
for _path in _dir.glob("*/__init__.py"):
|
|
||||||
tag = _path.parent.stem
|
def discover_services() -> tuple[list[Path], list[str]]:
|
||||||
if tag in seen:
|
"""Resolve the configured service dirs and glob every ``TAG/__init__.py``.
|
||||||
SHADOWED.append(f"{tag}: using {redact_path(str(seen[tag]))}, ignoring duplicate {redact_path(str(_path))}")
|
|
||||||
|
Priority IS list order: the first source to define a tag is the source, and discovery
|
||||||
|
shadows a later source (local or repo) with the same tag. List local last to make it a
|
||||||
|
fallback.
|
||||||
|
Returns the sorted service paths and the human-readable shadow lines.
|
||||||
|
"""
|
||||||
|
raw = config.directories.services
|
||||||
|
if not isinstance(raw, list):
|
||||||
|
raw = [raw]
|
||||||
|
service_dirs: list[Path] = []
|
||||||
|
for entry in raw:
|
||||||
|
if isinstance(entry, str) and is_repo_spec(entry):
|
||||||
|
try:
|
||||||
|
resolved = resolve_service_repo(entry, force=config.services_repo_force)
|
||||||
|
except DirtyServiceRepo as e:
|
||||||
|
# local edits in the clone - record it and let check_load_errors() exit cleanly
|
||||||
|
if (dirty := redact_path(str(e.path))) not in DIRTY_REPOS:
|
||||||
|
DIRTY_REPOS.append(dirty)
|
||||||
|
resolved = None
|
||||||
|
if resolved:
|
||||||
|
service_dirs.append(resolved)
|
||||||
else:
|
else:
|
||||||
seen[tag] = _path
|
service_dirs.append(Path(entry))
|
||||||
SERVICES = sorted(seen.values(), key=lambda x: x.parent.stem)
|
|
||||||
|
seen: dict[str, Path] = {}
|
||||||
|
shadowed: list[str] = []
|
||||||
|
for service_dir in service_dirs:
|
||||||
|
for path in service_dir.glob("*/__init__.py"):
|
||||||
|
tag = path.parent.stem
|
||||||
|
if tag in seen:
|
||||||
|
shadowed.append(
|
||||||
|
f"{tag}: using {redact_path(str(seen[tag]))}, ignoring duplicate {redact_path(str(path))}"
|
||||||
|
)
|
||||||
|
else:
|
||||||
|
seen[tag] = path
|
||||||
|
return sorted(seen.values(), key=lambda x: x.parent.stem), shadowed
|
||||||
|
|
||||||
|
|
||||||
|
SERVICES, SHADOWED = discover_services()
|
||||||
|
|
||||||
|
|
||||||
def load_service(path: Path) -> object:
|
def load_service(path: Path) -> object:
|
||||||
@@ -88,10 +106,110 @@ MODULES, LOAD_ERRORS = load_services(SERVICES)
|
|||||||
|
|
||||||
ALIASES = {tag: getattr(module, "ALIASES", ()) for tag, module in MODULES.items()}
|
ALIASES = {tag: getattr(module, "ALIASES", ()) for tag, module in MODULES.items()}
|
||||||
|
|
||||||
|
PENDING: set[str] = set()
|
||||||
|
RELOAD_LOCK = threading.Lock()
|
||||||
|
|
||||||
|
|
||||||
|
def reload_services(tags: Iterable[str]) -> list[str]:
|
||||||
|
"""Re-import the given service tags after a repo refresh; returns load errors.
|
||||||
|
|
||||||
|
Mutates every module global in place so importers that bound ``SERVICES``,
|
||||||
|
``MODULES`` or ``ALIASES`` by name see the swap. Drops a tag that discovery no
|
||||||
|
longer finds instead of reloading it.
|
||||||
|
"""
|
||||||
|
wanted = set(tags)
|
||||||
|
if not wanted:
|
||||||
|
return []
|
||||||
|
with RELOAD_LOCK:
|
||||||
|
for tag in wanted:
|
||||||
|
for name in [n for n in sys.modules if n == tag or n.startswith(f"{tag}.")]:
|
||||||
|
del sys.modules[name]
|
||||||
|
LOAD_ERRORS[:] = [err for err in LOAD_ERRORS if not err.startswith(f"{tag}: ")]
|
||||||
|
|
||||||
|
paths, shadowed = discover_services()
|
||||||
|
SERVICES[:] = paths
|
||||||
|
SHADOWED[:] = shadowed
|
||||||
|
|
||||||
|
errors: list[str] = []
|
||||||
|
found = {path.parent.stem: path for path in paths}
|
||||||
|
for tag in wanted:
|
||||||
|
if tag not in found:
|
||||||
|
MODULES.pop(tag, None)
|
||||||
|
ALIASES.pop(tag, None)
|
||||||
|
continue
|
||||||
|
try:
|
||||||
|
MODULES[tag] = load_service(found[tag])
|
||||||
|
ALIASES[tag] = getattr(MODULES[tag], "ALIASES", ())
|
||||||
|
except Exception as e:
|
||||||
|
errors.append(str(e))
|
||||||
|
LOAD_ERRORS.extend(errors)
|
||||||
|
PENDING.difference_update(wanted)
|
||||||
|
return errors
|
||||||
|
|
||||||
|
|
||||||
|
def repo_specs() -> list[str]:
|
||||||
|
"""The git repo specs configured under ``directories.services``."""
|
||||||
|
raw = config.directories.services
|
||||||
|
entries: list[Any] = raw if isinstance(raw, list) else [raw]
|
||||||
|
return [e for e in entries if isinstance(e, str) and is_repo_spec(e)]
|
||||||
|
|
||||||
|
|
||||||
|
def refresh_and_reload(busy: set[str] | None = None) -> list[dict[str, Any]]:
|
||||||
|
"""Force-sync every configured service repo and re-import the services that changed.
|
||||||
|
|
||||||
|
Tags in ``busy`` (services with a download job that has not finished) go to ``PENDING``
|
||||||
|
instead and appear as ``deferred``; ``apply_pending`` swaps them in later.
|
||||||
|
Blocking (git subprocess): callers wrap it in ``asyncio.to_thread``.
|
||||||
|
"""
|
||||||
|
busy = busy or set()
|
||||||
|
repos: list[dict[str, Any]] = []
|
||||||
|
for spec in repo_specs():
|
||||||
|
dest, changes = refresh_repo(spec)
|
||||||
|
tags = {line[1:] for line in changes if line[:1] in "+~-!"}
|
||||||
|
if dest and "cloned (new)" in changes:
|
||||||
|
tags.update(p.parent.stem for p in dest.glob("*/__init__.py"))
|
||||||
|
deferred = sorted(tags & busy)
|
||||||
|
PENDING.update(deferred)
|
||||||
|
errors = reload_services(tags - busy)
|
||||||
|
repos.append(
|
||||||
|
{
|
||||||
|
"spec": spec,
|
||||||
|
"updated": dest is not None,
|
||||||
|
"changes": list(changes),
|
||||||
|
"deferred": deferred,
|
||||||
|
"load_errors": errors,
|
||||||
|
}
|
||||||
|
)
|
||||||
|
return repos
|
||||||
|
|
||||||
|
|
||||||
|
def apply_pending(busy: set[str] | None = None) -> list[str]:
|
||||||
|
"""Reload every staged tag whose service is no longer busy; returns the tags applied."""
|
||||||
|
ready = sorted(PENDING - (busy or set()))
|
||||||
|
if ready:
|
||||||
|
for err in reload_services(ready):
|
||||||
|
log.error("Service reload failed: %s", err)
|
||||||
|
return ready
|
||||||
|
|
||||||
|
|
||||||
SUMMARY_LOGGED = False
|
SUMMARY_LOGGED = False
|
||||||
|
|
||||||
|
|
||||||
|
def log_load_issues() -> None:
|
||||||
|
"""Log every service load problem without raising - for long-running hosts like serve.
|
||||||
|
|
||||||
|
The loader skips broken services (the rest keep working), so the admin needs the log
|
||||||
|
line the CLI would otherwise render as an error box.
|
||||||
|
"""
|
||||||
|
log.info(f"Loaded {len(MODULES)} services" + (f" ({len(SHADOWED)} duplicate(s) ignored)" if SHADOWED else ""))
|
||||||
|
for shadow in SHADOWED:
|
||||||
|
log.debug("%s", shadow)
|
||||||
|
for path in DIRTY_REPOS:
|
||||||
|
log.warning(f"Service repo has local changes, not refreshed: {path}")
|
||||||
|
for err in LOAD_ERRORS:
|
||||||
|
log.error(f"Service skipped: {err}")
|
||||||
|
|
||||||
|
|
||||||
def check_load_errors() -> None:
|
def check_load_errors() -> None:
|
||||||
"""Log a one-line load summary (once) and raise a single clean error on failures.
|
"""Log a one-line load summary (once) and raise a single clean error on failures.
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user