mpmetrics

multiprocess-safe metrics

mpmetrics implements metrics suitable for use with OpenMetrics. It provides multiprocess-safe replacements for prometheus_client’s Counter, Gauge, Summary, Histogram, and Enum. To use it, just import these classes from this module instead of from prometheus_client.

mpmetrics.Counter = <mpmetrics.metrics.CollectorFactory object>

A Counter tracks counts of events or running totals.

Example use cases for Counters:

  • Number of requests processed

  • Number of items that were inserted into a queue

  • Total amount of data that a system has processed

Counters can only go up (and are reset when the process restarts). If your use case can go down, you should use a Gauge instead.

An example for a Counter:

from mpmetrics import Counter

c = Counter('my_failures_total', 'Description of counter')
c.inc()     # Increment by 1
c.inc(1.6)  # Increment by given value

There are also utilities to count exceptions raised:

@c.count_exceptions()
def f():
    pass

with c.count_exceptions():
    pass

# Count only one type of exception
with c.count_exceptions(ValueError):
    pass

For more information about the parameters used when creating a Counter, refer to CollectorFactory.

Counter.inc(amount=1, exemplar=None)

Increment by the given amount.

Counter.count_exceptions(exception=<class 'Exception'>)

Count exceptions in a block of code or function.

Can be used as a function decorator or context manager. Increments the counter when an exception of the given type is raised up out of the code.

mpmetrics.Gauge = <mpmetrics.metrics.CollectorFactory object>

Gauge metric, to report instantaneous values.

Examples of Gauges include:

  • In-progress requests

  • Number of items in a queue

  • Free memory

  • Total memory

  • Temperature

Gauges can go both up and down:

from mpmetrics import Gauge

g = Gauge('my_inprogress_requests', 'Description of gauge')
g.inc()      # Increment by 1
g.dec(10)    # Decrement by given value
g.set(4.2)   # Set to a given value

There are utilities for common use cases:

g.set_to_current_time()   # Set to current unix time

# Increment when entered, decrement when exited.
@g.track_inprogress()
def f():
    pass

with g.track_inprogress():
    pass

A Gauge can also take its value from a callback:

d = Gauge('data_objects', 'Number of objects')
my_dict = {}
d.set_function(lambda: len(my_dict))

For more information about the parameters used when creating a Gauge, refer to CollectorFactory.

Gauge.inc(amount=1)

Increment by the given amount.

Gauge.dec(amount=1)

Decrement by the given amount.

Gauge.set(amount)

Set to the given amount.

Gauge.set_to_current_time()

Set to the current time in seconds since the Epoch.

Gauge.track_inprogress()

Track in-progress blocks of code or functions.

Can be used as a function decorator or context manager. Increments the gauge when the code is entered, and decrements when it is exited.

Gauge.time()

Time a block of code or function, and set the duration in seconds.

Can be used as a function decorator or context manager.

mpmetrics.Summary = <mpmetrics.metrics.CollectorFactory object>

A Summary tracks the size and number of events.

Example use cases for Summaries:

  • Response latency

  • Request size

Example for a Summary:

from mpmetrics import Summary

s = Summary('request_size_bytes', 'Request size (bytes)')
s.observe(512)  # Observe 512 (bytes)

Example for a Summary using time:

from mpmetrics import Summary

REQUEST_TIME = Summary('response_latency_seconds', 'Response latency (seconds)')

@REQUEST_TIME.time()
def create_response(request):
  '''A dummy function'''
  time.sleep(1)

Example for using the same Summary object as a context manager:

with REQUEST_TIME.time():
    pass  # Logic to be timed

For more information about the parameters used when creating a Summary, refer to CollectorFactory.

Summary.observe(amount)

Observe the given amount.

The amount is usually positive or zero. Negative values are accepted but prevent current versions of Prometheus from properly detecting counter resets in the sum of observations. See https://prometheus.io/docs/practices/histograms/#count-and-sum-of-observations for details.

Summary.time()

Time a block of code or function, and observe the duration in seconds.

Can be used as a function decorator or context manager.

mpmetrics.Histogram = <mpmetrics.metrics.CollectorFactory object>

A Histogram tracks the size and number of events in buckets.

You can use Histograms for aggregatable calculation of quantiles.

Example use cases:

  • Response latency

  • Request size

Example for a Histogram:

from mpmetrics import Histogram

h = Histogram('request_size_bytes', 'Request size (bytes)')
h.observe(512)  # Observe 512 (bytes)

Example for a Histogram using time:

