Skip to content

Commit

Permalink
Add support for OTLP v0.5.0
Browse files Browse the repository at this point in the history 
Update metrics and trace
  • Loading branch information
@ocelotl
Alex Boten authored and ocelotl committed Sep 25, 2020
1 parent 90d7400 commit 8fd1ca2
Show file tree
Hide file tree
Showing 17 changed files with 2,697 additions and 588 deletions.
170 changes: 100 additions & 70 deletions .../opentelemetry-exporter-otlp/src/opentelemetry/exporter/otlp/metrics_exporter/__init__.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -15,14 +15,13 @@
"""OTLP Metrics Exporter"""

import logging
from typing import List, Sequence, Type, TypeVar, Union
from typing import List, Sequence, Type, TypeVar

# pylint: disable=duplicate-code
from opentelemetry.exporter.otlp.exporter import (
OTLPExporterMixin,
_get_resource_data,
)
from opentelemetry.metrics import InstrumentT
from opentelemetry.proto.collector.metrics.v1.metrics_service_pb2 import (
ExportMetricsServiceRequest,
)
Expand All @@ -31,17 +30,17 @@
)
from opentelemetry.proto.common.v1.common_pb2 import StringKeyValue
from opentelemetry.proto.metrics.v1.metrics_pb2 import (
AggregationTemporality,
DoubleDataPoint,
DoubleGauge,
DoubleSum,
InstrumentationLibraryMetrics,
Int64DataPoint,
)
from opentelemetry.proto.metrics.v1.metrics_pb2 import (
Metric as CollectorMetric,
)
from opentelemetry.proto.metrics.v1.metrics_pb2 import (
MetricDescriptor,
ResourceMetrics,
IntDataPoint,
IntGauge,
IntSum,
)
from opentelemetry.proto.metrics.v1.metrics_pb2 import Metric as OTLPMetric
from opentelemetry.proto.metrics.v1.metrics_pb2 import ResourceMetrics
from opentelemetry.sdk.metrics import (
Counter,
SumObserver,
Expand All @@ -57,7 +56,7 @@
)

logger = logging.getLogger(__name__)
DataPointT = TypeVar("DataPointT", Int64DataPoint, DoubleDataPoint)
DataPointT = TypeVar("DataPointT", IntDataPoint, DoubleDataPoint)


def _get_data_points(
Expand Down Expand Up @@ -93,45 +92,6 @@ def _get_data_points(
return data_points


def _get_temporality(
instrument: InstrumentT,
) -> "MetricDescriptor.TemporalityValue":
# pylint: disable=no-member
if isinstance(instrument, (Counter, UpDownCounter)):
temporality = MetricDescriptor.Temporality.DELTA
elif isinstance(instrument, (ValueRecorder, ValueObserver)):
temporality = MetricDescriptor.Temporality.INSTANTANEOUS
elif isinstance(instrument, (SumObserver, UpDownSumObserver)):
temporality = MetricDescriptor.Temporality.CUMULATIVE
else:
raise Exception(
"No temporality defined for instrument type {}".format(
type(instrument)
)
)

return temporality


def _get_type(value_type: Union[int, float]) -> "MetricDescriptor.TypeValue":
# pylint: disable=no-member
if value_type is int: # type: ignore[comparison-overlap]
type_ = MetricDescriptor.Type.INT64

elif value_type is float: # type: ignore[comparison-overlap]
type_ = MetricDescriptor.Type.DOUBLE

# FIXME What are the types that correspond with
# MetricDescriptor.Type.HISTOGRAM and
# MetricDescriptor.Type.SUMMARY?
else:
raise Exception(
"No type defined for valie type {}".format(type(value_type))
)

return type_


class OTLPMetricsExporter(
MetricsExporter,
OTLPExporterMixin[
Expand All @@ -150,6 +110,7 @@ class OTLPMetricsExporter(
_stub = MetricsServiceStub
_result = MetricsExportResult

# pylint: disable=no-self-use
def _translate_data(
self, data: Sequence[MetricRecord]
) -> ExportMetricsServiceRequest:
Expand All @@ -158,6 +119,22 @@ def _translate_data(

sdk_resource_instrumentation_library_metrics = {}

# The criteria to decide how to translate data is based on this table
# taken directly from OpenTelemetry Proto v0.5.0:

# TODO: Update table after the decision on:
# https://github.com/open-telemetry/opentelemetry-specification/issues/731.
# By default, metrics recording using the OpenTelemetry API are exported as
# (the table does not include MeasurementValueType to avoid extra rows):
#
# Instrument Type
# ----------------------------------------------
# Counter Sum(aggregation_temporality=delta;is_monotonic=true)
# UpDownCounter Sum(aggregation_temporality=delta;is_monotonic=false)
# ValueRecorder TBD
# SumObserver Sum(aggregation_temporality=cumulative;is_monotonic=true)
# UpDownSumObserver Sum(aggregation_temporality=cumulative;is_monotonic=false)
# ValueObserver Gauge()
for sdk_metric in data:

if sdk_metric.instrument.meter.resource not in (
Expand All @@ -167,37 +144,90 @@ def _translate_data(
sdk_metric.instrument.meter.resource
] = InstrumentationLibraryMetrics()

self._metric_descriptor_kwargs = {}
type_class = {
int: {
"sum": {"class": IntSum, "argument": "int_sum"},
"gauge": {"class": IntGauge, "argument": "int_gauge"},
"data_point_class": IntDataPoint,
},
float: {
"sum": {"class": DoubleSum, "argument": "double_sum"},
"gauge": {
"class": DoubleGauge,
"argument": "double_gauge",
},
"data_point_class": DoubleDataPoint,
},
}

value_type = sdk_metric.instrument.value_type

sum_class = type_class[value_type]["sum"]["class"]
gauge_class = type_class[value_type]["gauge"]["class"]
data_point_class = type_class[value_type]["data_point_class"]

if isinstance(sdk_metric.instrument, Counter):
otlp_metric_data = sum_class(
data_points=_get_data_points(sdk_metric, data_point_class),
aggregation_temporality=(
AggregationTemporality.AGGREGATION_TEMPORALITY_DELTA
),
is_monotonic=True,
)
argument = type_class[value_type]["sum"]["argument"]

metric_descriptor = MetricDescriptor(
name=sdk_metric.instrument.name,
description=sdk_metric.instrument.description,
unit=sdk_metric.instrument.unit,
type=_get_type(sdk_metric.instrument.value_type),
temporality=_get_temporality(sdk_metric.instrument),
)
elif isinstance(sdk_metric.instrument, UpDownCounter):
otlp_metric_data = sum_class(
data_points=_get_data_points(sdk_metric, data_point_class),
aggregation_temporality=(
AggregationTemporality.AGGREGATION_TEMPORALITY_DELTA
),
is_monotonic=False,
)
argument = type_class[value_type]["sum"]["argument"]

if metric_descriptor.type == MetricDescriptor.Type.INT64:
elif isinstance(sdk_metric.instrument, (ValueRecorder)):
logger.warning("Skipping exporting of ValueRecorder metric")
continue

collector_metric = CollectorMetric(
metric_descriptor=metric_descriptor,
int64_data_points=_get_data_points(
sdk_metric, Int64DataPoint
elif isinstance(sdk_metric.instrument, SumObserver):
otlp_metric_data = sum_class(
data_points=_get_data_points(sdk_metric, data_point_class),
aggregation_temporality=(
AggregationTemporality.AGGREGATION_TEMPORALITY_CUMULATIVE
),
is_monotonic=True,
)
argument = type_class[value_type]["sum"]["argument"]

elif metric_descriptor.type == MetricDescriptor.Type.DOUBLE:

collector_metric = CollectorMetric(
metric_descriptor=metric_descriptor,
double_data_points=_get_data_points(
sdk_metric, DoubleDataPoint
elif isinstance(sdk_metric.instrument, UpDownSumObserver):
otlp_metric_data = sum_class(
data_points=_get_data_points(sdk_metric, data_point_class),
aggregation_temporality=(
AggregationTemporality.AGGREGATION_TEMPORALITY_CUMULATIVE
),
is_monotonic=False,
)
argument = type_class[value_type]["sum"]["argument"]

elif isinstance(sdk_metric.instrument, (ValueObserver)):
otlp_metric_data = gauge_class(
data_points=_get_data_points(sdk_metric, data_point_class)
)
argument = type_class[value_type]["gauge"]["argument"]

sdk_resource_instrumentation_library_metrics[
sdk_metric.instrument.meter.resource
].metrics.append(collector_metric)
].metrics.append(
OTLPMetric(
**{
"name": sdk_metric.instrument.name,
"description": sdk_metric.instrument.description,
"unit": sdk_metric.instrument.unit,
argument: otlp_metric_data,
}
)
)

return ExportMetricsServiceRequest(
resource_metrics=_get_resource_data(
Expand Down
3 changes: 2 additions & 1 deletion ...er/opentelemetry-exporter-otlp/src/opentelemetry/exporter/otlp/trace_exporter/__init__.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -208,7 +208,8 @@ def _translate_data(
self._translate_status(sdk_span)

self._collector_span_kwargs["kind"] = getattr(
CollectorSpan.SpanKind, sdk_span.kind.name
CollectorSpan.SpanKind,
"SPAN_KIND_{}".format(sdk_span.kind.name),
)

sdk_resource_instrumentation_library_spans[
Expand Down
53 changes: 25 additions & 28 deletions exporter/opentelemetry-exporter-otlp/tests/test_otlp_metric_exporter.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -25,18 +25,15 @@
StringKeyValue,
)
from opentelemetry.proto.metrics.v1.metrics_pb2 import (
AggregationTemporality,
InstrumentationLibraryMetrics,
Int64DataPoint,
)
from opentelemetry.proto.metrics.v1.metrics_pb2 import (
Metric as CollectorMetric,
)
from opentelemetry.proto.metrics.v1.metrics_pb2 import (
MetricDescriptor,
ResourceMetrics,
IntDataPoint,
IntSum,
)
from opentelemetry.proto.metrics.v1.metrics_pb2 import Metric as OTLPMetric
from opentelemetry.proto.metrics.v1.metrics_pb2 import ResourceMetrics
from opentelemetry.proto.resource.v1.resource_pb2 import (
Resource as CollectorResource,
Resource as OTLPResource,
)
from opentelemetry.sdk.metrics import Counter, MeterProvider
from opentelemetry.sdk.metrics.export import MetricRecord
Expand Down Expand Up @@ -71,7 +68,7 @@ def test_translate_metrics(self):
expected = ExportMetricsServiceRequest(
resource_metrics=[
ResourceMetrics(
resource=CollectorResource(
resource=OTLPResource(
attributes=[
KeyValue(key="a", value=AnyValue(int_value=1)),
KeyValue(
Expand All @@ -82,26 +79,26 @@ def test_translate_metrics(self):
instrumentation_library_metrics=[
InstrumentationLibraryMetrics(
metrics=[
CollectorMetric(
metric_descriptor=MetricDescriptor(
name="a",
description="b",
unit="c",
type=MetricDescriptor.Type.INT64,
temporality=(
MetricDescriptor.Temporality.DELTA
OTLPMetric(
name="a",
description="b",
unit="c",
int_sum=IntSum(
data_points=[
IntDataPoint(
labels=[
StringKeyValue(
key="a", value="b"
)
],
value=1,
)
],
aggregation_temporality=(
AggregationTemporality.AGGREGATION_TEMPORALITY_DELTA
),
is_monotonic=True,
),
int64_data_points=[
Int64DataPoint(
labels=[
StringKeyValue(
key="a", value="b"
)
],
value=1,
)
],
)
]
)
Expand Down
Loading

0 comments on commit 8fd1ca2

Please sign in to comment.