Skip to content

Commit 1b1bf36

Browse files
herin049herin049
committed
Add 'opentelemetry-exporter-otlp-json-http' package.
1 parent f800e97 commit 1b1bf36

31 files changed

Lines changed: 4010 additions & 116 deletions

File tree

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

Lines changed: 254 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,9 @@
1515
from __future__ import annotations
1616

1717
import logging
18-
from collections.abc import Collection
18+
from collections.abc import Collection, Iterable
19+
from dataclasses import replace
20+
from os import environ
1921

2022
from opentelemetry.exporter.otlp.json.common._internal import (
2123
_encode_attributes,
@@ -56,8 +58,24 @@
5658
from opentelemetry.proto_json.resource.v1.resource import (
5759
Resource as JSONResource,
5860
)
61+
from opentelemetry.sdk._configuration.models import Aggregation
62+
from opentelemetry.sdk.environment_variables import (
63+
OTEL_EXPORTER_OTLP_METRICS_DEFAULT_HISTOGRAM_AGGREGATION,
64+
OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE,
65+
)
5966
from opentelemetry.sdk.metrics import (
67+
Counter,
6068
Exemplar,
69+
Histogram,
70+
ObservableCounter,
71+
ObservableGauge,
72+
ObservableUpDownCounter,
73+
UpDownCounter,
74+
)
75+
from opentelemetry.sdk.metrics._internal.aggregation import (
76+
AggregationTemporality,
77+
ExplicitBucketHistogramAggregation,
78+
ExponentialBucketHistogramAggregation,
6179
)
6280
from opentelemetry.sdk.metrics._internal.point import (
6381
ExponentialHistogramDataPoint,
@@ -68,15 +86,31 @@
6886
ScopeMetrics,
6987
)
7088
from opentelemetry.sdk.metrics.export import (
71-
ExponentialHistogram,
72-
Gauge,
73-
Histogram,
89+
ExponentialHistogram as ExponentialHistogramMetricData,
90+
)
91+
from opentelemetry.sdk.metrics.export import (
92+
Gauge as GaugeMetricData,
93+
)
94+
from opentelemetry.sdk.metrics.export import (
95+
Histogram as HistogramMetricData,
96+
)
97+
from opentelemetry.sdk.metrics.export import (
7498
MetricsData,
75-
Sum,
99+
)
100+
from opentelemetry.sdk.metrics.export import (
101+
Sum as SumMetricData,
76102
)
77103

78104
_logger = logging.getLogger(__name__)
79105

106+
_METRIC_DATA_FIELDS = (
107+
"gauge",
108+
"sum",
109+
"histogram",
110+
"exponential_histogram",
111+
"summary",
112+
)
113+
80114

81115
def encode_metrics(data: MetricsData) -> JSONExportMetricsServiceRequest:
82116
return JSONExportMetricsServiceRequest(
@@ -114,29 +148,29 @@ def _encode_metric(metric: Metric) -> JSONMetric:
114148
description=metric.description,
115149
unit=metric.unit,
116150
)
117-
if isinstance(metric.data, Gauge):
151+
if isinstance(metric.data, GaugeMetricData):
118152
json_metric.gauge = JSONGauge(
119153
data_points=[
120154
_encode_gauge_data_point(pt) for pt in metric.data.data_points
121155
]
122156
)
123-
elif isinstance(metric.data, Histogram):
157+
elif isinstance(metric.data, HistogramMetricData):
124158
json_metric.histogram = JSONHistogram(
125159
data_points=[
126160
_encode_histogram_data_point(pt)
127161
for pt in metric.data.data_points
128162
],
129163
aggregation_temporality=metric.data.aggregation_temporality,
130164
)
131-
elif isinstance(metric.data, Sum):
165+
elif isinstance(metric.data, SumMetricData):
132166
json_metric.sum = JSONSum(
133167
data_points=[
134168
_encode_sum_data_point(pt) for pt in metric.data.data_points
135169
],
136170
aggregation_temporality=metric.data.aggregation_temporality,
137171
is_monotonic=metric.data.is_monotonic,
138172
)
139-
elif isinstance(metric.data, ExponentialHistogram):
173+
elif isinstance(metric.data, ExponentialHistogramMetricData):
140174
json_metric.exponential_histogram = JSONExponentialHistogram(
141175
data_points=[
142176
_encode_exponential_histogram_data_point(pt)
@@ -269,3 +303,214 @@ def _encode_exemplars(
269303
json_exemplars.append(json_exemplar)
270304

271305
return json_exemplars
306+
307+
308+
def split_metrics_data(
309+
metrics_data: JSONExportMetricsServiceRequest,
310+
max_export_batch_size: int | None,
311+
) -> Iterable[JSONExportMetricsServiceRequest]:
312+
"""Split an ExportMetricsServiceRequest into batches of at most
313+
max_export_batch_size data points, preserving resource/scope hierarchy.
314+
"""
315+
if max_export_batch_size is None:
316+
yield metrics_data
317+
return
318+
319+
batch_size: int = 0
320+
resource_metrics_batch: list[JSONResourceMetrics] = []
321+
scope_metrics_batch: list[JSONScopeMetrics] = []
322+
metrics_batch: list[JSONMetric] = []
323+
324+
for (
325+
resource_metrics,
326+
scope_metrics,
327+
metric,
328+
field_name,
329+
data_points,
330+
) in _iter_metric_data_points(metrics_data):
331+
if (
332+
not resource_metrics_batch
333+
or resource_metrics_batch[-1].resource
334+
is not resource_metrics.resource
335+
):
336+
scope_metrics_batch = []
337+
resource_metrics_batch.append(
338+
replace(resource_metrics, scope_metrics=scope_metrics_batch)
339+
)
340+
341+
if (
342+
not scope_metrics_batch
343+
or scope_metrics_batch[-1].scope is not scope_metrics.scope
344+
):
345+
metrics_batch = []
346+
scope_metrics_batch.append(
347+
replace(scope_metrics, metrics=metrics_batch)
348+
)
349+
350+
data_points_batch: list = []
351+
metrics_batch.append(
352+
_build_metric_with_data_points(
353+
metric, field_name, data_points_batch
354+
)
355+
)
356+
357+
for data_point in data_points:
358+
if batch_size >= max_export_batch_size:
359+
yield JSONExportMetricsServiceRequest(
360+
resource_metrics=resource_metrics_batch
361+
)
362+
(
363+
resource_metrics_batch,
364+
scope_metrics_batch,
365+
metrics_batch,
366+
data_points_batch,
367+
) = _build_empty_metric_batches(
368+
resource_metrics, scope_metrics, metric, field_name
369+
)
370+
batch_size = 0
371+
data_points_batch.append(data_point)
372+
batch_size += 1
373+
374+
if batch_size > 0:
375+
yield JSONExportMetricsServiceRequest(
376+
resource_metrics=resource_metrics_batch
377+
)
378+
379+
380+
def _get_metric_data_field_name(metric: JSONMetric) -> str | None:
381+
return next(
382+
(f for f in _METRIC_DATA_FIELDS if getattr(metric, f) is not None),
383+
None,
384+
)
385+
386+
387+
def _iter_metric_data_points(
388+
metrics_data: JSONExportMetricsServiceRequest,
389+
) -> Iterable[
390+
tuple[JSONResourceMetrics, JSONScopeMetrics, JSONMetric, str, list]
391+
]:
392+
for resource_metrics in metrics_data.resource_metrics:
393+
for scope_metrics in resource_metrics.scope_metrics:
394+
for metric in scope_metrics.metrics:
395+
field_name = _get_metric_data_field_name(metric)
396+
if field_name is None:
397+
_logger.warning(
398+
"Tried to split and export an unsupported metric type. Skipping."
399+
)
400+
continue
401+
yield (
402+
resource_metrics,
403+
scope_metrics,
404+
metric,
405+
field_name,
406+
getattr(metric, field_name).data_points,
407+
)
408+
409+
410+
def _build_metric_with_data_points(
411+
metric: JSONMetric,
412+
field_name: str,
413+
data_points: list,
414+
) -> JSONMetric:
415+
new_data = replace(getattr(metric, field_name), data_points=data_points)
416+
return replace(metric, **{field_name: new_data})
417+
418+
419+
def _build_empty_metric_batches(
420+
resource_metrics: JSONResourceMetrics,
421+
scope_metrics: JSONScopeMetrics,
422+
metric: JSONMetric,
423+
field_name: str,
424+
) -> tuple[
425+
list[JSONResourceMetrics], list[JSONScopeMetrics], list[JSONMetric], list
426+
]:
427+
data_points_batch = []
428+
metrics_batch = [
429+
_build_metric_with_data_points(metric, field_name, data_points_batch)
430+
]
431+
scope_metrics_batch = [replace(scope_metrics, metrics=metrics_batch)]
432+
resource_metrics_batch = [
433+
replace(resource_metrics, scope_metrics=scope_metrics_batch)
434+
]
435+
return (
436+
resource_metrics_batch,
437+
scope_metrics_batch,
438+
metrics_batch,
439+
data_points_batch,
440+
)
441+
442+
443+
def _get_temporality(
444+
preferred_temporality: dict[type, AggregationTemporality] | None,
445+
) -> dict[type, AggregationTemporality]:
446+
preference = (
447+
environ.get(
448+
OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE, "CUMULATIVE"
449+
)
450+
.upper()
451+
.strip()
452+
)
453+
454+
if preference == "DELTA":
455+
instrument_class_temporality = {
456+
Counter: AggregationTemporality.DELTA,
457+
UpDownCounter: AggregationTemporality.CUMULATIVE,
458+
Histogram: AggregationTemporality.DELTA,
459+
ObservableCounter: AggregationTemporality.DELTA,
460+
ObservableUpDownCounter: AggregationTemporality.CUMULATIVE,
461+
ObservableGauge: AggregationTemporality.CUMULATIVE,
462+
}
463+
elif preference == "LOWMEMORY":
464+
instrument_class_temporality = {
465+
Counter: AggregationTemporality.DELTA,
466+
UpDownCounter: AggregationTemporality.CUMULATIVE,
467+
Histogram: AggregationTemporality.DELTA,
468+
ObservableCounter: AggregationTemporality.CUMULATIVE,
469+
ObservableUpDownCounter: AggregationTemporality.CUMULATIVE,
470+
ObservableGauge: AggregationTemporality.CUMULATIVE,
471+
}
472+
else:
473+
if preference != "CUMULATIVE":
474+
_logger.warning(
475+
"Unrecognized OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE"
476+
" value found: %s, using CUMULATIVE",
477+
preference,
478+
)
479+
instrument_class_temporality = {
480+
Counter: AggregationTemporality.CUMULATIVE,
481+
UpDownCounter: AggregationTemporality.CUMULATIVE,
482+
Histogram: AggregationTemporality.CUMULATIVE,
483+
ObservableCounter: AggregationTemporality.CUMULATIVE,
484+
ObservableUpDownCounter: AggregationTemporality.CUMULATIVE,
485+
ObservableGauge: AggregationTemporality.CUMULATIVE,
486+
}
487+
488+
instrument_class_temporality.update(preferred_temporality or {})
489+
return instrument_class_temporality
490+
491+
492+
def _get_aggregation(
493+
preferred_aggregation: dict[type, Aggregation] | None,
494+
) -> dict[type, Aggregation]:
495+
default_histogram_aggregation = environ.get(
496+
OTEL_EXPORTER_OTLP_METRICS_DEFAULT_HISTOGRAM_AGGREGATION,
497+
"explicit_bucket_histogram",
498+
)
499+
500+
if default_histogram_aggregation == "base2_exponential_bucket_histogram":
501+
instrument_class_aggregation: dict[type, Aggregation] = {
502+
Histogram: ExponentialBucketHistogramAggregation(),
503+
}
504+
else:
505+
if default_histogram_aggregation != "explicit_bucket_histogram":
506+
_logger.warning(
507+
"Invalid value for %s: %s, using explicit bucket histogram aggregation",
508+
OTEL_EXPORTER_OTLP_METRICS_DEFAULT_HISTOGRAM_AGGREGATION,
509+
default_histogram_aggregation,
510+
)
511+
instrument_class_aggregation = {
512+
Histogram: ExplicitBucketHistogramAggregation(),
513+
}
514+
515+
instrument_class_aggregation.update(preferred_aggregation or {})
516+
return instrument_class_aggregation

‎exporter/opentelemetry-exporter-otlp-json-common/src/opentelemetry/exporter/otlp/json/common/metrics_encoder.py‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@
1414

1515
from opentelemetry.exporter.otlp.json.common._internal.metrics_encoder import (
1616
encode_metrics,
17+
split_metrics_data,
1718
)
1819

19-
__all__ = ["encode_metrics"]
20+
__all__ = ["encode_metrics", "split_metrics_data"]

0 commit comments

Comments
 (0)