from mpmetrics import Histogram

REQUEST_TIME = Histogram('response_latency_seconds', 'Response latency (seconds)')

@REQUEST_TIME.time()
def create_response(request):
  '''A dummy function'''
  time.sleep(1)

Example of using the same Histogram object as a context manager:

with REQUEST_TIME.time():
    pass  # Logic to be timed

For more information about the parameters used when creating a Summary, refer to CollectorFactory.

The default buckets are intended to cover a typical web/rpc request from milliseconds to seconds. They can be overridden by passing buckets keyword argument to Histogram.

Histogram.observe(amount, exemplar=None)

Observe the given amount.

The amount is usually positive or zero. Negative values are accepted but prevent current versions of Prometheus from properly detecting counter resets in the sum of observations. See https://prometheus.io/docs/practices/histograms/#count-and-sum-of-observations for details.

Histogram.time()

Time a block of code or function, and observe the duration in seconds.

Can be used as a function decorator or context manager.

mpmetrics.Enum = <mpmetrics.metrics.CollectorFactory object>

Enum metric, which has one selected state in a set

Example usage:

from mpmetrics import Enum

e = Enum('task_state', 'Description of enum',
         states=['starting', 'running', 'stopped'])
e.state('running')

The first listed state will be the default.

For more information about the parameters used when creating a Summary, refer to CollectorFactory.

Enum.state(state)

Select a given state.

mpmetrics.metrics

Metric implementations

class mpmetrics.metrics.CollectorFactory(metric)

A factory for creating new metrics.

This class is used by metrics to create the appropriate collector based on the constructor arguments.

__call__(name, documentation, labelnames=(), namespace='', subsystem='', unit='', registry=<prometheus_client.registry.CollectorRegistry object>, **kwargs)

Create a new metric.

Parameters:

  • name (str) – The name of the metric

  • documentation (str) – Documentation for the metric. This will be displayed as a HELP comment before the metric.

  • labelnames (Iterable[str]) – A list of labels to be used with the metric

  • namespace (str) – A global namespace for the metric. This will be prepended to name.

  • subsystem (str) – A subsystem name for the metric. This will be prepended to name after namespace.

  • unit (str) – The unit of measurement for the metric. This will be appended to name.

  • registry (prometheus_client.registry.CollectorRegistry) – The registry to register this metric with. It will collect data from the metric.

  • **kwargs – Any additional arguments are passed to the metric itself.

Returns:

A new metric

Return type:

Option[Collector, LabeledCollector]

The name of the metric is roughly:

name = f"{namespace}_{subsystem}_{name}_{unit}"

with unnecessary underscores ommitted.

If labelnames is truthy, then a LabeledCollector for the metric will be returned. Otherwise a Collector will be returned.

class mpmetrics.metrics.Collector(metric, name, docs, registry, heap, kwargs)

A basic collector for non-labeled metrics.

Attributes are proxied to the underlying metric.

collect()

Collect samples from the metric

Returns:

An iterator yielding one metric with collected samples.

Return type:

Iterator[prometheus_client.metrics_core.Metric]

describe()

Describe the metric

Returns:

An iterator yielding one metric with no samples

Return type:

Iterator[prometheus_client.metrics_core.Metric]

class mpmetrics.metrics.LabeledCollector(mem, metric, name, docs, labelnames, registry, kwargs, heap)

A collector supporting labeled metrics.

labels() must be called to get individual metrics.

collect()

Collect samples from the metric

Returns:

An iterator yielding one metric with samples collected from each label.

Return type:

Iterator[prometheus_client.metrics_core.Metric]

describe()

Describe the metric

Returns:

An iterator yielding one metric with no samples

Return type:

Iterator[prometheus_client.metrics_core.Metric]

labels(*values, **labels)

Return the child for the given labelset.

All metrics can have labels, allowing grouping of related time series. Taking a counter as an example:

from mpmetrics import Counter

c = Counter('my_requests_total', 'HTTP Failures', ['method', 'endpoint'])
c.labels('get', '/').inc()
c.labels('post', '/submit').inc()

Labels can also be provided as keyword arguments:

from mpmetrics import Counter

c = Counter('my_requests_total', 'HTTP Failures', ['method', 'endpoint'])
c.labels(method='get', endpoint='/').inc()
c.labels(method='post', endpoint='/submit').inc()

See the best practices on naming and labels.

mpmetrics.flask

Import this module to monkey-patch prometheus_flask_exporter to use mpmetrics’ metrics.