Support 413 handling

sdk/monitor/azure-monitor-opentelemetry-exporter/CHANGELOG.md

@@ -3,6 +3,10 @@
 ## 1.0.0b58 (Unreleased)
 
 ### Features Added
+- Handle `413 Payload Too Large` responses by splitting the batch and persisting each batch to
+  local storage for retry at a smaller size, instead of dropping the entire batch. A single
+  envelope that is too large to split is dropped.
+  ([#48548](https://github.com/Azure/azure-sdk-for-python/pull/48548))
 
 ### Breaking Changes
 

sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_constants.py

@@ -76,7 +76,7 @@
     439,  # Quota, too Many Requests over extended time (legacy)
 )
 
-_REACHED_INGESTION_STATUS_CODES = (200, 206, 402, 408, 429, 439, 500)
+_REACHED_INGESTION_STATUS_CODES = (200, 206, 402, 408, 413, 429, 439, 500)
 
 # Envelope constants
 

sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_utils.py

@@ -14,14 +14,15 @@
 import time
 import warnings
 import hashlib
-from typing import Callable, Dict, Any, Optional
+from typing import Callable, Dict, Any, List, Optional
 
 from opentelemetry.semconv.resource import ResourceAttributes
 from opentelemetry.sdk.resources import Resource
 from opentelemetry.util.types import Attributes
 
 from azure.core.pipeline.policies import BearerTokenCredentialPolicy
 from azure.monitor.opentelemetry.exporter._generated.exporter.models import ContextTagKeys, TelemetryItem
+from azure.monitor.opentelemetry.exporter._generated.exporter._utils.model_base import SdkJSONEncoder
 from azure.monitor.opentelemetry.exporter._version import VERSION as ext_version
 from azure.monitor.opentelemetry.exporter._connection_string_parser import ConnectionStringParser
 from azure.monitor.opentelemetry.exporter._constants import (
@@ -47,6 +48,15 @@
 # Workaround for missing version file
 opentelemetry_version = version("opentelemetry-sdk")
 
+# Maximum ingestion limit over wire
+_MAX_INGESTION_PAYLOAD_SIZE_BYTES = 30 * 1000 * 1000
+
+# Marker to check the per-item size limit messages
+_ITEM_TOO_LARGE_MESSAGE_MARKERS = (
+    "length must not exceed",
+    "too large",
+)
+
 
 # Azure App Service
 
@@ -574,3 +584,47 @@ def _get_retry_delay_from_headers(headers: Any) -> Optional[int]:
     except ValueError:
         return None
     return None
+
+
+def _get_envelope_serialized_size(envelope: TelemetryItem) -> int:
+    # Return the serialized wire size (in bytes) of a single envelope on the wire.
+    try:
+        serialized = json.dumps(envelope, cls=SdkJSONEncoder, exclude_readonly=True)
+        return len(serialized.encode("utf-8"))
+    except Exception:  # pylint: disable=broad-except
+        return _MAX_INGESTION_PAYLOAD_SIZE_BYTES
+
+
+def _split_oversized_batch(envelopes: List[TelemetryItem]) -> List[List[TelemetryItem]]:
+    # Split an oversized batch into sub-batches that each fit under the ingestion size limit.
+    chunks: List[List[TelemetryItem]] = []
+    current: List[TelemetryItem] = []
+    current_size = 2  # account for the enclosing "[" and "]" of the JSON array
+    for envelope in envelopes:
+        envelope_size = _get_envelope_serialized_size(envelope)
+        # Default json.dumps (used by the transport) separates array elements with ", " (2 bytes).
+        separator_size = 2 if current else 0
+        if current and current_size + separator_size + envelope_size > _MAX_INGESTION_PAYLOAD_SIZE_BYTES:
+            chunks.append(current)
+            current = []
+            current_size = 2
+            separator_size = 0
+        current.append(envelope)
+        current_size += separator_size + envelope_size
+    if current:
+        chunks.append(current)
+    # If size-based packing produced a single chunk (e.g. because the byte
+    # estimate under-reported), fall back to halving by count so repeated 413s keep shrinking the
+    # batch.
+    if len(chunks) < 2 and len(envelopes) > 1:
+        midpoint = len(envelopes) // 2
+        chunks = [envelopes[:midpoint], envelopes[midpoint:]]
+    return chunks
+
+
+def _is_item_too_large(message: Optional[str]) -> bool:
+    # Determine if a per-item error indicates the envelope exceeded the ingestion size limit.
+    if not message:
+        return False
+    lowered = message.lower()
+    return any(marker in lowered for marker in _ITEM_TOO_LARGE_MESSAGE_MARKERS)

sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/export/_base.py

@@ -49,7 +49,7 @@
 from azure.monitor.opentelemetry.exporter._connection_string_parser import (
     ConnectionStringParser,
 )
-from azure.monitor.opentelemetry.exporter._storage import LocalFileStorage
+from azure.monitor.opentelemetry.exporter._storage import LocalFileStorage, StorageExportResult
 from azure.monitor.opentelemetry.exporter._configuration._state import (
     get_configuration_manager,
 )
@@ -63,6 +63,8 @@
     _get_os_name,
     _get_rp_name,
     _get_attach_type_name,
+    _is_item_too_large,
+    _split_oversized_batch,
 )
 from azure.monitor.opentelemetry.exporter._version import VERSION as ext_version
 from azure.monitor.opentelemetry.exporter.export._rate_limiter import (
@@ -317,6 +319,45 @@ def _handle_transmit_from_storage(self, envelopes: List[TelemetryItem], result:
             if self._should_collect_customer_sdkstats():
                 track_dropped_items(envelopes, DropCode.CLIENT_STORAGE_DISABLED)
 
+    def _handle_payload_too_large(
+        self, envelopes: List[TelemetryItem], response_error: HttpResponseError
+    ) -> ExportResult:
+        # Handle a 413 (Payload Too Large) response by splitting and persisting for retry.
+        if not self.storage:
+            if not self._is_stats_exporter():
+                logger.debug(
+                    "Payload too large but offline storage is disabled; dropping %d envelope(s): %s.",
+                    len(envelopes),
+                    response_error.message,
+                )
+            return ExportResult.FAILED_NOT_RETRYABLE
+        if len(envelopes) <= 1:
+            if not self._is_stats_exporter():
+                logger.warning(
+                    "Payload too large and cannot be split further; dropping %d envelope(s): %s.",
+                    len(envelopes),
+                    response_error.message,
+                )
+                if self._should_collect_customer_sdkstats() and isinstance(response_error.status_code, int):
+                    track_dropped_items(envelopes, response_error.status_code)
+            return ExportResult.FAILED_NOT_RETRYABLE
+
+        for sub_batch in _split_oversized_batch(envelopes):
+            result_from_storage_put = self.storage.put([x.as_dict() for x in sub_batch])
+            if result_from_storage_put == StorageExportResult.LOCAL_FILE_BLOB_SUCCESS:
+                if not self._is_stats_exporter() and self._should_collect_customer_sdkstats():
+                    track_retry_items(sub_batch, response_error)
+            else:
+                if not self._is_stats_exporter():
+                    logger.debug(
+                        "Failed to persist a split sub-batch of %d envelope(s) after a 413; "
+                        "these envelopes are dropped.",
+                        len(sub_batch),
+                    )
+                    if self._should_collect_customer_sdkstats():
+                        track_dropped_items_from_storage(result_from_storage_put, sub_batch)
+        return ExportResult.FAILED_NOT_RETRYABLE
+
     # pylint: disable=too-many-branches
     # pylint: disable=too-many-nested-blocks
     # pylint: disable=too-many-statements
@@ -446,11 +487,17 @@ def _transmit(self, envelopes: List[TelemetryItem], _skip_rate_limit: bool = Fal
                                         and isinstance(error.status_code, int)
                                     ):
                                         track_dropped_items([envelopes[error.index]], error.status_code)
+                                dropped_envelope = envelopes[error.index] if error.index is not None else None
+                                # When the server rejects an item for exceeding the per-item size limit
+                                # (detected via _is_item_too_large), omit its payload from the log: it is
+                                # too large to dump and may contain PII, which would leak and flood the logs.
+                                if dropped_envelope is None or _is_item_too_large(error.message):
+                                    dropped_envelope = ""  # type: ignore[assignment]
                                 logger.error(
                                     "Data drop %s: %s %s.",
                                     error.status_code,
                                     error.message,
-                                    (envelopes[error.index] if error.index is not None else ""),
+                                    dropped_envelope,
                                 )
                     storage = self.storage
                     if storage and resend_envelopes:
@@ -577,6 +624,11 @@ def _transmit(self, envelopes: List[TelemetryItem], _skip_rate_limit: bool = Fal
                         if self._should_collect_stats():
                             _update_requests_map(_REQ_EXCEPTION_NAME[1], value="Circular Redirect")
                         result = ExportResult.FAILED_NOT_RETRYABLE
+                elif response_error.status_code == 413:
+                    if self._should_collect_stats():
+                        _update_requests_map(_REQ_FAILURE_NAME[1], value=response_error.status_code)
+                    result = self._handle_payload_too_large(envelopes, response_error)
+
                 else:
                     # Any other status code counts as failure (non-retryable)
                     # 400 - Invalid - The server cannot or will not process the request due to the invalid telemetry (invalid data, iKey, etc.)

sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/statsbeat/customer/_utils.py

@@ -111,7 +111,7 @@ def _determine_client_retry_code(
         ConnectionError,
         OSError,
     )
-    if hasattr(error, "status_code") and error.status_code in [401, 403, 408, 429, 500, 502, 503, 504]:
+    if hasattr(error, "status_code") and error.status_code in [401, 403, 408, 413, 429, 500, 502, 503, 504]:
         # For specific status codes, preserve the custom message if available
         error_message = getattr(error, "message", None) if hasattr(error, "message") else None
         return (error.status_code, error_message or _UNKNOWN)

sdk/monitor/azure-monitor-opentelemetry-exporter/tests/customer_sdk_stats/test_utlities.py

@@ -286,7 +286,7 @@ def test_categorize_status_code_other(self):
 
     def test_determine_client_retry_code_http_errors(self):
         """Test determining retry code for HTTP errors with status codes."""
-        test_status_codes = [401, 403, 408, 429, 500, 502, 503, 504]
+        test_status_codes = [401, 403, 408, 413, 429, 500, 502, 503, 504]
 
         for status_code in test_status_codes:
             with self.subTest(status_code=status_code):