From 3d82e6844f71ce51e2bfd2dc0567f2f0b60fb7ec Mon Sep 17 00:00:00 2001 From: wakonig_k Date: Fri, 22 May 2026 17:46:18 +0200 Subject: [PATCH] feat: add console redis log and error log --- bec_ipython_client/bec_ipython_client/main.py | 21 +++- .../tests/client_tests/test_bec_client.py | 56 ++++++++++ bec_lib/bec_lib/logger.py | 101 ++++++++++++------ bec_lib/bec_lib/messages.py | 12 ++- bec_lib/tests/test_bec_logger.py | 31 ++++++ 5 files changed, 186 insertions(+), 35 deletions(-) diff --git a/bec_ipython_client/bec_ipython_client/main.py b/bec_ipython_client/bec_ipython_client/main.py index d92655aed..ccf549127 100644 --- a/bec_ipython_client/bec_ipython_client/main.py +++ b/bec_ipython_client/bec_ipython_client/main.py @@ -5,6 +5,7 @@ import functools import os import sys +import traceback from typing import Iterable, Literal, Tuple import IPython @@ -147,7 +148,7 @@ def _configure_ipython(self): self._ip.prompts = BECClientPrompt(ip=self._ip, client=self._client, username="unknown") self._refresh_ipython_username() self._load_magics() - self._ip.events.register("post_run_cell", log_console) + self._ip.events.register("pre_run_cell", log_console) self._ip.set_custom_exc((Exception,), self._create_exception_handler()) # represent objects using __str__, if overwritten, otherwise use __repr__ self._ip.display_formatter.formatters["text/plain"].for_type( @@ -252,25 +253,31 @@ def _ip_exception_handler( ): if issubclass(etype, AlarmBase): parent._alarm_history.append((etype, evalue, tb, tb_offset)) + log_console_error(etype, evalue, tb) print("\x1b[31m BEC alarm:\x1b[0m") evalue.pretty_print() print("For more details, use 'bec.show_last_alarm()'") return if issubclass(etype, ValidationError): + log_console_error(etype, evalue, tb) pretty_print_pydantic_validation_error(evalue) return if issubclass(etype, (ScanInterruption, DeviceConfigError)): + log_console_error(etype, evalue, tb, f"{evalue.__class__.__name__}: {evalue}") print(f"\x1b[31m {evalue.__class__.__name__}:\x1b[0m {evalue}") return if issubclass(etype, redis.exceptions.NoPermissionError): # pylint: disable=protected-access msg = f"The current user ({bec._client.username}) does not have the required permissions.\n {evalue}" + log_console_error(etype, evalue, tb, f"Unauthorized: {msg}") logger.info(f"Unauthorized: {msg}") print(f"\x1b[31m Unauthorized:\x1b[0m {msg}") return if issubclass(etype, ExceptionWithErrorInfo): + log_console_error(etype, evalue, tb) evalue.pretty_print() return + log_console_error(etype, evalue, tb) self.showtraceback((etype, evalue, tb), tb_offset=None) # standard IPython's printout @@ -322,7 +329,17 @@ def username(self, value): def log_console(execution_info): """log the console input""" - logger.log("CONSOLE_LOG", f"{execution_info.info.raw_cell}") + logger.log("CONSOLE_LOG", f"{execution_info.raw_cell}") + + +def log_console_error(etype, evalue, tb=None, message: str | None = None): + """Log console errors to the shared log stream.""" + if message is None: + if tb is not None: + message = "".join(traceback.format_exception(etype, evalue, tb)).rstrip() + else: + message = f"{etype.__name__}: {evalue}" + logger.log("CONSOLE_LOG_ERROR", message) # pylint: disable=wrong-import-position diff --git a/bec_ipython_client/tests/client_tests/test_bec_client.py b/bec_ipython_client/tests/client_tests/test_bec_client.py index 58e19a48b..0c7ae15cc 100644 --- a/bec_ipython_client/tests/client_tests/test_bec_client.py +++ b/bec_ipython_client/tests/client_tests/test_bec_client.py @@ -4,10 +4,12 @@ import IPython import pytest +import redis.exceptions from bec_ipython_client import BECIPythonClient, main from bec_lib import messages from bec_lib.alarm_handler import AlarmBase, AlarmHandler, Alarms +from bec_lib.bec_errors import DeviceConfigError from bec_lib.redis_connector import RedisConnector from bec_lib.service_config import ServiceConfig @@ -238,3 +240,57 @@ def test_bec_ipython_client_show_last_no_alarm(ipython_client, capsys): client.show_last_alarm() captured = capsys.readouterr() assert "No alarm has been raised in this session." in captured.out + + +def test_ipython_exception_handler_logs_console_error_for_device_config_error(): + shell = mock.MagicMock() + with mock.patch.object(main.logger, "log") as mock_log: + main._ip_exception_handler( + shell, + DeviceConfigError, + DeviceConfigError("bad config"), + None, + parent=mock.MagicMock(), + ) + + mock_log.assert_called_once_with("CONSOLE_LOG_ERROR", "DeviceConfigError: bad config") + shell.showtraceback.assert_not_called() + + +def test_ipython_exception_handler_logs_console_error_for_unhandled_exception(): + shell = mock.MagicMock() + try: + raise RuntimeError("boom") + except RuntimeError as exc: + with mock.patch.object(main.logger, "log") as mock_log: + main._ip_exception_handler(shell, RuntimeError, exc, exc.__traceback__, parent=mock.MagicMock()) + + mock_log.assert_called_once() + assert mock_log.call_args.args[0] == "CONSOLE_LOG_ERROR" + assert "RuntimeError: boom" in mock_log.call_args.args[1] + shell.showtraceback.assert_called_once() + + +def test_ipython_exception_handler_logs_console_error_for_permission_error(): + shell = mock.MagicMock() + parent = mock.MagicMock() + with ( + mock.patch.object(main.logger, "log") as mock_log, + mock.patch.object(main.logger, "info") as mock_info, + mock.patch.object( + main, "bec", mock.MagicMock(_client=mock.MagicMock(username="alice")), create=True + ), + ): + main._ip_exception_handler( + shell, + redis.exceptions.NoPermissionError, + redis.exceptions.NoPermissionError("denied"), + None, + parent=parent, + ) + + mock_log.assert_called_once() + assert mock_log.call_args.args[0] == "CONSOLE_LOG_ERROR" + assert "Unauthorized:" in mock_log.call_args.args[1] + mock_info.assert_called_once() + shell.showtraceback.assert_not_called() diff --git a/bec_lib/bec_lib/logger.py b/bec_lib/bec_lib/logger.py index dad1fd610..34245f118 100644 --- a/bec_lib/bec_lib/logger.py +++ b/bec_lib/bec_lib/logger.py @@ -42,6 +42,7 @@ class LogLevel(int, enum.Enum): ERROR = 40 CRITICAL = 50 CONSOLE_LOG = 21 + CONSOLE_LOG_ERROR = 22 class BECLogger: @@ -201,30 +202,6 @@ def _update_base_path(self, service_config: dict | None = None): self._base_path = self.writer_mixin.directory self.writer_mixin.create_directory(self._base_path) - def _logger_callback(self, msg): - if not self._configured: - return - if self.connector is None: - return - msg = json.loads(msg) - msg["service_name"] = self.service_name - try: - self.connector.xadd( - topic=MessageEndpoints.log(), - msg_dict={ - "data": bec_lib.messages.LogMessage( - log_type=msg["record"]["level"]["name"].lower(), log_msg=msg - ) - }, - max_size=10000, - ) - except Exception: - # connector disconnected? - # just ignore the error here... - # Exception is not explicitly specified, - # because it depends on the connector - pass - def get_format(self, level: LogLevel = None, is_stderr=False, is_container=False) -> str: """ Get the format for a specific log level. @@ -298,23 +275,38 @@ def filter(self, is_console: bool = False): Args: is_console (bool, optional): Whether the log is for the console. Defaults to False. - Returns: function: Filter function. """ def _filter(record): - if record["name"] in self._disabled_modules: + if self._is_disabled_record(record): return False - for module in self._disabled_modules: - if record["name"].startswith(module): - return False - if not is_console and record["level"].no == LogLevel.CONSOLE_LOG: + if not is_console and self._is_console_level(record["level"].no): return False return True return _filter + def _is_disabled_record(self, record) -> bool: + if record["name"] in self._disabled_modules: + return True + return any(record["name"].startswith(module) for module in self._disabled_modules) + + def _is_console_level(self, level_no: int) -> bool: + return level_no in (LogLevel.CONSOLE_LOG, LogLevel.CONSOLE_LOG_ERROR) + + def filter_console_redis_log(self, record) -> bool: + """ + Filter function for console redis log messages, which are used to send log messages redis. + + Args: + record (dict): Log record. + Returns: + bool: True if the log message should be sent to the console via redis, False otherwise + """ + return not self._is_disabled_record(record) and self._is_console_level(record["level"].no) + def add_sys_stderr(self, level: LogLevel): """ Add a sink to stderr. @@ -355,6 +347,7 @@ def add_console_log(self): """ try: self.logger.level("CONSOLE_LOG", no=21, color="", icon="📣") + self.logger.level("CONSOLE_LOG_ERROR", no=22, color="", icon="📣") except (TypeError, ValueError): # level with same severity already exists: already configured pass @@ -379,6 +372,7 @@ def add_console_log(self): opener=self._file_opener, ) self._console_log = True + self.add_console_redis_log() def add_redis_log(self, level: LogLevel): """ @@ -388,13 +382,58 @@ def add_redis_log(self, level: LogLevel): level (LogLevel): Log level. """ self.logger.add( - self._logger_callback, + self._publish_log_message, serialize=True, level=level, format=self.formatting(), filter=self.filter(), ) + def add_console_redis_log(self): + """ + Add a sink to the console redis log. + It deviates from the regular redis log in that it only includes messages with + level CONSOLE_LOG and CONSOLE_LOG_ERROR. + """ + self.logger.add( + self._console_redis_logger_callback, + serialize=True, + level=LogLevel.CONSOLE_LOG, + format=self.formatting(is_stderr=True), + filter=self.filter_console_redis_log, + ) + + def _console_redis_logger_callback(self, msg): + if not self._configured or self.connector is None: + return + self._publish_log_message(msg, service_name=f"{self.service_name}_CONSOLE") + + def _decode_log_payload(self, msg: str | dict) -> dict: + return json.loads(msg) if isinstance(msg, str) else dict(msg) + + def _publish_log_message(self, msg: str | dict, service_name: str | None = None) -> bool: + if not self._configured or self.connector is None: + return False + payload = self._decode_log_payload(msg) + payload["service_name"] = self.service_name if service_name is None else service_name + try: + self.connector.xadd( + topic=MessageEndpoints.log(), + msg_dict={ + "data": bec_lib.messages.LogMessage( + log_type=payload["record"]["level"]["name"].lower(), log_msg=payload + ) + }, + max_size=10000, + ) + return True + except Exception: + # connector disconnected? + # just ignore the error here... + # Exception is not explicitly specified, + # because it depends on the connector + return False + @property def disabled_modules(self) -> set[str]: """ diff --git a/bec_lib/bec_lib/messages.py b/bec_lib/bec_lib/messages.py index 76d351568..10bf6bb22 100644 --- a/bec_lib/bec_lib/messages.py +++ b/bec_lib/bec_lib/messages.py @@ -1008,7 +1008,7 @@ class LogMessage(BECMessage): """Log message Args: - log_type (Literal["trace", "debug", "info", "success", "warning", "error", "critical", "console_log"]) : Log type. + log_type (Literal["trace", "debug", "info", "success", "warning", "error", "critical", "console_log", "console_log_error"]) : Log type. log_msg (dict or str): Log message. metadata (dict, optional): Additional metadata. @@ -1016,7 +1016,15 @@ class LogMessage(BECMessage): msg_type: ClassVar[str] = "log_message" log_type: Literal[ - "trace", "debug", "info", "success", "warning", "error", "critical", "console_log" + "trace", + "debug", + "info", + "success", + "warning", + "error", + "critical", + "console_log", + "console_log_error", ] log_msg: dict | str diff --git a/bec_lib/tests/test_bec_logger.py b/bec_lib/tests/test_bec_logger.py index 392b65033..ece505cbf 100644 --- a/bec_lib/tests/test_bec_logger.py +++ b/bec_lib/tests/test_bec_logger.py @@ -1,3 +1,4 @@ +import json import os from pathlib import Path from unittest import mock @@ -98,3 +99,33 @@ def test_set_log_level(logger, log_level, sink, expected_level): logger.set_log_level(log_level, sink) for key, value in expected_level.items(): assert getattr(logger, key) == value + + +def test_console_redis_callback_publishes_to_log_endpoint_with_console_service_name(logger): + logger._configured = True + logger.service_name = "test" + logger.connector = mock.MagicMock(spec=RedisConnector) + + logger._console_redis_logger_callback( + json.dumps({"record": {"level": {"name": "CONSOLE_LOG"}}, "text": "hello"}) + ) + + logger.connector.xadd.assert_called_once() + kwargs = logger.connector.xadd.call_args.kwargs + assert kwargs["topic"] == logger.connector.xadd.call_args.kwargs["topic"] + assert kwargs["topic"].endpoint == "info/log" + assert kwargs["msg_dict"]["data"].log_type == "console_log" + assert kwargs["msg_dict"]["data"].log_msg["service_name"] == "test_CONSOLE" + + +def test_console_redis_callback_ignores_publish_failures(logger): + logger._configured = True + logger.service_name = "test" + logger.connector = mock.MagicMock(spec=RedisConnector) + logger.connector.xadd.side_effect = RuntimeError("redis unavailable") + + logger._console_redis_logger_callback( + json.dumps({"record": {"level": {"name": "CONSOLE_LOG_ERROR"}}, "text": "oops"}) + ) + + logger.connector.xadd.assert_called_once()