feat(observability): Datadog observability — metrics, readiness, log export, dashboards and alerts (CHOO-2807)

.env.example

@@ -91,6 +91,39 @@ DB_NAME=switch
 # every unattributed line look like that tenant's own.
 # TENANT_ID=default
 
+# ── Observability ────────────────────────────────────────────────────────────
+# Where this deployment reports metrics and logs. Unset — the default — means
+# it reports nowhere, and nothing measured leaves the process. Readiness
+# checking runs either way: /health/ready answers whether or not this is set.
+#
+# A base URL with no path. `/v1/metrics` and `/v1/logs` are appended, the same
+# convention OTEL_EXPORTER_OTLP_ENDPOINT follows, so pasting a full signal URL
+# here is refused at startup rather than posting to /v1/logs/v1/metrics.
+# OTLP_ENDPOINT=https://telemetry.example.com
+#
+# A UUID naming this deployment, stable across restarts. Required whenever the
+# endpoint is set: the collector drops payloads it cannot attribute, silently
+# and with a 200, so without one the server would report nothing while looking
+# correctly configured — which is why it refuses to start instead. Generate it
+# once with `uuidgen`; a fresh one per restart makes a single deployment look
+# like an endless population of new installs.
+# DEPLOYMENT_ID=
+#
+# Per signal. Metrics are the point and are on as soon as an endpoint is named.
+# Logs are off because they already reach this container's output, where a
+# cluster's log agent can read them — turning this on sends a second copy over
+# the network. Traces are off because the relay Switch reports to does not
+# serve /v1/traces yet, so enabling them means every export failing.
+# OTLP_METRICS_ENABLED=true
+# OTLP_LOGS_ENABLED=false
+# OTLP_TRACES_ENABLED=false
+#
+# `key=value` pairs, comma-separated, on every request. Needed for a collector
+# that authenticates; the relay does not.
+# OTLP_HEADERS=
+# OTLP_TIMEOUT_SECONDS=10
+# OTLP_EXPORT_INTERVAL_SECONDS=60
+
 # ── Client identity ──────────────────────────────────────────────────────────
 # The server half of every client's `@localpart:server` id. Nothing is
 # contacted at it; the ids are stable public handles.

CLAUDE.md

@@ -238,6 +238,12 @@ Tests live in `core/tests/switch_core/` mirroring the module structure. Uses pyt
 - `docs/old/LOCAL_DEVELOPMENT.md` — running Switch locally for development:
   `just` recipes, which port serves what, connecting Switch Console to a
   local server
+- `docs/old/observability.md` — what switch-core reports about itself and how
+  to turn it on: the metric catalogue and why attributes are declared, the
+  split between the liveness and readiness routes and why only the database
+  gates readiness, what the process reports in place of an infrastructure
+  agent, and what tracing still needs. Dashboards and alerts live in
+  `deploy/observability/`.
 - `docs/old/multi-tenancy.md` — why Switch is multi-tenant the way it is: the
   tenant model, sign-in and onboarding, one official messaging app per
   platform, and the phased plan the work follows. Phases 0 and 1 are built;

core/switch_core/bridges/agent/app.py

@@ -30,6 +30,7 @@
 from switch_core.db.stores.external_user_store import ExternalUserStore
 from switch_core.db.stores.room_store import RoomStore
 from switch_core.db.stores.task_store import TaskStore
+from switch_core.observability.http import MetricsMiddleware
 from switch_core.request_context import RequestContextMiddleware
 from switch_core.room_service import RoomService
 
@@ -160,6 +161,10 @@ async def log_validation_errors(
         api_key_cache=api_key_cache,
         session_factory=session_factory,  # type: ignore[arg-type]
     )
+    # Outside the bearer middleware, so a request rejected for bad credentials
+    # is still counted and timed — an authentication failure is traffic, and a
+    # spike of it is the thing you most want a dashboard to show.
+    app.add_middleware(MetricsMiddleware)
     # Added last, so it wraps the bearer middleware: a request rejected for bad
     # credentials is logged with a request id like any other.
     app.add_middleware(RequestContextMiddleware)

core/switch_core/bridges/collaboration/bridge_core.py

@@ -4,7 +4,8 @@
 import logging
 import re
 import uuid
