Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified	Actions
📁 ..	-	2025-05-25 12:26
📄 __init__.py # 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. import weakref from atexit import register, unregister from logging import getLogger from os import environ from threading import Lock from time import time_ns from typing import Optional, Sequence # This kind of import is needed to avoid Sphinx errors. import opentelemetry.sdk.metrics from opentelemetry.metrics import Counter as APICounter from opentelemetry.metrics import Histogram as APIHistogram from opentelemetry.metrics import Meter as APIMeter from opentelemetry.metrics import MeterProvider as APIMeterProvider from opentelemetry.metrics import NoOpMeter from opentelemetry.metrics import ObservableCounter as APIObservableCounter from opentelemetry.metrics import ObservableGauge as APIObservableGauge from opentelemetry.metrics import ( ObservableUpDownCounter as APIObservableUpDownCounter, ) from opentelemetry.metrics import UpDownCounter as APIUpDownCounter from opentelemetry.metrics import _Gauge as APIGauge from opentelemetry.sdk.environment_variables import ( OTEL_METRICS_EXEMPLAR_FILTER, OTEL_SDK_DISABLED, ) from opentelemetry.sdk.metrics._internal.exceptions import MetricsTimeoutError from opentelemetry.sdk.metrics._internal.exemplar import ( AlwaysOffExemplarFilter, AlwaysOnExemplarFilter, ExemplarFilter, TraceBasedExemplarFilter, ) from opentelemetry.sdk.metrics._internal.instrument import ( _Counter, _Gauge, _Histogram, _ObservableCounter, _ObservableGauge, _ObservableUpDownCounter, _UpDownCounter, ) from opentelemetry.sdk.metrics._internal.measurement_consumer import ( MeasurementConsumer, SynchronousMeasurementConsumer, ) from opentelemetry.sdk.metrics._internal.sdk_configuration import ( SdkConfiguration, ) from opentelemetry.sdk.resources import Resource from opentelemetry.sdk.util.instrumentation import InstrumentationScope from opentelemetry.util._once import Once from opentelemetry.util.types import ( Attributes, ) _logger = getLogger(__name__) class Meter(APIMeter): """See `opentelemetry.metrics.Meter`.""" def __init__( self, instrumentation_scope: InstrumentationScope, measurement_consumer: MeasurementConsumer, ): super().__init__( name=instrumentation_scope.name, version=instrumentation_scope.version, schema_url=instrumentation_scope.schema_url, ) self._instrumentation_scope = instrumentation_scope self._measurement_consumer = measurement_consumer self._instrument_id_instrument = {} self._instrument_id_instrument_lock = Lock() def create_counter(self, name, unit="", description="") -> APICounter: status = self._register_instrument(name, _Counter, unit, description) if status.conflict: # FIXME #2558 go through all views here and check if this # instrument registration conflict can be fixed. If it can be, do # not log the following warning. self._log_instrument_registration_conflict( name, APICounter.__name__, unit, description, status, ) if status.already_registered: with self._instrument_id_instrument_lock: return self._instrument_id_instrument[status.instrument_id] instrument = _Counter( name, self._instrumentation_scope, self._measurement_consumer, unit, description, ) with self._instrument_id_instrument_lock: self._instrument_id_instrument[status.instrument_id] = instrument return instrument def create_up_down_counter( self, name, unit="", description="" ) -> APIUpDownCounter: status = self._register_instrument( name, _UpDownCounter, unit, description ) if status.conflict: # FIXME #2558 go through all views here and check if this # instrument registration conflict can be fixed. If it can be, do # not log the following warning. self._log_instrument_registration_conflict( name, APIUpDownCounter.__name__, unit, description, status, ) if status.already_registered: with self._instrument_id_instrument_lock: return self._instrument_id_instrument[status.instrument_id] instrument = _UpDownCounter( name, self._instrumentation_scope, self._measurement_consumer, unit, description, ) with self._instrument_id_instrument_lock: self._instrument_id_instrument[status.instrument_id] = instrument return instrument def create_observable_counter( self, name, callbacks=None, unit="", description="", ) -> APIObservableCounter: status = self._register_instrument( name, _ObservableCounter, unit, description ) if status.conflict: # FIXME #2558 go through all views here and check if this # instrument registration conflict can be fixed. If it can be, do # not log the following warning. self._log_instrument_registration_conflict( name, APIObservableCounter.__name__, unit, description, status, ) if status.already_registered: with self._instrument_id_instrument_lock: return self._instrument_id_instrument[status.instrument_id] instrument = _ObservableCounter( name, self._instrumentation_scope, self._measurement_consumer, callbacks, unit, description, ) self._measurement_consumer.register_asynchronous_instrument(instrument) with self._instrument_id_instrument_lock: self._instrument_id_instrument[status.instrument_id] = instrument return instrument def create_histogram( self, name: str, unit: str = "", description: str = "", , explicit_bucket_boundaries_advisory: Optional[Sequence[float]] = None, ) -> APIHistogram: if explicit_bucket_boundaries_advisory is not None: invalid_advisory = False if isinstance(explicit_bucket_boundaries_advisory, Sequence): try: invalid_advisory = not ( all( isinstance(e, (float, int)) for e in explicit_bucket_boundaries_advisory ) ) except (KeyError, TypeError): invalid_advisory = True else: invalid_advisory = True if invalid_advisory: explicit_bucket_boundaries_advisory = None _logger.warning( "explicit_bucket_boundaries_advisory must be a sequence of numbers" ) status = self._register_instrument( name, _Histogram, unit, description, explicit_bucket_boundaries_advisory, ) if status.conflict: # FIXME #2558 go through all views here and check if this # instrument registration conflict can be fixed. If it can be, do # not log the following warning. self._log_instrument_registration_conflict( name, APIHistogram.__name__, unit, description, status, ) if status.already_registered: with self._instrument_id_instrument_lock: return self._instrument_id_instrument[status.instrument_id] instrument = _Histogram( name, self._instrumentation_scope, self._measurement_consumer, unit, description, explicit_bucket_boundaries_advisory, ) with self._instrument_id_instrument_lock: self._instrument_id_instrument[status.instrument_id] = instrument return instrument def create_gauge(self, name, unit="", description="") -> APIGauge: status = self._register_instrument(name, _Gauge, unit, description) if status.conflict: # FIXME #2558 go through all views here and check if this # instrument registration conflict can be fixed. If it can be, do # not log the following warning. self._log_instrument_registration_conflict( name, APIGauge.__name__, unit, description, status, ) if status.already_registered: with self._instrument_id_instrument_lock: return self._instrument_id_instrument[status.instrument_id] instrument = _Gauge( name, self._instrumentation_scope, self._measurement_consumer, unit, description, ) with self._instrument_id_instrument_lock: self._instrument_id_instrument[status.instrument_id] = instrument return instrument def create_observable_gauge( self, name, callbacks=None, unit="", description="" ) -> APIObservableGauge: status = self._register_instrument( name, _ObservableGauge, unit, description ) if status.conflict: # FIXME #2558 go through all views here and check if this # instrument registration conflict can be fixed. If it can be, do # not log the following warning. self._log_instrument_registration_conflict( name, APIObservableGauge.__name__, unit, description, status, ) if status.already_registered: with self._instrument_id_instrument_lock: return self._instrument_id_instrument[status.instrument_id] instrument = _ObservableGauge( name, self._instrumentation_scope, self._measurement_consumer, callbacks, unit, description, ) self._measurement_consumer.register_asynchronous_instrument(instrument) with self._instrument_id_instrument_lock: self._instrument_id_instrument[status.instrument_id] = instrument return instrument def create_observable_up_down_counter( self, name, callbacks=None, unit="", description="" ) -> APIObservableUpDownCounter: status = self._register_instrument( name, _ObservableUpDownCounter, unit, description ) if status.conflict: # FIXME #2558 go through all views here and check if this # instrument registration conflict can be fixed. If it can be, do # not log the following warning. self._log_instrument_registration_conflict( name, APIObservableUpDownCounter.__name__, unit, description, status, ) if status.already_registered: with self._instrument_id_instrument_lock: return self._instrument_id_instrument[status.instrument_id] instrument = _ObservableUpDownCounter( name, self._instrumentation_scope, self._measurement_consumer, callbacks, unit, description, ) self._measurement_consumer.register_asynchronous_instrument(instrument) with self._instrument_id_instrument_lock: self._instrument_id_instrument[status.instrument_id] = instrument return instrument def _get_exemplar_filter(exemplar_filter: str) -> ExemplarFilter: if exemplar_filter == "trace_based": return TraceBasedExemplarFilter() if exemplar_filter == "always_on": return AlwaysOnExemplarFilter() if exemplar_filter == "always_off": return AlwaysOffExemplarFilter() msg = f"Unknown exemplar filter '{exemplar_filter}'." raise ValueError(msg) class MeterProvider(APIMeterProvider): r"""See `opentelemetry.metrics.MeterProvider`. Args: metric_readers: Register metric readers to collect metrics from the SDK on demand. Each :class:`opentelemetry.sdk.metrics.export.MetricReader` is completely independent and will collect separate streams of metrics. TODO: reference ``PeriodicExportingMetricReader`` usage with push exporters here. resource: The resource representing what the metrics emitted from the SDK pertain to. shutdown_on_exit: If true, registers an `atexit` handler to call `MeterProvider.shutdown` views: The views to configure the metric output the SDK By default, instruments which do not match any :class:`opentelemetry.sdk.metrics.view.View` (or if no :class:`opentelemetry.sdk.metrics.view.View`\ s are provided) will report metrics with the default aggregation for the instrument's kind. To disable instruments by default, configure a match-all :class:`opentelemetry.sdk.metrics.view.View` with `DropAggregation` and then create :class:`opentelemetry.sdk.metrics.view.View`\ s to re-enable individual instruments: .. code-block:: python :caption: Disable default views MeterProvider( views=[ View(instrument_name="", aggregation=DropAggregation()), View(instrument_name="mycounter"), ], # ... ) """ _all_metric_readers_lock = Lock() _all_metric_readers = weakref.WeakSet() def __init__( self, metric_readers: Sequence[ "opentelemetry.sdk.metrics.export.MetricReader" ] = (), resource: Optional[Resource] = None, exemplar_filter: Optional[ExemplarFilter] = None, shutdown_on_exit: bool = True, views: Sequence["opentelemetry.sdk.metrics.view.View"] = (), ): self._lock = Lock() self._meter_lock = Lock() self._atexit_handler = None if resource is None: resource = Resource.create({}) self._sdk_config = SdkConfiguration( exemplar_filter=( exemplar_filter or _get_exemplar_filter( environ.get(OTEL_METRICS_EXEMPLAR_FILTER, "trace_based") ) ), resource=resource, metric_readers=metric_readers, views=views, ) self._measurement_consumer = SynchronousMeasurementConsumer( sdk_config=self._sdk_config ) disabled = environ.get(OTEL_SDK_DISABLED, "") self._disabled = disabled.lower().strip() == "true" if shutdown_on_exit: self._atexit_handler = register(self.shutdown) self._meters = {} self._shutdown_once = Once() self._shutdown = False for metric_reader in self._sdk_config.metric_readers: with self._all_metric_readers_lock: if metric_reader in self._all_metric_readers: # pylint: disable=broad-exception-raised raise Exception( f"MetricReader {metric_reader} has been registered " "already in other MeterProvider instance" ) self._all_metric_readers.add(metric_reader) metric_reader._set_collect_callback( self._measurement_consumer.collect ) def force_flush(self, timeout_millis: float = 10_000) -> bool: deadline_ns = time_ns() + timeout_millis * 106 metric_reader_error = {} for metric_reader in self._sdk_config.metric_readers: current_ts = time_ns() try: if current_ts >= deadline_ns: raise MetricsTimeoutError( "Timed out while flushing metric readers" ) metric_reader.force_flush( timeout_millis=(deadline_ns - current_ts) / 106 ) # pylint: disable=broad-exception-caught except Exception as error: metric_reader_error[metric_reader] = error if metric_reader_error: metric_reader_error_string = "\n".join( [ f"{metric_reader.__class__.__name__}: {repr(error)}" for metric_reader, error in metric_reader_error.items() ] ) # pylint: disable=broad-exception-raised raise Exception( "MeterProvider.force_flush failed because the following " "metric readers failed during collect:\n" f"{metric_reader_error_string}" ) return True def shutdown(self, timeout_millis: float = 30_000): deadline_ns = time_ns() + timeout_millis * 106 def _shutdown(): self._shutdown = True did_shutdown = self._shutdown_once.do_once(_shutdown) if not did_shutdown: _logger.warning("shutdown can only be called once") return metric_reader_error = {} for metric_reader in self._sdk_config.metric_readers: current_ts = time_ns() try: if current_ts >= deadline_ns: # pylint: disable=broad-exception-raised raise Exception( "Didn't get to execute, deadline already exceeded" ) metric_reader.shutdown( timeout_millis=(deadline_ns - current_ts) / 106 ) # pylint: disable=broad-exception-caught except Exception as error: metric_reader_error[metric_reader] = error if self._atexit_handler is not None: unregister(self._atexit_handler) self._atexit_handler = None if metric_reader_error: metric_reader_error_string = "\n".join( [ f"{metric_reader.__class__.__name__}: {repr(error)}" for metric_reader, error in metric_reader_error.items() ] ) # pylint: disable=broad-exception-raised raise Exception( ( "MeterProvider.shutdown failed because the following " "metric readers failed during shutdown:\n" f"{metric_reader_error_string}" ) ) def get_meter( self, name: str, version: Optional[str] = None, schema_url: Optional[str] = None, attributes: Optional[Attributes] = None, ) -> Meter: if self._disabled: return NoOpMeter(name, version=version, schema_url=schema_url) if self._shutdown: _logger.warning( "A shutdown `MeterProvider` can not provide a `Meter`" ) return NoOpMeter(name, version=version, schema_url=schema_url) if not name: _logger.warning("Meter name cannot be None or empty.") return NoOpMeter(name, version=version, schema_url=schema_url) info = InstrumentationScope(name, version, schema_url, attributes) with self._meter_lock: if not self._meters.get(info): # FIXME #2558 pass SDKConfig object to meter so that the meter # has access to views. self._meters[info] = Meter( info, self._measurement_consumer, ) return self._meters[info]	20.3 KB	2025-05-25 12:26
📁 __pycache__	-	2025-05-25 12:26
📄 _view_instrument_match.py # 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. from logging import getLogger from threading import Lock from time import time_ns from typing import Dict, List, Optional, Sequence from opentelemetry.metrics import Instrument from opentelemetry.sdk.metrics._internal.aggregation import ( Aggregation, DefaultAggregation, _Aggregation, _SumAggregation, ) from opentelemetry.sdk.metrics._internal.export import AggregationTemporality from opentelemetry.sdk.metrics._internal.measurement import Measurement from opentelemetry.sdk.metrics._internal.point import DataPointT from opentelemetry.sdk.metrics._internal.view import View _logger = getLogger(__name__) class _ViewInstrumentMatch: def __init__( self, view: View, instrument: Instrument, instrument_class_aggregation: Dict[type, Aggregation], ): self._view = view self._instrument = instrument self._attributes_aggregation: Dict[frozenset, _Aggregation] = {} self._lock = Lock() self._instrument_class_aggregation = instrument_class_aggregation self._name = self._view._name or self._instrument.name self._description = ( self._view._description or self._instrument.description ) if not isinstance(self._view._aggregation, DefaultAggregation): self._aggregation = self._view._aggregation._create_aggregation( self._instrument, None, self._view._exemplar_reservoir_factory, 0, ) else: self._aggregation = self._instrument_class_aggregation[ self._instrument.__class__ ]._create_aggregation( self._instrument, None, self._view._exemplar_reservoir_factory, 0, ) def conflicts(self, other: "_ViewInstrumentMatch") -> bool: # pylint: disable=protected-access result = ( self._name == other._name and self._instrument.unit == other._instrument.unit # The aggregation class is being used here instead of data point # type since they are functionally equivalent. and self._aggregation.__class__ == other._aggregation.__class__ ) if isinstance(self._aggregation, _SumAggregation): result = ( result and self._aggregation._instrument_is_monotonic == other._aggregation._instrument_is_monotonic and self._aggregation._instrument_aggregation_temporality == other._aggregation._instrument_aggregation_temporality ) return result # pylint: disable=protected-access def consume_measurement( self, measurement: Measurement, should_sample_exemplar: bool = True ) -> None: if self._view._attribute_keys is not None: attributes = {} for key, value in (measurement.attributes or {}).items(): if key in self._view._attribute_keys: attributes[key] = value elif measurement.attributes is not None: attributes = measurement.attributes else: attributes = {} aggr_key = frozenset(attributes.items()) if aggr_key not in self._attributes_aggregation: with self._lock: if aggr_key not in self._attributes_aggregation: if not isinstance( self._view._aggregation, DefaultAggregation ): aggregation = ( self._view._aggregation._create_aggregation( self._instrument, attributes, self._view._exemplar_reservoir_factory, time_ns(), ) ) else: aggregation = self._instrument_class_aggregation[ self._instrument.__class__ ]._create_aggregation( self._instrument, attributes, self._view._exemplar_reservoir_factory, time_ns(), ) self._attributes_aggregation[aggr_key] = aggregation self._attributes_aggregation[aggr_key].aggregate( measurement, should_sample_exemplar ) def collect( self, collection_aggregation_temporality: AggregationTemporality, collection_start_nanos: int, ) -> Optional[Sequence[DataPointT]]: data_points: List[DataPointT] = [] with self._lock: for aggregation in self._attributes_aggregation.values(): data_point = aggregation.collect( collection_aggregation_temporality, collection_start_nanos ) if data_point is not None: data_points.append(data_point) # Returning here None instead of an empty list because the caller # does not consume a sequence and to be consistent with the rest of # collect methods that also return None. return data_points or None	5.79 KB	2025-05-25 12:26
📄 aggregation.py # 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 functools import partial from logging import getLogger from math import inf from threading import Lock from typing import ( Callable, Generic, List, Optional, Sequence, Type, TypeVar, ) from opentelemetry.metrics import ( Asynchronous, Counter, Histogram, Instrument, ObservableCounter, ObservableGauge, ObservableUpDownCounter, Synchronous, UpDownCounter, _Gauge, ) from opentelemetry.sdk.metrics._internal.exemplar import ( Exemplar, ExemplarReservoirBuilder, ) from opentelemetry.sdk.metrics._internal.exponential_histogram.buckets import ( Buckets, ) from opentelemetry.sdk.metrics._internal.exponential_histogram.mapping import ( Mapping, ) from opentelemetry.sdk.metrics._internal.exponential_histogram.mapping.exponent_mapping import ( ExponentMapping, ) from opentelemetry.sdk.metrics._internal.exponential_histogram.mapping.logarithm_mapping import ( LogarithmMapping, ) from opentelemetry.sdk.metrics._internal.measurement import Measurement from opentelemetry.sdk.metrics._internal.point import Buckets as BucketsPoint from opentelemetry.sdk.metrics._internal.point import ( ExponentialHistogramDataPoint, HistogramDataPoint, NumberDataPoint, Sum, ) from opentelemetry.sdk.metrics._internal.point import Gauge as GaugePoint from opentelemetry.sdk.metrics._internal.point import ( Histogram as HistogramPoint, ) from 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, reservoir_builder: ExemplarReservoirBuilder, ): self._lock = Lock() self._attributes = attributes self._reservoir = reservoir_builder() self._previous_point = None @abstractmethod def aggregate( self, measurement: Measurement, should_sample_exemplar: bool = True ) -> None: """Aggregate a measurement. Args: measurement: Measurement to aggregate should_sample_exemplar: Whether the measurement should be sampled by the exemplars reservoir or not. """ @abstractmethod def collect( self, collection_aggregation_temporality: AggregationTemporality, collection_start_nano: int, ) -> Optional[_DataPointVarT]: pass def _collect_exemplars(self) -> Sequence[Exemplar]: """Returns the collected exemplars. Returns: The exemplars collected by the reservoir """ return self._reservoir.collect(self._attributes) def _sample_exemplar( self, measurement: Measurement, should_sample_exemplar: bool ) -> None: """Offer the measurement to the exemplar reservoir for sampling. It should be called within the each :ref:`aggregate` call. Args: measurement: The new measurement should_sample_exemplar: Whether the measurement should be sampled by the exemplars reservoir or not. """ if should_sample_exemplar: self._reservoir.offer( measurement.value, measurement.time_unix_nano, measurement.attributes, measurement.context, ) class _DropAggregation(_Aggregation): def aggregate( self, measurement: Measurement, should_sample_exemplar: bool = True ) -> None: pass def collect( self, collection_aggregation_temporality: AggregationTemporality, collection_start_nano: int, ) -> Optional[_DataPointVarT]: pass class _SumAggregation(_Aggregation[Sum]): def __init__( self, attributes: Attributes, instrument_is_monotonic: bool, instrument_aggregation_temporality: AggregationTemporality, start_time_unix_nano: int, reservoir_builder: ExemplarReservoirBuilder, ): super().__init__(attributes, reservoir_builder) self._start_time_unix_nano = start_time_unix_nano self._instrument_aggregation_temporality = ( instrument_aggregation_temporality ) self._instrument_is_monotonic = instrument_is_monotonic self._value = None self._previous_collection_start_nano = self._start_time_unix_nano self._previous_value = 0 def aggregate( self, measurement: Measurement, should_sample_exemplar: bool = True ) -> None: with self._lock: if self._value is None: self._value = 0 self._value = self._value + measurement.value self._sample_exemplar(measurement, should_sample_exemplar) def collect( self, collection_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. Synchronous instruments have a method which is called directly with increments for a given quantity: For example, an instrument that counts the amount of passengers in every vehicle that crosses a certain point in a highway: synchronous_instrument.add(2) collect(...) # 2 passengers are counted synchronous_instrument.add(3) collect(...) # 3 passengers are counted synchronous_instrument.add(1) collect(...) # 1 passenger is counted In this case the instrument aggregation temporality is DELTA because every value represents an increment to the count, Asynchronous instruments have a callback which returns the total value of a given quantity: For example, an instrument that measures the amount of bytes written to a certain hard drive: callback() -> 1352 collect(...) # 1352 bytes have been written so far callback() -> 2324 collect(...) # 2324 bytes have been written so far callback() -> 4542 collect(...) # 4542 bytes have been written so far In this case the instrument aggregation temporality is CUMULATIVE because every value represents the total of the measurement. There is also the collection aggregation temporality, which is passed to this method. The collection aggregation temporality defines the nature of the returned value by this aggregation. When the collection aggregation temporality matches the instrument aggregation temporality, then this method returns the current value directly: synchronous_instrument.add(2) collect(DELTA) -> 2 synchronous_instrument.add(3) collect(DELTA) -> 3 synchronous_instrument.add(1) collect(DELTA) -> 1 callback() -> 1352 collect(CUMULATIVE) -> 1352 callback() -> 2324 collect(CUMULATIVE) -> 2324 callback() -> 4542 collect(CUMULATIVE) -> 4542 When the collection aggregation temporality does not match the instrument aggregation temporality, then a conversion is made. For this purpose, this aggregation keeps a private attribute, self._previous_value. When the instrument is synchronous: self._previous_value is the sum of every previously collected (delta) value. In this case, the returned (cumulative) value will be: self._previous_value + value synchronous_instrument.add(2) collect(CUMULATIVE) -> 2 synchronous_instrument.add(3) collect(CUMULATIVE) -> 5 synchronous_instrument.add(1) collect(CUMULATIVE) -> 6 Also, as a diagram: time -> self._previous_value \|-------------\| value (delta) \|----\| returned value (cumulative) \|------------------\| When the instrument is asynchronous: self._previous_value is the value of the previously collected (cumulative) value. In this case, the returned (delta) value will be: value - self._previous_value callback() -> 1352 collect(DELTA) -> 1352 callback() -> 2324 collect(DELTA) -> 972 callback() -> 4542 collect(DELTA) -> 2218 Also, as a diagram: time -> self._previous_value \|-------------\| value (cumulative) \|------------------\| returned value (delta) \|----\| """ with self._lock: value = self._value self._value = None if ( self._instrument_aggregation_temporality is AggregationTemporality.DELTA ): # This happens when the corresponding instrument for this # aggregation is synchronous. if ( collection_aggregation_temporality is AggregationTemporality.DELTA ): previous_collection_start_nano = ( self._previous_collection_start_nano ) self._previous_collection_start_nano = ( collection_start_nano ) if value is None: return None return NumberDataPoint( attributes=self._attributes, exemplars=self._collect_exemplars(), start_time_unix_nano=previous_collection_start_nano, time_unix_nano=collection_start_nano, value=value, ) if value is None: value = 0 self._previous_value = value + self._previous_value return NumberDataPoint( attributes=self._attributes, exemplars=self._collect_exemplars(), start_time_unix_nano=self._start_time_unix_nano, time_unix_nano=collection_start_nano, value=self._previous_value, ) # This happens when the corresponding instrument for this # aggregation is asynchronous. if value is None: # This happens when the corresponding instrument callback # does not produce measurements. return None if ( collection_aggregation_temporality is AggregationTemporality.DELTA ): result_value = value - self._previous_value self._previous_value = value previous_collection_start_nano = ( self._previous_collection_start_nano ) self._previous_collection_start_nano = collection_start_nano return NumberDataPoint( attributes=self._attributes, exemplars=self._collect_exemplars(), start_time_unix_nano=previous_collection_start_nano, time_unix_nano=collection_start_nano, value=result_value, ) return NumberDataPoint( attributes=self._attributes, exemplars=self._collect_exemplars(), start_time_unix_nano=self._start_time_unix_nano, time_unix_nano=collection_start_nano, value=value, ) class _LastValueAggregation(_Aggregation[GaugePoint]): def __init__( self, attributes: Attributes, reservoir_builder: ExemplarReservoirBuilder, ): super().__init__(attributes, reservoir_builder) self._value = None def aggregate( self, measurement: Measurement, should_sample_exemplar: bool = True ): with self._lock: self._value = measurement.value self._sample_exemplar(measurement, should_sample_exemplar) def collect( self, collection_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 exemplars = self._collect_exemplars() return NumberDataPoint( attributes=self._attributes, exemplars=exemplars, start_time_unix_nano=None, time_unix_nano=collection_start_nano, value=value, ) _DEFAULT_EXPLICIT_BUCKET_HISTOGRAM_AGGREGATION_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, ) class _ExplicitBucketHistogramAggregation(_Aggregation[HistogramPoint]): def __init__( self, attributes: Attributes, instrument_aggregation_temporality: AggregationTemporality, start_time_unix_nano: int, reservoir_builder: ExemplarReservoirBuilder, boundaries: Optional[Sequence[float]] = None, record_min_max: bool = True, ): if boundaries is None: boundaries = ( _DEFAULT_EXPLICIT_BUCKET_HISTOGRAM_AGGREGATION_BOUNDARIES ) super().__init__( attributes, reservoir_builder=partial( reservoir_builder, boundaries=boundaries ), ) self._instrument_aggregation_temporality = ( instrument_aggregation_temporality ) self._start_time_unix_nano = start_time_unix_nano self._boundaries = tuple(boundaries) self._record_min_max = record_min_max self._value = None self._min = inf self._max = -inf self._sum = 0 self._previous_value = None self._previous_min = inf self._previous_max = -inf self._previous_sum = 0 self._previous_collection_start_nano = self._start_time_unix_nano def _get_empty_bucket_counts(self) -> List[int]: return [0] * (len(self._boundaries) + 1) def aggregate( self, measurement: Measurement, should_sample_exemplar: bool = True ) -> None: with self._lock: if self._value is None: self._value = self._get_empty_bucket_counts() measurement_value = measurement.value self._sum += measurement_value if self._record_min_max: self._min = min(self._min, measurement_value) self._max = max(self._max, measurement_value) self._value[bisect_left(self._boundaries, measurement_value)] += 1 self._sample_exemplar(measurement, should_sample_exemplar) def collect( self, collection_aggregation_temporality: AggregationTemporality, collection_start_nano: int, ) -> Optional[_DataPointVarT]: """ Atomically return a point for the current value of the metric. """ with self._lock: value = self._value sum_ = self._sum min_ = self._min max_ = self._max self._value = None self._sum = 0 self._min = inf self._max = -inf if ( self._instrument_aggregation_temporality is AggregationTemporality.DELTA ): # This happens when the corresponding instrument for this # aggregation is synchronous. if ( collection_aggregation_temporality is AggregationTemporality.DELTA ): previous_collection_start_nano = ( self._previous_collection_start_nano ) self._previous_collection_start_nano = ( collection_start_nano ) if value is None: return None return HistogramDataPoint( attributes=self._attributes, exemplars=self._collect_exemplars(), start_time_unix_nano=previous_collection_start_nano, time_unix_nano=collection_start_nano, count=sum(value), sum=sum_, bucket_counts=tuple(value), explicit_bounds=self._boundaries, min=min_, max=max_, ) if value is None: value = self._get_empty_bucket_counts() if self._previous_value is None: self._previous_value = self._get_empty_bucket_counts() self._previous_value = [ value_element + previous_value_element for ( value_element, previous_value_element, ) in zip(value, self._previous_value) ] self._previous_min = min(min_, self._previous_min) self._previous_max = max(max_, self._previous_max) self._previous_sum = sum_ + self._previous_sum return HistogramDataPoint( attributes=self._attributes, exemplars=self._collect_exemplars(), start_time_unix_nano=self._start_time_unix_nano, time_unix_nano=collection_start_nano, count=sum(self._previous_value), sum=self._previous_sum, bucket_counts=tuple(self._previous_value), explicit_bounds=self._boundaries, min=self._previous_min, max=self._previous_max, ) return None # 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, reservoir_builder: ExemplarReservoirBuilder, instrument_aggregation_temporality: AggregationTemporality, 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 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, max_scale: int = 20, ): # max_size is the maximum capacity of the positive and negative # buckets. # _sum is the sum of all the values aggregated by this aggregator. # _count is the count of all calls to aggregate. # _zero_count is the count of all the calls to aggregate when the value # to be aggregated is exactly 0. # _min is the smallest value aggregated by this aggregator. # _max is the smallest value aggregated by this aggregator. # _positive holds the positive values. # _negative holds the negative values by their absolute value. 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}" ) if max_scale > 20: _logger.warning( "max_scale is set to %s which is " "larger than the recommended value of 20", max_scale, ) # This aggregation is analogous to _ExplicitBucketHistogramAggregation, # the only difference is that with every call to aggregate, the size # and amount of buckets can change (in # _ExplicitBucketHistogramAggregation both size and amount of buckets # remain constant once it is instantiated). super().__init__( attributes, reservoir_builder=partial( reservoir_builder, size=min(20, max_size) ), ) self._instrument_aggregation_temporality = ( instrument_aggregation_temporality ) self._start_time_unix_nano = start_time_unix_nano self._max_size = max_size self._max_scale = max_scale self._value_positive = None self._value_negative = None self._min = inf self._max = -inf self._sum = 0 self._count = 0 self._zero_count = 0 self._scale = None self._previous_value_positive = None self._previous_value_negative = None self._previous_min = inf self._previous_max = -inf self._previous_sum = 0 self._previous_count = 0 self._previous_zero_count = 0 self._previous_scale = None self._previous_collection_start_nano = self._start_time_unix_nano self._mapping = self._new_mapping(self._max_scale) def aggregate( self, measurement: Measurement, should_sample_exemplar: bool = True ) -> None: # pylint: disable=too-many-branches,too-many-statements, too-many-locals with self._lock: if self._value_positive is None: self._value_positive = Buckets() if self._value_negative is None: self._value_negative = Buckets() measurement_value = measurement.value self._sum += measurement_value self._min = min(self._min, measurement_value) self._max = max(self._max, measurement_value) self._count += 1 if measurement_value == 0: self._zero_count += 1 if self._count == self._zero_count: self._scale = 0 return if measurement_value > 0: value = self._value_positive else: measurement_value = -measurement_value value = self._value_negative # The following code finds out if it is necessary to change the # buckets to hold the incoming measurement_value, changes them if # necessary. This process does not exist in # _ExplicitBucketHistogram aggregation because the buckets there # are constant in size and amount. index = self._mapping.map_to_index(measurement_value) is_rescaling_needed = False low, high = 0, 0 if len(value) == 0: value.index_start = index value.index_end = index value.index_base = index elif ( index < value.index_start and (value.index_end - index) >= self._max_size ): is_rescaling_needed = True low = index high = value.index_end elif ( index > value.index_end and (index - value.index_start) >= self._max_size ): is_rescaling_needed = True low = value.index_start high = index if is_rescaling_needed: scale_change = self._get_scale_change(low, high) self._downscale( scale_change, self._value_positive, self._value_negative, ) self._mapping = self._new_mapping( self._mapping.scale - scale_change ) index = self._mapping.map_to_index(measurement_value) self._scale = self._mapping.scale if index < value.index_start: span = value.index_end - index if span >= len(value.counts): value.grow(span + 1, self._max_size) value.index_start = index elif index > value.index_end: span = index - value.index_start if span >= len(value.counts): value.grow(span + 1, self._max_size) value.index_end = index bucket_index = index - value.index_base if bucket_index < 0: bucket_index += len(value.counts) # Now the buckets have been changed if needed and bucket_index will # be used to increment the counter of the bucket that needs to be # incremented. # This is analogous to # self._value[bisect_left(self._boundaries, measurement_value)] += 1 # in _ExplicitBucketHistogramAggregation.aggregate value.increment_bucket(bucket_index) self._sample_exemplar(measurement, should_sample_exemplar) def collect( self, collection_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: value_positive = self._value_positive value_negative = self._value_negative sum_ = self._sum min_ = self._min max_ = self._max count = self._count zero_count = self._zero_count scale = self._scale self._value_positive = None self._value_negative = None self._sum = 0 self._min = inf self._max = -inf self._count = 0 self._zero_count = 0 self._scale = None if ( self._instrument_aggregation_temporality is AggregationTemporality.DELTA ): # This happens when the corresponding instrument for this # aggregation is synchronous. if ( collection_aggregation_temporality is AggregationTemporality.DELTA ): previous_collection_start_nano = ( self._previous_collection_start_nano ) self._previous_collection_start_nano = ( collection_start_nano ) if value_positive is None and value_negative is None: return None return ExponentialHistogramDataPoint( attributes=self._attributes, exemplars=self._collect_exemplars(), start_time_unix_nano=previous_collection_start_nano, time_unix_nano=collection_start_nano, count=count, sum=sum_, scale=scale, zero_count=zero_count, positive=BucketsPoint( offset=value_positive.offset, bucket_counts=(value_positive.get_offset_counts()), ), negative=BucketsPoint( offset=value_negative.offset, bucket_counts=(value_negative.get_offset_counts()), ), # FIXME: Find the right value for flags flags=0, min=min_, max=max_, ) # Here collection_temporality is CUMULATIVE. # instrument_temporality is always DELTA for the time being. # Here we need to handle the case where: # collect is called after at least one other call to collect # (there is data in previous buckets, a call to merge is needed # to handle possible differences in bucket sizes). # collect is called without another call previous call to # collect was made (there is no previous buckets, previous, # empty buckets that are the same scale of the current buckets # need to be made so that they can be cumulatively aggregated # to the current buckets). if ( value_positive is None and self._previous_value_positive is None ): # This happens if collect is called for the first time # and aggregate has not yet been called. value_positive = Buckets() self._previous_value_positive = value_positive.copy_empty() if ( value_negative is None and self._previous_value_negative is None ): value_negative = Buckets() self._previous_value_negative = value_negative.copy_empty() if scale is None and self._previous_scale is None: scale = self._mapping.scale self._previous_scale = scale if ( value_positive is not None and self._previous_value_positive is None ): # This happens when collect is called the very first time # and aggregate has been called before. # We need previous buckets to add them to the current ones. # When collect is called for the first time, there are no # previous buckets, so we need to create empty buckets to # add them to the current ones. The addition of empty # buckets to the current ones will result in the current # ones unchanged. # The way the previous buckets are generated here is # different from the explicit bucket histogram where # the size and amount of the buckets does not change once # they are instantiated. Here, the size and amount of the # buckets can change with every call to aggregate. In order # to get empty buckets that can be added to the current # ones resulting in the current ones unchanged we need to # generate empty buckets that have the same size and amount # as the current ones, this is what copy_empty does. self._previous_value_positive = value_positive.copy_empty() if ( value_negative is not None and self._previous_value_negative is None ): self._previous_value_negative = value_negative.copy_empty() if scale is not None and self._previous_scale is None: self._previous_scale = scale if ( value_positive is None and self._previous_value_positive is not None ): value_positive = self._previous_value_positive.copy_empty() if ( value_negative is None and self._previous_value_negative is not None ): value_negative = self._previous_value_negative.copy_empty() if scale is None and self._previous_scale is not None: scale = self._previous_scale min_scale = min(self._previous_scale, scale) low_positive, high_positive = ( self._get_low_high_previous_current( self._previous_value_positive, value_positive, scale, min_scale, ) ) low_negative, high_negative = ( self._get_low_high_previous_current( self._previous_value_negative, value_negative, scale, 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), ) self._downscale( self._previous_scale - min_scale, self._previous_value_positive, self._previous_value_negative, ) # self._merge adds the values from value to # self._previous_value, this is analogous to # self._previous_value = [ # value_element + previous_value_element # for ( # value_element, # previous_value_element, # ) in zip(value, self._previous_value) # ] # in _ExplicitBucketHistogramAggregation.collect. self._merge( self._previous_value_positive, value_positive, scale, min_scale, collection_aggregation_temporality, ) self._merge( self._previous_value_negative, value_negative, scale, min_scale, collection_aggregation_temporality, ) self._previous_min = min(min_, self._previous_min) self._previous_max = max(max_, self._previous_max) self._previous_sum = sum_ + self._previous_sum self._previous_count = count + self._previous_count self._previous_zero_count = ( zero_count + self._previous_zero_count ) self._previous_scale = min_scale return ExponentialHistogramDataPoint( attributes=self._attributes, exemplars=self._collect_exemplars(), start_time_unix_nano=self._start_time_unix_nano, time_unix_nano=collection_start_nano, count=self._previous_count, sum=self._previous_sum, scale=self._previous_scale, zero_count=self._previous_zero_count, positive=BucketsPoint( offset=self._previous_value_positive.offset, bucket_counts=( self._previous_value_positive.get_offset_counts() ), ), negative=BucketsPoint( offset=self._previous_value_negative.offset, bucket_counts=( self._previous_value_negative.get_offset_counts() ), ), # FIXME: Find the right value for flags flags=0, min=self._previous_min, max=self._previous_max, ) return None def _get_low_high_previous_current( self, previous_point_buckets, current_point_buckets, current_scale, min_scale, ): (previous_point_low, previous_point_high) = self._get_low_high( previous_point_buckets, self._previous_scale, min_scale ) (current_point_low, current_point_high) = self._get_low_high( current_point_buckets, current_scale, 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 @staticmethod def _get_low_high(buckets, scale, min_scale): if buckets.counts == [0]: return 0, -1 shift = scale - min_scale return buckets.index_start >> shift, buckets.index_end >> shift @staticmethod def _new_mapping(scale: int) -> Mapping: if scale <= 0: return ExponentMapping(scale) return LogarithmMapping(scale) 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 @staticmethod def _downscale(change: int, positive, negative): if change == 0: return if change < 0: # pylint: disable=broad-exception-raised raise Exception("Invalid change of scale") positive.downscale(change) negative.downscale(change) def _merge( self, previous_buckets: Buckets, current_buckets: 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. current_index = current_buckets.index_base + current_bucket_index if current_index > current_buckets.index_end: current_index -= len(current_buckets.counts) index = current_index >> current_change if index < previous_buckets.index_start: span = previous_buckets.index_end - index if span >= self._max_size: # pylint: disable=broad-exception-raised 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_start if span >= self._max_size: # pylint: disable=broad-exception-raised 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, reservoir_factory: Callable[ [Type[_Aggregation]], ExemplarReservoirBuilder ], 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 ==================================================== ==================================== `opentelemetry.sdk.metrics.Counter` `SumAggregation` `opentelemetry.sdk.metrics.UpDownCounter` `SumAggregation` `opentelemetry.sdk.metrics.ObservableCounter` `SumAggregation` `opentelemetry.sdk.metrics.ObservableUpDownCounter` `SumAggregation` `opentelemetry.sdk.metrics.Histogram` `ExplicitBucketHistogramAggregation` `opentelemetry.sdk.metrics.ObservableGauge` `LastValueAggregation` ==================================================== ==================================== """ def _create_aggregation( self, instrument: Instrument, attributes: Attributes, reservoir_factory: Callable[ [Type[_Aggregation]], ExemplarReservoirBuilder ], start_time_unix_nano: int, ) -> _Aggregation: # pylint: disable=too-many-return-statements if isinstance(instrument, Counter): return _SumAggregation( attributes, reservoir_builder=reservoir_factory(_SumAggregation), instrument_is_monotonic=True, instrument_aggregation_temporality=( AggregationTemporality.DELTA ), start_time_unix_nano=start_time_unix_nano, ) if isinstance(instrument, UpDownCounter): return _SumAggregation( attributes, reservoir_builder=reservoir_factory(_SumAggregation), instrument_is_monotonic=False, instrument_aggregation_temporality=( AggregationTemporality.DELTA ), start_time_unix_nano=start_time_unix_nano, ) if isinstance(instrument, ObservableCounter): return _SumAggregation( attributes, reservoir_builder=reservoir_factory(_SumAggregation), instrument_is_monotonic=True, instrument_aggregation_temporality=( AggregationTemporality.CUMULATIVE ), start_time_unix_nano=start_time_unix_nano, ) if isinstance(instrument, ObservableUpDownCounter): return _SumAggregation( attributes, reservoir_builder=reservoir_factory(_SumAggregation), instrument_is_monotonic=False, instrument_aggregation_temporality=( AggregationTemporality.CUMULATIVE ), start_time_unix_nano=start_time_unix_nano, ) if isinstance(instrument, Histogram): boundaries = instrument._advisory.explicit_bucket_boundaries return _ExplicitBucketHistogramAggregation( attributes, reservoir_builder=reservoir_factory( _ExplicitBucketHistogramAggregation ), instrument_aggregation_temporality=( AggregationTemporality.DELTA ), boundaries=boundaries, start_time_unix_nano=start_time_unix_nano, ) if isinstance(instrument, ObservableGauge): return _LastValueAggregation( attributes, reservoir_builder=reservoir_factory(_LastValueAggregation), ) if isinstance(instrument, _Gauge): return _LastValueAggregation( attributes, reservoir_builder=reservoir_factory(_LastValueAggregation), ) # pylint: disable=broad-exception-raised raise Exception(f"Invalid instrument type {type(instrument)} found") class ExponentialBucketHistogramAggregation(Aggregation): def __init__( self, max_size: int = 160, max_scale: int = 20, ): self._max_size = max_size self._max_scale = max_scale def _create_aggregation( self, instrument: Instrument, attributes: Attributes, reservoir_factory: Callable[ [Type[_Aggregation]], ExemplarReservoirBuilder ], start_time_unix_nano: int, ) -> _Aggregation: instrument_aggregation_temporality = AggregationTemporality.UNSPECIFIED if isinstance(instrument, Synchronous): instrument_aggregation_temporality = AggregationTemporality.DELTA elif isinstance(instrument, Asynchronous): instrument_aggregation_temporality = ( AggregationTemporality.CUMULATIVE ) return _ExponentialBucketHistogramAggregation( attributes, reservoir_factory(_ExponentialBucketHistogramAggregation), instrument_aggregation_temporality, start_time_unix_nano, max_size=self._max_size, max_scale=self._max_scale, ) 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: Optional[Sequence[float]] = None, record_min_max: bool = True, ) -> None: self._boundaries = boundaries self._record_min_max = record_min_max def _create_aggregation( self, instrument: Instrument, attributes: Attributes, reservoir_factory: Callable[ [Type[_Aggregation]], ExemplarReservoirBuilder ], start_time_unix_nano: int, ) -> _Aggregation: instrument_aggregation_temporality = AggregationTemporality.UNSPECIFIED if isinstance(instrument, Synchronous): instrument_aggregation_temporality = AggregationTemporality.DELTA elif isinstance(instrument, Asynchronous): instrument_aggregation_temporality = ( AggregationTemporality.CUMULATIVE ) if self._boundaries is not None: boundaries = self._boundaries else: boundaries = instrument._advisory.explicit_bucket_boundaries return _ExplicitBucketHistogramAggregation( attributes, instrument_aggregation_temporality, start_time_unix_nano, reservoir_factory(_ExplicitBucketHistogramAggregation), 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, reservoir_factory: Callable[ [Type[_Aggregation]], ExemplarReservoirBuilder ], start_time_unix_nano: int, ) -> _Aggregation: instrument_aggregation_temporality = AggregationTemporality.UNSPECIFIED if isinstance(instrument, Synchronous): instrument_aggregation_temporality = AggregationTemporality.DELTA elif isinstance(instrument, Asynchronous): instrument_aggregation_temporality = ( AggregationTemporality.CUMULATIVE ) return _SumAggregation( attributes, isinstance(instrument, (Counter, ObservableCounter)), instrument_aggregation_temporality, start_time_unix_nano, reservoir_factory(_SumAggregation), ) 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, reservoir_factory: Callable[ [Type[_Aggregation]], ExemplarReservoirBuilder ], start_time_unix_nano: int, ) -> _Aggregation: return _LastValueAggregation( attributes, reservoir_builder=reservoir_factory(_LastValueAggregation), ) class DropAggregation(Aggregation): """Using this aggregation will make all measurements be ignored.""" def _create_aggregation( self, instrument: Instrument, attributes: Attributes, reservoir_factory: Callable[ [Type[_Aggregation]], ExemplarReservoirBuilder ], start_time_unix_nano: int, ) -> _Aggregation: return _DropAggregation( attributes, reservoir_factory(_DropAggregation) )	50.27 KB	2025-05-25 12:26
📄 exceptions.py # 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. class MetricsTimeoutError(Exception): """Raised when a metrics function times out"""	675 B	2025-05-25 12:26
📁 exemplar	-	2025-05-25 12:26
📁 exponential_histogram	-	2025-05-25 12:26
📁 export	-	2025-05-25 12:26
📄 instrument.py # 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-ancestors, unused-import from __future__ import annotations from logging import getLogger from time import time_ns from typing import Generator, Iterable, List, Sequence, Union # This kind of import is needed to avoid Sphinx errors. import opentelemetry.sdk.metrics from opentelemetry.context import Context, get_current from opentelemetry.metrics import CallbackT from opentelemetry.metrics import Counter as APICounter from opentelemetry.metrics import Histogram as APIHistogram from opentelemetry.metrics import ObservableCounter as APIObservableCounter from opentelemetry.metrics import ObservableGauge as APIObservableGauge from opentelemetry.metrics import ( ObservableUpDownCounter as APIObservableUpDownCounter, ) from opentelemetry.metrics import UpDownCounter as APIUpDownCounter from opentelemetry.metrics import _Gauge as APIGauge from opentelemetry.metrics._internal.instrument import ( CallbackOptions, _MetricsHistogramAdvisory, ) from opentelemetry.sdk.metrics._internal.measurement import Measurement from opentelemetry.sdk.util.instrumentation import InstrumentationScope _logger = getLogger(__name__) _ERROR_MESSAGE = ( "Expected ASCII string of maximum length 63 characters but got {}" ) class _Synchronous: def __init__( self, name: str, instrumentation_scope: InstrumentationScope, measurement_consumer: "opentelemetry.sdk.metrics.MeasurementConsumer", unit: str = "", description: str = "", ): # pylint: disable=no-member result = self._check_name_unit_description(name, unit, description) if result["name"] is None: # pylint: disable=broad-exception-raised raise Exception(_ERROR_MESSAGE.format(name)) if result["unit"] is None: # pylint: disable=broad-exception-raised raise Exception(_ERROR_MESSAGE.format(unit)) name = result["name"] unit = result["unit"] description = result["description"] self.name = name.lower() self.unit = unit self.description = description self.instrumentation_scope = instrumentation_scope self._measurement_consumer = measurement_consumer super().__init__(name, unit=unit, description=description) class _Asynchronous: def __init__( self, name: str, instrumentation_scope: InstrumentationScope, measurement_consumer: "opentelemetry.sdk.metrics.MeasurementConsumer", callbacks: Iterable[CallbackT] \| None = None, unit: str = "", description: str = "", ): # pylint: disable=no-member result = self._check_name_unit_description(name, unit, description) if result["name"] is None: # pylint: disable=broad-exception-raised raise Exception(_ERROR_MESSAGE.format(name)) if result["unit"] is None: # pylint: disable=broad-exception-raised raise Exception(_ERROR_MESSAGE.format(unit)) name = result["name"] unit = result["unit"] description = result["description"] self.name = name.lower() self.unit = unit self.description = description self.instrumentation_scope = instrumentation_scope self._measurement_consumer = measurement_consumer super().__init__(name, callbacks, unit=unit, description=description) self._callbacks: List[CallbackT] = [] if callbacks is not None: for callback in callbacks: if isinstance(callback, Generator): # advance generator to it's first yield next(callback) def inner( options: CallbackOptions, callback=callback, ) -> Iterable[Measurement]: try: return callback.send(options) except StopIteration: return [] self._callbacks.append(inner) else: self._callbacks.append(callback) def callback( self, callback_options: CallbackOptions ) -> Iterable[Measurement]: for callback in self._callbacks: try: for api_measurement in callback(callback_options): yield Measurement( api_measurement.value, time_unix_nano=time_ns(), instrument=self, context=api_measurement.context or get_current(), attributes=api_measurement.attributes, ) except Exception: # pylint: disable=broad-exception-caught _logger.exception( "Callback failed for instrument %s.", self.name ) class Counter(_Synchronous, APICounter): def __new__(cls, args, kwargs): if cls is Counter: raise TypeError("Counter must be instantiated via a meter.") return super().__new__(cls) def add( self, amount: Union[int, float], attributes: dict[str, str] \| None = None, context: Context \| None = None, ): if amount < 0: _logger.warning( "Add amount must be non-negative on Counter %s.", self.name ) return time_unix_nano = time_ns() self._measurement_consumer.consume_measurement( Measurement( amount, time_unix_nano, self, context or get_current(), attributes, ) ) class UpDownCounter(_Synchronous, APIUpDownCounter): def __new__(cls, args, *kwargs): if cls is UpDownCounter: raise TypeError("UpDownCounter must be instantiated via a meter.") return super().__new__(cls) def add( self, amount: Union[int, float], attributes: dict[str, str] \| None = None, context: Context \| None = None, ): time_unix_nano = time_ns() self._measurement_consumer.consume_measurement( Measurement( amount, time_unix_nano, self, context or get_current(), attributes, ) ) class ObservableCounter(_Asynchronous, APIObservableCounter): def __new__(cls, args, *kwargs): if cls is ObservableCounter: raise TypeError( "ObservableCounter must be instantiated via a meter." ) return super().__new__(cls) class ObservableUpDownCounter(_Asynchronous, APIObservableUpDownCounter): def __new__(cls, args, *kwargs): if cls is ObservableUpDownCounter: raise TypeError( "ObservableUpDownCounter must be instantiated via a meter." ) return super().__new__(cls) class Histogram(_Synchronous, APIHistogram): def __init__( self, name: str, instrumentation_scope: InstrumentationScope, measurement_consumer: "opentelemetry.sdk.metrics.MeasurementConsumer", unit: str = "", description: str = "", explicit_bucket_boundaries_advisory: Sequence[float] \| None = None, ): super().__init__( name, unit=unit, description=description, instrumentation_scope=instrumentation_scope, measurement_consumer=measurement_consumer, ) self._advisory = _MetricsHistogramAdvisory( explicit_bucket_boundaries=explicit_bucket_boundaries_advisory ) def __new__(cls, args, *kwargs): if cls is Histogram: raise TypeError("Histogram must be instantiated via a meter.") return super().__new__(cls) def record( self, amount: Union[int, float], attributes: dict[str, str] \| None = None, context: Context \| None = None, ): if amount < 0: _logger.warning( "Record amount must be non-negative on Histogram %s.", self.name, ) return time_unix_nano = time_ns() self._measurement_consumer.consume_measurement( Measurement( amount, time_unix_nano, self, context or get_current(), attributes, ) ) class Gauge(_Synchronous, APIGauge): def __new__(cls, args, *kwargs): if cls is Gauge: raise TypeError("Gauge must be instantiated via a meter.") return super().__new__(cls) def set( self, amount: Union[int, float], attributes: dict[str, str] \| None = None, context: Context \| None = None, ): time_unix_nano = time_ns() self._measurement_consumer.consume_measurement( Measurement( amount, time_unix_nano, self, context or get_current(), attributes, ) ) class ObservableGauge(_Asynchronous, APIObservableGauge): def __new__(cls, args, **kwargs): if cls is ObservableGauge: raise TypeError( "ObservableGauge must be instantiated via a meter." ) return super().__new__(cls) # Below classes exist to prevent the direct instantiation class _Counter(Counter): pass class _UpDownCounter(UpDownCounter): pass class _ObservableCounter(ObservableCounter): pass class _ObservableUpDownCounter(ObservableUpDownCounter): pass class _Histogram(Histogram): pass class _Gauge(Gauge): pass class _ObservableGauge(ObservableGauge): pass	10.26 KB	2025-05-25 12:26
📄 measurement.py # 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. from dataclasses import dataclass from typing import Union from opentelemetry.context import Context from opentelemetry.metrics import Instrument from opentelemetry.util.types import Attributes @dataclass(frozen=True) class Measurement: """ Represents a data point reported via the metrics API to the SDK. Attributes value: Measured value time_unix_nano: The time the API call was made to record the Measurement instrument: The instrument that produced this `Measurement`. context: The active Context of the Measurement at API call time. attributes: Measurement attributes """ # TODO Fix doc - if using valid Google `Attributes:` key, the attributes are duplicated # one will come from napoleon extension and the other from autodoc extension. This # will raise an sphinx error of duplicated object description # See https://github.com/sphinx-doc/sphinx/issues/8664 value: Union[int, float] time_unix_nano: int instrument: Instrument context: Context attributes: Attributes = None	1.62 KB	2025-05-25 12:26
📄 measurement_consumer.py # 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=unused-import from abc import ABC, abstractmethod from threading import Lock from time import time_ns from typing import Iterable, List, Mapping, Optional # This kind of import is needed to avoid Sphinx errors. import opentelemetry.sdk.metrics import opentelemetry.sdk.metrics._internal.instrument import opentelemetry.sdk.metrics._internal.sdk_configuration from opentelemetry.metrics._internal.instrument import CallbackOptions from opentelemetry.sdk.metrics._internal.exceptions import MetricsTimeoutError from opentelemetry.sdk.metrics._internal.measurement import Measurement from opentelemetry.sdk.metrics._internal.metric_reader_storage import ( MetricReaderStorage, ) from opentelemetry.sdk.metrics._internal.point import Metric class MeasurementConsumer(ABC): @abstractmethod def consume_measurement(self, measurement: Measurement) -> None: pass @abstractmethod def register_asynchronous_instrument( self, instrument: ( "opentelemetry.sdk.metrics._internal.instrument_Asynchronous" ), ): pass @abstractmethod def collect( self, metric_reader: "opentelemetry.sdk.metrics.MetricReader", timeout_millis: float = 10_000, ) -> Optional[Iterable[Metric]]: pass class SynchronousMeasurementConsumer(MeasurementConsumer): def __init__( self, sdk_config: "opentelemetry.sdk.metrics._internal.SdkConfiguration", ) -> None: self._lock = Lock() self._sdk_config = sdk_config # should never be mutated self._reader_storages: Mapping[ "opentelemetry.sdk.metrics.MetricReader", MetricReaderStorage ] = { reader: MetricReaderStorage( sdk_config, reader._instrument_class_temporality, reader._instrument_class_aggregation, ) for reader in sdk_config.metric_readers } self._async_instruments: List[ "opentelemetry.sdk.metrics._internal.instrument._Asynchronous" ] = [] def consume_measurement(self, measurement: Measurement) -> None: should_sample_exemplar = ( self._sdk_config.exemplar_filter.should_sample( measurement.value, measurement.time_unix_nano, measurement.attributes, measurement.context, ) ) for reader_storage in self._reader_storages.values(): reader_storage.consume_measurement( measurement, should_sample_exemplar ) def register_asynchronous_instrument( self, instrument: ( "opentelemetry.sdk.metrics._internal.instrument._Asynchronous" ), ) -> None: with self._lock: self._async_instruments.append(instrument) def collect( self, metric_reader: "opentelemetry.sdk.metrics.MetricReader", timeout_millis: float = 10_000, ) -> Optional[Iterable[Metric]]: with self._lock: metric_reader_storage = self._reader_storages[metric_reader] # for now, just use the defaults callback_options = CallbackOptions() deadline_ns = time_ns() + (timeout_millis * 1e6) default_timeout_ns = 10000 * 1e6 for async_instrument in self._async_instruments: remaining_time = deadline_ns - time_ns() if remaining_time < default_timeout_ns: callback_options = CallbackOptions( timeout_millis=remaining_time / 1e6 ) measurements = async_instrument.callback(callback_options) if time_ns() >= deadline_ns: raise MetricsTimeoutError( "Timed out while executing callback" ) for measurement in measurements: should_sample_exemplar = ( self._sdk_config.exemplar_filter.should_sample( measurement.value, measurement.time_unix_nano, measurement.attributes, measurement.context, ) ) metric_reader_storage.consume_measurement( measurement, should_sample_exemplar ) result = self._reader_storages[metric_reader].collect() return result	5.04 KB	2025-05-25 12:26
📄 metric_reader_storage.py # 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. from logging import getLogger from threading import RLock from time import time_ns from typing import Dict, List, Optional from opentelemetry.metrics import ( Asynchronous, Counter, Instrument, ObservableCounter, ) from opentelemetry.sdk.metrics._internal._view_instrument_match import ( _ViewInstrumentMatch, ) from opentelemetry.sdk.metrics._internal.aggregation import ( Aggregation, ExplicitBucketHistogramAggregation, _DropAggregation, _ExplicitBucketHistogramAggregation, _ExponentialBucketHistogramAggregation, _LastValueAggregation, _SumAggregation, ) from opentelemetry.sdk.metrics._internal.export import AggregationTemporality from opentelemetry.sdk.metrics._internal.measurement import Measurement from opentelemetry.sdk.metrics._internal.point import ( ExponentialHistogram, Gauge, Histogram, Metric, MetricsData, ResourceMetrics, ScopeMetrics, Sum, ) from opentelemetry.sdk.metrics._internal.sdk_configuration import ( SdkConfiguration, ) from opentelemetry.sdk.metrics._internal.view import View from opentelemetry.sdk.util.instrumentation import InstrumentationScope _logger = getLogger(__name__) _DEFAULT_VIEW = View(instrument_name="") class MetricReaderStorage: """The SDK's storage for a given reader""" def __init__( self, sdk_config: SdkConfiguration, instrument_class_temporality: Dict[type, AggregationTemporality], instrument_class_aggregation: Dict[type, Aggregation], ) -> None: self._lock = RLock() self._sdk_config = sdk_config self._instrument_view_instrument_matches: Dict[ Instrument, List[_ViewInstrumentMatch] ] = {} self._instrument_class_temporality = instrument_class_temporality self._instrument_class_aggregation = instrument_class_aggregation def _get_or_init_view_instrument_match( self, instrument: Instrument ) -> List[_ViewInstrumentMatch]: # Optimistically get the relevant views for the given instrument. Once set for a given # instrument, the mapping will never change if instrument in self._instrument_view_instrument_matches: return self._instrument_view_instrument_matches[instrument] with self._lock: # double check if it was set before we held the lock if instrument in self._instrument_view_instrument_matches: return self._instrument_view_instrument_matches[instrument] # not present, hold the lock and add a new mapping view_instrument_matches = [] self._handle_view_instrument_match( instrument, view_instrument_matches ) # if no view targeted the instrument, use the default if not view_instrument_matches: view_instrument_matches.append( _ViewInstrumentMatch( view=_DEFAULT_VIEW, instrument=instrument, instrument_class_aggregation=( self._instrument_class_aggregation ), ) ) self._instrument_view_instrument_matches[instrument] = ( view_instrument_matches ) return view_instrument_matches def consume_measurement( self, measurement: Measurement, should_sample_exemplar: bool = True ) -> None: for view_instrument_match in self._get_or_init_view_instrument_match( measurement.instrument ): view_instrument_match.consume_measurement( measurement, should_sample_exemplar ) def collect(self) -> Optional[MetricsData]: # Use a list instead of yielding to prevent a slow reader from holding # SDK locks # While holding the lock, new _ViewInstrumentMatch can't be added from # another thread (so we are sure we collect all existing view). # However, instruments can still send measurements that will make it # into the individual aggregations; collection will acquire those locks # iteratively to keep locking as fine-grained as possible. One side # effect is that end times can be slightly skewed among the metric # streams produced by the SDK, but we still align the output timestamps # for a single instrument. collection_start_nanos = time_ns() with self._lock: instrumentation_scope_scope_metrics: Dict[ InstrumentationScope, ScopeMetrics ] = {} for ( instrument, view_instrument_matches, ) in self._instrument_view_instrument_matches.items(): aggregation_temporality = self._instrument_class_temporality[ instrument.__class__ ] metrics: List[Metric] = [] for view_instrument_match in view_instrument_matches: data_points = view_instrument_match.collect( aggregation_temporality, collection_start_nanos ) if data_points is None: continue if isinstance( # pylint: disable=protected-access view_instrument_match._aggregation, _SumAggregation, ): data = Sum( aggregation_temporality=aggregation_temporality, data_points=data_points, is_monotonic=isinstance( instrument, (Counter, ObservableCounter) ), ) elif isinstance( # pylint: disable=protected-access view_instrument_match._aggregation, _LastValueAggregation, ): data = Gauge(data_points=data_points) elif isinstance( # pylint: disable=protected-access view_instrument_match._aggregation, _ExplicitBucketHistogramAggregation, ): data = Histogram( data_points=data_points, aggregation_temporality=aggregation_temporality, ) elif isinstance( # pylint: disable=protected-access view_instrument_match._aggregation, _DropAggregation, ): continue elif isinstance( # pylint: disable=protected-access view_instrument_match._aggregation, _ExponentialBucketHistogramAggregation, ): data = ExponentialHistogram( data_points=data_points, aggregation_temporality=aggregation_temporality, ) metrics.append( Metric( # pylint: disable=protected-access # pylint: disable=possibly-used-before-assignment name=view_instrument_match._name, description=view_instrument_match._description, unit=view_instrument_match._instrument.unit, data=data, ) ) if metrics: if instrument.instrumentation_scope not in ( instrumentation_scope_scope_metrics ): instrumentation_scope_scope_metrics[ instrument.instrumentation_scope ] = ScopeMetrics( scope=instrument.instrumentation_scope, metrics=metrics, schema_url=instrument.instrumentation_scope.schema_url, ) else: instrumentation_scope_scope_metrics[ instrument.instrumentation_scope ].metrics.extend(metrics) if instrumentation_scope_scope_metrics: return MetricsData( resource_metrics=[ ResourceMetrics( resource=self._sdk_config.resource, scope_metrics=list( instrumentation_scope_scope_metrics.values() ), schema_url=self._sdk_config.resource.schema_url, ) ] ) return None def _handle_view_instrument_match( self, instrument: Instrument, view_instrument_matches: List["_ViewInstrumentMatch"], ) -> None: for view in self._sdk_config.views: # pylint: disable=protected-access if not view._match(instrument): continue if not self._check_view_instrument_compatibility(view, instrument): continue new_view_instrument_match = _ViewInstrumentMatch( view=view, instrument=instrument, instrument_class_aggregation=( self._instrument_class_aggregation ), ) for ( existing_view_instrument_matches ) in self._instrument_view_instrument_matches.values(): for ( existing_view_instrument_match ) in existing_view_instrument_matches: if existing_view_instrument_match.conflicts( new_view_instrument_match ): _logger.warning( "Views %s and %s will cause conflicting " "metrics identities", existing_view_instrument_match._view, new_view_instrument_match._view, ) view_instrument_matches.append(new_view_instrument_match) @staticmethod def _check_view_instrument_compatibility( view: View, instrument: Instrument ) -> bool: """ Checks if a view and an instrument are compatible. Returns `true` if they are compatible and a `_ViewInstrumentMatch` object should be created, `false` otherwise. """ result = True # pylint: disable=protected-access if isinstance(instrument, Asynchronous) and isinstance( view._aggregation, ExplicitBucketHistogramAggregation ): _logger.warning( "View %s and instrument %s will produce " "semantic errors when matched, the view " "has not been applied.", view, instrument, ) result = False return result	11.77 KB	2025-05-25 12:26
📄 point.py # 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=unused-import from dataclasses import asdict, dataclass, field from json import dumps, loads from typing import Optional, Sequence, Union # This kind of import is needed to avoid Sphinx errors. import opentelemetry.sdk.metrics._internal from opentelemetry.sdk.metrics._internal.exemplar import Exemplar from opentelemetry.sdk.resources import Resource from opentelemetry.sdk.util.instrumentation import InstrumentationScope from opentelemetry.util.types import Attributes @dataclass(frozen=True) class NumberDataPoint: """Single data point in a timeseries that describes the time-varying scalar value of a metric. """ attributes: Attributes start_time_unix_nano: int time_unix_nano: int value: Union[int, float] exemplars: Sequence[Exemplar] = field(default_factory=list) def to_json(self, indent: Optional[int] = 4) -> str: return dumps(asdict(self), indent=indent) @dataclass(frozen=True) class HistogramDataPoint: """Single data point in a timeseries that describes the time-varying scalar value of a metric. """ attributes: Attributes start_time_unix_nano: int time_unix_nano: int count: int sum: Union[int, float] bucket_counts: Sequence[int] explicit_bounds: Sequence[float] min: float max: float exemplars: Sequence[Exemplar] = field(default_factory=list) def to_json(self, indent: Optional[int] = 4) -> str: return dumps(asdict(self), indent=indent) @dataclass(frozen=True) class Buckets: offset: int bucket_counts: Sequence[int] @dataclass(frozen=True) class ExponentialHistogramDataPoint: """Single data point in a timeseries whose boundaries are defined by an exponential function. This timeseries describes the time-varying scalar value of a metric. """ attributes: Attributes start_time_unix_nano: int time_unix_nano: int count: int sum: Union[int, float] scale: int zero_count: int positive: Buckets negative: Buckets flags: int min: float max: float exemplars: Sequence[Exemplar] = field(default_factory=list) def to_json(self, indent: Optional[int] = 4) -> str: return dumps(asdict(self), indent=indent) @dataclass(frozen=True) class ExponentialHistogram: """Represents the type of a metric that is calculated by aggregating as an ExponentialHistogram of all reported measurements over a time interval. """ data_points: Sequence[ExponentialHistogramDataPoint] aggregation_temporality: ( "opentelemetry.sdk.metrics.export.AggregationTemporality" ) def to_json(self, indent: Optional[int] = 4) -> str: return dumps( { "data_points": [ loads(data_point.to_json(indent=indent)) for data_point in self.data_points ], "aggregation_temporality": self.aggregation_temporality, }, indent=indent, ) @dataclass(frozen=True) class Sum: """Represents the type of a scalar metric that is calculated as a sum of all reported measurements over a time interval.""" data_points: Sequence[NumberDataPoint] aggregation_temporality: ( "opentelemetry.sdk.metrics.export.AggregationTemporality" ) is_monotonic: bool def to_json(self, indent: Optional[int] = 4) -> str: return dumps( { "data_points": [ loads(data_point.to_json(indent=indent)) for data_point in self.data_points ], "aggregation_temporality": self.aggregation_temporality, "is_monotonic": self.is_monotonic, }, indent=indent, ) @dataclass(frozen=True) class Gauge: """Represents the type of a scalar metric that always exports the current value for every data point. It should be used for an unknown aggregation.""" data_points: Sequence[NumberDataPoint] def to_json(self, indent: Optional[int] = 4) -> str: return dumps( { "data_points": [ loads(data_point.to_json(indent=indent)) for data_point in self.data_points ], }, indent=indent, ) @dataclass(frozen=True) class Histogram: """Represents the type of a metric that is calculated by aggregating as a histogram of all reported measurements over a time interval.""" data_points: Sequence[HistogramDataPoint] aggregation_temporality: ( "opentelemetry.sdk.metrics.export.AggregationTemporality" ) def to_json(self, indent: Optional[int] = 4) -> str: return dumps( { "data_points": [ loads(data_point.to_json(indent=indent)) for data_point in self.data_points ], "aggregation_temporality": self.aggregation_temporality, }, indent=indent, ) # pylint: disable=invalid-name DataT = Union[Sum, Gauge, Histogram, ExponentialHistogram] DataPointT = Union[ NumberDataPoint, HistogramDataPoint, ExponentialHistogramDataPoint ] @dataclass(frozen=True) class Metric: """Represents a metric point in the OpenTelemetry data model to be exported.""" name: str description: Optional[str] unit: Optional[str] data: DataT def to_json(self, indent: Optional[int] = 4) -> str: return dumps( { "name": self.name, "description": self.description or "", "unit": self.unit or "", "data": loads(self.data.to_json(indent=indent)), }, indent=indent, ) @dataclass(frozen=True) class ScopeMetrics: """A collection of Metrics produced by a scope""" scope: InstrumentationScope metrics: Sequence[Metric] schema_url: str def to_json(self, indent: Optional[int] = 4) -> str: return dumps( { "scope": loads(self.scope.to_json(indent=indent)), "metrics": [ loads(metric.to_json(indent=indent)) for metric in self.metrics ], "schema_url": self.schema_url, }, indent=indent, ) @dataclass(frozen=True) class ResourceMetrics: """A collection of ScopeMetrics from a Resource""" resource: Resource scope_metrics: Sequence[ScopeMetrics] schema_url: str def to_json(self, indent: Optional[int] = 4) -> str: return dumps( { "resource": loads(self.resource.to_json(indent=indent)), "scope_metrics": [ loads(scope_metrics.to_json(indent=indent)) for scope_metrics in self.scope_metrics ], "schema_url": self.schema_url, }, indent=indent, ) @dataclass(frozen=True) class MetricsData: """An array of ResourceMetrics""" resource_metrics: Sequence[ResourceMetrics] def to_json(self, indent: Optional[int] = 4) -> str: return dumps( { "resource_metrics": [ loads(resource_metrics.to_json(indent=indent)) for resource_metrics in self.resource_metrics ] }, indent=indent, )	7.91 KB	2025-05-25 12:26
📄 sdk_configuration.py # 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=unused-import from dataclasses import dataclass from typing import Sequence # This kind of import is needed to avoid Sphinx errors. import opentelemetry.sdk.metrics import opentelemetry.sdk.resources @dataclass class SdkConfiguration: exemplar_filter: "opentelemetry.sdk.metrics.ExemplarFilter" resource: "opentelemetry.sdk.resources.Resource" metric_readers: Sequence["opentelemetry.sdk.metrics.MetricReader"] views: Sequence["opentelemetry.sdk.metrics.View"]	1.06 KB	2025-05-25 12:26
📄 view.py # 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. from fnmatch import fnmatch from logging import getLogger from typing import Callable, Optional, Set, Type from opentelemetry.metrics import Instrument from opentelemetry.sdk.metrics._internal.aggregation import ( Aggregation, DefaultAggregation, _Aggregation, _ExplicitBucketHistogramAggregation, _ExponentialBucketHistogramAggregation, ) from opentelemetry.sdk.metrics._internal.exemplar import ( AlignedHistogramBucketExemplarReservoir, ExemplarReservoirBuilder, SimpleFixedSizeExemplarReservoir, ) _logger = getLogger(__name__) def _default_reservoir_factory( aggregation_type: Type[_Aggregation], ) -> ExemplarReservoirBuilder: """Default reservoir factory per aggregation.""" if issubclass(aggregation_type, _ExplicitBucketHistogramAggregation): return AlignedHistogramBucketExemplarReservoir if issubclass(aggregation_type, _ExponentialBucketHistogramAggregation): return SimpleFixedSizeExemplarReservoir return SimpleFixedSizeExemplarReservoir class View: """ A `View` configuration parameters can be used for the following purposes: 1. Match instruments: When an instrument matches a view, measurements received by that instrument will be processed. 2. Customize metric streams: A metric stream is identified by a match between a view and an instrument and a set of attributes. The metric stream can be customized by certain attributes of the corresponding view. The attributes documented next serve one of the previous two purposes. Args: instrument_type: This is an instrument matching attribute: the class the instrument must be to match the view. instrument_name: This is an instrument matching attribute: the name the instrument must have to match the view. Wild card characters are supported. Wild card characters should not be used with this attribute if the view has also a ``name`` defined. meter_name: This is an instrument matching attribute: the name the instrument meter must have to match the view. meter_version: This is an instrument matching attribute: the version the instrument meter must have to match the view. meter_schema_url: This is an instrument matching attribute: the schema URL the instrument meter must have to match the view. name: This is a metric stream customizing attribute: the name of the metric stream. If `None`, the name of the instrument will be used. description: This is a metric stream customizing attribute: the description of the metric stream. If `None`, the description of the instrument will be used. attribute_keys: This is a metric stream customizing attribute: this is a set of attribute keys. If not `None` then only the measurement attributes that are in ``attribute_keys`` will be used to identify the metric stream. aggregation: This is a metric stream customizing attribute: the aggregation instance to use when data is aggregated for the corresponding metrics stream. If `None` an instance of `DefaultAggregation` will be used. exemplar_reservoir_factory: This is a metric stream customizing attribute: the exemplar reservoir factory instrument_unit: This is an instrument matching attribute: the unit the instrument must have to match the view. This class is not intended to be subclassed by the user. """ _default_aggregation = DefaultAggregation() def __init__( self, instrument_type: Optional[Type[Instrument]] = None, instrument_name: Optional[str] = None, meter_name: Optional[str] = None, meter_version: Optional[str] = None, meter_schema_url: Optional[str] = None, name: Optional[str] = None, description: Optional[str] = None, attribute_keys: Optional[Set[str]] = None, aggregation: Optional[Aggregation] = None, exemplar_reservoir_factory: Optional[ Callable[[Type[_Aggregation]], ExemplarReservoirBuilder] ] = None, instrument_unit: Optional[str] = None, ): if ( instrument_type is instrument_name is instrument_unit is meter_name is meter_version is meter_schema_url is None ): # pylint: disable=broad-exception-raised raise Exception( "Some instrument selection " f"criteria must be provided for View {name}" ) if ( name is not None and instrument_name is not None and ("*" in instrument_name or "?" in instrument_name) ): # pylint: disable=broad-exception-raised raise Exception( f"View {name} declared with wildcard " "characters in instrument_name" ) # _name, _description, _aggregation, _exemplar_reservoir_factory and # _attribute_keys will be accessed when instantiating a _ViewInstrumentMatch. self._name = name self._instrument_type = instrument_type self._instrument_name = instrument_name self._instrument_unit = instrument_unit self._meter_name = meter_name self._meter_version = meter_version self._meter_schema_url = meter_schema_url self._description = description self._attribute_keys = attribute_keys self._aggregation = aggregation or self._default_aggregation self._exemplar_reservoir_factory = ( exemplar_reservoir_factory or _default_reservoir_factory ) # pylint: disable=too-many-return-statements # pylint: disable=too-many-branches def _match(self, instrument: Instrument) -> bool: if self._instrument_type is not None: if not isinstance(instrument, self._instrument_type): return False if self._instrument_name is not None: if not fnmatch(instrument.name, self._instrument_name): return False if self._instrument_unit is not None: if not fnmatch(instrument.unit, self._instrument_unit): return False if self._meter_name is not None: if instrument.instrumentation_scope.name != self._meter_name: return False if self._meter_version is not None: if instrument.instrumentation_scope.version != self._meter_version: return False if self._meter_schema_url is not None: if ( instrument.instrumentation_scope.schema_url != self._meter_schema_url ): return False return True	7.35 KB	2025-05-25 12:26

Terminal