Enhance logging and debugging capabilities

- Added a `--debug` flag to the logs command for enabling debug mode.
- Updated logs command to display debug mode status.
- Enhanced logging in the SqliteDatasetLoggerAdapter and SqliteEventBus for better traceability of events and errors.
- Implemented debug logging in the WebSocketManager and LogsServer to provide detailed insights during WebSocket connections and broadcasts.
- Enabled debug mode for all relevant loggers in the logs server system.

Author: Dylan Huang

eval_protocol/cli.py

@@ -300,6 +300,7 @@ def parse_args(args=None):
     # Logs command
     logs_parser = subparsers.add_parser("logs", help="Serve logs with file watching and real-time updates")
     logs_parser.add_argument("--port", type=int, default=8000, help="Port to bind to (default: 8000)")
+    logs_parser.add_argument("--debug", action="store_true", help="Enable debug mode")
 
     # Upload command
     upload_parser = subparsers.add_parser(

eval_protocol/cli_commands/logs.py

@@ -16,6 +16,7 @@ def logs_command(args):
     print(f"🌐 URL: http://localhost:{port}")
     print(f"🔌 WebSocket: ws://localhost:{port}/ws")
     print(f"👀 Watching paths: {['current directory']}")
+    print(f"🔍 Debug mode: {args.debug}")
     print("Press Ctrl+C to stop the server")
     print("-" * 50)
 
@@ -25,7 +26,7 @@ def logs_command(args):
     elasticsearch_config = ElasticsearchSetup().setup_elasticsearch()
 
     try:
-        serve_logs(port=args.port, elasticsearch_config=elasticsearch_config)
+        serve_logs(port=args.port, elasticsearch_config=elasticsearch_config, debug=args.debug)
         return 0
     except KeyboardInterrupt:
         print("\n🛑 Server stopped by user")

eval_protocol/dataset_logger/sqlite_dataset_logger_adapter.py

@@ -23,12 +23,19 @@ def __init__(self, db_path: Optional[str] = None, store: Optional[SqliteEvaluati
 
     def log(self, row: "EvaluationRow") -> None:
         data = row.model_dump(exclude_none=True, mode="json")
+        rollout_id = data.get("execution_metadata", {}).get("rollout_id", "unknown")
+        logger.debug(f"[EVENT_BUS_EMIT] Starting to log row with rollout_id: {rollout_id}")
+
         self._store.upsert_row(data=data)
+        logger.debug(f"[EVENT_BUS_EMIT] Successfully stored row in database for rollout_id: {rollout_id}")
+
         try:
+            logger.debug(f"[EVENT_BUS_EMIT] Emitting event '{LOG_EVENT_TYPE}' for rollout_id: {rollout_id}")
             event_bus.emit(LOG_EVENT_TYPE, EvaluationRow(**data))
+            logger.debug(f"[EVENT_BUS_EMIT] Successfully emitted event for rollout_id: {rollout_id}")
         except Exception as e:
             # Avoid breaking storage due to event emission issues
-            logger.error(f"Failed to emit row_upserted event: {e}")
+            logger.error(f"[EVENT_BUS_EMIT] Failed to emit row_upserted event for rollout_id {rollout_id}: {e}")
             pass
 
     def read(self, rollout_id: Optional[str] = None) -> List["EvaluationRow"]:

eval_protocol/event_bus/sqlite_event_bus.py

@@ -35,75 +35,106 @@ def emit(self, event_type: str, data: Any) -> None:
             event_type: Type of event (e.g., "log")
             data: Event data
         """
+        logger.debug(f"[CROSS_PROCESS_EMIT] Emitting event type: {event_type}")
+
         # Call local listeners immediately
+        logger.debug(f"[CROSS_PROCESS_EMIT] Calling {len(self._listeners)} local listeners")
         super().emit(event_type, data)
+        logger.debug("[CROSS_PROCESS_EMIT] Completed local listener calls")
 
         # Publish to cross-process subscribers
+        logger.debug("[CROSS_PROCESS_EMIT] Publishing to cross-process subscribers")
         self._publish_cross_process(event_type, data)
+        logger.debug("[CROSS_PROCESS_EMIT] Completed cross-process publish")
 
     def _publish_cross_process(self, event_type: str, data: Any) -> None:
         """Publish event to cross-process subscribers via database."""
-        self._db.publish_event(event_type, data, self._process_id)
+        logger.debug(f"[CROSS_PROCESS_PUBLISH] Publishing event {event_type} to database")
+        try:
+            self._db.publish_event(event_type, data, self._process_id)
+            logger.debug(f"[CROSS_PROCESS_PUBLISH] Successfully published event {event_type} to database")
+        except Exception as e:
+            logger.error(f"[CROSS_PROCESS_PUBLISH] Failed to publish event {event_type} to database: {e}")
 
     def start_listening(self) -> None:
         """Start listening for cross-process events."""
         if self._running:
+            logger.debug("[CROSS_PROCESS_LISTEN] Already listening, skipping start")
             return
 
+        logger.debug("[CROSS_PROCESS_LISTEN] Starting cross-process event listening")
         self._running = True
         self._start_database_listener()
+        logger.debug("[CROSS_PROCESS_LISTEN] Started database listener thread")
 
     def stop_listening(self) -> None:
         """Stop listening for cross-process events."""
+        logger.debug("[CROSS_PROCESS_LISTEN] Stopping cross-process event listening")
         self._running = False
         if self._listener_thread and self._listener_thread.is_alive():
+            logger.debug("[CROSS_PROCESS_LISTEN] Waiting for listener thread to stop")
             self._listener_thread.join(timeout=1)
+            logger.debug("[CROSS_PROCESS_LISTEN] Listener thread stopped")
 
     def _start_database_listener(self) -> None:
         """Start database-based event listener."""
 
         def database_listener():
+            logger.debug("[CROSS_PROCESS_LISTENER] Starting database listener loop")
             last_cleanup = time.time()
 
             while self._running:
                 try:
                     # Get unprocessed events from other processes
                     events = self._db.get_unprocessed_events(self._process_id)
+                    if events:
+                        logger.debug(f"[CROSS_PROCESS_LISTENER] Found {len(events)} unprocessed events")
 
                     for event in events:
                         if not self._running:
                             break
 
                         try:
+                            logger.debug(
+                                f"[CROSS_PROCESS_LISTENER] Processing event {event['event_id']} of type {event['event_type']}"
+                            )
                             # Handle the event
                             self._handle_cross_process_event(event["event_type"], event["data"])
+                            logger.debug(f"[CROSS_PROCESS_LISTENER] Successfully processed event {event['event_id']}")
 
                             # Mark as processed
                             self._db.mark_event_processed(event["event_id"])
+                            logger.debug(f"[CROSS_PROCESS_LISTENER] Marked event {event['event_id']} as processed")
 
                         except Exception as e:
-                            logger.debug(f"Failed to process event {event['event_id']}: {e}")
+                            logger.debug(f"[CROSS_PROCESS_LISTENER] Failed to process event {event['event_id']}: {e}")
 
                     # Clean up old events every hour
                     current_time = time.time()
                     if current_time - last_cleanup >= 3600:
+                        logger.debug("[CROSS_PROCESS_LISTENER] Cleaning up old events")
                         self._db.cleanup_old_events()
                         last_cleanup = current_time
 
                     # Small sleep to prevent busy waiting
                     time.sleep(0.1)
 
                 except Exception as e:
-                    logger.debug(f"Database listener error: {e}")
+                    logger.debug(f"[CROSS_PROCESS_LISTENER] Database listener error: {e}")
                     time.sleep(1)
 
         self._listener_thread = threading.Thread(target=database_listener, daemon=True)
         self._listener_thread.start()
 
     def _handle_cross_process_event(self, event_type: str, data: Any) -> None:
         """Handle events received from other processes."""
-        for listener in self._listeners:
+        logger.debug(f"[CROSS_PROCESS_HANDLE] Handling cross-process event type: {event_type}")
+        logger.debug(f"[CROSS_PROCESS_HANDLE] Calling {len(self._listeners)} listeners")
+
+        for i, listener in enumerate(self._listeners):
             try:
+                logger.debug(f"[CROSS_PROCESS_HANDLE] Calling listener {i}")
                 listener(event_type, data)
+                logger.debug(f"[CROSS_PROCESS_HANDLE] Successfully called listener {i}")
             except Exception as e:
-                logger.debug(f"Cross-process event listener failed for {event_type}: {e}")
+                logger.debug(f"[CROSS_PROCESS_HANDLE] Cross-process event listener {i} failed for {event_type}: {e}")