mirror of
https://github.com/pewdiepie-archdaemon/odysseus.git
synced 2026-11-01 18:52:21 +01:00
`_gather_inputs()` fails soft: a source that raises leaves its inventory as
`[]`. Every probe reads an empty inventory as "not configured" and answers
`disabled`, so a broken integrations file, email store, or endpoint database is
indistinguishable from a subsystem the operator deliberately turned off. With
the remaining subsystems disabled, `GET /api/diagnostics/services` then reports
`overall: "ok"` while three of its inputs failed to load — the exceptions appear
only in DEBUG logs.
Record each failure alongside the empty value (`failed`: source -> controlled
category) and report the affected subsystem through `_source_unavailable()`
rather than probing it with an inventory that is empty for the wrong reason.
- `degraded`, not `disabled`: the configured state is unknown, not known-off.
Not `down` either — nothing reached the subsystem and failed; the inventory
could not be read. Severity is non-zero either way, so a report containing a
failed source can no longer roll up to `ok`.
- A source that loads and is genuinely empty still reports `disabled`.
- One broken source degrades one entry; the others are still probed.
- `meta` carries the source name and the controlled `_classify_error()`
category only, never exception text, credentials, or database detail.
- `inputs.get("failed")` so an inputs payload without the key still probes
normally.
Verified with the reporter's harness: settings loading with
`search_provider: "disabled"` while the other three sources raise now yields
`overall: "degraded"` with `searxng` still `disabled`; before, all five read
`disabled` and `overall` was `ok`.
This commit is contained in:
+63
-8
@@ -76,9 +76,23 @@ _ERROR_DETAIL = {
|
|||||||
"auth_or_protocol_error": "authentication or protocol error",
|
"auth_or_protocol_error": "authentication or protocol error",
|
||||||
"no_models": "endpoint returned no models",
|
"no_models": "endpoint returned no models",
|
||||||
"no_host": "no host configured",
|
"no_host": "no host configured",
|
||||||
|
"config_source_error": "configuration source could not be read",
|
||||||
"error": "probe failed",
|
"error": "probe failed",
|
||||||
}
|
}
|
||||||
|
|
||||||
|
# Config/inventory source behind each network subsystem. `_gather_inputs()`
|
||||||
|
# fails soft, so a source that could not be read hands its probe an empty
|
||||||
|
# inventory — indistinguishable from "nothing configured", which every probe
|
||||||
|
# answers with `disabled`. The collector therefore reports the source failure
|
||||||
|
# instead of running a probe whose verdict would be a lie.
|
||||||
|
CONFIG_SOURCE_ERROR = "config_source_error"
|
||||||
|
_SUBSYSTEM_SOURCE = {
|
||||||
|
"searxng": "settings",
|
||||||
|
"ntfy": "integrations",
|
||||||
|
"email": "accounts",
|
||||||
|
"providers": "endpoints",
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
def _svc(name: str, status: str, detail: str, **meta: Any) -> Dict[str, Any]:
|
def _svc(name: str, status: str, detail: str, **meta: Any) -> Dict[str, Any]:
|
||||||
return {"name": name, "status": status, "detail": detail, "meta": dict(meta)}
|
return {"name": name, "status": status, "detail": detail, "meta": dict(meta)}
|
||||||
@@ -411,26 +425,33 @@ def _gather_inputs() -> Dict[str, Any]:
|
|||||||
"""Pull live config/account/endpoint lists from the app's data sources.
|
"""Pull live config/account/endpoint lists from the app's data sources.
|
||||||
|
|
||||||
Each lookup fails soft: a broken source yields an empty/neutral value so a
|
Each lookup fails soft: a broken source yields an empty/neutral value so a
|
||||||
single failure can't take down the whole health report.
|
single failure can't take down the whole health report. The failure is also
|
||||||
|
*recorded* under `failed` (source -> controlled category), because the empty
|
||||||
|
value alone cannot be told apart from a source that loaded and is genuinely
|
||||||
|
empty — and the probes read empty as `disabled`.
|
||||||
"""
|
"""
|
||||||
settings: Dict[str, Any] = {}
|
settings: Dict[str, Any] = {}
|
||||||
integrations: List[Dict[str, Any]] = []
|
integrations: List[Dict[str, Any]] = []
|
||||||
accounts: List[Dict[str, Any]] = []
|
accounts: List[Dict[str, Any]] = []
|
||||||
endpoints: List[Dict[str, Any]] = []
|
endpoints: List[Dict[str, Any]] = []
|
||||||
|
failed: Dict[str, str] = {}
|
||||||
try:
|
try:
|
||||||
from src.settings import load_settings
|
from src.settings import load_settings
|
||||||
settings = load_settings() or {}
|
settings = load_settings() or {}
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
|
failed["settings"] = _classify_error(e)
|
||||||
logger.debug(f"service_health: settings load failed: {e}")
|
logger.debug(f"service_health: settings load failed: {e}")
|
||||||
try:
|
try:
|
||||||
from src.integrations import load_integrations
|
from src.integrations import load_integrations
|
||||||
integrations = load_integrations() or []
|
integrations = load_integrations() or []
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
|
failed["integrations"] = _classify_error(e)
|
||||||
logger.debug(f"service_health: integrations load failed: {e}")
|
logger.debug(f"service_health: integrations load failed: {e}")
|
||||||
try:
|
try:
|
||||||
from routes.email_helpers import _list_email_accounts
|
from routes.email_helpers import _list_email_accounts
|
||||||
accounts = _list_email_accounts() or []
|
accounts = _list_email_accounts() or []
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
|
failed["accounts"] = _classify_error(e)
|
||||||
logger.debug(f"service_health: email accounts load failed: {e}")
|
logger.debug(f"service_health: email accounts load failed: {e}")
|
||||||
try:
|
try:
|
||||||
from core.database import SessionLocal, ModelEndpoint
|
from core.database import SessionLocal, ModelEndpoint
|
||||||
@@ -443,9 +464,10 @@ def _gather_inputs() -> Dict[str, Any]:
|
|||||||
finally:
|
finally:
|
||||||
db.close()
|
db.close()
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
|
failed["endpoints"] = _classify_error(e)
|
||||||
logger.debug(f"service_health: endpoint load failed: {e}")
|
logger.debug(f"service_health: endpoint load failed: {e}")
|
||||||
return {"settings": settings, "integrations": integrations,
|
return {"settings": settings, "integrations": integrations,
|
||||||
"accounts": accounts, "endpoints": endpoints}
|
"accounts": accounts, "endpoints": endpoints, "failed": failed}
|
||||||
|
|
||||||
|
|
||||||
async def _run_subsystem(name: str, fn: Callable, *args: Any) -> Dict[str, Any]:
|
async def _run_subsystem(name: str, fn: Callable, *args: Any) -> Dict[str, Any]:
|
||||||
@@ -464,6 +486,24 @@ async def _run_subsystem(name: str, fn: Callable, *args: Any) -> Dict[str, Any]:
|
|||||||
return _svc(name, DOWN, _detail_for(category), error=category)
|
return _svc(name, DOWN, _detail_for(category), error=category)
|
||||||
|
|
||||||
|
|
||||||
|
async def _source_unavailable(name: str, source: str,
|
||||||
|
category: str) -> Dict[str, Any]:
|
||||||
|
"""Controlled entry for a subsystem whose config source could not be read.
|
||||||
|
|
||||||
|
Deliberately not `disabled`: the configured state is *unknown*, not known to
|
||||||
|
be off. `degraded` rather than `down` because the failure is in reading the
|
||||||
|
inventory, not a probe that reached the subsystem and failed — but it still
|
||||||
|
carries non-zero severity, so a report containing one can never roll up to
|
||||||
|
`ok`. `meta` names the source and the controlled category of the underlying
|
||||||
|
exception; never its text.
|
||||||
|
"""
|
||||||
|
return _svc(name, DEGRADED,
|
||||||
|
f"Could not read configuration "
|
||||||
|
f"({_detail_for(CONFIG_SOURCE_ERROR)}).",
|
||||||
|
error=CONFIG_SOURCE_ERROR, source=source,
|
||||||
|
source_error=category)
|
||||||
|
|
||||||
|
|
||||||
async def collect_service_health(rag_manager: Any = None,
|
async def collect_service_health(rag_manager: Any = None,
|
||||||
memory_vector: Any = None) -> Dict[str, Any]:
|
memory_vector: Any = None) -> Dict[str, Any]:
|
||||||
"""Run every probe and return {overall, services, timestamp}.
|
"""Run every probe and return {overall, services, timestamp}.
|
||||||
@@ -472,6 +512,10 @@ async def collect_service_health(rag_manager: Any = None,
|
|||||||
four network subsystems run concurrently, each under `_SUBSYSTEM_DEADLINE`,
|
four network subsystems run concurrently, each under `_SUBSYSTEM_DEADLINE`,
|
||||||
with an overall `_AGGREGATE_DEADLINE` backstop. Per-item probes inside
|
with an overall `_AGGREGATE_DEADLINE` backstop. Per-item probes inside
|
||||||
providers/email are themselves bounded by `_FANOUT_BUDGET`.
|
providers/email are themselves bounded by `_FANOUT_BUDGET`.
|
||||||
|
|
||||||
|
A subsystem whose config/inventory source failed to load is reported via
|
||||||
|
`_source_unavailable()` instead of being probed with an empty inventory,
|
||||||
|
which every probe would otherwise report as `disabled`.
|
||||||
"""
|
"""
|
||||||
from datetime import datetime, timezone
|
from datetime import datetime, timezone
|
||||||
|
|
||||||
@@ -482,12 +526,23 @@ async def collect_service_health(rag_manager: Any = None,
|
|||||||
chroma = chromadb_health(rag_manager, memory_vector)
|
chroma = chromadb_health(rag_manager, memory_vector)
|
||||||
|
|
||||||
names = ["searxng", "ntfy", "email", "providers"]
|
names = ["searxng", "ntfy", "email", "providers"]
|
||||||
coros = [
|
probes = {
|
||||||
_run_subsystem("searxng", searxng_health, settings),
|
"searxng": (searxng_health, (settings,)),
|
||||||
_run_subsystem("ntfy", ntfy_health, inputs["integrations"], settings),
|
"ntfy": (ntfy_health, (inputs["integrations"], settings)),
|
||||||
_run_subsystem("email", email_health, inputs["accounts"]),
|
"email": (email_health, (inputs["accounts"],)),
|
||||||
_run_subsystem("providers", providers_health, inputs["endpoints"]),
|
"providers": (providers_health, (inputs["endpoints"],)),
|
||||||
]
|
}
|
||||||
|
# A subsystem whose inventory could not be read is reported as such. The
|
||||||
|
# others are still probed normally — one broken source degrades one entry.
|
||||||
|
failed = inputs.get("failed") or {}
|
||||||
|
coros = []
|
||||||
|
for name in names:
|
||||||
|
source = _SUBSYSTEM_SOURCE[name]
|
||||||
|
if source in failed:
|
||||||
|
coros.append(_source_unavailable(name, source, failed[source]))
|
||||||
|
else:
|
||||||
|
fn, args = probes[name]
|
||||||
|
coros.append(_run_subsystem(name, fn, *args))
|
||||||
try:
|
try:
|
||||||
results = await asyncio.wait_for(asyncio.gather(*coros),
|
results = await asyncio.wait_for(asyncio.gather(*coros),
|
||||||
timeout=_AGGREGATE_DEADLINE)
|
timeout=_AGGREGATE_DEADLINE)
|
||||||
|
|||||||
@@ -137,3 +137,170 @@ def test_collect_aggregate_deadline_yields_controlled_result(monkeypatch):
|
|||||||
net = [s for s in out["services"] if s["name"] != "chromadb"]
|
net = [s for s in out["services"] if s["name"] != "chromadb"]
|
||||||
assert all(s["status"] == sh.DOWN and s["meta"].get("error") == "timeout"
|
assert all(s["status"] == sh.DOWN and s["meta"].get("error") == "timeout"
|
||||||
for s in net)
|
for s in net)
|
||||||
|
|
||||||
|
|
||||||
|
# ── Discovery failure is not "disabled" (#6154) ──
|
||||||
|
#
|
||||||
|
# `_gather_inputs()` fails soft to an empty inventory, and every probe reads an
|
||||||
|
# empty inventory as `disabled`. Without a record of *why* it is empty, a broken
|
||||||
|
# integrations file / email store / endpoint DB is reported as an intentionally
|
||||||
|
# unconfigured subsystem — and, with the rest disabled, `overall: "ok"`.
|
||||||
|
|
||||||
|
def _boom(*_a, **_k):
|
||||||
|
raise RuntimeError("secret-bearing detail: postgres://u:p@db/x")
|
||||||
|
|
||||||
|
|
||||||
|
def test_gather_inputs_records_each_failed_source(monkeypatch):
|
||||||
|
import core.database
|
||||||
|
import routes.email_helpers
|
||||||
|
import src.integrations
|
||||||
|
import src.settings
|
||||||
|
|
||||||
|
monkeypatch.setattr(src.settings, "load_settings", lambda: {"a": 1})
|
||||||
|
monkeypatch.setattr(src.integrations, "load_integrations", _boom)
|
||||||
|
monkeypatch.setattr(routes.email_helpers, "_list_email_accounts", _boom)
|
||||||
|
monkeypatch.setattr(core.database, "SessionLocal", _boom)
|
||||||
|
|
||||||
|
out = sh._gather_inputs()
|
||||||
|
assert out["settings"] == {"a": 1}
|
||||||
|
assert set(out["failed"]) == {"integrations", "accounts", "endpoints"}
|
||||||
|
# Controlled categories only — never the exception text.
|
||||||
|
assert set(out["failed"].values()) <= set(sh._ERROR_DETAIL)
|
||||||
|
assert "postgres" not in repr(out)
|
||||||
|
|
||||||
|
|
||||||
|
def test_gather_inputs_reports_no_failures_when_sources_load(monkeypatch):
|
||||||
|
import core.database
|
||||||
|
import routes.email_helpers
|
||||||
|
import src.integrations
|
||||||
|
import src.settings
|
||||||
|
|
||||||
|
class _DB:
|
||||||
|
def query(self, *_a):
|
||||||
|
return self
|
||||||
|
|
||||||
|
def filter(self, *_a):
|
||||||
|
return self
|
||||||
|
|
||||||
|
def all(self):
|
||||||
|
return []
|
||||||
|
|
||||||
|
def close(self):
|
||||||
|
pass
|
||||||
|
|
||||||
|
monkeypatch.setattr(src.settings, "load_settings", lambda: {})
|
||||||
|
monkeypatch.setattr(src.integrations, "load_integrations", lambda: [])
|
||||||
|
monkeypatch.setattr(routes.email_helpers, "_list_email_accounts", lambda: [])
|
||||||
|
monkeypatch.setattr(core.database, "SessionLocal", lambda: _DB())
|
||||||
|
|
||||||
|
assert sh._gather_inputs()["failed"] == {}
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("source,service", [
|
||||||
|
("settings", "searxng"),
|
||||||
|
("integrations", "ntfy"),
|
||||||
|
("accounts", "email"),
|
||||||
|
("endpoints", "providers"),
|
||||||
|
])
|
||||||
|
def test_failed_source_is_not_reported_as_disabled(monkeypatch, source, service):
|
||||||
|
import asyncio
|
||||||
|
monkeypatch.setattr(sh, "_gather_inputs", lambda: {
|
||||||
|
"settings": {"search_provider": "disabled"},
|
||||||
|
"integrations": [], "accounts": [], "endpoints": [],
|
||||||
|
"failed": {source: "network_error"},
|
||||||
|
})
|
||||||
|
out = asyncio.run(sh.collect_service_health(None, None))
|
||||||
|
entry = next(s for s in out["services"] if s["name"] == service)
|
||||||
|
assert entry["status"] != sh.DISABLED
|
||||||
|
assert entry["meta"]["error"] == sh.CONFIG_SOURCE_ERROR
|
||||||
|
assert entry["meta"]["source"] == source
|
||||||
|
assert entry["meta"]["source_error"] == "network_error"
|
||||||
|
# A failed source must reach the aggregate verdict.
|
||||||
|
assert out["overall"] != sh.OK
|
||||||
|
assert set(out) == {"overall", "services", "timestamp"}
|
||||||
|
|
||||||
|
|
||||||
|
def test_all_sources_failing_cannot_report_overall_ok(monkeypatch):
|
||||||
|
# The exact shape from the report: settings loads and says search is off,
|
||||||
|
# the other three sources raise. Before the fix this was overall "ok" with
|
||||||
|
# five "disabled" entries.
|
||||||
|
import asyncio
|
||||||
|
monkeypatch.setattr(sh, "_gather_inputs", lambda: {
|
||||||
|
"settings": {"search_provider": "disabled"},
|
||||||
|
"integrations": [], "accounts": [], "endpoints": [],
|
||||||
|
"failed": {"integrations": "error", "accounts": "error",
|
||||||
|
"endpoints": "error"},
|
||||||
|
})
|
||||||
|
out = asyncio.run(sh.collect_service_health(None, None))
|
||||||
|
by_name = {s["name"]: s for s in out["services"]}
|
||||||
|
assert out["overall"] == sh.DEGRADED
|
||||||
|
# Loaded-and-genuinely-empty still reads as disabled.
|
||||||
|
assert by_name["searxng"]["status"] == sh.DISABLED
|
||||||
|
for name in ("ntfy", "email", "providers"):
|
||||||
|
assert by_name[name]["status"] == sh.DEGRADED
|
||||||
|
|
||||||
|
|
||||||
|
def test_empty_inventory_without_failure_stays_disabled(monkeypatch):
|
||||||
|
import asyncio
|
||||||
|
monkeypatch.setattr(sh, "_gather_inputs", lambda: {
|
||||||
|
"settings": {"search_provider": "disabled"},
|
||||||
|
"integrations": [], "accounts": [], "endpoints": [], "failed": {},
|
||||||
|
})
|
||||||
|
out = asyncio.run(sh.collect_service_health(_Store(True), _Store(True)))
|
||||||
|
assert out["overall"] == sh.OK
|
||||||
|
assert all(s["status"] == sh.DISABLED
|
||||||
|
for s in out["services"] if s["name"] != "chromadb")
|
||||||
|
|
||||||
|
|
||||||
|
def test_one_failed_source_still_probes_the_others(monkeypatch):
|
||||||
|
# A broken endpoint DB must not stop ntfy/email from being probed.
|
||||||
|
import asyncio
|
||||||
|
probed = []
|
||||||
|
|
||||||
|
def _probe(name):
|
||||||
|
def _fn(*_a, **_k):
|
||||||
|
probed.append(name)
|
||||||
|
return sh._svc(name, sh.OK, "probed")
|
||||||
|
return _fn
|
||||||
|
|
||||||
|
monkeypatch.setattr(sh, "_gather_inputs", lambda: {
|
||||||
|
"settings": {}, "integrations": [], "accounts": [], "endpoints": [],
|
||||||
|
"failed": {"endpoints": "timeout"},
|
||||||
|
})
|
||||||
|
monkeypatch.setattr(sh, "searxng_health", _probe("searxng"))
|
||||||
|
monkeypatch.setattr(sh, "ntfy_health", _probe("ntfy"))
|
||||||
|
monkeypatch.setattr(sh, "email_health", _probe("email"))
|
||||||
|
monkeypatch.setattr(sh, "providers_health", _probe("providers"))
|
||||||
|
|
||||||
|
out = asyncio.run(sh.collect_service_health(None, None))
|
||||||
|
assert sorted(probed) == ["email", "ntfy", "searxng"]
|
||||||
|
assert "providers" not in probed
|
||||||
|
by_name = {s["name"]: s for s in out["services"]}
|
||||||
|
assert by_name["providers"]["status"] == sh.DEGRADED
|
||||||
|
assert by_name["ntfy"]["status"] == sh.OK
|
||||||
|
|
||||||
|
|
||||||
|
def test_source_failure_entry_carries_no_raw_detail(monkeypatch):
|
||||||
|
import asyncio
|
||||||
|
monkeypatch.setattr(sh, "_gather_inputs", lambda: {
|
||||||
|
"settings": {}, "integrations": [], "accounts": [], "endpoints": [],
|
||||||
|
"failed": {"accounts": "auth_or_protocol_error"},
|
||||||
|
})
|
||||||
|
out = asyncio.run(sh.collect_service_health(None, None))
|
||||||
|
entry = next(s for s in out["services"] if s["name"] == "email")
|
||||||
|
assert entry["detail"] == (
|
||||||
|
"Could not read configuration (configuration source could not be read).")
|
||||||
|
assert set(entry["meta"]) == {"error", "source", "source_error"}
|
||||||
|
|
||||||
|
|
||||||
|
def test_collect_tolerates_inputs_without_a_failed_key(monkeypatch):
|
||||||
|
# `_gather_inputs()` is monkeypatched in a number of existing tests (and by
|
||||||
|
# anything calling the collector with a stub); a payload predating `failed`
|
||||||
|
# must still probe normally rather than raise.
|
||||||
|
import asyncio
|
||||||
|
monkeypatch.setattr(sh, "_gather_inputs", lambda: {
|
||||||
|
"settings": {"search_provider": "disabled"},
|
||||||
|
"integrations": [], "accounts": [], "endpoints": [],
|
||||||
|
})
|
||||||
|
out = asyncio.run(sh.collect_service_health(_Store(True), _Store(True)))
|
||||||
|
assert out["overall"] == sh.OK
|
||||||
|
|||||||
Reference in New Issue
Block a user