opmonlib.publisher.OpMonPublisher Class Reference

Inheritance diagram for opmonlib.publisher.OpMonPublisher:

Collaboration diagram for opmonlib.publisher.OpMonPublisher:

Public Member Functions
None	__init__ (self, OpMonConf conf, int|str log_level=logging.INFO, bool rich_handler=False)
None	publish_message (self, logging.Logger logger, int|str level, str message)
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	__init__ (self)
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)

Additional Inherited Members
Public Attributes inherited from opmonlib.publisher_base.OpMonPublisherBase
	ts = Timestamp()
	substructure = None
	log
	conf
	publisher
	default_topic

Detailed Description

Publish operational monitoring metrics to file or stream.

Definition at line 17 of file publisher.py.

Constructor & Destructor Documentation

__init__()

None opmonlib.publisher.OpMonPublisher.__init__	(		self,
		OpMonConf	conf,
		int | str	log_level = logging.INFO,
		bool	rich_handler = False )

Construct the object to publish OpMon metrics to stdout.

Definition at line 20 of file publisher.py.

    ) -> None:
        """Construct the object to publish OpMon metrics to stdout."""
        super().__init__()
        self.log = logging.getLogger("OpMonPublisher")
        if not isinstance(log_level, int):
            log_level = self.log_level_to_int(log_level)
        self.log.setLevel(log_level)
        self.log.addHandler(setup_rich_handler())
 
        self.conf = conf
        self.conf.level = self.log_level_to_int(self.conf.level)
 
        if self.conf.opmon_type == "stdout":
            if rich_handler:
                handler = setup_rich_handler()
            else:
                handler = logging.StreamHandler(sys.stdout)
                handler.setFormatter(LoggingFormatter(fmt=full_log_format))
        elif self.conf.opmon_type == "file":
            self.conf.path = extract_opmon_file_path(self.conf.path)
            handler = logging.FileHandler(self.conf.path)
        elif self.conf.opmon_type == "stream":
            self.log.error("Type must not be stream to use file or stdout handling.")
            sys.exit(1)
        else:
            self.log.error("Unsupported OpMon type.")
            sys.exit(1)
 
        self.default_topic = "monitoring." + self.conf.topic
        self.publisher = logging.getLogger(self.default_topic)
        self.publisher.addHandler(handler)
        self.publisher.setLevel(self.conf.level)
 
        super().__post_init__()
        return
 

Member Function Documentation

publish()

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

Publish the message to either a file or the terminal.

Reimplemented from opmonlib.publisher_base.OpMonPublisherBase.

Definition at line 69 of file publisher.py.

    ) -> None:
        """Publish the message to either a file or the terminal."""
        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 not isinstance(level, int):
            level = self.log_level_to_int(level)
        if level < self.conf.level:
            return
 
        metric = self.to_entry(message=message, custom_origin=custom_origin)
        target_topic = self.extract_topic(message)
        publishing_logger = logging.getLogger(f"{self.publisher.name}.{target_topic}")
        self.publish_message(publishing_logger, level, MessageToJson(metric))
        return

publish_message()

None opmonlib.publisher.OpMonPublisher.publish_message	(		self,
		logging.Logger	logger,
		int | str	level,
		str	message )

Log the metric with the appropriate level.

Definition at line 61 of file publisher.py.

    ) -> None:
        """Log the metric with the appropriate level."""
        method = getattr(logger, self.log_level_to_str(level).lower(), logger.info)
        method(message)
        return
 

---

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

• /github/workspace/dunedaq/sourcecode/opmonlib/python/opmonlib/publisher.py