CMDP Module

SPDX-FileCopyrightText: 2024 DESY and the Constellation authors SPDX-License-Identifier: EUPL-1.2

Module implementing the Constellation Monitoring Distribution Protocol.

class core.cmdp.CMDP1Header(name: str)

Bases: object

Class implementing a Constellation CMDP1 message header.

decode(header: Any) → tuple[str, Timestamp, dict[str, Any] | None]

Decode header string and return host, timestamp and meta map.

encode(meta: dict[str, Any] | None = None) → memoryview

Generate and return a header as list.

Additional keyword arguments are required for protocols specifying additional fields.

protocol = 'CMDP\x01'

send(socket: ~zmq.sugar.socket.Socket, flags: int = <Flag.SNDMORE: 2>, meta: dict[str, ~typing.Any] | None = None) → None

Send a message header via socket.

meta is an optional dictionary that is sent as a map of string/value pairs with the header.

class core.cmdp.CMDPPublisher(name: str, socket: Socket | None)

Bases: CMDPTransmitter

Class for publishing Constellation monitoring messages via ZMQ.

has_log_subscribers(record: LogRecord) → bool

Return whether or not we have subscribers for the given log topic.

has_metric_subscribers(metric_name: str) → bool

Return whether or not we have subscribers for the given metric data topic.

register_log(topic: str, description: str | None = '') → None

Register a LOG topic that subscribers should be notified about.

register_stat(topic: str, description: str | None = '') → None

Register a STAT topic that subscribers should be notified about.

update_subscriptions() → None

Receive a Constellation subscription message and handle that.

class core.cmdp.CMDPTransmitter(name: str, socket: Socket | None)

Bases: object

Class for sending Constellation monitoring messages via ZMQ.

close() → None

Close the socket.

closed() → bool

Return whether socket is closed or not.

recv(flags: int = 0) → LogRecord | Metric | Notification | None

Receive a Constellation monitoring message and return log or metric.

send(data: LogRecord | Metric) → None

Send a LogRecord or a Metric.

send_log(record: LogRecord) → None

Send a LogRecord via an ZMQ socket.

Follows the Constellation Monitoring Distribution Protocol.

record: LogRecord to send.

send_metric(metric: Metric) → None

Send a metric via a ZMQ socket.

class core.cmdp.Metric(name: str, unit: str, value: Any = None)

Bases: object

Class to hold information for a Constellation metric.

pack() → bytes

Pack metric for CMDP payload.

class core.cmdp.Notification(prefix: str)

Bases: object

Class to hold information for a Constellation CMDP notification.

core.cmdp.decode_log(name: str, topic: str, msg: list[bytes]) → LogRecord

Receive a Constellation log message.

core.cmdp.decode_metric(name: str, topic: str, msg: list[Any]) → Metric

Receive a Constellation STATS message and return a Metric.

core.cmdp.decode_notification(name: str, topic: str, msg: list[bytes]) → Notification

Receive a Constellation log message.