feat(core): 统一可观测性记录与后端查询

.changes/observability-unification.added.md

@@ -0,0 +1 @@
+统一可观测性记录与查询：默认写入内置信息库，可选择 Loki 双向后端；提供精确 Peer 查询/摄取能力、Job 生命周期和操作摘要，并明确外部故障、历史范围与资源关闭责任。

.env.example

@@ -1,6 +1,6 @@
-# Server Configuration
-HOST=0.0.0.0
-PORT=8000
+# Server Configuration
+HOST=0.0.0.0
+PORT=8000
 PEER_ID=00000000-0000-4000-8000-000000000002
 PEER_NAME=core-py-development
 
@@ -22,20 +22,18 @@ INKCRE_DATABASE_ENVIRONMENT=development
 # Set `true` if your db will scale to 0 when idle
 # This will disable listen to pgsql notifications and enable pool_pre_ping
 DATABASE_SCALE_0=false
-
-# Logging Configuration (Optional)
-# postgresql, logtail
-OBSRV__LOGGING_BACKEND=postgresql 
-# level lower than this will not be sent to backend
-OBSRV__LOGGING_BACKEND_LEVEL=20 
-
-# Logtail As Logging Backend Configuration
-OBSRV__LOGTAIL_SOURCE_TOKEN=your_token_here
-OBSRV__LOGTAIL_HOST=https://in.logtail.com
-
-# PostgreSQL Configuration (Docker Deployment)
-POSTGRES_DB=inkcre
-POSTGRES_USER=postgres
+
+# Ordinary log threshold; backend selection lives in deployment config
+# inkcre.observability.backend / inkcre.observability.backend.v1
+OBSRV__LOGGING_BACKEND_LEVEL=20
+# Private Loki configuration, only used when the deployment selects Loki.
+# OBSRV__LOKI_URL=https://your-loki-host
+# OBSRV__LOKI_HEADERS='{"Authorization":"Basic YOUR_ENCODED_VALUE"}'
+OBSRV__EXTERNAL_CONTENT_ENABLED=false
+
+# PostgreSQL Configuration (Docker Deployment)
+POSTGRES_DB=inkcre
+POSTGRES_USER=postgres
 POSTGRES_PASSWORD=postgres
 POSTGREST_PORT=3000
 CORE_PORT=8000

app/observability.py

@@ -2,6 +2,12 @@
 
 import asyncio
 
+import pydantic
+
+from inkcre_core.context import current_core
+from inkcre_core.libs.obsrv.backends import LokiBackend, PostgreSQLBackend
+from inkcre_core.libs.obsrv.runtime import ObservationRuntime
+
 from opentelemetry.trace import SpanKind
 from starlette.types import ASGIApp, Message, Receive, Scope, Send
 
@@ -10,7 +16,14 @@
   DeploymentConfigService,
 )
 from app.settings import settings
-from inkcre_core.schemas.observability import CONFIG_KEY, CONFIG_SCHEMA, ObservabilityConfig
+from inkcre_core.schemas.observability import (
+  CONFIG_KEY,
+  CONFIG_SCHEMA,
+  ObservabilityConfig,
+  BACKEND_CONFIG_KEY,
+  BACKEND_CONFIG_SCHEMA,
+  ObservabilityBackendConfig,
+)
 from app.version import APPLICATION_VERSION
 from inkcre_core.libs.obsrv.main import get_logger
 from inkcre_core.libs.obsrv.telemetry import is_enabled, operation, start_telemetry
@@ -29,24 +42,67 @@
 )
 
 
+DeploymentConfigManager.register_schema(
+  BACKEND_CONFIG_SCHEMA, ObservabilityBackendConfig, keys=(BACKEND_CONFIG_KEY,)
+)
+
+
 async def initialize_telemetry() -> None:
-  """Read once after database admission; telemetry cannot fail business readiness."""
+  """Bind the selected recording/query backend; OTLP is an independent extra exit."""
+  core = current_core()
+  backend = None
+  selected = "postgresql"
+  error_message = None
+  config = None
+  try:
+    async with asyncio.timeout(5):
+      selection = await DeploymentConfigService.get(BACKEND_CONFIG_KEY)
+      if selection is not None and not isinstance(selection, ObservabilityBackendConfig):
+        raise ValueError("unexpected observability backend schema")
+      selected = selection.backend if selection is not None else "postgresql"
+      if selected == "loki" or settings.obsrv.telemetry_enabled:
+        config = await DeploymentConfigService.get(CONFIG_KEY)
+    if selected == "postgresql":
+      backend = PostgreSQLBackend(core.sessions)
+    else:
+      if not isinstance(config, ObservabilityConfig) or settings.obsrv.loki_url is None:
+        raise ValueError("external backend needs deployment identity and private URL")
+      url = settings.obsrv.loki_url
+      if url.username or url.password or url.query or url.fragment:
+        raise ValueError("private Loki URL cannot contain credentials/query/fragment")
+      headers = pydantic.TypeAdapter(dict[str, str]).validate_json(
+        settings.obsrv.loki_headers.get_secret_value()
+      )
+      backend = LokiBackend(
+        url=str(url), deployment_id=str(config.deployment_id), headers=headers
+      )
+  except Exception:
+    # Missing selection is DB. A failed/invalid selection read is not permission
+    # to silently choose DB or redirect writes from an intended external backend.
+    error_message = "Observability configuration unavailable or incomplete"
+  if core.observability is not None:
+    await core.observability.aclose()
+  core.observability = ObservationRuntime(
+    backend,
+    selected=selected,
+    peer_id=str(core.options.peer_id),
+    error=error_message,
+    allow_external_content=settings.obsrv.external_content_enabled,
+  )
+  core.observability.start()
   if not settings.obsrv.telemetry_enabled:
     return
