From abd4d22a67f702f634fc766d54fdcc3fae8003ae Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=80=BB=E5=B7=A5?= Date: Mon, 14 Sep 2026 01:10:58 +0800 Subject: [PATCH] =?UTF-8?q?HEL-543=20=E8=BF=94=E5=B7=A5:=20=E8=A1=A5?= =?UTF-8?q?=E5=85=85=20observability=20=E8=BF=90=E8=A1=8C=E6=97=B6?= =?UTF-8?q?=E5=8F=AF=E5=85=B3=E9=97=AD=E5=BC=80=E5=85=B3=20(DATAHUB=5FOBSE?= =?UTF-8?q?RVABILITY)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 总工复核 🔴：安全边界要求新增观测功能必须可关闭、关闭后现有功能完全照旧， 但此前实现没有任何运行时开关。修复： - settings.py: 新增 Settings.observability_enabled 字段，沿用既有 DATAHUB_SCHEDULER 的环境变量模式，读取 DATAHUB_OBSERVABILITY (0/false/off 关闭，默认开启)。 - hub.py: Hub.__init__ 把 settings.observability_enabled 挂到 self.db 上，让 pipeline/realtime_serve/steward/admin_api 已经 在传的 db 参数直接带上开关，零额外改造。 - observability.py: 新增 is_enabled(db)，缺失该属性时默认按启用处理 (向后兼容裸 HubDB 用例/测试)。关闭时 observe() 变成纯 透传（不计时、不分类、不碰数据库)，record_call() 直接 no-op。 - admin_api.py: 4 个新只读端点关闭时返回明确的 {"enabled": false, ...空结构} 而不是静默返回旧数据。 - 新增 10 个测试：开关默认值/环境变量解析、关闭后 observe() 的透传语义 (含异常原样重新抛出)、关闭后 record_call() 零写入、关闭后重新开启恢复 记录、4 个 admin 端点在关闭态的响应结构。 全量测试 183/183 通过（新增 10 个，含此前 173 个零回归）。 Co-authored-by: Cursor Co-authored-by: multica-agent --- xiaobai-datahub/datahub/admin_api.py | 22 ++++- xiaobai-datahub/datahub/hub.py | 6 ++ xiaobai-datahub/datahub/observability.py | 23 ++++- xiaobai-datahub/datahub/settings.py | 6 ++ .../tests/test_admin_observability_api.py | 35 +++++++ xiaobai-datahub/tests/test_observability.py | 92 +++++++++++++++++++ 6 files changed, 178 insertions(+), 6 deletions(-) diff --git a/xiaobai-datahub/datahub/admin_api.py b/xiaobai-datahub/datahub/admin_api.py index 01a488c..78dfa7e 100644 --- a/xiaobai-datahub/datahub/admin_api.py +++ b/xiaobai-datahub/datahub/admin_api.py @@ -7,6 +7,7 @@ from datahub.adapters import RESERVED from datahub.auth import AuthService from datahub.db import HubDB from datahub import lineage as lineage_module +from datahub import observability from datahub.pipeline import OFFICIAL_DATASETS, STOCKS_DATASET, Pipeline from datahub.scheduler import Scheduler from datahub.serving import ApiError @@ -114,6 +115,8 @@ class AdminAPI: # static registries in source_catalog.py / lineage.py. # ------------------------------------------------------------------ def providers_status(self, provider: str = "", limit: int = 50) -> dict[str, Any]: + if not observability.is_enabled(self.db): + return {"enabled": False, "health": [], "recent_calls": []} limit = max(1, min(int(limit or 50), 200)) health_sql = "SELECT * FROM provider_health" params: tuple[Any, ...] = () @@ -127,19 +130,30 @@ class AdminAPI: calls_sql += " WHERE provider = ?" calls_sql += " ORDER BY id DESC LIMIT ?" recent = self.db.fetchall(calls_sql, (*params, limit)) - return {"health": health, "recent_calls": recent} + return {"enabled": True, "health": health, "recent_calls": recent} def source_catalog(self) -> dict[str, Any]: - return {"items": source_catalog.snapshot(self.db, self.auth)} + if not observability.is_enabled(self.db): + return {"enabled": False, "items": []} + return {"enabled": True, "items": source_catalog.snapshot(self.db, self.auth)} def lineage(self, trade_date: str = "") -> dict[str, Any]: day = yyyymmdd(trade_date) if trade_date else yyyymmdd(now_shanghai()) - return {"trade_date": day, "items": lineage_module.snapshot(self.db, day)} + if not observability.is_enabled(self.db): + return {"enabled": False, "trade_date": day, "items": []} + return {"enabled": True, "trade_date": day, "items": lineage_module.snapshot(self.db, day)} def lineage_affected(self, provider: str = "", interface: str = "") -> dict[str, Any]: if not provider: raise ApiError("INVALID_ARGUMENT", "provider is required") - return {"provider": provider, "interface": interface, "items": lineage_module.affected(self.db, provider, interface)} + if not observability.is_enabled(self.db): + return {"enabled": False, "provider": provider, "interface": interface, "items": []} + return { + "enabled": True, + "provider": provider, + "interface": interface, + "items": lineage_module.affected(self.db, provider, interface), + } def jobs(self) -> dict[str, Any]: runs = self.db.fetchall("SELECT * FROM job_runs ORDER BY id DESC LIMIT 100") diff --git a/xiaobai-datahub/datahub/hub.py b/xiaobai-datahub/datahub/hub.py index ba0d924..1f58370 100644 --- a/xiaobai-datahub/datahub/hub.py +++ b/xiaobai-datahub/datahub/hub.py @@ -24,6 +24,12 @@ class Hub: raise SystemExit("DATAHUB_ENCRYPTION_KEY 未配置") self.settings = settings self.db = HubDB(settings.db_path) + # HEL-543: carry the observability kill switch on the db handle so + # every call site that already threads `db` through (pipeline, + # realtime_serve, steward, admin_api) picks it up for free with no + # extra plumbing. Missing this attribute (e.g. a bare HubDB built + # directly in tests) defaults to enabled — see observability.is_enabled. + self.db.observability_enabled = settings.observability_enabled self.vault = SecretVault(settings.encryption_key) self.auth = AuthService(self.db, self.vault, settings.api_token, settings.admin_password) token = settings.tushare_token or self.auth.load_credential("tushare_token") diff --git a/xiaobai-datahub/datahub/observability.py b/xiaobai-datahub/datahub/observability.py index 395cdae..72bfd9a 100644 --- a/xiaobai-datahub/datahub/observability.py +++ b/xiaobai-datahub/datahub/observability.py @@ -81,6 +81,21 @@ def _logger(): return get_logger() +def is_enabled(db: Any) -> bool: + """Runtime kill switch (``Settings.observability_enabled`` / + ``DATAHUB_OBSERVABILITY``, wired onto the db handle in ``Hub.__init__``). + + Defaults to enabled when the attribute is absent — e.g. a bare ``HubDB`` + built directly in a test, or any call site that predates HEL-543 — so + this can never accidentally disable an existing deployment. Never + raises. + """ + try: + return bool(getattr(db, "observability_enabled", True)) + except Exception: # pragma: no cover - defensive, must never raise + return True + + def classify_error(message: str) -> tuple[str, str]: """Best-effort, side-reading classification of an exception message. @@ -171,7 +186,7 @@ def record_call( ) -> None: """Fail-open recorder. Never raises; a write failure here must never be able to take down a real, otherwise-successful data path.""" - if db is None: + if db is None or not is_enabled(db): return try: now = isoformat(now_shanghai()) @@ -282,8 +297,12 @@ def observe( Returns exactly what ``fn()`` returns and re-raises exactly what ``fn()`` raises. ``db`` may be ``None`` (e.g. in call sites that are not wired to a database yet); in that case this is a transparent passthrough - with no recording at all. + with no recording at all. Same when the ``DATAHUB_OBSERVABILITY`` kill + switch is off (see ``is_enabled``): this becomes ``return fn()`` with no + timing, no classification, and no db access whatsoever. """ + if db is not None and not is_enabled(db): + return fn() started = time.perf_counter() try: result = fn() diff --git a/xiaobai-datahub/datahub/settings.py b/xiaobai-datahub/datahub/settings.py index 3aa86b0..ff96ddc 100644 --- a/xiaobai-datahub/datahub/settings.py +++ b/xiaobai-datahub/datahub/settings.py @@ -33,6 +33,11 @@ class Settings: quality: dict[str, Any] = field(default_factory=dict) log_level: str = "INFO" scheduler_enabled: bool = True + # HEL-543 kill switch: off disables the provider_call_log/provider_health + # side channel entirely (observe()/record_call() become no-ops and the + # new read-only admin endpoints report {"enabled": false}). Default on; + # existing routing/fetch/publish behavior is identical either way. + observability_enabled: bool = True @property def tushare_rate_per_minute(self) -> int: @@ -126,4 +131,5 @@ def load_settings( quality=_load_quality(quality_path), log_level=environ.get("DATAHUB_LOG_LEVEL") or "INFO", scheduler_enabled=str(environ.get("DATAHUB_SCHEDULER") or "1") not in {"0", "false", "False"}, + observability_enabled=str(environ.get("DATAHUB_OBSERVABILITY") or "1") not in {"0", "false", "False", "off", "OFF"}, ) diff --git a/xiaobai-datahub/tests/test_admin_observability_api.py b/xiaobai-datahub/tests/test_admin_observability_api.py index cf7f85a..a9723d1 100644 --- a/xiaobai-datahub/tests/test_admin_observability_api.py +++ b/xiaobai-datahub/tests/test_admin_observability_api.py @@ -124,6 +124,41 @@ class AdminObservabilityApiTests(unittest.TestCase): self.assertEqual(body["provider"], "tushare") self.assertEqual(body["interface"], "daily") + def test_disabled_kill_switch_reports_enabled_false_with_empty_structure(self) -> None: + # HEL-543 total-review 🔴: flip the runtime kill switch the same way + # Hub.__init__ wires Settings.observability_enabled onto the db + # handle, then confirm every new endpoint reports disabled with an + # explicit empty structure rather than silently going quiet. + self.hub.db.observability_enabled = False + try: + pipeline = self.hub.pipeline + pipeline.ingest_reference(TRADE_DATE) + pipeline.run_dataset("daily", TRADE_DATE) # must still fully succeed + + status, body, _ = self._get("/admin/api/providers/status") + self.assertEqual(status, 200) + self.assertEqual(body, {"enabled": False, "health": [], "recent_calls": []}) + + status, body, _ = self._get("/admin/api/source-catalog") + self.assertEqual(status, 200) + self.assertEqual(body, {"enabled": False, "items": []}) + + status, body, _ = self._get("/admin/api/lineage") + self.assertEqual(status, 200) + self.assertFalse(body["enabled"]) + self.assertEqual(body["items"], []) + + status, body, _ = self._get("/admin/api/lineage/affected?provider=tushare") + self.assertEqual(status, 200) + self.assertEqual( + body, {"enabled": False, "provider": "tushare", "interface": "", "items": []} + ) + + # Nothing was ever written while disabled. + self.assertEqual(self.hub.db.fetchall("SELECT * FROM provider_call_log"), []) + finally: + self.hub.db.observability_enabled = True + def test_must_change_password_blocks_new_endpoints_too(self) -> None: from urllib.error import HTTPError diff --git a/xiaobai-datahub/tests/test_observability.py b/xiaobai-datahub/tests/test_observability.py index 301b74f..37accd8 100644 --- a/xiaobai-datahub/tests/test_observability.py +++ b/xiaobai-datahub/tests/test_observability.py @@ -210,5 +210,97 @@ class ObserveTests(unittest.TestCase): self.assertIs(result, sentinel) +class _ToggleDB(HubDB): + """A real HubDB subclass so we can flip the HEL-543 kill switch the same + way Hub.__init__ does, without needing a full Hub/Settings wiring.""" + + +class KillSwitchTests(unittest.TestCase): + """HEL-543 total-review 🔴: the observability side channel must be + disable-able at runtime, and disabling it must leave existing behavior + completely unchanged (pure passthrough, zero db access).""" + + def setUp(self) -> None: + self.tmp = tempfile.TemporaryDirectory() + self.db = _ToggleDB(Path(self.tmp.name) / "hub.db") + + def tearDown(self) -> None: + self.tmp.cleanup() + + def test_is_enabled_defaults_true_when_attribute_absent(self): + # A bare HubDB (as used throughout the rest of this test suite, and + # by any pre-HEL-543 call site) must default to enabled. + self.assertTrue(observability.is_enabled(self.db)) + self.assertTrue(observability.is_enabled(None)) + + def test_disabled_record_call_writes_nothing(self): + self.db.observability_enabled = False + observability.record_call(self.db, "eastmoney", "indices", status="ok", latency_ms=1) + self.assertEqual(self.db.fetchall("SELECT * FROM provider_call_log"), []) + self.assertEqual(self.db.fetchall("SELECT * FROM provider_health"), []) + + def test_disabled_observe_is_a_pure_passthrough_on_success(self): + self.db.observability_enabled = False + sentinel = {"ts_code": "600000.SH"} + calls = {"n": 0} + + def fn(): + calls["n"] += 1 + return sentinel + + result = observability.observe(self.db, "eastmoney", "indices", fn) + self.assertIs(result, sentinel) + self.assertEqual(calls["n"], 1) + self.assertEqual(self.db.fetchall("SELECT * FROM provider_call_log"), []) + + def test_disabled_observe_still_reraises_the_exact_exception(self): + self.db.observability_enabled = False + boom = RuntimeError("upstream exploded") + + def fn(): + raise boom + + with self.assertRaises(RuntimeError) as ctx: + observability.observe(self.db, "eastmoney", "indices", fn) + self.assertIs(ctx.exception, boom) + self.assertEqual(self.db.fetchall("SELECT * FROM provider_call_log"), []) + + def test_re_enabling_resumes_recording(self): + self.db.observability_enabled = False + observability.observe(self.db, "eastmoney", "indices", lambda: {"ok": True}) + self.assertEqual(self.db.fetchall("SELECT * FROM provider_call_log"), []) + self.db.observability_enabled = True + observability.observe(self.db, "eastmoney", "indices", lambda: {"ok": True}) + self.assertEqual(len(self.db.fetchall("SELECT * FROM provider_call_log")), 1) + + +class SettingsToggleTests(unittest.TestCase): + """The kill switch follows the same env-var pattern as DATAHUB_SCHEDULER.""" + + def test_defaults_to_enabled(self): + from datahub.settings import load_settings + + settings = load_settings(env={}) + self.assertTrue(settings.observability_enabled) + + def test_datahub_observability_zero_disables(self): + from datahub.settings import load_settings + + settings = load_settings(env={"DATAHUB_OBSERVABILITY": "0"}) + self.assertFalse(settings.observability_enabled) + + def test_datahub_observability_off_disables(self): + from datahub.settings import load_settings + + settings = load_settings(env={"DATAHUB_OBSERVABILITY": "off"}) + self.assertFalse(settings.observability_enabled) + + def test_datahub_observability_one_keeps_enabled(self): + from datahub.settings import load_settings + + settings = load_settings(env={"DATAHUB_OBSERVABILITY": "1"}) + self.assertTrue(settings.observability_enabled) + + if __name__ == "__main__": unittest.main()