Add exemplars feature (open-telemetry#4094)

Author: fcollonval

CHANGELOG.md

@@ -11,6 +11,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#4154](https://github.com/open-telemetry/opentelemetry-python/pull/4154))
 - sdk: Add support for log formatting
   ([#4137](https://github.com/open-telemetry/opentelemetry-python/pull/4166))
+- sdk: Implementation of exemplars
+  ([#4094](https://github.com/open-telemetry/opentelemetry-python/pull/4094))
 - Implement events sdk
   ([#4176](https://github.com/open-telemetry/opentelemetry-python/pull/4176))
 

docs/conf.py

@@ -138,6 +138,18 @@
         "py:class",
         "opentelemetry.proto.collector.logs.v1.logs_service_pb2.ExportLogsServiceRequest",
     ),
+    (
+        "py:class",
+        "opentelemetry.sdk.metrics._internal.exemplar.exemplar_reservoir.FixedSizeExemplarReservoirABC",
+    ),
+    (
+        "py:class",
+        "opentelemetry.sdk.metrics._internal.exemplar.exemplar.Exemplar",
+    ),
+    (
+        "py:class",
+        "opentelemetry.sdk.metrics._internal.aggregation._Aggregation",
+    ),
 ]
 
 # Add any paths that contain templates here, relative to this directory.

docs/examples/metrics/reader/README.rst

@@ -5,6 +5,7 @@ These examples show how to customize the metrics that are output by the SDK usin
 
 * preferred_aggregation.py: Shows how to configure the preferred aggregation for metric instrument types.
 * preferred_temporality.py: Shows how to configure the preferred temporality for metric instrument types.
+* preferred_exemplarfilter.py: Shows how to configure the exemplar filter.
 
 The source files of these examples are available :scm_web:`here <docs/examples/metrics/reader/>`.
 

docs/examples/metrics/reader/preferred_exemplarfilter.py

@@ -0,0 +1,62 @@
+# 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 time
+
+from opentelemetry import trace
+from opentelemetry.metrics import get_meter_provider, set_meter_provider
+from opentelemetry.sdk.metrics import MeterProvider
+from opentelemetry.sdk.metrics._internal.exemplar import AlwaysOnExemplarFilter
+from opentelemetry.sdk.metrics.export import (
+    ConsoleMetricExporter,
+    PeriodicExportingMetricReader,
+)
+from opentelemetry.sdk.trace import TracerProvider
+
+# Create an ExemplarFilter instance
+# Available values are AlwaysOffExemplarFilter, AlwaysOnExemplarFilter
+# and TraceBasedExemplarFilter.
+# The default value is `TraceBasedExemplarFilter`.
+#
+# You can also use the environment variable `OTEL_METRICS_EXEMPLAR_FILTER`
+# to change the default value.
+#
+# You can also define your own filter by implementing the abstract class
+# `ExemplarFilter`
+exemplar_filter = AlwaysOnExemplarFilter()
+
+exporter = ConsoleMetricExporter()
+
+reader = PeriodicExportingMetricReader(
+    exporter,
+    export_interval_millis=5_000,
+)
+
+# Set up the MeterProvider with the ExemplarFilter
+provider = MeterProvider(
+    metric_readers=[reader],
+    exemplar_filter=exemplar_filter,  # Pass the ExemplarFilter to the MeterProvider
+)
+set_meter_provider(provider)
+
+meter = get_meter_provider().get_meter("exemplar-filter-example", "0.1.2")
+counter = meter.create_counter("my-counter")
+
+# Create a trace and span as the default exemplar filter `TraceBasedExemplarFilter`
+# will only store exemplar if a context exists
+trace.set_tracer_provider(TracerProvider())
+tracer = trace.get_tracer(__name__)
+with tracer.start_as_current_span("foo"):
+    for value in range(10):
+        counter.add(value)
+        time.sleep(2.0)

docs/examples/metrics/views/README.rst

@@ -7,6 +7,7 @@ These examples show how to customize the metrics that are output by the SDK usin
 * change_name.py: Shows how to change the name of a metric.
 * limit_num_of_attrs.py: Shows how to limit the number of attributes that are output for a metric.
 * drop_metrics_from_instrument.py: Shows how to drop measurements from an instrument.
+* change_reservoir_factory.py: Shows how to use your own ``ExemplarReservoir``
 
 The source files of these examples are available :scm_web:`here <docs/examples/metrics/views/>`.
 

docs/examples/metrics/views/change_reservoir_factory.py

@@ -0,0 +1,90 @@
+# 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 random
+import time
+from typing import Type
+
+from opentelemetry import trace
+from opentelemetry.metrics import get_meter_provider, set_meter_provider
+from opentelemetry.sdk.metrics import MeterProvider
+from opentelemetry.sdk.metrics._internal.aggregation import (
+    DefaultAggregation,
+    _Aggregation,
+    _ExplicitBucketHistogramAggregation,
+)
+from opentelemetry.sdk.metrics._internal.exemplar import (
+    AlignedHistogramBucketExemplarReservoir,
+    ExemplarReservoirBuilder,
+    SimpleFixedSizeExemplarReservoir,
+)
+from opentelemetry.sdk.metrics.export import (
+    ConsoleMetricExporter,
+    PeriodicExportingMetricReader,
+)
+from opentelemetry.sdk.metrics.view import View
+from opentelemetry.sdk.trace import TracerProvider
+
+
+# Create a custom reservoir factory with specified parameters
+def custom_reservoir_factory(
+    aggregationType: Type[_Aggregation],
+) -> ExemplarReservoirBuilder:
+    if issubclass(aggregationType, _ExplicitBucketHistogramAggregation):
+        return AlignedHistogramBucketExemplarReservoir
+    else:
+        # Custom reservoir must accept `**kwargs` that may set the `size` for
+        # _ExponentialBucketHistogramAggregation or the `boundaries` for
+        # _ExplicitBucketHistogramAggregation
+        return lambda **kwargs: SimpleFixedSizeExemplarReservoir(
+            size=10,
+            **{k: v for k, v in kwargs.items() if k != "size"},
+        )
+
+
+# Create a view with the custom reservoir factory
+change_reservoir_factory_view = View(
+    instrument_name="my.counter",
+    name="name",
+    aggregation=DefaultAggregation(),
+    exemplar_reservoir_factory=custom_reservoir_factory,
+)
+
+# Use console exporter for the example
+exporter = ConsoleMetricExporter()
+
+# Create a metric reader with stdout exporter
+reader = PeriodicExportingMetricReader(exporter, export_interval_millis=1_000)
+provider = MeterProvider(
+    metric_readers=[
+        reader,
+    ],
+    views=[
+        change_reservoir_factory_view,
+    ],
+)
+set_meter_provider(provider)
+
+meter = get_meter_provider().get_meter("reservoir-factory-change", "0.1.2")
+
+my_counter = meter.create_counter("my.counter")
+
+# Create a trace and span as the default exemplar filter `TraceBasedExemplarFilter`
+# will only store exemplar if a context exists
+trace.set_tracer_provider(TracerProvider())
+tracer = trace.get_tracer(__name__)
+with tracer.start_as_current_span("foo"):
+    while 1:
+        my_counter.add(random.randint(1, 10))
+        time.sleep(random.random())

exporter/opentelemetry-exporter-otlp-proto-common/src/opentelemetry/exporter/otlp/proto/common/_internal/metrics_encoder/__init__.py

@@ -28,6 +28,8 @@
 )
 from opentelemetry.exporter.otlp.proto.common._internal import (
     _encode_attributes,
+    _encode_span_id,
+    _encode_trace_id,
 )
 from opentelemetry.sdk.environment_variables import (
     OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE,
@@ -254,6 +256,7 @@ def _encode_metric(metric, pb2_metric):
             pt = pb2.NumberDataPoint(
                 attributes=_encode_attributes(data_point.attributes),
                 time_unix_nano=data_point.time_unix_nano,
+                exemplars=_encode_exemplars(data_point.exemplars),
             )
             if isinstance(data_point.value, int):
                 pt.as_int = data_point.value
@@ -267,6 +270,7 @@ def _encode_metric(metric, pb2_metric):
                 attributes=_encode_attributes(data_point.attributes),
                 time_unix_nano=data_point.time_unix_nano,
                 start_time_unix_nano=data_point.start_time_unix_nano,
+                exemplars=_encode_exemplars(data_point.exemplars),
                 count=data_point.count,
                 sum=data_point.sum,
                 bucket_counts=data_point.bucket_counts,
@@ -285,6 +289,7 @@ def _encode_metric(metric, pb2_metric):
                 attributes=_encode_attributes(data_point.attributes),
                 start_time_unix_nano=data_point.start_time_unix_nano,
                 time_unix_nano=data_point.time_unix_nano,
+                exemplars=_encode_exemplars(data_point.exemplars),
             )
             if isinstance(data_point.value, int):
                 pt.as_int = data_point.value
@@ -322,6 +327,7 @@ def _encode_metric(metric, pb2_metric):
                 attributes=_encode_attributes(data_point.attributes),
                 time_unix_nano=data_point.time_unix_nano,
                 start_time_unix_nano=data_point.start_time_unix_nano,
+                exemplars=_encode_exemplars(data_point.exemplars),
                 count=data_point.count,
                 sum=data_point.sum,
                 scale=data_point.scale,
@@ -342,3 +348,35 @@ def _encode_metric(metric, pb2_metric):
             "unsupported data type %s",
             metric.data.__class__.__name__,
         )
+
+
+def _encode_exemplars(sdk_exemplars: list) -> list:
+    """
+    Converts a list of SDK Exemplars into a list of protobuf Exemplars.
+
+    Args:
+        sdk_exemplars (list): The list of exemplars from the OpenTelemetry SDK.
+
+    Returns:
+        list: A list of protobuf exemplars.
+    """
+    pb_exemplars = []
+    for sdk_exemplar in sdk_exemplars:
+        pb_exemplar = pb2.Exemplar(
+            time_unix_nano=sdk_exemplar.time_unix_nano,
+            span_id=_encode_span_id(sdk_exemplar.span_id),
+            trace_id=_encode_trace_id(sdk_exemplar.trace_id),
+            filtered_attributes=_encode_attributes(
+                sdk_exemplar.filtered_attributes
+            ),
+        )
+        # Assign the value based on its type in the SDK exemplar
+        if isinstance(sdk_exemplar.value, float):
+            pb_exemplar.as_double = sdk_exemplar.value
+        elif isinstance(sdk_exemplar.value, int):
+            pb_exemplar.as_int = sdk_exemplar.value
+        else:
+            raise ValueError("Exemplar value must be an int or float")
+        pb_exemplars.append(pb_exemplar)
+
+    return pb_exemplars