-from collections.abc import Awaitable, Callable, Mapping
+from collections.abc import Awaitable, Callable, Iterator, Mapping
+from contextlib import contextmanager
 from dataclasses import dataclass, field
 from typing import TYPE_CHECKING, Any, TypeVar
 
@@ -44,6 +45,12 @@
 from switch_core.db.tenant_lookup import tenant_of_room
 from switch_core.events import AgentRuntimeStateEvent
 from switch_core.logging_context import log_context
+from switch_core.observability.catalogue import (
+    BRIDGE_ERRORS,
+    BRIDGE_EVENTS_IN,
+    BRIDGE_EVENTS_OUT,
+)
+from switch_core.observability.metrics import metrics
 from switch_core.provisioning import Provisioning
 from switch_core.room_service import RoomCreateConfig
 from switch_core.tenant_context import no_tenant, tenant_scope
@@ -228,11 +235,31 @@ def adapter(self) -> CollaborationAdapter:
     def tenant_id(self) -> str:
         return self._bridge_tenant_id
 
+    @contextmanager
+    def _counted_outbound(self, kind: str) -> Iterator[None]:
+        """Count one relay out to the platform, and its failure if it fails.
+
+        Placed around the relay call rather than at the top of the handler: a
+        handler returns early for a puppet's own echo and for a room with no
+        channel mapping, and neither of those is a message anybody sent
+        outwards.
+        """
+        metrics().increment(
+            BRIDGE_EVENTS_OUT, {"platform": self._bridge_type, "kind": kind}
+        )
+        try:
+            yield
+        except Exception:
+            metrics().increment(
+                BRIDGE_ERRORS, {"platform": self._bridge_type, "direction": "outbound"}
+            )
+            raise
+
     def _traced(
-        self, handler: Callable[[_InboundEventT], Awaitable[None]]
+        self, kind: str, handler: Callable[[_InboundEventT], Awaitable[None]]
     ) -> Callable[[_InboundEventT], Awaitable[None]]:
-        """Give each inbound platform event its own id in the logs, and bind
-        the tenant the event belongs to.
+        """Give each inbound platform event its own id in the logs, count it,
+        and bind the tenant the event belongs to.
 
         An event fans out across room lookup, identity provisioning and the
         transport, so without this the lines from two events arriving at once
