From 0fec0b5c8ff08ecb1cd2784a22ab3fdecf44576c Mon Sep 17 00:00:00 2001 From: Diego Hurtado Date: Wed, 22 Jul 2026 00:57:59 -0500 Subject: [PATCH 1/2] Add max_export_batch_size to PeriodicExportingMetricReader Add an optional max_export_batch_size parameter to PeriodicExportingMetricReader. When set, collected metrics are split into batches so that no single exporter.export() call carries more than the configured number of data points, preserving the resource/scope/metric grouping. When unset (None), behavior is unchanged and the full collection is exported in a single call. Batching happens under the existing export lock. --- .changelog/0000.added | 3 + .../sdk/metrics/_internal/export/__init__.py | 120 +++++++++++- .../test_periodic_exporting_metric_reader.py | 180 ++++++++++++++++++ 3 files changed, 299 insertions(+), 4 deletions(-) create mode 100644 .changelog/0000.added diff --git a/.changelog/0000.added b/.changelog/0000.added new file mode 100644 index 00000000000..58f6a21a450 --- /dev/null +++ b/.changelog/0000.added @@ -0,0 +1,3 @@ +Add optional `max_export_batch_size` to `PeriodicExportingMetricReader` to +chunk collected metrics into batches whose data point count never exceeds the +configured limit diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/export/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/export/__init__.py index aee356f78d2..bb2da8e48d9 100644 --- a/opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/export/__init__.py +++ b/opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/export/__init__.py @@ -55,7 +55,12 @@ _ObservableUpDownCounter, _UpDownCounter, ) -from opentelemetry.sdk.metrics._internal.point import MetricsData +from opentelemetry.sdk.metrics._internal.point import ( + Metric, + MetricsData, + ResourceMetrics, + ScopeMetrics, +) from opentelemetry.semconv._incubating.attributes.otel_attributes import ( OtelComponentTypeValues, ) @@ -66,6 +71,85 @@ _logger = getLogger(__name__) +def _batch_metrics_data( + metrics_data: MetricsData, max_export_batch_size: int +) -> Iterable[MetricsData]: + """Split a `MetricsData` into a sequence of `MetricsData` batches where no + single batch contains more than ``max_export_batch_size`` data points. + + The resource/scope/metric grouping structure is preserved: every emitted + batch reproduces the resource and scope objects that own the data points it + carries. Metrics whose data point count on its own exceeds + ``max_export_batch_size`` are emitted whole in their own batch rather than + being split across the metric boundary, so no data point is dropped or + duplicated. + """ + current_resource_metrics: list[ResourceMetrics] = [] + current_scope_metrics: list[ScopeMetrics] = [] + current_metrics: list[Metric] = [] + current_count = 0 + + def _build_batch() -> MetricsData: + scope_metrics = current_scope_metrics + if current_metrics: + scope_metrics = [ + *scope_metrics, + ScopeMetrics( + scope=scope_metrics_source.scope, + metrics=current_metrics, + schema_url=scope_metrics_source.schema_url, + ), + ] + resource_metrics = current_resource_metrics + if scope_metrics: + resource_metrics = [ + *resource_metrics, + ResourceMetrics( + resource=resource_metrics_source.resource, + scope_metrics=scope_metrics, + schema_url=resource_metrics_source.schema_url, + ), + ] + return MetricsData(resource_metrics=resource_metrics) + + for resource_metrics_source in metrics_data.resource_metrics: + for scope_metrics_source in resource_metrics_source.scope_metrics: + for metric in scope_metrics_source.metrics: + metric_count = len(metric.data.data_points) + if ( + current_count + and current_count + metric_count > max_export_batch_size + ): + yield _build_batch() + current_resource_metrics = [] + current_scope_metrics = [] + current_metrics = [] + current_count = 0 + current_metrics.append(metric) + current_count += metric_count + if current_metrics: + current_scope_metrics.append( + ScopeMetrics( + scope=scope_metrics_source.scope, + metrics=current_metrics, + schema_url=scope_metrics_source.schema_url, + ) + ) + current_metrics = [] + if current_scope_metrics: + current_resource_metrics.append( + ResourceMetrics( + resource=resource_metrics_source.resource, + scope_metrics=current_scope_metrics, + schema_url=resource_metrics_source.schema_url, + ) + ) + current_scope_metrics = [] + + if current_resource_metrics: + yield MetricsData(resource_metrics=current_resource_metrics) + + class MetricExportResult(Enum): """Result of exporting a metric @@ -471,6 +555,19 @@ class PeriodicExportingMetricReader(MetricReader): The configured exporter's :py:meth:`~MetricExporter.export` method will not be called concurrently. + + Args: + exporter: The exporter used to send the collected metrics. + export_interval_millis: The time interval in milliseconds between two + consecutive collections. + export_timeout_millis: The maximum amount of time in milliseconds a + collection is allowed to take. + max_export_batch_size: The maximum number of data points to include in + a single :py:meth:`~MetricExporter.export` call. When set, the + collected metrics are split into batches so that no single export + call carries more than this many data points, preserving the + resource/scope/metric grouping. When ``None`` (the default) the + whole collection is exported in a single call. """ def __init__( @@ -478,6 +575,7 @@ def __init__( exporter: MetricExporter, export_interval_millis: float | None = None, export_timeout_millis: float | None = None, + max_export_batch_size: int | None = None, ) -> None: # PeriodicExportingMetricReader defers to exporter for configuration super().__init__( @@ -514,6 +612,12 @@ def __init__( export_timeout_millis = 30000 self._export_interval_millis = export_interval_millis self._export_timeout_millis = export_timeout_millis + if max_export_batch_size is not None and max_export_batch_size <= 0: + raise ValueError( + f"max_export_batch_size value {max_export_batch_size} is " + "invalid and needs to be larger than zero." + ) + self._max_export_batch_size = max_export_batch_size self._shutdown = False self._shutdown_event = Event() self._shutdown_once = Once() @@ -578,9 +682,17 @@ def _receive_metrics( # pylint: disable=broad-exception-caught,invalid-name try: with self._export_lock: - self._exporter.export( - metrics_data, timeout_millis=timeout_millis - ) + if self._max_export_batch_size is None: + self._exporter.export( + metrics_data, timeout_millis=timeout_millis + ) + else: + for batch in _batch_metrics_data( + metrics_data, self._max_export_batch_size + ): + self._exporter.export( + batch, timeout_millis=timeout_millis + ) except Exception: _logger.exception("Exception while exporting metrics") detach(token) diff --git a/opentelemetry-sdk/tests/metrics/test_periodic_exporting_metric_reader.py b/opentelemetry-sdk/tests/metrics/test_periodic_exporting_metric_reader.py index ab8877c21c3..84ce44d6872 100644 --- a/opentelemetry-sdk/tests/metrics/test_periodic_exporting_metric_reader.py +++ b/opentelemetry-sdk/tests/metrics/test_periodic_exporting_metric_reader.py @@ -22,6 +22,7 @@ MetricsTimeoutError, ) from opentelemetry.sdk.metrics._internal import _Counter +from opentelemetry.sdk.metrics._internal.export import _batch_metrics_data from opentelemetry.sdk.metrics._internal.point import ( HistogramDataPoint, MetricsData, @@ -147,6 +148,25 @@ def collect(self, timeout_millis: float = 10_000) -> None: ) +def _metrics_data_point_count(metrics_data: MetricsData) -> int: + return sum( + len(metric.data.data_points) + for resource_metrics in metrics_data.resource_metrics + for scope_metrics in resource_metrics.scope_metrics + for metric in scope_metrics.metrics + ) + + +def _flattened_metric_names(batches) -> list: + return [ + metric.name + for batch in batches + for resource_metrics in batch.resource_metrics + for scope_metrics in resource_metrics.scope_metrics + for metric in scope_metrics.metrics + ] + + class TestPeriodicExportingMetricReader(ConcurrencyTestBase): def test_defaults(self): pmr = PeriodicExportingMetricReader(FakeMetricsExporter()) @@ -307,6 +327,166 @@ def test_metric_exporer_gc(self): "The PeriodicExportingMetricReader object created by this test wasn't garbage collected", ) + def test_no_batching_by_default(self): + exporter = FakeMetricsExporter() + pmr = self._create_periodic_reader(metrics, exporter) + pmr._receive_metrics(metrics) + self.assertEqual(len(exporter.metrics), 1) + self.assertEqual(exporter.metrics[0], metrics) + pmr.shutdown() + + def test_max_export_batch_size_none_single_export(self): + exporter = FakeMetricsExporter() + pmr = PeriodicExportingMetricReader( + exporter, max_export_batch_size=None + ) + pmr._set_collect_callback(lambda reader, timeout_millis: metrics) + pmr._receive_metrics(metrics) + self.assertEqual(len(exporter.metrics), 1) + self.assertEqual(exporter.metrics[0], metrics) + pmr.shutdown() + + def test_max_export_batch_size_invalid(self): + for invalid in (0, -1): + with self.assertRaises(ValueError): + PeriodicExportingMetricReader( + FakeMetricsExporter(), + max_export_batch_size=invalid, + ) + + def test_max_export_batch_size_chunks_data_points(self): + # metrics contains one sum and one gauge, each with a single data point + # (two data points total in a single scope of a single resource). + exporter = FakeMetricsExporter() + pmr = PeriodicExportingMetricReader(exporter, max_export_batch_size=1) + pmr._set_collect_callback(lambda reader, timeout_millis: metrics) + pmr._receive_metrics(metrics) + + # Two data points with a batch size of 1 -> two export calls. + self.assertEqual(len(exporter.metrics), 2) + for exported in exporter.metrics: + self.assertEqual(_metrics_data_point_count(exported), 1) + + # No dropped or duplicated data points, structure preserved. + self.assertEqual( + _flattened_metric_names(exporter.metrics), + [ + "sum_name", + "gauge_name", + ], + ) + for exported in exporter.metrics: + resource_metrics = exported.resource_metrics[0] + self.assertEqual( + resource_metrics.resource, + metrics.resource_metrics[0].resource, + ) + scope_metrics = resource_metrics.scope_metrics[0] + self.assertEqual( + scope_metrics.scope, + metrics.resource_metrics[0].scope_metrics[0].scope, + ) + pmr.shutdown() + + def test_max_export_batch_size_larger_than_total(self): + exporter = FakeMetricsExporter() + pmr = PeriodicExportingMetricReader( + exporter, max_export_batch_size=100 + ) + pmr._set_collect_callback(lambda reader, timeout_millis: metrics) + pmr._receive_metrics(metrics) + self.assertEqual(len(exporter.metrics), 1) + self.assertEqual(_metrics_data_point_count(exporter.metrics[0]), 2) + pmr.shutdown() + + def test_batch_metrics_data_many_points_multiple_scopes(self): + # Build a MetricsData with multiple resources, scopes and metrics, + # each metric carrying several data points, then verify batching + # preserves every data point exactly once and never exceeds the limit. + def _number_metric(name, num_points): + return Metric( + name=name, + description="", + unit="", + data=Sum( + data_points=[ + NumberDataPoint( + attributes={"i": i}, + start_time_unix_nano=time_ns(), + time_unix_nano=time_ns(), + value=i, + ) + for i in range(num_points) + ], + aggregation_temporality=1, + is_monotonic=True, + ), + ) + + source = MetricsData( + resource_metrics=[ + ResourceMetrics( + resource=Resource.create({"r": 0}), + scope_metrics=[ + ScopeMetrics( + scope=InstrumentationScope(name="scope_a"), + metrics=[ + _number_metric("m_a1", 3), + _number_metric("m_a2", 4), + ], + schema_url="", + ), + ScopeMetrics( + scope=InstrumentationScope(name="scope_b"), + metrics=[_number_metric("m_b1", 5)], + schema_url="", + ), + ], + schema_url="", + ), + ResourceMetrics( + resource=Resource.create({"r": 1}), + scope_metrics=[ + ScopeMetrics( + scope=InstrumentationScope(name="scope_c"), + metrics=[_number_metric("m_c1", 2)], + schema_url="", + ), + ], + schema_url="", + ), + ] + ) + + total_points = _metrics_data_point_count(source) + self.assertEqual(total_points, 3 + 4 + 5 + 2) + + batches = list(_batch_metrics_data(source, 3)) + + # Total data points preserved. + self.assertEqual( + sum(_metrics_data_point_count(b) for b in batches), + total_points, + ) + # All metric names present exactly once, in original order. + self.assertEqual( + _flattened_metric_names(batches), + ["m_a1", "m_a2", "m_b1", "m_c1"], + ) + # No batch exceeds the limit unless a single metric already does. + for batch in batches: + count = _metrics_data_point_count(batch) + metric_counts = [ + len(metric.data.data_points) + for rm in batch.resource_metrics + for sm in rm.scope_metrics + for metric in sm.metrics + ] + self.assertTrue( + count <= 3 or (len(metric_counts) == 1), + f"batch with {count} points and metric_counts {metric_counts}", + ) + @patch.dict( "os.environ", {OTEL_PYTHON_SDK_INTERNAL_METRICS_ENABLED: "true"} ) From a2b99dc396544497722a3ce62f46728097ae9366 Mon Sep 17 00:00:00 2001 From: Diego Hurtado Date: Wed, 22 Jul 2026 08:37:44 -0500 Subject: [PATCH 2/2] Rename changelog fragment to match PR number --- .changelog/{0000.added => 25.added} | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename .changelog/{0000.added => 25.added} (100%) diff --git a/.changelog/0000.added b/.changelog/25.added similarity index 100% rename from .changelog/0000.added rename to .changelog/25.added