Add app health metrics and log previously silent collab failures

server/controllers/api/v1/logs_viewer.py

@@ -12,6 +12,9 @@
 import json
 import logging
 
+from tornado.iostream import StreamClosedError
+
+from digi_server.logger import get_logger
 from utils.log_buffer import get_client_buffer, get_server_buffer
 from utils.web.base_controller import BaseAPIController
 from utils.web.route import ApiRoute, ApiVersion
@@ -234,7 +237,18 @@ async def get(self):
                 if matched:
                     self.write(f"data: {json.dumps(matched[0])}\n\n")
                     await self.flush()
-        except Exception:  # noqa: BLE001
-            pass
+        except StreamClosedError:
+            # The client dropped the stream mid-write; expected, not an error.
+            get_logger().debug(
+                f"Log stream ({source}) from {self.request.remote_ip} closed "
+                f"by the client"
+            )
+        except Exception:
+            # Filtering or serialising an entry failed: a bug, not a disconnect.
+            get_logger().warning(
+                f"Log stream ({source}) from {self.request.remote_ip} ended "
+                f"on an unexpected error",
+                exc_info=True,
+            )
         finally:
             unsubscribe()

server/controllers/ws_controller.py

@@ -30,6 +30,7 @@
 from models.show import Act, Show
 from models.user import User
 from rbac.role import Role
