HEL-543 返工: 补充 observability 运行时可关闭开关 (DATAHUB_OBSERVABILITY)

总工复核 🔴：安全边界要求新增观测功能必须可关闭、关闭后现有功能完全照旧，
但此前实现没有任何运行时开关。修复：

- 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 <cursoragent@cursor.com>
Co-authored-by: multica-agent <github@multica.ai>

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")

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")

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()

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"},
     )

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
 

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()