Metrics

The Metrics API provides a preconfigured client for the StatsD protocol, as well as several useful utilities for reporting metrics.

Basic Usage

>>> from canvas_core import metrics

# Get a preconfigured StatsD client.
>>> statsd = metrics.get_statsd_client()

# Use any method that the StatsD interface provides.
>>> statsd.incr("my.metric.identifier")
>>> statsd.timing("my.metric.timing", 10)
>>> ...

# Use the mkstat utility to add tags to your metrics.
>>> stat = metrics.mkstat("my.metric.identifier", tags={"my": "tag", "another": "tag"})
>>> statsd.incr(stat)

# Use the measure context manager to automatically measure blocks of code.
#
# The context manager will automatically measure call count, successful
# calls, errored calls, and executing timing.
#
>>> with measure("my.function.call", extra_tags={"my": "tag"}):
...    my_function()

# You can also used the @measured decorator on any function, which applies
# the measure context manager to the function call. It automatically tags
# the metrics with the import path of the decorated function.
>>> @metrics.measured
... def my_function() -> None:
...     pass

# The @measured decorator accepts the same keyword arguments as the measure
# context manager.
>>> @metrics.measured(extra_tags={"my": "tags"})
... def my_function() -> None:
...     pass

Metrics reported by the measure context manager

The metrics collected by the measured context manager are:

metrics.measurements.executions

The number of times the block within the context manager was executed.

Each execution is tagged with a status tag (either success for successful executions, or error for executions that resulted in an exception being raised).

metrics.measurements.timings

The amount of time elapsed during execution of the block (in milliseconds).

Each timing is tagged with a status tag (either success for successful executions, or error for executions that resulted in an exception being raised).

API Reference

canvas_core.metrics.get_statsd_client(*args: Any, **kwargs: Any) → StatsClient

Return a configured StatsD client.

canvas_core.metrics.measure(name: str, extra_tags: dict[str, str] | None = None, client: StatsClientBase | None = None) → Generator[PipelineBase, None, None]

A context manager for collecting metrics about a context block.

Parameters:

• name – The name of the block being measured (added as a StatsD tag)

• extra_tags – A dict of extra tags to be added to all recorded metrics.

• client – An optional alternate StatsD client.

Yields:

A pipeline for collecting additional metrics in the same batch.

canvas_core.metrics.measured(fn: F | None = None, **options: Any) → Callable[[F], F] | F

Collect metrics about the decorated function.

Parameters:

• fn – The decorated function.

• client – An optional alternate StatsD client.

Returns:

A decorated function if called without arguments (@measured), or a decorator if called with arguments (@measured(client=…))

canvas_core.metrics.mkstat(metric_name: str, tags: dict[str, str] | None = None) → str

Generate a valid statsd string from the given metric name and tags.

>>> _statsd_string("metric.name", tags={"my": "tag1", "another": "tag2"})
metric.name,my=tag1,another=tag2