tbp.monty.telemetry#

Structured telemetry framework built upon the logging module.

Provides a structured telemetry module that emits inline telemetry events routed through standard Python logging mechanics. Telemetry schemas are passed as the log message (msg) to logging.Logger.log, so handlers receive them as record.msg. record.getMessage() returns schema.kind.

The telemetry level must be configured via the experiment config YAML. Easiest is adding “ - /telemetry: info” under “defaults:”. Available configs are “info” and “warning”.

The global level is defined via the telemetry.tbp.monty logger. It can be overridden on a per-module basis.

Config example:

experiment:
  config:
    telemetry:
      loggers:
        # Global level
        telemetry.tbp.monty:
          level: CRITICAL
        # Module-specific level
        telemetry.tbp.monty.frameworks.models.graph_matching:
          level: INFO

Usage example:

from tbp.monty import telemetry
from tbp.monty.telemetry.schemas import TelemetryEvent

telemeter = telemetry.getTelemeter(__name__)
telemeter.info(TelemetryEvent(kind="CustomEvent", your_key="your_value", ...))
telemeter.debug(TelemetryEvent(kind="DebugEvent", ...))

Handler example:

class MyHandler(logging.Handler):
    def emit(self, record):
        event = record.msg  # the TelemetryEvent instance
        ...
getTelemeter(name: str) → TelemetryPublisher[source]#

Returns a telemetry logger with the specified name.

This method is essentially a wrapper for logger.getLogger. It prefixes the logger name with telemetry. if not present, which routes logger creation through _TelemetryLoggerFactory to return a TelemetryPublisher instance.

All calls to this function with a given name return the same logger instance.

Example:

telemeter = telemetry.getTelemeter(__name__)
telemeter.info(TelemetryEvent(...))
Return type:

TelemetryPublisher

Returns:

The logger instance of the telemetry publisher.

Raises:

TypeError – If the logger factory fails to return a telemetry publisher.

tbp.monty.telemetry.formatters#

class JsonFormatter(fmt=None, datefmt=None, style='%', validate=True)[source]#

Bases: Formatter

Serializes a telemetry LogRecord instance as a JSON string.

format(record: LogRecord) → str[source]#

Format the specified record as text.

