admin/SolarManager
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
05ce19447826dab412f04ae44623d52fe6630ec3
SolarManager/mysql/opentelemetry/sdk/metrics/_internal/aggregation.py
T
adminandClaude Opus 5 79843aa2ae SolarManager unter Versionsverwaltung 
Erster Stand der Hintergrundprozesse, die auf der Synology unter
/volume1/homes/wagner/SolarManager laufen: der Manager selbst, die Sammler
je Geraet, die MQTT-Bruecke, der Wecker und - neu hinzugezogen - der
AutoAction-Runner, der als Hintergrundprozess hierher gehoert und nicht ins
Web-Verzeichnis.

Zugangsdaten stehen nicht mehr im Quelltext, sondern in config.ini, die
nicht mit eingecheckt wird. Vorlage ist config.ini.example, gelesen wird sie
von konfig.py. Betroffen waren solarManager.py (Datenbank und Wattpilot),
zeit.py, gatherWaterData.py, wecker.py und skoda_testdaten.py, das sich das
Passwort bisher aus dem Quelltext eines anderen Moduls herausgesucht hat.

Die Kia-Anbindung ist mit dem Fahrzeug entfallen: kiaTest.py,
gatherCarData.py und hyundai_kia_connect_api sind nicht mehr dabei, ebenso
gatherInverterData.py, auf das nur noch eine auskommentierte Zeile zeigte.

Die mitgelieferten Bibliotheken bleiben im Repository - die NAS hat kein
pip, sie muessen neben den Skripten liegen.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-09-02 20:46:59 +02:00

1035 lines
34 KiB
Python
Raw Blame History