-  attributes: dict[str, str] = {"inkcre.peer.id": str(settings.peer_id)}
-  endpoints: dict[str, str] = {}
-  try:
-    async with asyncio.timeout(2):
-      config = await DeploymentConfigService.get(CONFIG_KEY)
-    if not isinstance(config, ObservabilityConfig):
-      raise ValueError("missing or unexpected observability schema")
-    attributes["inkcre.deployment.id"] = str(config.deployment_id)
-    if config.otlp_http_endpoints is not None:
-      endpoints = config.otlp_http_endpoints.model_dump(mode="json", exclude_none=True)
-  except Exception as error:
-    # Configuration may contain credentials despite the admitted schema. Never render it.
-    get_logger().warning("Telemetry configuration unavailable (%s)", type(error).__name__)
+  if not isinstance(config, ObservabilityConfig):
+    get_logger().warning("Telemetry configuration unavailable")
     return
+  attributes: dict[str, str] = {
+    "inkcre.peer.id": str(settings.peer_id),
+    "inkcre.deployment.id": str(config.deployment_id),
+  }
+  endpoints: dict[str, str] = {}
+  if config.otlp_http_endpoints is not None:
+    endpoints = config.otlp_http_endpoints.model_dump(mode="json", exclude_none=True)
+
   try:
     start_telemetry(
       enabled=True,

app/routes/observability.py

@@ -0,0 +1,60 @@
+"""Authenticated, bounded observability capability; no caller-selected destination."""
+
+import asyncio
+
+import fastapi
+from pydantic import ValidationError
+
+from inkcre_core.business.peer import PeerHTTPInbound
+from inkcre_core.libs.obsrv.backends import ObservabilityUnavailable
+from inkcre_core.libs.obsrv.records import ObservationRequest
+from inkcre_core.libs.obsrv.runtime import current_observability
+
+ROUTER = fastapi.APIRouter(tags=["observability"])
+PEER_INBOUND = PeerHTTPInbound(
+  capability="core.observability.v1", method="POST", path="/observability"
+)
+_MAX_BYTES = 256 * 1024
+
+
+@ROUTER.post("/observability")
+async def observability(request: fastapi.Request):
+  runtime = current_observability()
+  if runtime is None:
+    raise fastapi.HTTPException(503, "Observability is not initialized")
+  if runtime.requests_in_flight >= 4:
+    raise fastapi.HTTPException(429, "Observability request capacity reached")
+  runtime.requests_in_flight += 1
+  try:
+    async with asyncio.timeout(15):
+      payload = bytearray()
+      async for chunk in request.stream():
+        payload.extend(chunk)
+        if len(payload) > _MAX_BYTES:
+          raise fastapi.HTTPException(413, "Observability request exceeds 256 KiB")
+      try:
+        body = ObservationRequest.model_validate_json(payload)
+      except ValidationError as error:
+        # Validation details can contain the original body. Do not reflect log contents.
+        raise fastapi.HTTPException(422, "Invalid observability request") from error
+      if body.expected_backend is not None and body.expected_backend != runtime.selected:
+        raise fastapi.HTTPException(409, "Observability backend selection changed")
+      if body.action == "status":
+        return runtime.status()
+      if body.action == "query":
+        assert body.query is not None
+        try:
+          return await runtime.query(body.query)
+        except ObservabilityUnavailable as error:
+          raise fastapi.HTTPException(
+            503, "Selected observability backend is unavailable"
+          ) from error
+      assert body.records is not None
+      if not runtime.status().recording:
+        raise fastapi.HTTPException(503, "Selected observability backend is not recording")
+      accepted = sum(runtime.emit(record) for record in body.records)
+      return {"accepted": accepted, "dropped": len(body.records) - accepted}
+  except TimeoutError as error:
+    raise fastapi.HTTPException(408, "Observability request timed out") from error
+  finally:
+    runtime.requests_in_flight -= 1