oss_19_periodic_metric_reader_batch_size

.changelog/25.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

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,13 +555,27 @@ 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__(
         self,
         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)

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"}
     )