OpMonPublisher.OpMonPublisher Class Reference

Inheritance diagram for OpMonPublisher.OpMonPublisher:

Collaboration diagram for OpMonPublisher.OpMonPublisher:

Public Member Functions
None	__init__ (self, OpMonConf conf)
str	extract_key (self, OpMonEntry opmon_entry)
None	publish (self, Msg message, dict[str, str]|None custom_origin=None, int|str|None level=None)
Public Member Functions inherited from opmonlib.publisher_base.OpMonPublisherBase
None	__post_init__ (self)
None	check_publisher (self)
str	extract_topic (self, Msg message)
OpMonId	make_origin (self, str session, str app)
dict[str, str]	validate_custom_origin (self, dict[str, str]|None custom_origin=None)
dict	make_data (self, Msg message, str top_block="")
OpMonValue	to_map (self, int|float|bool|str value, int field_type)
OpMonEntry	to_entry (self, Msg message, dict[str, str]|None custom_origin)
int	log_level_to_int (self, str|int level)
str	log_level_to_str (self, str|int level)

Public Attributes
	log = logging.getLogger("OpMonPublisher")
	conf = conf
	opmon_producer = None
str	default_topic = "monitoring." + self.conf.topic
	publisher
Public Attributes inherited from opmonlib.publisher_base.OpMonPublisherBase
	ts = Timestamp()
	substructure = None
	log
	conf
	publisher = "OpMon configuration must be of type OpMonConf."
	default_topic
tuple	publisher

Detailed Description

Tool for publishing operational monitoring metrics to kafka.

Definition at line 14 of file OpMonPublisher.py.

Constructor & Destructor Documentation

__init__()

None OpMonPublisher.OpMonPublisher.__init__	(		self,
		OpMonConf	conf )

Construct the object to publish OpMon metrics to kafka.

Reimplemented from opmonlib.publisher_base.OpMonPublisherBase.

Definition at line 17 of file OpMonPublisher.py.

    def __init__(self, conf: OpMonConf) -> None:
        """Construct the object to publish OpMon metrics to kafka."""
        super().__init__()
        self.log = logging.getLogger("OpMonPublisher")
        self.conf = conf
        if isinstance(self.conf.level, str):
            self.conf.level = logging_log_level_from_str(self.conf.level)
        self.log.setLevel(self.conf.level)
        self.log.addHandler(setup_rich_handler())
 
        if self.conf.opmon_type != "stream":
            self.log.error("Type must be stream to publish to kafka.")
            sys.exit(1)
 
        if self.conf.bootstrap == "":
            self.log.warning(
                "There is no boostrap provided, not initializing publisher to topic %s",
                self.conf.default_topic,
            )
            self.opmon_producer = None
            return
 
        self.default_topic = "monitoring." + self.conf.topic
        self.publisher = KafkaProducer(
            bootstrap_servers=conf.bootstrap,
            value_serializer=lambda v: v.SerializeToString(),
            key_serializer=lambda k: str(k).encode("utf-8"),
        )
 
        super().__post_init__()
        return
 

Member Function Documentation

extract_key()

str OpMonPublisher.OpMonPublisher.extract_key	(		self,
		OpMonEntry	opmon_entry )

Extract  the key from the OpMonEntry.

Definition at line 49 of file OpMonPublisher.py.

    def extract_key(self, opmon_entry: OpMonEntry) -> str:
        """Extract  the key from the OpMonEntry."""
        self.check_publisher()
        key = str(opmon_entry.origin.session)
        if opmon_entry.origin.application != "":
            key += "." + opmon_entry.origin.application
        if opmon_entry.origin.substructure:
            for substructure_id in opmon_entry.origin.substructure:
                key += "." + substructure_id
        key += "/" + str(opmon_entry.measurement)
        return key
 

publish()

None OpMonPublisher.OpMonPublisher.publish	(		self,
		Msg	message,
		dict[str, str] | None	custom_origin = None,
		int | str | None	level = None )

Send an OpMonEntry to Kafka.

Reimplemented from opmonlib.publisher_base.OpMonPublisherBase.

Definition at line 61 of file OpMonPublisher.py.

    ) -> None:
        """Send an OpMonEntry to Kafka."""
        if not isinstance(message, Msg):
            self.log.error("Passed message needs to be of type google.protobuf.message")
            return
        if not level:
            level = self.conf.level
        if isinstance(level, str):
            level = logging_log_level_from_str(level)
        if level < self.conf.level:
            return
 
        metric = self.to_entry(message=message, custom_origin=custom_origin)
 
        if len(metric.data) == 0:
            self.log.warning("OpMonEntry of type %s has no data", message.__name__)
            return
 
        target_topic = self.extract_topic(message)
        target_key = self.extract_key(metric)
 
        self.publisher.send(target_topic, value=metric, key=target_key)
        return

Member Data Documentation

conf

OpMonPublisher.OpMonPublisher.conf = conf

Definition at line 21 of file OpMonPublisher.py.

default_topic

str OpMonPublisher.OpMonPublisher.default_topic = "monitoring." + self.conf.topic

Definition at line 39 of file OpMonPublisher.py.

log

OpMonPublisher.OpMonPublisher.log = logging.getLogger("OpMonPublisher")

Definition at line 20 of file OpMonPublisher.py.

opmon_producer

OpMonPublisher.OpMonPublisher.opmon_producer = None

Definition at line 36 of file OpMonPublisher.py.

publisher

OpMonPublisher.OpMonPublisher.publisher

Initial value:

=  KafkaProducer(
            bootstrap_servers=conf.bootstrap,
            value_serializer=lambda v: v.SerializeToString(),
            key_serializer=lambda k: str(k).encode("utf-8"),
        )

Definition at line 40 of file OpMonPublisher.py.

---

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

• /github/workspace/dunedaq/sourcecode/kafkaopmon/python/kafkaopmon/OpMonPublisher.py