GitHub

metrics

PreviousNext

API reference for metrics.

pyfect metrics as OpenTelemetry metrics (Effect's OtelMetrics).

layer(lambda: reader) hands the current metric registry to an SDK MetricReader (a PeriodicExportingMetricReader exporting over OTLP, say), which collects it whenever it reads::

metrics = otel.metrics.layer(lambda: PeriodicExportingMetricReader(exporter))
program.pipe(fx.provide(metrics.pipe(layer.provide(otel.resource.layer_empty))))

Readers are bound to pyfect's producer instead of a MeterProvider; a reader can be bound once. Counters, frequencies and summaries become sums, gauges gauges, histograms explicit-bucket histograms.

Classes

MetricProducer

Functions

make_producer

make_producer(temporality='cumulative') -> Effect[MetricProducer, Never, OtelResource]

A producer of the current metric registry, with the OtelResource.

ParameterTypeDescription
temporalityTemporalityPreference = 'cumulative'

Returns Effect[MetricProducer, Never, OtelResource]

def make_producer(
    temporality: TemporalityPreference = "cumulative",
) -> Effect[MetricProducer, Never, OtelResource]:
    """A producer of the current metric registry, with the ``OtelResource``."""

    @gen
    def make():
        resource = yield from OtelResource()
        services = yield from context()
        return MetricProducer(resource, services, temporality)

    return make

register_producer

register_producer(producer, readers, *, shutdown_timeout=DEFAULT_SHUTDOWN_TIMEOUT) -> Effect[list[MetricReader], Never, Scope]

Bind each reader to its own fork of producer, and shut them down when the scope closes. If a reader cannot be bound (it is bound already), the readers bound so far are shut down and the effect dies.

ParameterTypeDescription
producerMetricProducer
readersReaders
shutdown_timeoutDuration = DEFAULT_SHUTDOWN_TIMEOUT

Returns Effect[list[MetricReader], Never, Scope]

def register_producer(
    producer: MetricProducer,
    readers: Readers,
    *,
    shutdown_timeout: Duration = DEFAULT_SHUTDOWN_TIMEOUT,
) -> Effect[list[MetricReader], Never, Scope]:
    """Bind each reader to its own fork of ``producer``, and shut them down
    when the scope closes. If a reader cannot be bound (it is bound
    already), the readers bound so far are shut down and the effect dies."""

    def acquire() -> Effect[list[MetricReader]]:
        made = readers()
        given = [made] if isinstance(made, MetricReader) else list(made)
        bound: list[MetricReader] = []
        try:
            for reader in given:
                _bind(reader, producer.fork())
                bound.append(reader)
        except Exception as error:
            failure = error
            return _shut_down(bound, shutdown_timeout).flat_map(lambda _: die(failure))
        return sync(lambda: bound)

    return acquire_release(
        suspend(acquire), lambda bound, _exit: _shut_down(bound, shutdown_timeout)
    )

layer

layer(readers, *, temporality='cumulative', shutdown_timeout=DEFAULT_SHUTDOWN_TIMEOUT) -> Layer[Never, Never, OtelResource]

Bind the readers readers() makes to a producer of the current registry while the layer is in use.

ParameterTypeDescription
readersReaders
temporalityTemporalityPreference = 'cumulative'
shutdown_timeoutDuration = DEFAULT_SHUTDOWN_TIMEOUT

Returns Layer[Never, Never, OtelResource]

def layer(
    readers: Readers,
    *,
    temporality: TemporalityPreference = "cumulative",
    shutdown_timeout: Duration = DEFAULT_SHUTDOWN_TIMEOUT,
) -> Layer[Never, Never, OtelResource]:
    """Bind the readers ``readers()`` makes to a producer of the current
    registry while the layer is in use."""
    registered: Effect[Any, Never, OtelResource | Scope] = make_producer(temporality).flat_map(
        lambda producer: register_producer(producer, readers, shutdown_timeout=shutdown_timeout)
    )
    return _layer.effect_discard(registered)