# 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 __future__ import annotations
from dataclasses import replace
from logging import getLogger
from os import environ
from typing import Iterable, List, Tuple, Union
from typing import Sequence as TypingSequence
from grpc import ChannelCredentials, Compression
from opentelemetry.exporter.otlp.proto.common._internal.metrics_encoder import (
OTLPMetricExporterMixin,
)
from opentelemetry.exporter.otlp.proto.common.metrics_encoder import (
encode_metrics,
)
from opentelemetry.exporter.otlp.proto.grpc.exporter import ( # noqa: F401
OTLPExporterMixin,
_get_credentials,
environ_to_compression,
get_resource_data,
)
from opentelemetry.proto.collector.metrics.v1.metrics_service_pb2 import (
ExportMetricsServiceRequest,
)
from opentelemetry.proto.collector.metrics.v1.metrics_service_pb2_grpc import (
MetricsServiceStub,
)
from opentelemetry.proto.common.v1.common_pb2 import ( # noqa: F401
InstrumentationScope,
)
from opentelemetry.proto.metrics.v1 import metrics_pb2 as pb2 # noqa: F401
from opentelemetry.sdk.environment_variables import (
_OTEL_PYTHON_EXPORTER_OTLP_GRPC_METRICS_CREDENTIAL_PROVIDER,
OTEL_EXPORTER_OTLP_METRICS_CERTIFICATE,
OTEL_EXPORTER_OTLP_METRICS_CLIENT_CERTIFICATE,
OTEL_EXPORTER_OTLP_METRICS_CLIENT_KEY,
OTEL_EXPORTER_OTLP_METRICS_COMPRESSION,
OTEL_EXPORTER_OTLP_METRICS_ENDPOINT,
OTEL_EXPORTER_OTLP_METRICS_HEADERS,
OTEL_EXPORTER_OTLP_METRICS_INSECURE,
OTEL_EXPORTER_OTLP_METRICS_TIMEOUT,
)
from opentelemetry.sdk.metrics._internal.aggregation import Aggregation
from opentelemetry.sdk.metrics.export import ( # noqa: F401
AggregationTemporality,
DataPointT,
Gauge,
Metric,
MetricExporter,
MetricExportResult,
MetricsData,
ResourceMetrics,
ScopeMetrics,
Sum,
)
from opentelemetry.sdk.metrics.export import ( # noqa: F401
ExponentialHistogram as ExponentialHistogramType,
)
from opentelemetry.sdk.metrics.export import ( # noqa: F401
Histogram as HistogramType,
)
_logger = getLogger(__name__)
[docs]
class OTLPMetricExporter(
MetricExporter,
OTLPExporterMixin[
MetricsData,
ExportMetricsServiceRequest,
MetricExportResult,
MetricsServiceStub,
],
OTLPMetricExporterMixin,
):
"""OTLP metric exporter
Args:
endpoint: Target URL to which the exporter is going to send metrics
max_export_batch_size: Maximum number of data points to export in a single request. This is to deal with
gRPC's 4MB message size limit. If not set there is no limit to the number of data points in a request.
If it is set and the number of data points exceeds the max, the request will be split.
"""
def __init__(
self,
endpoint: str | None = None,
insecure: bool | None = None,
credentials: ChannelCredentials | None = None,
headers: Union[TypingSequence[Tuple[str, str]], dict[str, str], str]
| None = None,
timeout: float | None = None,
compression: Compression | None = None,
preferred_temporality: dict[type, AggregationTemporality]
| None = None,
preferred_aggregation: dict[type, Aggregation] | None = None,
max_export_batch_size: int | None = None,
channel_options: Tuple[Tuple[str, str]] | None = None,
):
insecure_metrics = environ.get(OTEL_EXPORTER_OTLP_METRICS_INSECURE)
if insecure is None and insecure_metrics is not None:
insecure = insecure_metrics.lower() == "true"
if (
not insecure
and environ.get(OTEL_EXPORTER_OTLP_METRICS_CERTIFICATE) is not None
):
credentials = _get_credentials(
credentials,
_OTEL_PYTHON_EXPORTER_OTLP_GRPC_METRICS_CREDENTIAL_PROVIDER,
OTEL_EXPORTER_OTLP_METRICS_CERTIFICATE,
OTEL_EXPORTER_OTLP_METRICS_CLIENT_KEY,
OTEL_EXPORTER_OTLP_METRICS_CLIENT_CERTIFICATE,
)
environ_timeout = environ.get(OTEL_EXPORTER_OTLP_METRICS_TIMEOUT)
environ_timeout = (
float(environ_timeout) if environ_timeout is not None else None
)
compression = (
environ_to_compression(OTEL_EXPORTER_OTLP_METRICS_COMPRESSION)
if compression is None
else compression
)
self._common_configuration(
preferred_temporality, preferred_aggregation
)
OTLPExporterMixin.__init__(
self,
stub=MetricsServiceStub,
result=MetricExportResult,
endpoint=endpoint
or environ.get(OTEL_EXPORTER_OTLP_METRICS_ENDPOINT),
insecure=insecure,
credentials=credentials,
headers=headers or environ.get(OTEL_EXPORTER_OTLP_METRICS_HEADERS),
timeout=timeout or environ_timeout,
compression=compression,
channel_options=channel_options,
)
self._max_export_batch_size: int | None = max_export_batch_size
def _translate_data( # type: ignore [reportIncompatibleMethodOverride]
self, data: MetricsData
) -> ExportMetricsServiceRequest:
return encode_metrics(data)
[docs]
def export(
self,
metrics_data: MetricsData,
timeout_millis: float = 10_000,
**kwargs,
) -> MetricExportResult:
# TODO(#2663): OTLPExporterMixin should pass timeout to gRPC
if self._max_export_batch_size is None:
return self._export(data=metrics_data)
export_result = MetricExportResult.SUCCESS
for split_metrics_data in self._split_metrics_data(metrics_data):
split_export_result = self._export(data=split_metrics_data)
if split_export_result is MetricExportResult.FAILURE:
export_result = MetricExportResult.FAILURE
return export_result
def _split_metrics_data(
self,
metrics_data: MetricsData,
) -> Iterable[MetricsData]:
assert self._max_export_batch_size is not None
batch_size: int = 0
split_resource_metrics: List[ResourceMetrics] = []
for resource_metrics in metrics_data.resource_metrics:
split_scope_metrics: List[ScopeMetrics] = []
split_resource_metrics.append(
replace(
resource_metrics,
scope_metrics=split_scope_metrics,
)
)
for scope_metrics in resource_metrics.scope_metrics:
split_metrics: List[Metric] = []
split_scope_metrics.append(
replace(
scope_metrics,
metrics=split_metrics,
)
)
for metric in scope_metrics.metrics:
split_data_points: List[DataPointT] = []
split_metrics.append(
replace(
metric,
data=replace(
metric.data,
data_points=split_data_points,
),
)
)
for data_point in metric.data.data_points:
split_data_points.append(data_point)
batch_size += 1
if batch_size >= self._max_export_batch_size:
yield MetricsData(
resource_metrics=split_resource_metrics
)
# Reset all the variables
batch_size = 0
split_data_points = []
split_metrics = [
replace(
metric,
data=replace(
metric.data,
data_points=split_data_points,
),
)
]
split_scope_metrics = [
replace(
scope_metrics,
metrics=split_metrics,
)
]
split_resource_metrics = [
replace(
resource_metrics,
scope_metrics=split_scope_metrics,
)
]
if not split_data_points:
# If data_points is empty remove the whole metric
split_metrics.pop()
if not split_metrics:
# If metrics is empty remove the whole scope_metrics
split_scope_metrics.pop()
if not split_scope_metrics:
# If scope_metrics is empty remove the whole resource_metrics
split_resource_metrics.pop()
if batch_size > 0:
yield MetricsData(resource_metrics=split_resource_metrics)
[docs]
def shutdown(self, timeout_millis: float = 30_000, **kwargs) -> None:
OTLPExporterMixin.shutdown(self, timeout_millis=timeout_millis)
@property
def _exporting(self) -> str:
return "metrics"
[docs]
def force_flush(self, timeout_millis: float = 10_000) -> bool:
"""Nothing is buffered in this exporter, so this method does nothing."""
return True