The record’s attribute dictionary is used as the operand to a string formatting operation which yields the returned string. Before formatting the dictionary, a couple of preparatory steps are carried out. The message attribute of the record is computed using LogRecord.getMessage(). If the formatting string uses the time (as determined by a call to usesTime(), formatTime() is called to format the event time. If there is exception information, it is formatted using formatException() and appended to the message.

Return type:

str

tbp.monty.telemetry.publishers#

class TelemetryPublisher(*args, **kwargs) → None[source]#

Bases: Logger

Structured telemetry publisher.

Subclasses logging.Logger and emits TelemetrySchema as structured LogRecord instances routed through the logging pipeline to telemetry handlers.

The TelemetryEvent is passed as the log message, so it is available to handlers as record.msg; record.getMessage() returns event.kind.

Do not instantiate this class directly; obtain it via telemetry.getTelemeter.

Example:

telemeter = telemetry.getTelemeter(__name__)
telemeter.info(TelemetryEvent(...))
__init__(*args, **kwargs) → None[source]#

Initializes the logger; do not instantiate this class outside its module.

critical(msg: object, *args, **kwargs) → None[source]#

Emits a telemetry event at CRITICAL log level.

Parameters:

  • msg (object) – The TelemetryEvent instance.

  • *args – Passed forward to Logger.log method.

  • **kwargs – Passed forward to Logger.log method.

Return type:

None

debug(msg: object, *args, **kwargs) → None[source]#

Emits a telemetry event at DEBUG log level.

Parameters:

  • msg (object) – The TelemetryEvent instance.

  • *args – Passed forward to Logger.log method.

  • **kwargs – Passed forward to Logger.log method.

Return type:

None

emit(level: int, event: TelemetryEvent, *args, **kwargs) → None[source]#

Emits a telemetry event at the specified log level.

Equivalent of log method, type-hinted for convenience.

Parameters:

  • level (int) – The log level.

  • event (TelemetryEvent) – The TelemetryEvent instance.

  • *args – Passed forward to Logger.log method.

  • **kwargs – Passed forward to Logger.log method.

Return type:

None

error(msg: object, *args, **kwargs) → None[source]#

Emits a telemetry event at ERROR log level.

Parameters:

  • msg (object) – The TelemetryEvent instance.

  • *args – Passed forward to Logger.log method.

  • **kwargs – Passed forward to Logger.log method.

Return type:

None

exception(msg: object, *args, exc_info=True, **kwargs) → None[source]#

Emits a telemetry event at ERROR log level with exception info attached.

Parameters:

  • msg (object) – The TelemetryEvent instance.

  • *args – Passed forward to Logger.log method.

  • exc_info – Passed forward to Logger.log method.

  • **kwargs – Passed forward to Logger.log method.

Return type:

None

fatal(msg: object, *args, **kwargs) → None#

Emits a telemetry event at CRITICAL log level.

Parameters:

  • msg (object) – The TelemetryEvent instance.

  • *args – Passed forward to Logger.log method.

  • **kwargs – Passed forward to Logger.log method.

Return type:

None

info(msg: object, *args, **kwargs) → None[source]#

Emits a telemetry event at INFO log level.

Parameters:

  • msg (object) – The TelemetryEvent instance.

  • *args – Passed forward to Logger.log method.

  • **kwargs – Passed forward to Logger.log method.

Return type:

None

log(level: int, msg: object, *args, **kwargs) → None[source]#

Emits a telemetry event at the specified log level.

Parameters:

  • level (int) – The log level.

  • msg (object) – The TelemetryEvent instance.

  • *args – Passed forward to Logger.log method.

  • **kwargs – Passed forward to Logger.log method.

Return type:

None

warn(msg: object, *args, **kwargs) → None[source]#

Emits a telemetry event at WARNING log level.

Parameters:

  • msg (object) – The TelemetryEvent instance.

  • *args – Passed forward to Logger.log method.

  • **kwargs – Passed forward to Logger.log method.

Return type:

None

warning(msg: object, *args, **kwargs) → None[source]#

Emits a telemetry event at WARNING log level.

Parameters:

  • msg (object) – The TelemetryEvent instance.

  • *args – Passed forward to Logger.log method.

  • **kwargs – Passed forward to Logger.log method.

Return type:

None

tbp.monty.telemetry.schemas#

class TelemetryEvent(**data: Any) → None[source]#

Bases: TelemetrySchema

Base model class for telemetry events; carries instantaneous data changes.

kind: Annotated[str, Field(validate_default=True)]#

Schema identifier used for event filtering by subscribed handlers. It is also used as the log message by loggers associated with telemetry. If empty, defaults to schema class name.

model_config: ClassVar[ConfigDict] = {'extra': 'allow'}#

Allows adding extra attributes to the model.

class TelemetrySchema(**data: Any) → None[source]#

Bases: BaseModel

Base model class for all telemetry schemas.

Subclasses add fields for their payload.

classmethod validate_kind(value: str) → str[source]#
Return type:

str

VERSION: Final[int] = 1#

Schema version number, incremented on backwards-incompatible field changes. For use by a model validator or discriminated union.

kind: Annotated[str, Field(validate_default=True)]#

Schema identifier used for event filtering by subscribed handlers. It is also used as the log message by loggers associated with telemetry. If empty, defaults to schema class name.

model_config: ClassVar[ConfigDict] = {}#

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

tbp.monty.telemetry.telemeter#

getTelemeter(name: str) → TelemetryPublisher[source]#

Returns a telemetry logger with the specified name.

This method is essentially a wrapper for logger.getLogger. It prefixes the logger name with telemetry. if not present, which routes logger creation through _TelemetryLoggerFactory to return a TelemetryPublisher instance.

All calls to this function with a given name return the same logger instance.

Example:

telemeter = telemetry.getTelemeter(__name__)
telemeter.info(TelemetryEvent(...))
Return type:

TelemetryPublisher

Returns:

The logger instance of the telemetry publisher.

Raises:

TypeError – If the logger factory fails to return a telemetry publisher.