# Copyright The OpenTelemetry Authors
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# pylint: disable=too-many-lines
from abc import ABC, abstractmethod
from bisect import bisect_left
from enum import IntEnum
from logging import getLogger
from math import inf
from threading import Lock
from typing import Generic, List, Optional, Sequence, TypeVar
from mysql.opentelemetry.metrics import (
Asynchronous,
Counter,
Histogram,
Instrument,
ObservableCounter,
ObservableGauge,
ObservableUpDownCounter,
Synchronous,
UpDownCounter,
)
from mysql.opentelemetry.sdk.metrics._internal.exponential_histogram.buckets import (
Buckets,
)
from mysql.opentelemetry.sdk.metrics._internal.exponential_histogram.mapping.exponent_mapping import (
ExponentMapping,
)
from mysql.opentelemetry.sdk.metrics._internal.exponential_histogram.mapping.logarithm_mapping import (
LogarithmMapping,
)
from mysql.opentelemetry.sdk.metrics._internal.measurement import Measurement
from mysql.opentelemetry.sdk.metrics._internal.point import (
Buckets as BucketsPoint,
ExponentialHistogramDataPoint,
Gauge,
Histogram as HistogramPoint,
HistogramDataPoint,
NumberDataPoint,
Sum,
)
from mysql.opentelemetry.util.types import Attributes
_DataPointVarT = TypeVar("_DataPointVarT", NumberDataPoint, HistogramDataPoint)
_logger = getLogger(__name__)
class AggregationTemporality(IntEnum):
"""
The temporality to use when aggregating data.
Can be one of the following values:
"""
UNSPECIFIED = 0
DELTA = 1
CUMULATIVE = 2
class _Aggregation(ABC, Generic[_DataPointVarT]):
def __init__(self, attributes: Attributes):
self._lock = Lock()
self._attributes = attributes
self._previous_point = None
@abstractmethod
def aggregate(self, measurement: Measurement) -> None:
pass
@abstractmethod
def collect(
self,
aggregation_temporality: AggregationTemporality,
collection_start_nano: int,
) -> Optional[_DataPointVarT]:
pass
class _DropAggregation(_Aggregation):
def aggregate(self, measurement: Measurement) -> None:
pass
def collect(
self,
aggregation_temporality: AggregationTemporality,
collection_start_nano: int,
) -> Optional[_DataPointVarT]:
pass
class _SumAggregation(_Aggregation[Sum]):
def __init__(
self,
attributes: Attributes,
instrument_is_monotonic: bool,
instrument_temporality: AggregationTemporality,
start_time_unix_nano: int,
):
super().__init__(attributes)
self._start_time_unix_nano = start_time_unix_nano
self._instrument_temporality = instrument_temporality
self._instrument_is_monotonic = instrument_is_monotonic
if self._instrument_temporality is AggregationTemporality.DELTA:
self._value = 0
else:
self._value = None
def aggregate(self, measurement: Measurement) -> None:
with self._lock:
if self._value is None:
self._value = 0
self._value = self._value + measurement.value
def collect(
self,
aggregation_temporality: AggregationTemporality,
collection_start_nano: int,
) -> Optional[NumberDataPoint]:
"""
Atomically return a point for the current value of the metric and
reset the aggregation value.
"""
if self._instrument_temporality is AggregationTemporality.DELTA:
with self._lock:
value = self._value
start_time_unix_nano = self._start_time_unix_nano
self._value = 0
self._start_time_unix_nano = collection_start_nano
else:
with self._lock:
if self._value is None:
return None
value = self._value
self._value = None
start_time_unix_nano = self._start_time_unix_nano
current_point = NumberDataPoint(
attributes=self._attributes,
start_time_unix_nano=start_time_unix_nano,
time_unix_nano=collection_start_nano,
value=value,
)
if self._previous_point is None or (
self._instrument_temporality is aggregation_temporality
):
# Output DELTA for a synchronous instrument
# Output CUMULATIVE for an asynchronous instrument
self._previous_point = current_point
return current_point
if aggregation_temporality is AggregationTemporality.DELTA:
# Output temporality DELTA for an asynchronous instrument
value = current_point.value - self._previous_point.value
output_start_time_unix_nano = self._previous_point.time_unix_nano
else:
# Output CUMULATIVE for a synchronous instrument
value = current_point.value + self._previous_point.value
output_start_time_unix_nano = self._previous_point.start_time_unix_nano
current_point = NumberDataPoint(
attributes=self._attributes,
start_time_unix_nano=output_start_time_unix_nano,
time_unix_nano=current_point.time_unix_nano,
value=value,
)
self._previous_point = current_point
return current_point
class _LastValueAggregation(_Aggregation[Gauge]):
def __init__(self, attributes: Attributes):
super().__init__(attributes)
self._value = None
def aggregate(self, measurement: Measurement):
with self._lock:
self._value = measurement.value
def collect(
self,
aggregation_temporality: AggregationTemporality,
collection_start_nano: int,
) -> Optional[_DataPointVarT]:
"""
Atomically return a point for the current value of the metric.
"""
with self._lock:
if self._value is None:
return None
value = self._value
self._value = None
return NumberDataPoint(
attributes=self._attributes,
start_time_unix_nano=0,
time_unix_nano=collection_start_nano,
value=value,
)
class _ExplicitBucketHistogramAggregation(_Aggregation[HistogramPoint]):
def __init__(
self,
attributes: Attributes,
start_time_unix_nano: int,
boundaries: Sequence[float] = (
0.0,
5.0,
10.0,
25.0,
50.0,
75.0,
100.0,
250.0,
500.0,
750.0,
1000.0,
2500.0,
5000.0,
7500.0,
10000.0,
),
record_min_max: bool = True,
):
super().__init__(attributes)
self._boundaries = tuple(boundaries)
self._bucket_counts = self._get_empty_bucket_counts()
self._min = inf
self._max = -inf
self._sum = 0
self._record_min_max = record_min_max
self._start_time_unix_nano = start_time_unix_nano
# It is assumed that the "natural" aggregation temporality for a
# Histogram instrument is DELTA, like the "natural" aggregation
# temporality for a Counter is DELTA and the "natural" aggregation
# temporality for an ObservableCounter is CUMULATIVE.
self._instrument_temporality = AggregationTemporality.DELTA
def _get_empty_bucket_counts(self) -> List[int]:
return [0] * (len(self._boundaries) + 1)
def aggregate(self, measurement: Measurement) -> None:
value = measurement.value
if self._record_min_max:
self._min = min(self._min, value)
self._max = max(self._max, value)
self._sum += value
self._bucket_counts[bisect_left(self._boundaries, value)] += 1
def collect(
self,
aggregation_temporality: AggregationTemporality,
collection_start_nano: int,
) -> Optional[_DataPointVarT]:
"""
Atomically return a point for the current value of the metric.
"""
with self._lock:
if not any(self._bucket_counts):
return None
bucket_counts = self._bucket_counts
start_time_unix_nano = self._start_time_unix_nano
sum_ = self._sum
max_ = self._max
min_ = self._min
self._bucket_counts = self._get_empty_bucket_counts()
self._start_time_unix_nano = collection_start_nano
self._sum = 0
self._min = inf
self._max = -inf
current_point = HistogramDataPoint(
attributes=self._attributes,
start_time_unix_nano=start_time_unix_nano,
time_unix_nano=collection_start_nano,
count=sum(bucket_counts),
sum=sum_,
bucket_counts=tuple(bucket_counts),
explicit_bounds=self._boundaries,
min=min_,
max=max_,
)
if self._previous_point is None or (
self._instrument_temporality is aggregation_temporality
):
self._previous_point = current_point
return current_point
max_ = current_point.max
min_ = current_point.min
if aggregation_temporality is AggregationTemporality.CUMULATIVE:
start_time_unix_nano = self._previous_point.start_time_unix_nano
sum_ = current_point.sum + self._previous_point.sum
# Only update min/max on delta -> cumulative
max_ = max(current_point.max, self._previous_point.max)
min_ = min(current_point.min, self._previous_point.min)
bucket_counts = [
curr_count + prev_count
for curr_count, prev_count in zip(
current_point.bucket_counts,
self._previous_point.bucket_counts,
)
]
else:
start_time_unix_nano = self._previous_point.time_unix_nano
sum_ = current_point.sum - self._previous_point.sum
bucket_counts = [
curr_count - prev_count
for curr_count, prev_count in zip(
current_point.bucket_counts,
self._previous_point.bucket_counts,
)
]
current_point = HistogramDataPoint(
attributes=self._attributes,
start_time_unix_nano=start_time_unix_nano,
time_unix_nano=current_point.time_unix_nano,
count=sum(bucket_counts),
sum=sum_,
bucket_counts=tuple(bucket_counts),
explicit_bounds=current_point.explicit_bounds,
min=min_,
max=max_,
)
self._previous_point = current_point
return current_point
# pylint: disable=protected-access
class _ExponentialBucketHistogramAggregation(_Aggregation[HistogramPoint]):
# _min_max_size and _max_max_size are the smallest and largest values
# the max_size parameter may have, respectively.
# _min_max_size is is the smallest reasonable value which is small enough
# to contain the entire normal floating point range at the minimum scale.
_min_max_size = 2
# _max_max_size is an arbitrary limit meant to limit accidental creation of
# giant exponential bucket histograms.
_max_max_size = 16384
def __init__(
self,
attributes: Attributes,
start_time_unix_nano: int,
# This is the default maximum number of buckets per positive or
# negative number range. The value 160 is specified by mysql.OpenTelemetry.
# See the derivation here:
# https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/metrics/sdk.md#exponential-bucket-histogram-aggregation)
max_size: int = 160,
):
super().__init__(attributes)
# max_size is the maximum capacity of the positive and negative
# buckets.
if max_size < self._min_max_size:
raise ValueError(
f"Buckets max size {max_size} is smaller than "
"minimum max size {self._min_max_size}"
)
if max_size > self._max_max_size:
raise ValueError(
f"Buckets max size {max_size} is larger than "
"maximum max size {self._max_max_size}"
)
self._max_size = max_size
# _sum is the sum of all the values aggregated by this aggregator.
self._sum = 0
# _count is the count of all calls to aggregate.
self._count = 0
# _zero_count is the count of all the calls to aggregate when the value
# to be aggregated is exactly 0.
self._zero_count = 0
# _min is the smallest value aggregated by this aggregator.
self._min = inf
# _max is the smallest value aggregated by this aggregator.
self._max = -inf
# _positive holds the positive values.
self._positive = Buckets()
# _negative holds the negative values by their absolute value.
self._negative = Buckets()
# _mapping corresponds to the current scale, is shared by both the
# positive and negative buckets.
self._mapping = LogarithmMapping(LogarithmMapping._max_scale)
self._instrument_temporality = AggregationTemporality.DELTA
self._start_time_unix_nano = start_time_unix_nano
self._previous_scale = None
self._previous_start_time_unix_nano = None
self._previous_sum = None
self._previous_max = None
self._previous_min = None
self._previous_positive = None
self._previous_negative = None
def aggregate(self, measurement: Measurement) -> None:
# pylint: disable=too-many-branches,too-many-statements, too-many-locals
with self._lock:
value = measurement.value
# 0. Set the following attributes:
# _min
# _max
# _count
# _zero_count
# _sum
if value < self._min:
self._min = value
if value > self._max:
self._max = value
self._count += 1
if value == 0:
self._zero_count += 1
# No need to do anything else if value is zero, just increment the
# zero count.
return
self._sum += value
# 1. Use the positive buckets for positive values and the negative
# buckets for negative values.
if value > 0:
buckets = self._positive
else:
# Both exponential and logarithm mappings use only positive values
# so the absolute value is used here.
value = -value
buckets = self._negative
# 2. Compute the index for the value at the current scale.
index = self._mapping.map_to_index(value)
# IncrementIndexBy starts here
# 3. Determine if a change of scale is needed.
is_rescaling_needed = False
if len(buckets) == 0:
buckets.index_start = index
buckets.index_end = index
buckets.index_base = index
elif (
index < buckets.index_start
and (buckets.index_end - index) >= self._max_size
):
is_rescaling_needed = True
low = index
high = buckets.index_end
elif (
index > buckets.index_end
and (index - buckets.index_start) >= self._max_size
):
is_rescaling_needed = True
low = buckets.index_start
high = index
# 4. Rescale the mapping if needed.
if is_rescaling_needed:
self._downscale(
self._get_scale_change(low, high),
self._positive,
self._negative,
)
index = self._mapping.map_to_index(value)
# 5. If the index is outside
# [buckets.index_start, buckets.index_end] readjust the buckets
# boundaries or add more buckets.
if index < buckets.index_start:
span = buckets.index_end - index
if span >= len(buckets.counts):
buckets.grow(span + 1, self._max_size)
buckets.index_start = index
elif index > buckets.index_end:
span = index - buckets.index_start
if span >= len(buckets.counts):
buckets.grow(span + 1, self._max_size)
buckets.index_end = index
# 6. Compute the index of the bucket to be incremented.
bucket_index = index - buckets.index_base
if bucket_index < 0:
bucket_index += len(buckets.counts)
# 7. Increment the bucket.
buckets.increment_bucket(bucket_index)
def collect(
self,
aggregation_temporality: AggregationTemporality,
collection_start_nano: int,
) -> Optional[_DataPointVarT]:
"""
Atomically return a point for the current value of the metric.
"""
# pylint: disable=too-many-statements, too-many-locals
with self._lock:
if self._count == 0:
return None
current_negative = self._negative
current_positive = self._positive
current_zero_count = self._zero_count
current_count = self._count
current_start_time_unix_nano = self._start_time_unix_nano
current_sum = self._sum
current_max = self._max
if current_max == -inf:
current_max = None
current_min = self._min
if current_min == inf:
current_min = None
if self._count == self._zero_count:
current_scale = 0
else:
current_scale = self._mapping.scale
self._negative = Buckets()
self._positive = Buckets()
self._start_time_unix_nano = collection_start_nano
self._sum = 0
self._count = 0
self._zero_count = 0
self._min = inf
self._max = -inf
current_point = ExponentialHistogramDataPoint(
attributes=self._attributes,
start_time_unix_nano=current_start_time_unix_nano,
time_unix_nano=collection_start_nano,
count=current_count,
sum=current_sum,
scale=current_scale,
zero_count=current_zero_count,
positive=BucketsPoint(
offset=current_positive.offset,
bucket_counts=current_positive.counts,
),
negative=BucketsPoint(
offset=current_negative.offset,
bucket_counts=current_negative.counts,
),
# FIXME: Find the right value for flags
flags=0,
min=current_min,
max=current_max,
)
if self._previous_scale is None or (
self._instrument_temporality is aggregation_temporality
):
self._previous_scale = current_scale
self._previous_start_time_unix_nano = current_start_time_unix_nano
self._previous_max = current_max
self._previous_min = current_min
self._previous_sum = current_sum
self._previous_positive = current_positive
self._previous_negative = current_negative
return current_point
min_scale = min(self._previous_scale, current_scale)
low_positive, high_positive = self._get_low_high_previous_current(
self._previous_positive, current_positive, min_scale
)
low_negative, high_negative = self._get_low_high_previous_current(
self._previous_negative, current_negative, min_scale
)
min_scale = min(
min_scale - self._get_scale_change(low_positive, high_positive),
min_scale - self._get_scale_change(low_negative, high_negative),
)
# FIXME Go implementation checks if the histogram (not the mapping
# but the histogram) has a count larger than zero, if not, scale
# (the histogram scale) would be zero. See exponential.go 191
self._downscale(
self._mapping.scale - min_scale,
self._previous_positive,
self._previous_negative,
)
if aggregation_temporality is AggregationTemporality.CUMULATIVE:
start_time_unix_nano = self._previous_start_time_unix_nano
sum_ = current_sum + self._previous_sum
# Only update min/max on delta -> cumulative
max_ = max(current_max, self._previous_max)
min_ = min(current_min, self._previous_min)
self._merge(
self._previous_positive,
current_positive,
current_scale,
min_scale,
aggregation_temporality,
)
self._merge(
self._previous_negative,
current_negative,
current_scale,
min_scale,
aggregation_temporality,
)
else:
start_time_unix_nano = self._previous_start_time_unix_nano
sum_ = current_sum - self._previous_sum
max_ = current_max
min_ = current_min
self._merge(
self._previous_positive,
current_positive,
current_scale,
min_scale,
aggregation_temporality,
)
self._merge(
self._previous_negative,
current_negative,
current_scale,
min_scale,
aggregation_temporality,
)
current_point = ExponentialHistogramDataPoint(
attributes=self._attributes,
start_time_unix_nano=start_time_unix_nano,
time_unix_nano=collection_start_nano,
count=current_count,
sum=sum_,
scale=current_scale,
zero_count=current_zero_count,
positive=BucketsPoint(
offset=current_positive.offset,
bucket_counts=current_positive.counts,
),
negative=BucketsPoint(
offset=current_negative.offset,
bucket_counts=current_negative.counts,
),
# FIXME: Find the right value for flags
flags=0,
min=min_,
max=max_,
)
self._previous_scale = current_scale
self._previous_positive = current_positive
self._previous_negative = current_negative
self._previous_start_time_unix_nano = current_start_time_unix_nano
self._previous_sum = current_sum
return current_point
def _get_low_high_previous_current(
self, previous_point_buckets, current_point_buckets, min_scale
):
(previous_point_low, previous_point_high) = self._get_low_high(
previous_point_buckets, min_scale
)
(current_point_low, current_point_high) = self._get_low_high(
current_point_buckets, min_scale
)
if current_point_low > current_point_high:
low = previous_point_low
high = previous_point_high
elif previous_point_low > previous_point_high:
low = current_point_low
high = current_point_high
else:
low = min(previous_point_low, current_point_low)
high = max(previous_point_high, current_point_high)
return low, high
def _get_low_high(self, buckets, min_scale):
if buckets.counts == [0]:
return 0, -1
shift = self._mapping._scale - min_scale
return buckets.index_start >> shift, buckets.index_end >> shift
def _get_scale_change(self, low, high):
change = 0
while high - low >= self._max_size:
high = high >> 1
low = low >> 1
change += 1
return change
def _downscale(self, change: int, positive, negative):
if change == 0:
return
if change < 0:
raise Exception("Invalid change of scale")
new_scale = self._mapping.scale - change
positive.downscale(change)
negative.downscale(change)
if new_scale <= 0:
mapping = ExponentMapping(new_scale)
else:
mapping = LogarithmMapping(new_scale)
self._mapping = mapping
def _merge(
self,
previous_buckets,
current_buckets,
current_scale,
min_scale,
aggregation_temporality,
):
current_change = current_scale - min_scale
for current_bucket_index, current_bucket in enumerate(current_buckets.counts):
if current_bucket == 0:
continue
# Not considering the case where len(previous_buckets) == 0. This
# would not happen because self._previous_point is only assigned to
# an ExponentialHistogramDataPoint object if self._count != 0.
index = (current_buckets.offset + current_bucket_index) >> current_change
if index < previous_buckets.index_start:
span = previous_buckets.index_end - index
if span >= self._max_size:
raise Exception("Incorrect merge scale")
if span >= len(previous_buckets.counts):
previous_buckets.grow(span + 1, self._max_size)
previous_buckets.index_start = index
if index > previous_buckets.index_end:
span = index - previous_buckets.index_end
if span >= self._max_size:
raise Exception("Incorrect merge scale")
if span >= len(previous_buckets.counts):
previous_buckets.grow(span + 1, self._max_size)
previous_buckets.index_end = index
bucket_index = index - previous_buckets.index_base
if bucket_index < 0:
bucket_index += len(previous_buckets.counts)
if aggregation_temporality is AggregationTemporality.DELTA:
current_bucket = -current_bucket
previous_buckets.increment_bucket(bucket_index, increment=current_bucket)
class Aggregation(ABC):
"""
Base class for all aggregation types.
"""
@abstractmethod
def _create_aggregation(
self,
instrument: Instrument,
attributes: Attributes,
start_time_unix_nano: int,
) -> _Aggregation:
"""Creates an aggregation"""
class DefaultAggregation(Aggregation):
"""
The default aggregation to be used in a `View`.
This aggregation will create an actual aggregation depending on the
instrument type, as specified next:
==================================================== ====================================
Instrument Aggregation
==================================================== ====================================
`mysql.opentelemetry.sdk.metrics.Counter` `SumAggregation`
`mysql.opentelemetry.sdk.metrics.UpDownCounter` `SumAggregation`
`mysql.opentelemetry.sdk.metrics.ObservableCounter` `SumAggregation`
`mysql.opentelemetry.sdk.metrics.ObservableUpDownCounter` `SumAggregation`
`mysql.opentelemetry.sdk.metrics.Histogram` `ExplicitBucketHistogramAggregation`
`mysql.opentelemetry.sdk.metrics.ObservableGauge` `LastValueAggregation`
==================================================== ====================================
"""
def _create_aggregation(
self,
instrument: Instrument,
attributes: Attributes,
start_time_unix_nano: int,
) -> _Aggregation:
# pylint: disable=too-many-return-statements
if isinstance(instrument, Counter):
return _SumAggregation(
attributes,
instrument_is_monotonic=True,
instrument_temporality=AggregationTemporality.DELTA,
start_time_unix_nano=start_time_unix_nano,
)
if isinstance(instrument, UpDownCounter):
return _SumAggregation(
attributes,
instrument_is_monotonic=False,
instrument_temporality=AggregationTemporality.DELTA,
start_time_unix_nano=start_time_unix_nano,
)
if isinstance(instrument, ObservableCounter):
return _SumAggregation(
attributes,
instrument_is_monotonic=True,
instrument_temporality=AggregationTemporality.CUMULATIVE,
start_time_unix_nano=start_time_unix_nano,
)
if isinstance(instrument, ObservableUpDownCounter):
return _SumAggregation(
attributes,
instrument_is_monotonic=False,
instrument_temporality=AggregationTemporality.CUMULATIVE,
start_time_unix_nano=start_time_unix_nano,
)
if isinstance(instrument, Histogram):
return _ExplicitBucketHistogramAggregation(attributes, start_time_unix_nano)
if isinstance(instrument, ObservableGauge):
return _LastValueAggregation(attributes)
raise Exception(f"Invalid instrument type {type(instrument)} found")
class ExponentialBucketHistogramAggregation(Aggregation):
def __init__(
self,
max_size: int = 160,
):
self._max_size = max_size
def _create_aggregation(
self,
instrument: Instrument,
attributes: Attributes,
start_time_unix_nano: int,
) -> _Aggregation:
return _ExponentialBucketHistogramAggregation(
attributes,
start_time_unix_nano,
max_size=self._max_size,
)
class ExplicitBucketHistogramAggregation(Aggregation):
"""This aggregation informs the SDK to collect:
- Count of Measurement values falling within explicit bucket boundaries.
- Arithmetic sum of Measurement values in population. This SHOULD NOT be collected when used with instruments that record negative measurements, e.g. UpDownCounter or ObservableGauge.
- Min (optional) Measurement value in population.
- Max (optional) Measurement value in population.
Args:
boundaries: Array of increasing values representing explicit bucket boundary values.
record_min_max: Whether to record min and max.
"""
def __init__(
self,
boundaries: Sequence[float] = (
0.0,
5.0,
10.0,
25.0,
50.0,
75.0,
100.0,
250.0,
500.0,
750.0,
1000.0,
2500.0,
5000.0,
7500.0,
10000.0,
),
record_min_max: bool = True,
) -> None:
self._boundaries = boundaries
self._record_min_max = record_min_max
def _create_aggregation(
self,
instrument: Instrument,
attributes: Attributes,
start_time_unix_nano: int,
) -> _Aggregation:
return _ExplicitBucketHistogramAggregation(
attributes,
start_time_unix_nano,
self._boundaries,
self._record_min_max,
)
class SumAggregation(Aggregation):
"""This aggregation informs the SDK to collect:
- The arithmetic sum of Measurement values.
"""
def _create_aggregation(
self,
instrument: Instrument,
attributes: Attributes,
start_time_unix_nano: int,
) -> _Aggregation:
temporality = AggregationTemporality.UNSPECIFIED
if isinstance(instrument, Synchronous):
temporality = AggregationTemporality.DELTA
elif isinstance(instrument, Asynchronous):
temporality = AggregationTemporality.CUMULATIVE
return _SumAggregation(
attributes,
isinstance(instrument, (Counter, ObservableCounter)),
temporality,
start_time_unix_nano,
)
class LastValueAggregation(Aggregation):
"""
This aggregation informs the SDK to collect:
- The last Measurement.
- The timestamp of the last Measurement.
"""
def _create_aggregation(
self,
instrument: Instrument,
attributes: Attributes,
start_time_unix_nano: int,
) -> _Aggregation:
return _LastValueAggregation(attributes)
class DropAggregation(Aggregation):
"""Using this aggregation will make all measurements be ignored."""
def _create_aggregation(
self,
instrument: Instrument,
attributes: Attributes,
start_time_unix_nano: int,
) -> _Aggregation:
return _DropAggregation(attributes)
Reference in New Issue View Git Blame Copy Permalink