+from utils.web import app_metrics
 from utils.web.base_controller import DatabaseMixin
 from utils.web.jwt_service import (
     TOKEN_EXPIRED,
@@ -92,8 +93,20 @@ class WebSocketController(WSMessageSizeMetricsMixin, DatabaseMixin, WebSocketHan
         "DISCARD_SCRIPT_DRAFT": "_op_discard_script_draft",
     }
 
+    _SESSION_OPS = frozenset({"NEW_CLIENT", "REFRESH_CLIENT", "LIVE_SHOW_JUMP_TO_PAGE"})
+    # Every OP the server handles; the only values used for the ``op`` metric
+    # label, so a client cannot create unlimited series.
+    _KNOWN_OPS = (
+        frozenset({"AUTHENTICATE", "REFRESH_TOKEN"})
+        | _SESSION_OPS
+        | frozenset(_LEADER_OP_HANDLERS)
+        | frozenset(_SCRIPT_ROOM_OP_HANDLERS)
+    )
+
     def __init__(self, application, request, **kwargs):
         super().__init__(application, request, **kwargs)
+        # True between open() counting this connection and on_close() releasing it.
+        self._counted_connection = False
         self.application: DigiScriptServer = application
         self.current_user_id = None
         self.current_username: str | None = None
@@ -110,6 +123,12 @@ def __init__(self, application, request, **kwargs):
         # not resumed (its row had gone); used for the late leadership reclaim.
         self._presented_uuid: Optional[str] = None
 
+    def _who(self) -> str:
+        """Return ``username (ip)`` (or just the ip) for log messages."""
+        if self.current_username:
+            return f"{self.current_username} ({self.request.remote_ip})"
+        return str(self.request.remote_ip)
+
     def update_session(self, user_id=None) -> bool:
         """Create or refresh the Session row for this connection's uuid.
 
@@ -157,6 +176,8 @@ def check_origin(self, origin):
 
     @gen.coroutine
     def open(self, *args: str, **kwargs: str) -> Optional[Awaitable[None]]:
+        app_metrics.ws_connections.inc()
+        self._counted_connection = True
         self.__setattr__("internal_id", str(uuid4()))
         self._placeholder_id = self.__getattribute__("internal_id")
         self.application.clients.append(self)
@@ -189,6 +210,9 @@ def on_close(self) -> None:
         if self._close_handled:
             return
         self._close_handled = True
+        if self._counted_connection:
+            self._counted_connection = False
+            app_metrics.ws_connections.dec()
 
         if self in self.application.clients:
             self.application.clients.remove(self)
@@ -487,12 +511,17 @@ async def on_message(self, message: Union[str, bytes]):
                 content = f"<{size} bytes>"
             else:
                 content = raw
+            app_metrics.ws_messages_total.labels(app_metrics.OTHER_OP).inc()
             get_logger().warning(
                 f"Malformed WS message from {self.request.remote_ip}: "
                 f"{type(e).__name__}: {e} {content}"
             )
             return
 
+        app_metrics.ws_messages_total.labels(
+            app_metrics.bounded_op(ws_op, self._KNOWN_OPS)
+        ).inc()
+
         # With redaction on never log the payload: AUTHENTICATE / REFRESH_TOKEN
         # carry live JWTs.
         if self.application.digi_settings.get_sync("log_redaction"):
@@ -1208,7 +1237,11 @@ async def _op_yjs_sync(self, room_manager, data: dict) -> None:
         try:
             decoded = base64.b64decode(payload)
         except Exception:
-            get_logger().warning("Invalid base64 in YJS_SYNC message")
+            get_logger().warning(
+                f"Invalid base64 in YJS_SYNC message for revision "
+                f"{room.revision_id} from {self._who()} "
+                f"(payload {len(str(payload))} chars)"
+            )
             return
 
         if step == 1:
@@ -1287,7 +1320,11 @@ async def _op_yjs_update(self, room_manager, data: dict) -> None:
         try:
             decoded = base64.b64decode(payload)
         except Exception:
-            get_logger().warning("Invalid base64 in YJS_UPDATE message")
+            get_logger().warning(
+                f"Invalid base64 in YJS_UPDATE message for revision "
+                f"{room.revision_id} from {self._who()} "
+                f"(payload {len(str(payload))} chars)"
+            )
             return
 
         if await self._is_live_session_active():
@@ -1307,6 +1344,10 @@ async def _op_yjs_update(self, room_manager, data: dict) -> None:
         try:
             follow_up = await room.apply_update(decoded)
         except Exception:
+            get_logger().exception(
+                f"YJS_UPDATE: Failed to apply update for revision "
+                f"{room.revision_id} from {self._who()}"
+            )
             await self.write_message(
                 {
                     "OP": "NOOP",
@@ -1330,7 +1371,11 @@ async def _op_yjs_awareness(self, room_manager, data: dict) -> None:
         try:
             decoded = base64.b64decode(payload)
         except Exception:
-            get_logger().warning("Invalid base64 in YJS_AWARENESS message")
+            get_logger().debug(
+                f"Invalid base64 in YJS_AWARENESS message for revision "
+                f"{room.revision_id} from {self._who()} "
+                f"(payload {len(str(payload))} chars)"
+            )
             return
 
         await room.broadcast_awareness(decoded, sender=self)

server/digi_server/app_server.py

@@ -43,6 +43,7 @@
 from utils.module_discovery import get_resource_path, is_frozen
 from utils.script_room_manager import RoomManager
 from utils.version_checker import VersionChecker
+from utils.web import app_metrics
 from utils.web.jwt_service import JWTService
 from utils.web.pending_disconnects import PendingDisconnects
 from utils.web.request_size_metrics import observe_http_request
@@ -346,6 +347,9 @@ class AlembicVersion(self._db.Model):
             cookie_secret=self._get_cookie_secret(),
             login_url="/login",
         )
+        # Fully constructed: let the metrics read this app's live state at scrape
+        # time (weakly referenced, so no teardown is needed).
+        app_metrics.register_application(self)
 
     _METRICS_WARN_INTERVAL = 60.0
     _last_metrics_warning = float("-inf")

server/test/controllers/api/v1/test_logs_viewer.py

@@ -1,9 +1,12 @@
 """Integration tests for GET /api/v1/logs/view."""
 
 import logging
+from unittest import mock
 
 from tornado import escape
+from tornado.iostream import StreamClosedError
 
+from digi_server.logger import get_logger
 from test.conftest import DigiScriptTestCase
 from utils.log_buffer import get_client_buffer, get_server_buffer
 
@@ -373,6 +376,52 @@ def test_stream_non_admin_returns_403(self):
         resp = self._fetch_stream_headers(token=user_token)
         self.assertEqual(403, resp.code)
 
+    def _run_stream_with_failing_filter(self, exc):
+        """Stream /logs/stream with a live-entry filter that raises ``exc``.
+
+        :returns: Tuple of (level name, message) for each ``Log stream`` record.
+        """
+        token = self._create_and_login_admin()
+        calls = []
+
+        def flaky_filter(entries, *args):
+            calls.append(1)
+            if len(calls) > 1:
+                raise exc
+            return []
+
+        self.io_loop.call_later(
+            0.3, lambda: self._inject_server_entry("sse_error_trigger")
+        )
+        with (
+            mock.patch("controllers.api.v1.logs_viewer._filter_entries", flaky_filter),
+            self.assertLogs(get_logger(), level="DEBUG") as captured,
+        ):
+            try:
+                self.fetch(
+                    "/api/v1/logs/stream?source=server",
+                    method="GET",
+                    headers={"Authorization": f"Bearer {token}"},
+                    raise_error=False,
+                    request_timeout=1.0,
+                    streaming_callback=lambda _chunk: None,
+                )
+            except Exception:
+                pass  # HTTPTimeoutError: the stream stays open otherwise
+        return [
+            (r.levelname, r.getMessage())
+            for r in captured.records
+            if "Log stream (server)" in r.getMessage()
+        ]
+
+    def test_stream_client_disconnect_is_logged_at_debug(self):
+        records = self._run_stream_with_failing_filter(StreamClosedError())
+        self.assertEqual(["DEBUG"], [level for level, _ in records])
+
+    def test_stream_unexpected_error_is_logged_at_warning(self):
+        records = self._run_stream_with_failing_filter(RuntimeError("boom"))
+        self.assertEqual(["WARNING"], [level for level, _ in records])
+
     def test_stream_backfill_contains_existing_entries(self):
         """Entries injected before connection must appear in the first chunk.