From 2c9540e83f571b646959956517f0b4866b4cb3c6 Mon Sep 17 00:00:00 2001
From: ahmedk <karim.ahmed@xfel.eu>
Date: Mon, 11 Nov 2024 17:49:19 +0100
Subject: [PATCH] feat: remove logging handler and save error and logging
directly
---
notebooks/test/test-logging.ipynb | 47 +++---
src/xfel_calibrate/setup_logging.py | 222 ++++++++++------------------
2 files changed, 93 insertions(+), 176 deletions(-)
diff --git a/notebooks/test/test-logging.ipynb b/notebooks/test/test-logging.ipynb
index 0d033847c..eee72a5a3 100644
--- a/notebooks/test/test-logging.ipynb
+++ b/notebooks/test/test-logging.ipynb
@@ -18,26 +18,10 @@
"metadata": {},
"outputs": [],
"source": [
- "import logging"
- ]
- },
- {
- "cell_type": "markdown",
- "id": "ba0663ca",
- "metadata": {},
- "source": [
- "## INFO/DEBUG"
- ]
- },
- {
- "cell_type": "code",
- "execution_count": null,
- "id": "a3b5e455",
- "metadata": {},
- "outputs": [],
- "source": [
- "# This wont be logged\n",
- "logging.info(\"Logging some (INFO)rmation\")"
+ "import logging\n",
+ "import warnings\n",
+ "\n",
+ "from cal_tools.warnings import CalibrationWarning"
]
},
{
@@ -55,20 +39,23 @@
"metadata": {},
"outputs": [],
"source": [
- "import warnings\n",
- "from cal_tools.warnings import CalibrationWarning\n",
- "\n",
"class TestCalWarning(CalibrationWarning):\n",
" \"\"\"Base class for custom user warnings\"\"\"\n",
" pass\n",
"\n",
"\n",
- "def warn_user(message, warning_class):\n",
- " warnings.warn(message, warning_class)\n",
- " \n",
- "\n",
"if 1 < 2:\n",
- " warn_user('This inequality is true!', TestCalWarning)"
+ " warnings.warn('This inequality is true!', TestCalWarning)"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "id": "6a4d4f01",
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "warnings.warn('This user warning will be ignored', UserWarning)"
]
},
{
@@ -79,7 +66,7 @@
"outputs": [],
"source": [
"# This wont be logged\n",
- "logging.warning(\"This is a warning message using logging standard library.\")"
+ "logging.warning(\"This is a warning message using logging standard library. It wont be logged\")"
]
},
{
@@ -98,7 +85,7 @@
"outputs": [],
"source": [
"# This wont be logged\n",
- "logging.error(\"Logging some (ERROR) without failing the notebook\")"
+ "logging.error(\"Logging some (ERROR) without failing the notebook. It wont be logged\")"
]
},
{
diff --git a/src/xfel_calibrate/setup_logging.py b/src/xfel_calibrate/setup_logging.py
index 465b07921..4714054ef 100644
--- a/src/xfel_calibrate/setup_logging.py
+++ b/src/xfel_calibrate/setup_logging.py
@@ -1,162 +1,92 @@
-import logging
import os
import sys
-import traceback
+import json
import warnings
import IPython
-from pythonjsonlogger import jsonlogger
+from datetime import datetime
+from typing import Type, Any
from cal_tools.warnings import CalibrationWarning
-NOTEBOOK_NAME = os.getenv('CAL_NOTEBOOK_NAME', 'Unknown notebook')
-JOB_ID = os.getenv('SLURM_JOB_ID', 'local')
-
-
-class ContextFilter(logging.Filter):
- def filter(self, record):
- # Only allow records that come from exception handlers
- if getattr(record, 'from_exception_handler', False):
- record.notebook = NOTEBOOK_NAME
- record.job_id = JOB_ID
- return True
- return False
-
-
-def get_class_hierarchy(cls):
- """Helper function to get the full class hierarchy"""
- class_hierarchy = []
- current_class = cls
- while current_class and current_class != object:
- class_hierarchy.append(current_class.__name__)
- current_class = current_class.__base__
- return '.'.join(reversed(class_hierarchy))
-
-
-class CustomJsonFormatter(jsonlogger.JsonFormatter):
- def add_fields(self, log_record, record, message_dict):
- super(CustomJsonFormatter, self).add_fields(
- log_record, record, message_dict)
- log_record['timestamp'] = self.formatTime(record, self.datefmt)
- log_record['level'] = record.levelname
- log_record['filename'] = record.filename
- log_record['lineno'] = record.lineno
-
- # Get log_class from extra parameters (set in our warning/error handlers)
- if hasattr(record, 'log_class'):
- log_record['log_class'] = record.log_class
-
- if record.exc_info:
- log_record['exc_info'] = self.formatException(record.exc_info)
- exc_class = record.exc_info[0]
- log_record['class'] = get_class_hierarchy(exc_class)
-
-
-# Create a logger
-logger = logging.getLogger()
-logger.setLevel(logging.INFO)
-
-# Define a custom JSON format
-formatter = CustomJsonFormatter(
- '%(timestamp)s %(level)s %(filename)s %(lineno)d '
- '%(notebook)s %(job_id)s %(log_class)s %(message)s')
-
-
-# Function to create a file handler with job-specific JSON log file
-def create_job_specific_handler(log_level, file_suffix):
- log_file = f'{file_suffix}_{JOB_ID}.log'
- handler = logging.FileHandler(log_file, delay=True)
- handler.setLevel(log_level)
- handler.setFormatter(formatter)
- return handler
-
-
-# Create job-specific file handlers
-error_handler = create_job_specific_handler(logging.ERROR, 'errors')
-warning_handler = create_job_specific_handler(logging.WARNING, 'warnings')
-
-# Keep console handler for notebook and slurm.out stdout
-console_handler = logging.StreamHandler()
-console_handler.setLevel(logging.INFO)
-console_formatter = logging.Formatter('%(levelname)s: %(message)s')
-console_handler.setFormatter(console_formatter)
-
-# Add the custom filter to handlers
-context_filter = ContextFilter()
-error_handler.addFilter(context_filter)
-warning_handler.addFilter(context_filter)
-warning_handler.addFilter(lambda record: record.levelno < logging.ERROR)
-console_handler.addFilter(context_filter)
-
-# Add handlers to logger
-logger.addHandler(error_handler)
-logger.addHandler(warning_handler)
-logger.addHandler(console_handler)
-
-handling_error = False
-
-
-def safe_handle_error(exc_type, exc_value, exc_traceback):
- global handling_error
-
- # Added this block to skip sys.exit()
- if exc_type in (SystemExit, KeyboardInterrupt):
- sys.__excepthook__(exc_type, exc_value, exc_traceback)
- return
-
- if handling_error: # Avoid infinite loop of errors.
- sys.stderr.write("Recursive error detected!\n")
- traceback.print_exception(
- exc_type, exc_value, exc_traceback, file=sys.stderr)
- return
- handling_error = True
+
+def get_class_hierarchy(cls: Type) -> str:
+ """Get the full class hierarchy of a class."""
+ return '.'.join(c.__name__ for c in cls.__mro__)
+
+
+def log_error(
+ exc_type: Type[Exception],
+ exc_value: Exception,
+ exc_traceback: Any
+) -> None:
+ """Log error information to file."""
try:
- # Log the error with the notebook name, job ID, and additional metadata
- logger.error(str(exc_value),
- extra={
- 'notebook': NOTEBOOK_NAME,
- 'job_id': JOB_ID,
- 'log_class': get_class_hierarchy(exc_type),
- 'from_exception_handler': True
- },
- exc_info=(exc_type, exc_value, exc_traceback)
- )
- except Exception as log_error:
- sys.stderr.write(f"Logging failed: {log_error}\n")
- traceback.print_exception(
- exc_type, exc_value, exc_traceback, file=sys.stderr)
- finally:
- handling_error = False
+ error_info = {
+ "timestamp": datetime.now().isoformat(),
+ "job_id": os.getenv('SLURM_JOB_ID', 'local'),
+ "level": "ERROR",
+ "message": str(exc_value),
+ "class": get_class_hierarchy(exc_type),
+ }
+ with open(f"errors_{error_info['job_id']}.log", "a") as error_log_file:
+ error_log_file.write(json.dumps(error_info) + "\n")
+
+ except Exception as e:
+ sys.stdout.write(f"Logging failed: {e}\n")
+
+
+def handle_warning(
+ message: Warning, category: Type[Warning],
+ # these are needed for `warnings.showwarning`
+ filename=None, lineno=None,
+ file=None, line=None,
+) -> None:
+ """Log and display warnings."""
-def handle_warning(message, category, filename, lineno, file=None, line=None):
try:
- logger.warning(
- "%s",
- message,
- extra={
- 'notebook': NOTEBOOK_NAME,
- 'job_id': JOB_ID,
- 'log_class': get_class_hierarchy(category),
- 'from_exception_handler': True
- }
- )
- except Exception as log_error:
- sys.stderr.write(f"Logging failed: {log_error}\n")
-
-
-# Replace the handlers with our custom ones
-sys.excepthook = safe_handle_error
-warnings.showwarning = handle_warning
+ warning_info = {
+ "timestamp": datetime.now().isoformat(),
+ "job_id": os.getenv('SLURM_JOB_ID', 'local'),
+ "level": "WARNING",
+ "message": str(message),
+ "class": get_class_hierarchy(category),
+ }
+
+ with open(
+ f"warnings_{warning_info['job_id']}.log", "a"
+ ) as warning_log_file:
+ warning_log_file.write(json.dumps(warning_info) + "\n")
+
+ except Exception as e:
+ sys.stdout.write(f"Warning logging failed: {e}\n")
+ finally:
+ # Ensure warning is displayed in notebook.
+ # TODO: use e.g. latex_warning from cal_tools
+ # to display colored warning textbox
+ print(f"\n{category.__name__}: {message}\n")
-# Set up warnings filter
-warnings.filterwarnings("ignore") # Ignore all warnings
-# Except CalibrationWarning and subclasses
-warnings.simplefilter("default", CalibrationWarning)
+
+# Get IPython shell
+shell = IPython.get_ipython()
+
+# Store original handlers
+original_showtraceback = shell.showtraceback
-# Override IPython's exception handling
-def custom_showtraceback(self, *args, **kwargs):
- return safe_handle_error(*sys.exc_info())
+# Custom exception handler for IPython
+def custom_showtraceback(*args, **kwargs):
+ etype, value, tb = sys.exc_info()
+ log_error(etype, value, tb)
+ original_showtraceback(*args, **kwargs)
+ if not args and not kwargs:
+ return etype, value, tb
-IPython.core.interactiveshell.InteractiveShell.showtraceback = custom_showtraceback # noqa
+# Install handlers
+shell.showtraceback = custom_showtraceback
+warnings.showwarning = handle_warning
+
+# Ignore all warnings globally, except CalibrationWarning
+warnings.filterwarnings("ignore")
+warnings.simplefilter("default", CalibrationWarning)
--
GitLab