Add a configurable max_export_batch_size to the gRPC metrics exporter

Author: overmeulen

Diff for: CHANGELOG.md

@@ -7,6 +7,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+- Add a configurable max_export_batch_size to the gRPC metrics exporter
+  ([#2809](https://github.com/open-telemetry/opentelemetry-python/pull/2809))
 - Remove support for 3.6
   ([#2763](https://github.com/open-telemetry/opentelemetry-python/pull/2763))
 - Update PeriodicExportingMetricReader to never call export() concurrently

Diff for: exporter/opentelemetry-exporter-otlp-proto-grpc/src/opentelemetry/exporter/otlp/proto/grpc/metric_exporter/__init__.py

@@ -13,7 +13,7 @@
 
 from logging import getLogger
 from os import environ
-from typing import Dict, Optional, Sequence
+from typing import Dict, Iterable, List, Optional, Sequence
 from grpc import ChannelCredentials, Compression
 from opentelemetry.sdk.metrics._internal.aggregation import Aggregation
 from opentelemetry.exporter.otlp.proto.grpc.exporter import (
@@ -42,12 +42,15 @@
 )
 from opentelemetry.sdk.metrics.export import (
     AggregationTemporality,
+    DataPointT,
     Gauge,
     Histogram as HistogramType,
     Metric,
     MetricExporter,
     MetricExportResult,
     MetricsData,
+    ResourceMetrics,
+    ScopeMetrics,
     Sum,
 )
 
@@ -71,6 +74,7 @@ def __init__(
         compression: Optional[Compression] = None,
         preferred_temporality: Dict[type, AggregationTemporality] = None,
         preferred_aggregation: Dict[type, Aggregation] = None,
+        max_export_batch_size: Optional[int] = None,
     ):
 
         if insecure is None:
@@ -122,6 +126,8 @@ def __init__(
             compression=compression,
         )
 
+        self._max_export_batch_size: Optional[int] = max_export_batch_size
+
     def _translate_data(
         self, data: MetricsData
     ) -> ExportMetricsServiceRequest:
@@ -223,8 +229,9 @@ def _translate_data(
                             )
                             pb2_metric.sum.data_points.append(pt)
                     else:
-                        _logger.warn(
-                            "unsupported datapoint type %s", metric.point
+                        _logger.warning(
+                            "unsupported data type %s",
+                            metric.data.__class__.__name__,
                         )
                         continue
 
@@ -245,7 +252,132 @@ def export(
         **kwargs,
     ) -> MetricExportResult:
         # TODO(#2663): OTLPExporterMixin should pass timeout to gRPC
-        return self._export(metrics_data)
+        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]:
+        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(
+                ResourceMetrics(
+                    resource=resource_metrics.resource,
+                    schema_url=resource_metrics.schema_url,
+                    scope_metrics=split_scope_metrics,
+                )
+            )
+            for scope_metrics in resource_metrics.scope_metrics:
+                split_metrics: List[Metric] = []
+                split_scope_metrics.append(
+                    ScopeMetrics(
+                        scope=scope_metrics.scope,
+                        schema_url=scope_metrics.schema_url,
+                        metrics=split_metrics,
+                    )
+                )
+                for metric in scope_metrics.metrics:
+                    split_data_points: List[DataPointT] = []
+                    split_metrics.append(
+                        self._create_metric_copy(
+                            metric=metric,
+                            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 = [
+                                self._create_metric_copy(
+                                    metric=metric,
+                                    data_points=split_data_points,
+                                ),
+                            ]
+                            split_scope_metrics = [
+                                ScopeMetrics(
+                                    scope=scope_metrics.scope,
+                                    schema_url=scope_metrics.schema_url,
+                                    metrics=split_metrics,
+                                )
+                            ]
+                            split_resource_metrics = [
+                                ResourceMetrics(
+                                    resource=resource_metrics.resource,
+                                    schema_url=resource_metrics.schema_url,
+                                    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)
+
+    @staticmethod
+    def _create_metric_copy(
+        metric: Metric,
+        data_points: List[DataPointT],
+    ) -> Metric:
+        if isinstance(metric.data, Sum):
+            empty_data = Sum(
+                aggregation_temporality=metric.data.aggregation_temporality,
+                is_monotonic=metric.data.is_monotonic,
+                data_points=data_points,
+            )
+        elif isinstance(metric.data, Gauge):
+            empty_data = Gauge(
+                data_points=data_points,
+            )
+        elif isinstance(metric.data, HistogramType):
+            empty_data = HistogramType(
+                aggregation_temporality=metric.data.aggregation_temporality,
+                data_points=data_points,
+            )
+        else:
+            _logger.warning(
+                "unsupported data type %s", metric.data.__class__.__name__
+            )
+            empty_data = None
+
+        return Metric(
+            name=metric.name,
+            description=metric.description,
+            unit=metric.unit,
+            data=empty_data,
+        )
 
     def shutdown(self, timeout_millis: float = 30_000, **kwargs) -> None:
         pass