ERSKafkaLogHandler.ERSKafkaLogHandler Class Reference

Inheritance diagram for ERSKafkaLogHandler.ERSKafkaLogHandler:

Collaboration diagram for ERSKafkaLogHandler.ERSKafkaLogHandler:

Public Member Functions
	__init__ (self, str session="Unknown", str kafka_address="monkafka.cern.ch:30092", str kafka_topic="ers_stream", str|None app_name=None)
None	emit (self, logging.LogRecord record)

Public Attributes
str	session = session
str	kafka_address = kafka_address
str	kafka_topic = kafka_topic
	app_name = app_name
	publisher

Static Protected Member Functions
SeverityLevel	_convert_logging_level_to_ers_level (int level)

Detailed Description

A logging handler that sends log messages to the ERS system via Kafka.

Note 1: You need to have Kafka to use this correctly, to see the message in Grafana, you will need to have the full ERS stack running.
Note 2: IMPORTANT!! you MUST NOT use this handler on the root logger. Use it on a logger that you have created yourself (this is because the root logger is used by Kafka, and it creates a circular dependency).

Example:
```python
import logging
from erskafka.ERSKafkaLogHandler import ERSKafkaLogHandler

logger = logging.getLogger("my_logger") # NOTE, THIS IS NOT THE ROOT LOGGER!!!
# logger = logging.getLogger() # FORBIDDEN!!! Will should raise an exception
logger.setLevel(logging.DEBUG)
handler = ERSKafkaLogHandler(session="test")
logger.addHandler(handler)
logger.debug("This is a debug message")
```

Definition at line 9 of file ERSKafkaLogHandler.py.

Constructor & Destructor Documentation

__init__()

ERSKafkaLogHandler.ERSKafkaLogHandler.__init__	(		self,
		str	session = "Unknown",
		str	kafka_address = "monkafka.cern.ch:30092",
		str	kafka_topic = "ers_stream",
		str | None	app_name = None )

Definition at line 29 of file ERSKafkaLogHandler.py.

    ):
        super().__init__()
        os.environ['DUNEDAQ_PARTITION'] = session
        self.session:str = session
        self.kafka_address:str = kafka_address
        self.kafka_topic:str = kafka_topic
        self.app_name = app_name
 
        self.publisher = ERSPublisher(
            bootstrap = kafka_address,
            topic = kafka_topic,
        )
 

Member Function Documentation

_convert_logging_level_to_ers_level()

SeverityLevel ERSKafkaLogHandler.ERSKafkaLogHandler._convert_logging_level_to_ers_level ( int level )

staticprotected

Definition at line 49 of file ERSKafkaLogHandler.py.

    def _convert_logging_level_to_ers_level(level:int) -> SeverityLevel:
        match level:
            case logging.DEBUG:
                return SeverityLevel.DEBUG
            case logging.INFO:
                return SeverityLevel.INFO
            case logging.WARNING:
                return SeverityLevel.WARNING
            case logging.ERROR:
                return SeverityLevel.ERROR
            case logging.CRITICAL:
                return SeverityLevel.FATAL
            case _:
                return SeverityLevel.INFO
 

emit()

None ERSKafkaLogHandler.ERSKafkaLogHandler.emit	(		self,
		logging.LogRecord	record )

Definition at line 64 of file ERSKafkaLogHandler.py.

    def emit(self, record:logging.LogRecord) -> None:
        ers_level = ERSKafkaLogHandler._convert_logging_level_to_ers_level(record.levelno)
 
        if record.name == 'root':
            raise ERSKafkaLogHandlerOnRootLogger('To avoid all sorts of undesired behaviours this logger cannot be use on the root logger. Use logging.getLogger("some_name").addHandler(ERSKafkaLogHandler()) instead.')
 
        success = self.publisher.publish(
            record.msg,
            severity = ers_level.name,
            context_kwargs = dict(
                package_name = str(record.module),
                application_name = self.app_name if self.app_name else str(record.name),
                line_number = record.lineno,
                file_name = str(record.pathname),
                function_name = str(record.funcName),
                process_id=record.process,
                thread_id=0 # TODO: have better way of handling this. It should come from drunc etc. 
            )
        )
        
        #! publish returns a future, so a flush ensures the message is sent out
        self.publisher.producer.flush()
        if not success:
            print(f'WARNING! Failed to publish: {record.msg} to Kafka')

Member Data Documentation

app_name

ERSKafkaLogHandler.ERSKafkaLogHandler.app_name = app_name

Definition at line 41 of file ERSKafkaLogHandler.py.

kafka_address

str ERSKafkaLogHandler.ERSKafkaLogHandler.kafka_address = kafka_address

Definition at line 39 of file ERSKafkaLogHandler.py.

kafka_topic

str ERSKafkaLogHandler.ERSKafkaLogHandler.kafka_topic = kafka_topic

Definition at line 40 of file ERSKafkaLogHandler.py.

publisher

ERSKafkaLogHandler.ERSKafkaLogHandler.publisher

Initial value:

=  ERSPublisher(
            bootstrap = kafka_address,
            topic = kafka_topic,
        )

Definition at line 43 of file ERSKafkaLogHandler.py.

session

str ERSKafkaLogHandler.ERSKafkaLogHandler.session = session

Definition at line 38 of file ERSKafkaLogHandler.py.

---

The documentation for this class was generated from the following file:

• /github/workspace/dunedaq/sourcecode/erskafka/python/erskafka/ERSKafkaLogHandler.py