@@ -260,14 +287,28 @@ def _traced(
         async def traced(event: _InboundEventT) -> None:
             event_id = uuid.uuid4().hex[:16]
             with log_context(request_id=f"{self._bridge_type}-{event_id}"):
+                metrics().increment(
+                    BRIDGE_EVENTS_IN, {"platform": self._bridge_type, "event": kind}
+                )
                 room_ids = self._channel_to_room.get(event.channel_id)
                 tenant_id = (
                     self._bridge_tenant_id
                     if room_ids is None
                     else await self._room_tenant(room_ids[0])
                 )
                 with tenant_scope(tenant_id):
-                    await handler(event)
+                    try:
+                        await handler(event)
+                    except Exception:
+                        # Counted here and re-raised unchanged: whoever handles
+                        # it above still does. An inbound handler that fails is
+                        # a message a person sent and nobody received, which is
+                        # invisible from the platform's side.
+                        metrics().increment(
+                            BRIDGE_ERRORS,
+                            {"platform": self._bridge_type, "direction": "inbound"},
+                        )
+                        raise
 
         return traced
 
@@ -277,11 +318,15 @@ async def start(self) -> None:
         self._adapter.set_channel_migration_handler(self._handle_channel_migrated)
         self._adapter.set_agent_presentation_resolver(self._agent_presentation)
         await self._adapter.start(
-            on_message=self._traced(self._handle_inbound_message),
-            on_command=self._traced(self._handle_inbound_command),
-            on_agent_joined=self._traced(self._handle_agent_joined_channel),
-            on_user_joined=self._traced(self._handle_user_joined_channel),
-            on_app_joined=self._traced(self._handle_app_joined_channel),
+            on_message=self._traced("message", self._handle_inbound_message),
+            on_command=self._traced("command", self._handle_inbound_command),
+            on_agent_joined=self._traced(
+                "agent_joined", self._handle_agent_joined_channel
+            ),
+            on_user_joined=self._traced(
+                "user_joined", self._handle_user_joined_channel
+            ),
+            on_app_joined=self._traced("app_joined", self._handle_app_joined_channel),
         )
         await self._ensure_channel_captures()
         # Deliberately not awaited. Provisioning is one call per agent against
@@ -1478,7 +1523,8 @@ async def handle_outbound_message(
 
         room_id, _ = self._channel_to_room[channel_id]
         with tenant_scope(await self._room_tenant(room_id)):
-            await self._relay_outbound_message(channel_id, event)
+            with self._counted_outbound("message"):
+                await self._relay_outbound_message(channel_id, event)
 
     async def _relay_outbound_message(
         self, channel_id: str, event: TransportMessage
@@ -1616,7 +1662,8 @@ async def handle_outbound_media(
 
         room_id, _ = self._channel_to_room[channel_id]
         with tenant_scope(await self._room_tenant(room_id)):
-            await self._relay_outbound_media(channel_id, event, client)
+            with self._counted_outbound("media"):
+                await self._relay_outbound_media(channel_id, event, client)
 
     async def _relay_outbound_media(
         self, channel_id: str, event: TransportMedia, client: ClientBase[Any]

core/switch_core/bridges/collaboration/lifecycle_service.py

@@ -90,6 +90,12 @@ def __init__(
         # (see CollaborationAdapter.exclusive_resource). Lets a second
         # claimant be refused by name instead of failing on the resource.
         self._held_resources: dict[str, str] = {}
+        # Bridges that were started and have not been stopped on purpose. A
+        # crash removes a bridge from `_bridges` and leaves it here, which is
+        # what makes "configured but no longer running" answerable at all —
+        # otherwise a crashed bridge is indistinguishable from one that was
+        # never set up, and the only evidence is a log line nobody reads.
+        self._started: set[str] = set()
         # Serialises registration. The exclusivity check reads the stored
         # bridges and the winner is not written until several awaits later,
         # so two concurrent registrations would both see a free resource and
@@ -533,6 +539,7 @@ async def start(self, bridge_id: str) -> None:
         )
         self._bridges[bridge_id] = bridge_core
         self._tasks[bridge_id] = task
+        self._started.add(bridge_id)
         if wanted is not None:
             self._held_resources[bridge_id] = wanted
 
@@ -636,6 +643,7 @@ async def stop(self, bridge_id: str) -> None:
 
         self._bridges.pop(bridge_id, None)
         self._held_resources.pop(bridge_id, None)
+        self._started.discard(bridge_id)
         logger.info("Stopped collaboration bridge %s", bridge_id)
 
     async def restart(self, bridge_id: str) -> None:
@@ -707,6 +715,24 @@ async def remove(self, bridge_id: str) -> None:
     def get(self, bridge_id: str) -> BridgeCore | None:
         return self._bridges.get(bridge_id)
 
+    def expected_count(self) -> int:
+        """Bridges that were started and have not been stopped deliberately."""
+        return len(self._started)
+
+    def running_count(self) -> int:
+        """Of those, how many still have a task that has not finished.
+
+        A bridge's task runs until shutdown, so a task that is *done* has
+        stopped serving whether it raised or returned — both are equally
+        invisible to anything that only checks membership of `_bridges`.
+        """
+        running = 0
+        for bridge_id in self._started:
+            task = self._tasks.get(bridge_id)
+            if bridge_id in self._bridges and task is not None and not task.done():
+                running += 1
+        return running
+
     def bridges_for_tenant(self, tenant_id: str) -> list[BridgeCore]:
         """Running bridges belonging to `tenant_id`, and none other.
 

core/switch_core/clients/client_lifecycle_service.py

@@ -268,6 +268,14 @@ async def remove(self, client_id: str) -> None:
     def get(self, client_id: str) -> ClientBase[ClientConfig] | None:
         return self._clients.get(client_id)
 
+    def running_count(self) -> int:
+        """Clients believed to be running right now.
+
+        A crashed client removes itself from the registry, so this falling is
+        the only signal that one did — there is no failure counter to read.
+        """
+        return len(self._clients)
+
     def get_by_agent_id(self, agent_id: str) -> ClientBase[ClientConfig] | None:
         for client in self._clients.values():
             if isinstance(client, AgentClient) and client._agent is not None: