Skip to content

Commit

Permalink
Refactor metric format
Browse files Browse the repository at this point in the history 
Fixes open-telemetry#2646
  • Loading branch information
@ocelotl
ocelotl committed May 6, 2022
1 parent 7397605 commit 6e47731
Show file tree
Hide file tree
Showing 11 changed files with 553 additions and 130 deletions.
196 changes: 116 additions & 80 deletions ...r-otlp-proto-grpc/src/opentelemetry/exporter/otlp/proto/grpc/_metric_exporter/__init__.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -11,9 +11,9 @@
# See the License for the specific language governing permissions and
# limitations under the License.

import logging
from logging import getLogger
from os import environ
from typing import Optional, Sequence
from typing import Optional, Sequence, Dict, Iterable
from grpc import ChannelCredentials, Compression
from opentelemetry.exporter.otlp.proto.grpc.exporter import (
OTLPExporterMixin,
Expand Down Expand Up @@ -41,8 +41,13 @@
MetricExporter,
MetricExportResult,
)
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.util.instrumentation import (
InstrumentationScope as SDKInstrumentationScope,
)
from opentelemetry.sdk._metrics.metric_reader import MetricReaderMetric

logger = logging.getLogger(__name__)
_logger = getLogger(__name__)


class OTLPMetricExporter(
Expand Down Expand Up @@ -79,87 +84,113 @@ def __init__(
)

def _translate_data(
self, data: Sequence[Metric]
self,
data: Dict[
Resource,
Dict[SDKInstrumentationScope, Iterable[MetricReaderMetric]],
],
) -> ExportMetricsServiceRequest:
sdk_resource_scope_metrics = {}

for metric in data:
resource = metric.resource
scope_map = sdk_resource_scope_metrics.get(resource, {})
if not scope_map:
sdk_resource_scope_metrics[resource] = scope_map

scope_metrics = scope_map.get(metric.instrumentation_scope)

if not scope_metrics:
if metric.instrumentation_scope is not None:
scope_map[metric.instrumentation_scope] = pb2.ScopeMetrics(
scope=InstrumentationScope(
name=metric.instrumentation_scope.name,
version=metric.instrumentation_scope.version,

sdk_resource_scope_metrics: Dict[
Resource, Dict[SDKInstrumentationScope, pb2.ScopeMetrics]
] = {}

for resource, instrumentation_scope_metrics in data.items():

if resource not in sdk_resource_scope_metrics:
sdk_resource_scope_metrics[resource] = {}

for (
instrumentation_scope,
metrics,
) in instrumentation_scope_metrics.items():
if instrumentation_scope not in sdk_resource_scope_metrics:
if instrumentation_scope is None:
sdk_resource_scope_metrics[resource][
instrumentation_scope
] = pb2.ScopeMetrics()
else:
sdk_resource_scope_metrics[resource][
instrumentation_scope
] = pb2.ScopeMetrics(
scope=InstrumentationScope(
name=instrumentation_scope.name,
version=instrumentation_scope.version,
)
)
scope_metrics = sdk_resource_scope_metrics[resource][
instrumentation_scope
]

for metric in metrics:

pbmetric = pb2.Metric(
name=metric.name,
description=metric.description,
unit=metric.unit,
)
else:
scope_map[
metric.instrumentation_scope
] = pb2.ScopeMetrics()
if isinstance(metric.point, Gauge):
pt = pb2.NumberDataPoint(
attributes=self._translate_attributes(
metric.attributes
),
time_unix_nano=metric.point.time_unix_nano,
)
if isinstance(metric.point.value, int):
pt.as_int = metric.point.value
else:
pt.as_double = metric.point.value
pbmetric.gauge.data_points.append(pt)
elif isinstance(metric.point, Histogram):
pt = pb2.HistogramDataPoint(
attributes=self._translate_attributes(
metric.attributes
),
time_unix_nano=metric.point.time_unix_nano,
start_time_unix_nano=(
metric.point.start_time_unix_nano
),
count=sum(metric.point.bucket_counts),
sum=metric.point.sum,
bucket_counts=metric.point.bucket_counts,
explicit_bounds=metric.point.explicit_bounds,
)
pbmetric.histogram.aggregation_temporality = (
metric.point.aggregation_temporality
)
pbmetric.histogram.data_points.append(pt)
elif isinstance(metric.point, Sum):
pt = pb2.NumberDataPoint(
attributes=self._translate_attributes(
metric.attributes
),
start_time_unix_nano=(
metric.point.start_time_unix_nano
),
time_unix_nano=metric.point.time_unix_nano,
)
if isinstance(metric.point.value, int):
pt.as_int = metric.point.value
else:
pt.as_double = metric.point.value
# note that because sum is a message type, the fields
# must be set individually rather than instantiating a
# pb2.Sum and setting it once
pbmetric.sum.aggregation_temporality = (
metric.point.aggregation_temporality
)
pbmetric.sum.is_monotonic = metric.point.is_monotonic
pbmetric.sum.data_points.append(pt)
else:
_logger.warn(
"unsupported datapoint type %s", metric.point
)
continue

scope_metrics = scope_map.get(metric.instrumentation_scope)
scope_metrics.metrics.append(
pbmetric,
)

pbmetric = pb2.Metric(
name=metric.name,
description=metric.description,
unit=metric.unit,
)
if isinstance(metric.point, Gauge):
pt = pb2.NumberDataPoint(
attributes=self._translate_attributes(metric.attributes),
time_unix_nano=metric.point.time_unix_nano,
)
if isinstance(metric.point.value, int):
pt.as_int = metric.point.value
else:
pt.as_double = metric.point.value
pbmetric.gauge.data_points.append(pt)
elif isinstance(metric.point, Histogram):
pt = pb2.HistogramDataPoint(
attributes=self._translate_attributes(metric.attributes),
time_unix_nano=metric.point.time_unix_nano,
start_time_unix_nano=metric.point.start_time_unix_nano,
count=sum(metric.point.bucket_counts),
sum=metric.point.sum,
bucket_counts=metric.point.bucket_counts,
explicit_bounds=metric.point.explicit_bounds,
)
pbmetric.histogram.aggregation_temporality = (
metric.point.aggregation_temporality
)
pbmetric.histogram.data_points.append(pt)
elif isinstance(metric.point, Sum):
pt = pb2.NumberDataPoint(
attributes=self._translate_attributes(metric.attributes),
start_time_unix_nano=metric.point.start_time_unix_nano,
time_unix_nano=metric.point.time_unix_nano,
)
if isinstance(metric.point.value, int):
pt.as_int = metric.point.value
else:
pt.as_double = metric.point.value
# note that because sum is a message type, the fields must be
# set individually rather than instantiating a pb2.Sum and setting
# it once
pbmetric.sum.aggregation_temporality = (
metric.point.aggregation_temporality
)
pbmetric.sum.is_monotonic = metric.point.is_monotonic
pbmetric.sum.data_points.append(pt)
else:
logger.warn("unsupported datapoint type %s", metric.point)
continue

scope_metrics.metrics.append(
pbmetric,
)
return ExportMetricsServiceRequest(
resource_metrics=get_resource_data(
sdk_resource_scope_metrics,
Expand All @@ -169,7 +200,12 @@ def _translate_data(
)

def export(
self, metrics: Sequence[Metric], *args, **kwargs
self,
metrics: Dict[
Resource, Dict[InstrumentationScope, Iterable[MetricReaderMetric]]
],
*args,
**kwargs
) -> MetricExportResult:
return self._export(metrics)

Expand Down
Loading

0 comments on commit 6e47731

Please sign in to comment.