Events and metrics

This page documents the base API for defining metric events and metrics.

Public API

class mvx.common.metrics.MetricEvent

Base class for metric event objects.

A metric event describes something that happened in runtime code and may affect one or more registered metrics. Recorders dispatch these events to metrics, while metrics decide whether a particular event is relevant to their own state.

abstract property event_type: str

Return the stable event type.

The event type identifies the kind of runtime occurrence represented by this event. Metrics use it to decide whether and how the event should affect their values.

Returns:

stable event type.

class mvx.common.metrics.Metric

Base class for metric value objects.

A metric owns one measured value or a related group of measured values. It receives MetricEvent objects from a recorder and decides whether each event affects its internal state.

Implementations are responsible for keeping their own state consistent and for returning a snapshot that can be consumed by monitoring, logging, tests, or diagnostics.

abstract property metric_name: str

Return the stable metric name.

The name identifies this metric inside a recorder and should not depend on the current metric value.

Returns:

stable metric name.

abstractmethod handle_event(event)

Apply a metric event to this metric.

The recorder calls this method for registered metrics when a new MetricEvent is dispatched. The method should update internal state only when the event is relevant to this metric.

Parameters:

event (MetricEvent) – metric event to apply.

Return type:

bool

Returns:

True if the metric value changed, False otherwise.

abstractmethod snapshot()

Return a serializable snapshot of the current metric value.

The returned mapping should contain log- and diagnostics-friendly values that describe the current metric state.

Return type:

Mapping[str, Any]

Returns:

current metric snapshot.

Minimal event example

A metric event is usually a small immutable object with an event_type and domain payload.

from dataclasses import dataclass
from enum import StrEnum

from mvx.common.metrics import MetricEvent


class DocumentSaveAttemptOutcome(StrEnum):
    SUCCESS = "SUCCESS"
    FAILURE = "FAILURE"


@dataclass(frozen=True, slots=True)
class DocumentSaveAttemptMetricEvent(MetricEvent):
    outcome: DocumentSaveAttemptOutcome

    @property
    def event_type(self) -> str:
        return "document_storage.save.attempt"

Minimal metric example

A metric accepts relevant events, updates its own state, and exposes a snapshot.

from typing import Any, Mapping

from mvx.common.metrics import Metric, MetricEvent


class DocumentSaveAttemptsMetric(Metric):
    def __init__(self) -> None:
        self._total = 0
        self._success_total = 0
        self._failure_total = 0

    @property
    def metric_name(self) -> str:
        return "document_storage.save.attempts"

    def handle_event(self, event: MetricEvent) -> bool:
        if not isinstance(event, DocumentSaveAttemptMetricEvent):
            return False

        self._total += 1

        if event.outcome is DocumentSaveAttemptOutcome.SUCCESS:
            self._success_total += 1

        elif event.outcome is DocumentSaveAttemptOutcome.FAILURE:
            self._failure_total += 1

        return True

    def snapshot(self) -> Mapping[str, Any]:
        return {
            "name": self.metric_name,
            "dimensions": {
                "total": self._total,
                "success_total": self._success_total,
                "failure_total": self._failure_total,
            },
        }

Contract notes

MetricEvent.event_type should be stable.

Metric.metric_name should be stable because recorder snapshots use it as the metric key.

Metric.handle_event() returns True only when the metric accepted the event and changed its state. Recorders use this return value for post-change processing.

Metric.snapshot() is the public read surface of the metric state. External code should use snapshots instead of metric internals.