GitHub

MetricProducer

PreviousNext

API reference for MetricProducer.

Collects the registry of context (captured when the producer was made) as MetricsData with resource. A delta producer remembers the values of the series it last collected, so give each reader its own fork().

Attributes

AttributeTypeDescription
resource= resource
context= context
temporalityTemporalityPreference = temporality

Functions

__init__

__init__(resource, context, temporality='cumulative') -> None
ParameterTypeDescription
resourceResource
contextContext
temporalityTemporalityPreference = 'cumulative'
def __init__(
    self,
    resource: Resource,
    context: Context,
    temporality: TemporalityPreference = "cumulative",
) -> None:
    self.resource = resource
    self.context = context
    self.temporality: TemporalityPreference = temporality
    self._start_times: dict[str, int] = {}
    self._start_time = time.time_ns()
    self._previous_export_time = self._start_time
    self._totals = _Totals()

fork

fork() -> MetricProducer

A fresh producer of the same registry, resource and temporality.

Returns MetricProducer

def fork(self) -> MetricProducer:
    """A fresh producer of the same registry, resource and temporality."""
    return MetricProducer(self.resource, self.context, self.temporality)

collect

collect() -> MetricsData

Every series of the registry now.

Returns MetricsData

def collect(self) -> MetricsData:
    """Every series of the registry now."""
    delta = self.temporality == "delta"
    run = _Collection(
        now=time.time_ns(),
        delta=delta,
        start_time=self._start_time,
        interval_start=self._previous_export_time if delta else self._start_time,
        start_times=self._start_times,
        previous=self._totals,
    )
    for snapshot in unsafe_snapshot(self.context):
        run.add(snapshot)
    # Only series in this snapshot carry their delta state forward, so a
    # removed series does not report a negative delta when recreated.
    self._totals = run.totals
    if delta:
        self._previous_export_time = run.now
    return MetricsData(
        [ResourceMetrics(self.resource, [ScopeMetrics(_SCOPE, run.metrics(), "")], "")]
    )