remotivelabs.topology.metrics

class Metric(builtins.str, enum.Enum):

Names of every metric the library can emit.

FRAMES_RECEIVED = <Metric.FRAMES_RECEIVED: 'frames.received'>
HANDLER_DURATION = <Metric.HANDLER_DURATION: 'handler.duration'>
class Metrics(typing.Protocol):

Sink for metric observations. Backends implement this protocol.

Metrics(*args, **kwargs)
def increment( self, metric: Metric, tags: Mapping[str, str]) -> None:

Increment a counter by one.

def counter( self, metric: Metric, tags: Mapping[str, str], value: int) -> None:

Add value to a counter.

def timing( self, metric: Metric, tags: Mapping[str, str], seconds: float) -> None:

Record a histogram observation in seconds.

def timed( self, metric: Metric, tags: Mapping[str, str]) -> contextlib.AbstractContextManager[None]:

Context manager that records elapsed seconds on exit.

class NoOpMetrics:

Default Metrics implementation — records nothing.

def increment( self, metric: Metric, tags: Mapping[str, str]) -> None:
def counter( self, metric: Metric, tags: Mapping[str, str], value: int) -> None:
def timing( self, metric: Metric, tags: Mapping[str, str], seconds: float) -> None:
def timed( self, metric: Metric, tags: Mapping[str, str]) -> contextlib.AbstractContextManager[None]:
Tags = typing.Mapping[str, str]
def get_metrics() -> Metrics:

Return the process-wide Metrics sink.

Without METRICS_PORT set, returns a no-op sink with zero overhead. With METRICS_PORT set, lazily configures an OpenTelemetry MeterProvider with a Prometheus exporter and serves /metrics on that port.

Raises:
  • ImportError: If METRICS_PORT is set but the metrics extras are not installed (opentelemetry / prometheus_client).
  • ValueError: If METRICS_PORT is not a valid port number.
Note:

By default, enabling metrics unregisters prometheus_client's default python_*/process_*/GC collectors so scrapes contain only application metrics. Set METRICS_KEEP_PROMETHEUS_DEFAULTS=1 to keep them.