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.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, handling: MetricsType, value: Any = None)

Bases: object

Class to hold information for a Constellation metric.

pack() → bytes

Pack metric for CMDP payload.

class core.cmdp.MetricsType(*values)

Bases: Enum

ACCUMULATE = 2

AVERAGE = 3

LAST_VALUE = 1

RATE = 4

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.