From 9581cc676e4179bbbb102c2b3041d119188ae14e Mon Sep 17 00:00:00 2001 From: Radhika Gupta Date: Wed, 5 Aug 2026 11:43:35 -0700 Subject: [PATCH 01/18] Add support for 413 handling --- .../CHANGELOG.md | 3 + .../opentelemetry/exporter/export/_base.py | 24 +++++++ .../tests/test_base_exporter.py | 72 +++++++++++++++++++ 3 files changed, 99 insertions(+) diff --git a/sdk/monitor/azure-monitor-opentelemetry-exporter/CHANGELOG.md b/sdk/monitor/azure-monitor-opentelemetry-exporter/CHANGELOG.md index b408f3b9d37c..73e75cfac91b 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-exporter/CHANGELOG.md +++ b/sdk/monitor/azure-monitor-opentelemetry-exporter/CHANGELOG.md @@ -33,6 +33,9 @@ ### Features Added - Retry payloads for which request is sent successfully but no response is received, per [SPEC.](https://github.com/aep-health-and-standards/Telemetry-Collection-Spec/pull/1018) ([#47870](https://github.com/Azure/azure-sdk-for-python/pull/47870)) +- Handle `413 Payload Too Large` responses by splitting the batch and persisting each half 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. ### Bugs Fixed - Propagate main agent attribute to child spans diff --git a/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/export/_base.py b/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/export/_base.py index c06142d68647..8cd55900ac39 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/export/_base.py +++ b/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/export/_base.py @@ -577,6 +577,30 @@ 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) + if self.storage and len(envelopes) > 1: + midpoint = len(envelopes) // 2 + for half in (envelopes[:midpoint], envelopes[midpoint:]): + envelopes_to_store = [x.as_dict() for x in half] + self.storage.put(envelopes_to_store) + if self._should_collect_customer_sdkstats(): + # These halves are persisted for retry at a smaller size. + track_retry_items(half, response_error) + else: + # Single envelope (or storage disabled): cannot split, so drop it. + if not self._is_stats_exporter(): + logger.error( + "Payload too large and cannot be split further: %s.", + response_error.message, + ) + if self._should_collect_customer_sdkstats() and isinstance( + response_error.status_code, int + ): + track_dropped_items(envelopes, response_error.status_code) + # Mark as not retryable because we already write to storage here + result = ExportResult.FAILED_NOT_RETRYABLE 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.) diff --git a/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/test_base_exporter.py b/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/test_base_exporter.py index 943b13251182..c1138ed6fe0f 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/test_base_exporter.py +++ b/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/test_base_exporter.py @@ -1299,6 +1299,78 @@ def test_transmission_408(self): result = self._base._transmit(self._envelopes_to_export) self.assertEqual(result, ExportResult.FAILED_RETRYABLE) + def test_transmission_413_splits_batch(self): + """A 413 with more than one envelope should split the batch in half and + persist each half as a separate blob so they are retried at a smaller size.""" + exporter = BaseExporter(disable_offline_storage=True) + exporter.storage = mock.Mock() + custom_envelopes_to_export = [ + TelemetryItem(name="Test1", time=datetime.now()), + TelemetryItem(name="Test2", time=datetime.now()), + TelemetryItem(name="Test3", time=datetime.now()), + TelemetryItem(name="Test4", time=datetime.now()), + ] + with mock.patch.object(AzureMonitorClient, "track", side_effect=_make_http_response_error(413)): + result = exporter._transmit(custom_envelopes_to_export) + self.assertEqual(result, ExportResult.FAILED_NOT_RETRYABLE) + # Two separate blobs (one per half) so the halves are retried independently. + self.assertEqual(exporter.storage.put.call_count, 2) + + def test_transmission_413_single_envelope_dropped(self): + """A 413 with a single envelope cannot be split further and should be dropped.""" + exporter = BaseExporter(disable_offline_storage=True) + exporter.storage = mock.Mock() + single_envelope = [TelemetryItem(name="Test", time=datetime.now())] + with mock.patch.object(AzureMonitorClient, "track", side_effect=_make_http_response_error(413)): + result = exporter._transmit(single_envelope) + self.assertEqual(result, ExportResult.FAILED_NOT_RETRYABLE) + exporter.storage.put.assert_not_called() + + def test_transmission_413_storage_disabled_dropped(self): + """A 413 with storage disabled cannot be persisted and should be dropped.""" + exporter = BaseExporter(disable_offline_storage=True) + exporter.storage = None + custom_envelopes_to_export = [ + TelemetryItem(name="Test1", time=datetime.now()), + TelemetryItem(name="Test2", time=datetime.now()), + ] + with mock.patch.object(AzureMonitorClient, "track", side_effect=_make_http_response_error(413)): + result = exporter._transmit(custom_envelopes_to_export) + self.assertEqual(result, ExportResult.FAILED_NOT_RETRYABLE) + + def test_transmission_413_persists_and_retries(self): + """End-to-end with real on-disk storage: a 413 persists the split halves to + disk, and a later successful drain reads them back and retries them.""" + exporter = self._base + clean_folder(exporter.storage._path) + custom_envelopes_to_export = [ + TelemetryItem(name="Test1", time=datetime.now()), + TelemetryItem(name="Test2", time=datetime.now()), + TelemetryItem(name="Test3", time=datetime.now()), + TelemetryItem(name="Test4", time=datetime.now()), + ] + # First transmit gets a 413 -> the batch is split and each half is persisted. + with mock.patch.object(AzureMonitorClient, "track", side_effect=_make_http_response_error(413)): + result = exporter._transmit(custom_envelopes_to_export) + self.assertEqual(result, ExportResult.FAILED_NOT_RETRYABLE) + # Two separate blobs were written to disk (one per half). + blobs = [f for f in os.listdir(exporter.storage._path) if ".blob" in f] + self.assertEqual(len(blobs), 2) + # Strip the retry lease so the freshly-persisted blobs are drainable now. + for name in os.listdir(exporter.storage._path): + if name.endswith(".lock"): + src = os.path.join(exporter.storage._path, name) + dst = os.path.join(exporter.storage._path, name[: name.rindex("@")]) + os.rename(src, dst) + # Now the endpoint accepts data -> draining retries the persisted halves. + with mock.patch.object(AzureMonitorClient, "track") as post: + post.return_value = TrackResponse(items_received=2, items_accepted=2, errors=[]) + exporter._transmit_from_storage() + # Both halves were re-sent and their blobs removed from disk. + self.assertEqual(post.call_count, 2) + remaining = [f for f in os.listdir(exporter.storage._path) if ".blob" in f] + self.assertEqual(len(remaining), 0) + def test_transmission_429(self): with mock.patch.object(AzureMonitorClient, "track", side_effect=_make_http_response_error(429)): result = self._base._transmit(self._envelopes_to_export) From 1a4cbb9a69af62e767c72b8be3260b2f6e096ad0 Mon Sep 17 00:00:00 2001 From: Radhika Gupta Date: Wed, 12 Aug 2026 10:16:33 -0700 Subject: [PATCH 02/18] Add support for 413 - payload too large handling --- .../opentelemetry/exporter/export/_base.py | 139 +++++++++++--- .../tests/test_base_exporter.py | 180 ++++++++++++++++++ 2 files changed, 292 insertions(+), 27 deletions(-) diff --git a/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/export/_base.py b/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/export/_base.py index 8cd55900ac39..37c6c186503c 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/export/_base.py +++ b/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/export/_base.py @@ -1,6 +1,7 @@ # Copyright (c) Microsoft Corporation. All rights reserved. # Licensed under the MIT License. import getpass +import json import logging import os import tempfile @@ -27,6 +28,9 @@ from azure.monitor.opentelemetry.exporter._generated.exporter.models import ( TelemetryItem, ) +from azure.monitor.opentelemetry.exporter._generated.exporter._utils.model_base import ( + SdkJSONEncoder, +) from azure.monitor.opentelemetry.exporter._constants import ( _ALLOWED_REDIRECT_DOMAIN_SUFFIXES, _AZURE_MONITOR_DISTRO_VERSION_ARG, @@ -95,6 +99,15 @@ _TEMPDIR_PREFIX = "opentelemetry-python-" _SERVICE_API_LATEST = "2020-09-15_Preview" +# Maximum request body size (in bytes) the ingestion endpoint accepts on the wire before +# responding with 413 (Payload Too Large). +_MAX_INGESTION_PAYLOAD_SIZE_BYTES = 30 * 1000 * 1000 + +# Maximum size (in bytes) of a single telemetry item the ingestion endpoint accepts. Items +# larger than this are rejected per-item (surfaced as a 400 within a 206 partial response) and +# cannot be split further. ~1.2 MiB. +_MAX_INGESTION_ITEM_SIZE_BYTES = 1258291 + class ExportResult(Enum): SUCCESS = 0 @@ -317,6 +330,41 @@ 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 len(envelopes) <= 1 or not self.storage: + 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): + try: + self.storage.put([x.as_dict() for x in sub_batch]) + except Exception: # pylint: disable=broad-except + # Persisting a split sub-batch failed (e.g. disk full or a serialization + # error). Drop just this sub-batch rather than aborting the whole split so the + # remaining sub-batches still get a chance to be persisted and retried. + if not self._is_stats_exporter(): + logger.warning( # pylint: disable=do-not-log-exceptions-if-not-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() and isinstance(response_error.status_code, int): + track_dropped_items(sub_batch, response_error.status_code) + continue + if self._should_collect_customer_sdkstats(): + track_retry_items(sub_batch, response_error) + return ExportResult.FAILED_NOT_RETRYABLE + # pylint: disable=too-many-branches # pylint: disable=too-many-nested-blocks # pylint: disable=too-many-statements @@ -446,12 +494,31 @@ 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) - logger.error( - "Data drop %s: %s %s.", - error.status_code, - error.message, - (envelopes[error.index] if error.index is not None else ""), - ) + dropped_envelope = envelopes[error.index] if error.index is not None else None + # An envelope that exceeds the per-item ingestion size limit (~1.2 MiB) + # cannot be split further and is dropped here. Dumping it would flood the + # log with a huge (and potentially PII-laden) payload, so log a concise + # summary instead. For all other drops keep the existing behavior of + # logging the full envelope to aid debugging. + if ( + dropped_envelope is not None + and _get_envelope_serialized_size(dropped_envelope) + > _MAX_INGESTION_ITEM_SIZE_BYTES + ): + logger.error( + "Data drop %s: %s. Envelope of type %s exceeds the per-item " + "ingestion size limit and is omitted from the log.", + error.status_code, + error.message, + dropped_envelope.name, + ) + else: + logger.error( + "Data drop %s: %s. %s", + error.status_code, + error.message, + (dropped_envelope if dropped_envelope is not None else ""), + ) storage = self.storage if storage and resend_envelopes: envelopes_to_store = [x.as_dict() for x in resend_envelopes] @@ -580,27 +647,8 @@ def _transmit(self, envelopes: List[TelemetryItem], _skip_rate_limit: bool = Fal elif response_error.status_code == 413: if self._should_collect_stats(): _update_requests_map(_REQ_FAILURE_NAME[1], value=response_error.status_code) - if self.storage and len(envelopes) > 1: - midpoint = len(envelopes) // 2 - for half in (envelopes[:midpoint], envelopes[midpoint:]): - envelopes_to_store = [x.as_dict() for x in half] - self.storage.put(envelopes_to_store) - if self._should_collect_customer_sdkstats(): - # These halves are persisted for retry at a smaller size. - track_retry_items(half, response_error) - else: - # Single envelope (or storage disabled): cannot split, so drop it. - if not self._is_stats_exporter(): - logger.error( - "Payload too large and cannot be split further: %s.", - response_error.message, - ) - if self._should_collect_customer_sdkstats() and isinstance( - response_error.status_code, int - ): - track_dropped_items(envelopes, response_error.status_code) - # Mark as not retryable because we already write to storage here - result = ExportResult.FAILED_NOT_RETRYABLE + 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.) @@ -830,6 +878,43 @@ def _host(netloc: str) -> str: return False +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) + # Every element after the first is preceded by a "," separator inside the JSON array. + separator_size = 1 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) + # Guarantee progress: 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. Only split when there is more than one envelope; a single envelope cannot be divided + # and is handled by the caller (which drops it), so this never produces an empty sub-batch. + if len(chunks) < 2 and len(envelopes) > 1: + midpoint = len(envelopes) // 2 + chunks = [envelopes[:midpoint], envelopes[midpoint:]] + return chunks + + def _is_invalid_code(response_code: Optional[int]) -> bool: """Determine if response is a invalid response. diff --git a/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/test_base_exporter.py b/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/test_base_exporter.py index c1138ed6fe0f..7bd5e72cf928 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/test_base_exporter.py +++ b/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/test_base_exporter.py @@ -16,6 +16,10 @@ BaseExporter, ExportResult, _get_storage_directory, + _get_envelope_serialized_size, + _split_oversized_batch, + _MAX_INGESTION_PAYLOAD_SIZE_BYTES, + _MAX_INGESTION_ITEM_SIZE_BYTES, ) from azure.monitor.opentelemetry.exporter._utils import _get_retry_delay_from_headers from azure.monitor.opentelemetry.exporter._storage import StorageExportResult @@ -1115,6 +1119,77 @@ def test_transmission_206_no_retry(self): self.assertEqual(result, ExportResult.FAILED_NOT_RETRYABLE) exporter.storage.put.assert_not_called() + def test_transmission_206_oversized_item_omits_envelope_from_log(self): + """An item that exceeds the per-item ingestion size limit is dropped with a concise log + message; the full (potentially large / PII-laden) envelope must NOT be dumped to the log.""" + exporter = BaseExporter(disable_offline_storage=True) + exporter.storage = mock.Mock() + oversized_envelope = TelemetryItem( + name="OversizedItem", + time=datetime.now(), + tags={"marker": "SECRET_PAYLOAD_MARKER"}, + ) + custom_envelopes_to_export = [oversized_envelope] + with mock.patch.object(AzureMonitorClient, "track") as post, mock.patch( + "azure.monitor.opentelemetry.exporter.export._base._get_envelope_serialized_size", + return_value=_MAX_INGESTION_ITEM_SIZE_BYTES + 1, + ): + post.return_value = TrackResponse( + items_received=1, + items_accepted=0, + errors=[ + TelemetryErrorDetails( + index=0, + status_code=400, + message="Telemetry item length must not exceed 1258291", + ), + ], + ) + with self.assertLogs( + "azure.monitor.opentelemetry.exporter.export._base", level="ERROR" + ) as cm: + result = exporter._transmit(custom_envelopes_to_export) + self.assertEqual(result, ExportResult.FAILED_NOT_RETRYABLE) + log_output = "\n".join(cm.output) + # Concise summary is logged (status, server message, and envelope type only). + self.assertIn("exceeds the per-item ingestion size limit and is omitted from the log", log_output) + self.assertIn("OversizedItem", log_output) + # The full envelope payload (its tags/data) must NOT be dumped to the log. + self.assertNotIn("SECRET_PAYLOAD_MARKER", log_output) + + def test_transmission_206_normal_drop_logs_full_envelope(self): + """A non-retryable drop for a normally-sized item keeps the existing behavior of logging + the full envelope to aid debugging.""" + exporter = BaseExporter(disable_offline_storage=True) + exporter.storage = mock.Mock() + dropped_envelope = TelemetryItem( + name="NormalItem", + time=datetime.now(), + tags={"marker": "SECRET_PAYLOAD_MARKER"}, + ) + custom_envelopes_to_export = [dropped_envelope] + with mock.patch.object(AzureMonitorClient, "track") as post: + post.return_value = TrackResponse( + items_received=1, + items_accepted=0, + errors=[ + TelemetryErrorDetails( + index=0, + status_code=400, + message="should drop", + ), + ], + ) + with self.assertLogs( + "azure.monitor.opentelemetry.exporter.export._base", level="ERROR" + ) as cm: + result = exporter._transmit(custom_envelopes_to_export) + self.assertEqual(result, ExportResult.FAILED_NOT_RETRYABLE) + log_output = "\n".join(cm.output) + # The full envelope (including its tags) is logged for a normally-sized drop. + self.assertIn("SECRET_PAYLOAD_MARKER", log_output) + self.assertNotIn("omitted from the log", log_output) + def test_transmission_206_sampling_rejection_not_retried(self): """Test that items rejected due to ingestion sampling are not retried or persisted.""" exporter = BaseExporter(disable_offline_storage=True) @@ -1371,6 +1446,111 @@ def test_transmission_413_persists_and_retries(self): remaining = [f for f in os.listdir(exporter.storage._path) if ".blob" in f] self.assertEqual(len(remaining), 0) + def test_transmission_413_sub_batch_persist_failure_drops_that_batch_only(self): + """If persisting one split sub-batch fails during a 413, that sub-batch is dropped + (with a warning) while the remaining sub-batches are still attempted and persisted.""" + exporter = BaseExporter(disable_offline_storage=True) + exporter.storage = mock.Mock() + # First sub-batch persist raises, second succeeds -> both are attempted (the failure + # must not abort the loop) and only the failed sub-batch is dropped. + exporter.storage.put.side_effect = [OSError("disk full"), None] + custom_envelopes_to_export = [ + TelemetryItem(name="Test1", time=datetime.now()), + TelemetryItem(name="Test2", time=datetime.now()), + TelemetryItem(name="Test3", time=datetime.now()), + TelemetryItem(name="Test4", time=datetime.now()), + ] + with mock.patch.object(AzureMonitorClient, "track", side_effect=_make_http_response_error(413)): + with self.assertLogs( + "azure.monitor.opentelemetry.exporter.export._base", level="WARNING" + ) as cm: + result = exporter._transmit(custom_envelopes_to_export) + self.assertEqual(result, ExportResult.FAILED_NOT_RETRYABLE) + # Both sub-batches were attempted despite the first failing (loop did not abort). + self.assertEqual(exporter.storage.put.call_count, 2) + # The failed sub-batch is reported via a warning. + self.assertTrue(any("Failed to persist a split sub-batch" in line for line in cm.output)) + + def test_get_envelope_serialized_size_matches_wire(self): + """The measured size must be > 0 and match the exact bytes the transport puts on the + wire (same encoder + read-only exclusion).""" + import json + from azure.monitor.opentelemetry.exporter._generated.exporter._utils.model_base import SdkJSONEncoder + + envelope = TelemetryItem(name="Test", time=datetime.now()) + expected = len(json.dumps(envelope, cls=SdkJSONEncoder, exclude_readonly=True).encode("utf-8")) + size = _get_envelope_serialized_size(envelope) + self.assertGreater(size, 0) + self.assertEqual(size, expected) + + def test_get_envelope_serialized_size_failure_returns_max(self): + """If an envelope cannot be serialized, sizing returns the max payload size so the item + is isolated into its own sub-batch instead of being counted as weightless.""" + envelope = TelemetryItem(name="Test", time=datetime.now()) + with mock.patch( + "azure.monitor.opentelemetry.exporter.export._base.json.dumps", + side_effect=TypeError("not serializable"), + ): + size = _get_envelope_serialized_size(envelope) + self.assertEqual(size, _MAX_INGESTION_PAYLOAD_SIZE_BYTES) + + def test_split_oversized_batch_empty(self): + """Splitting an empty batch yields no sub-batches (and never an empty half).""" + self.assertEqual(_split_oversized_batch([]), []) + + def test_split_oversized_batch_single_envelope_no_empty_half(self): + """A single envelope cannot be divided; it is returned as one sub-batch with no empty + second half.""" + single = [TelemetryItem(name="Test", time=datetime.now())] + chunks = _split_oversized_batch(single) + self.assertEqual(len(chunks), 1) + self.assertEqual(len(chunks[0]), 1) + + def test_split_oversized_batch_isolates_oversized_envelopes(self): + """When each envelope already fills the budget, every envelope is packed into its own + sub-batch.""" + envelopes = [ + TelemetryItem(name="Test1", time=datetime.now()), + TelemetryItem(name="Test2", time=datetime.now()), + TelemetryItem(name="Test3", time=datetime.now()), + ] + with mock.patch( + "azure.monitor.opentelemetry.exporter.export._base._get_envelope_serialized_size", + return_value=_MAX_INGESTION_PAYLOAD_SIZE_BYTES, + ): + chunks = _split_oversized_batch(envelopes) + self.assertEqual([len(c) for c in chunks], [1, 1, 1]) + + def test_split_oversized_batch_packs_multiple_chunks_by_size(self): + """Size-based packing groups multiple envelopes per sub-batch and starts a new + sub-batch once the running size would exceed the ingestion limit.""" + envelopes = [TelemetryItem(name="Test{}".format(i), time=datetime.now()) for i in range(5)] + # Each envelope reports ~40% of the limit -> at most 2 per chunk (2*0.4 fits, + # 3*0.4 exceeds), yielding chunks of [2, 2, 1]. + per_item = int(_MAX_INGESTION_PAYLOAD_SIZE_BYTES * 0.4) + with mock.patch( + "azure.monitor.opentelemetry.exporter.export._base._get_envelope_serialized_size", + return_value=per_item, + ): + chunks = _split_oversized_batch(envelopes) + self.assertEqual([len(c) for c in chunks], [2, 2, 1]) + + def test_split_oversized_batch_falls_back_to_count_halving(self): + """If the byte estimate under-reports and everything fits in one chunk, the batch is + still halved by count so repeated 413s keep shrinking it.""" + envelopes = [ + TelemetryItem(name="Test1", time=datetime.now()), + TelemetryItem(name="Test2", time=datetime.now()), + TelemetryItem(name="Test3", time=datetime.now()), + TelemetryItem(name="Test4", time=datetime.now()), + ] + with mock.patch( + "azure.monitor.opentelemetry.exporter.export._base._get_envelope_serialized_size", + return_value=1, + ): + chunks = _split_oversized_batch(envelopes) + self.assertEqual([len(c) for c in chunks], [2, 2]) + def test_transmission_429(self): with mock.patch.object(AzureMonitorClient, "track", side_effect=_make_http_response_error(429)): result = self._base._transmit(self._envelopes_to_export) From 4ec3c1dd9ff39cf3fa03cc18fe72f2c53895b98e Mon Sep 17 00:00:00 2001 From: Radhika Gupta Date: Wed, 12 Aug 2026 10:21:08 -0700 Subject: [PATCH 03/18] Add CHANGELOG --- sdk/monitor/azure-monitor-opentelemetry-exporter/CHANGELOG.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/sdk/monitor/azure-monitor-opentelemetry-exporter/CHANGELOG.md b/sdk/monitor/azure-monitor-opentelemetry-exporter/CHANGELOG.md index 73e75cfac91b..f49efbbcdd1e 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-exporter/CHANGELOG.md +++ b/sdk/monitor/azure-monitor-opentelemetry-exporter/CHANGELOG.md @@ -3,6 +3,8 @@ ## 1.0.0b58 (Unreleased) ### Features Added +- Retry payloads for which request is sent successfully but no response is received, per [SPEC.](https://github.com/aep-health-and-standards/Telemetry-Collection-Spec/pull/1018) + ([#47870](https://github.com/Azure/azure-sdk-for-python/pull/47870)) ### Breaking Changes From a1b8914fbddc70cd9e23dd64194f660a56e9a311 Mon Sep 17 00:00:00 2001 From: Radhika Gupta Date: Thu, 13 Aug 2026 12:41:21 -0700 Subject: [PATCH 04/18] Address feedback --- .../opentelemetry/exporter/export/_base.py | 38 +++++++++++-------- .../tests/test_base_exporter.py | 33 ++++++++++++---- 2 files changed, 47 insertions(+), 24 deletions(-) diff --git a/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/export/_base.py b/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/export/_base.py index 37c6c186503c..a881f80749c1 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/export/_base.py +++ b/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/export/_base.py @@ -53,7 +53,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, ) @@ -334,7 +334,17 @@ 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 len(envelopes) <= 1 or not self.storage: + 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, + ) + if self._should_collect_customer_sdkstats(): + track_dropped_items(envelopes, DropCode.CLIENT_STORAGE_DISABLED) + 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.", @@ -346,23 +356,19 @@ def _handle_payload_too_large( return ExportResult.FAILED_NOT_RETRYABLE for sub_batch in _split_oversized_batch(envelopes): - try: - self.storage.put([x.as_dict() for x in sub_batch]) - except Exception: # pylint: disable=broad-except - # Persisting a split sub-batch failed (e.g. disk full or a serialization - # error). Drop just this sub-batch rather than aborting the whole split so the - # remaining sub-batches still get a chance to be persisted and retried. + 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 self._should_collect_customer_sdkstats(): + track_retry_items(sub_batch, response_error) + else: if not self._is_stats_exporter(): - logger.warning( # pylint: disable=do-not-log-exceptions-if-not-debug + 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() and isinstance(response_error.status_code, int): - track_dropped_items(sub_batch, response_error.status_code) - continue - if self._should_collect_customer_sdkstats(): - track_retry_items(sub_batch, response_error) + 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 @@ -894,8 +900,8 @@ def _split_oversized_batch(envelopes: List[TelemetryItem]) -> List[List[Telemetr current_size = 2 # account for the enclosing "[" and "]" of the JSON array for envelope in envelopes: envelope_size = _get_envelope_serialized_size(envelope) - # Every element after the first is preceded by a "," separator inside the JSON array. - separator_size = 1 if current else 0 + # 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 = [] diff --git a/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/test_base_exporter.py b/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/test_base_exporter.py index 7bd5e72cf928..a01bda13fb2f 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/test_base_exporter.py +++ b/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/test_base_exporter.py @@ -23,6 +23,7 @@ ) from azure.monitor.opentelemetry.exporter._utils import _get_retry_delay_from_headers from azure.monitor.opentelemetry.exporter._storage import StorageExportResult +from azure.monitor.opentelemetry.exporter._constants import DropCode from azure.monitor.opentelemetry.exporter.statsbeat._state import ( _REQUESTS_MAP, _STATSBEAT_STATE, @@ -1379,6 +1380,7 @@ def test_transmission_413_splits_batch(self): persist each half as a separate blob so they are retried at a smaller size.""" exporter = BaseExporter(disable_offline_storage=True) exporter.storage = mock.Mock() + exporter.storage.put.return_value = StorageExportResult.LOCAL_FILE_BLOB_SUCCESS custom_envelopes_to_export = [ TelemetryItem(name="Test1", time=datetime.now()), TelemetryItem(name="Test2", time=datetime.now()), @@ -1402,7 +1404,8 @@ def test_transmission_413_single_envelope_dropped(self): exporter.storage.put.assert_not_called() def test_transmission_413_storage_disabled_dropped(self): - """A 413 with storage disabled cannot be persisted and should be dropped.""" + """A 413 with storage disabled cannot be persisted and should be dropped with the + CLIENT_STORAGE_DISABLED reason, consistent with other retryable codes.""" exporter = BaseExporter(disable_offline_storage=True) exporter.storage = None custom_envelopes_to_export = [ @@ -1410,8 +1413,17 @@ def test_transmission_413_storage_disabled_dropped(self): TelemetryItem(name="Test2", time=datetime.now()), ] with mock.patch.object(AzureMonitorClient, "track", side_effect=_make_http_response_error(413)): - result = exporter._transmit(custom_envelopes_to_export) + with mock.patch.object(exporter, "_should_collect_customer_sdkstats", return_value=True): + with mock.patch( + "azure.monitor.opentelemetry.exporter.export._base.track_dropped_items" + ) as mock_track_dropped: + result = exporter._transmit(custom_envelopes_to_export) self.assertEqual(result, ExportResult.FAILED_NOT_RETRYABLE) + # The batch is dropped with the storage-disabled reason, matching every other + # retryable failure when offline storage is turned off. + mock_track_dropped.assert_called_once_with( + custom_envelopes_to_export, DropCode.CLIENT_STORAGE_DISABLED + ) def test_transmission_413_persists_and_retries(self): """End-to-end with real on-disk storage: a 413 persists the split halves to @@ -1447,13 +1459,18 @@ def test_transmission_413_persists_and_retries(self): self.assertEqual(len(remaining), 0) def test_transmission_413_sub_batch_persist_failure_drops_that_batch_only(self): - """If persisting one split sub-batch fails during a 413, that sub-batch is dropped - (with a warning) while the remaining sub-batches are still attempted and persisted.""" + """storage.put reports failure by return value (not by raising). If persisting one + split sub-batch fails during a 413, that sub-batch is dropped (with a warning) while + the remaining sub-batches are still attempted and persisted.""" exporter = BaseExporter(disable_offline_storage=True) exporter.storage = mock.Mock() - # First sub-batch persist raises, second succeeds -> both are attempted (the failure - # must not abort the loop) and only the failed sub-batch is dropped. - exporter.storage.put.side_effect = [OSError("disk full"), None] + # First sub-batch persist reports a failure result, second succeeds -> both are + # attempted (a failing sub-batch must not abort the loop) and only the failed one + # is dropped rather than counted as a pending retry. + exporter.storage.put.side_effect = [ + StorageExportResult.CLIENT_PERSISTENCE_CAPACITY_REACHED, + StorageExportResult.LOCAL_FILE_BLOB_SUCCESS, + ] custom_envelopes_to_export = [ TelemetryItem(name="Test1", time=datetime.now()), TelemetryItem(name="Test2", time=datetime.now()), @@ -1462,7 +1479,7 @@ def test_transmission_413_sub_batch_persist_failure_drops_that_batch_only(self): ] with mock.patch.object(AzureMonitorClient, "track", side_effect=_make_http_response_error(413)): with self.assertLogs( - "azure.monitor.opentelemetry.exporter.export._base", level="WARNING" + "azure.monitor.opentelemetry.exporter.export._base", level="DEBUG" ) as cm: result = exporter._transmit(custom_envelopes_to_export) self.assertEqual(result, ExportResult.FAILED_NOT_RETRYABLE) From 5bff63371315e8e645e32daad5520bc1fa76e91c Mon Sep 17 00:00:00 2001 From: Radhika Gupta Date: Thu, 13 Aug 2026 13:58:08 -0700 Subject: [PATCH 05/18] Update changelog --- sdk/monitor/azure-monitor-opentelemetry-exporter/CHANGELOG.md | 3 --- 1 file changed, 3 deletions(-) diff --git a/sdk/monitor/azure-monitor-opentelemetry-exporter/CHANGELOG.md b/sdk/monitor/azure-monitor-opentelemetry-exporter/CHANGELOG.md index f49efbbcdd1e..173a8e5ece86 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-exporter/CHANGELOG.md +++ b/sdk/monitor/azure-monitor-opentelemetry-exporter/CHANGELOG.md @@ -35,9 +35,6 @@ ### Features Added - Retry payloads for which request is sent successfully but no response is received, per [SPEC.](https://github.com/aep-health-and-standards/Telemetry-Collection-Spec/pull/1018) ([#47870](https://github.com/Azure/azure-sdk-for-python/pull/47870)) -- Handle `413 Payload Too Large` responses by splitting the batch and persisting each half 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. ### Bugs Fixed - Propagate main agent attribute to child spans From 2a149fd3d57d4757aaad36ada92570872fe82d63 Mon Sep 17 00:00:00 2001 From: Radhika Gupta Date: Thu, 13 Aug 2026 16:43:13 -0700 Subject: [PATCH 06/18] Fix format --- .../opentelemetry/exporter/export/_base.py | 29 +++++++++++++++---- .../tests/test_base_exporter.py | 16 +++------- 2 files changed, 28 insertions(+), 17 deletions(-) diff --git a/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/export/_base.py b/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/export/_base.py index a881f80749c1..fcb093c2b698 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/export/_base.py +++ b/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/export/_base.py @@ -333,7 +333,15 @@ def _handle_transmit_from_storage(self, envelopes: List[TelemetryItem], result: 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.""" + """Handle a 413 (Payload Too Large) response by splitting and persisting for retry. + + :param envelopes: The telemetry envelopes that were rejected as too large. + :type envelopes: list[~azure.monitor.opentelemetry.exporter._generated.models.TelemetryItem] + :param response_error: The error returned for the oversized request. + :type response_error: ~azure.core.exceptions.HttpResponseError + :return: The export result for the oversized batch. + :rtype: ~azure.monitor.opentelemetry.exporter.export._base.ExportResult + """ if not self.storage: if not self._is_stats_exporter(): logger.debug( @@ -508,8 +516,7 @@ def _transmit(self, envelopes: List[TelemetryItem], _skip_rate_limit: bool = Fal # logging the full envelope to aid debugging. if ( dropped_envelope is not None - and _get_envelope_serialized_size(dropped_envelope) - > _MAX_INGESTION_ITEM_SIZE_BYTES + and _get_envelope_serialized_size(dropped_envelope) > _MAX_INGESTION_ITEM_SIZE_BYTES ): logger.error( "Data drop %s: %s. Envelope of type %s exceeds the per-item " @@ -885,7 +892,13 @@ def _host(netloc: str) -> str: def _get_envelope_serialized_size(envelope: TelemetryItem) -> int: - """Return the serialized wire size (in bytes) of a single envelope on the wire.""" + """Return the serialized wire size (in bytes) of a single envelope on the wire. + + :param envelope: The telemetry envelope to measure. + :type envelope: ~azure.monitor.opentelemetry.exporter._generated.models.TelemetryItem + :return: The serialized size of the envelope in bytes. + :rtype: int + """ try: serialized = json.dumps(envelope, cls=SdkJSONEncoder, exclude_readonly=True) return len(serialized.encode("utf-8")) @@ -894,7 +907,13 @@ def _get_envelope_serialized_size(envelope: TelemetryItem) -> int: 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.""" + """Split an oversized batch into sub-batches that each fit under the ingestion size limit. + + :param envelopes: The oversized batch of telemetry envelopes to split. + :type envelopes: list[~azure.monitor.opentelemetry.exporter._generated.models.TelemetryItem] + :return: A list of sub-batches, each expected to fit under the ingestion size limit. + :rtype: list[list[~azure.monitor.opentelemetry.exporter._generated.models.TelemetryItem]] + """ chunks: List[List[TelemetryItem]] = [] current: List[TelemetryItem] = [] current_size = 2 # account for the enclosing "[" and "]" of the JSON array diff --git a/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/test_base_exporter.py b/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/test_base_exporter.py index a01bda13fb2f..c82af2f69316 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/test_base_exporter.py +++ b/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/test_base_exporter.py @@ -1146,9 +1146,7 @@ def test_transmission_206_oversized_item_omits_envelope_from_log(self): ), ], ) - with self.assertLogs( - "azure.monitor.opentelemetry.exporter.export._base", level="ERROR" - ) as cm: + with self.assertLogs("azure.monitor.opentelemetry.exporter.export._base", level="ERROR") as cm: result = exporter._transmit(custom_envelopes_to_export) self.assertEqual(result, ExportResult.FAILED_NOT_RETRYABLE) log_output = "\n".join(cm.output) @@ -1181,9 +1179,7 @@ def test_transmission_206_normal_drop_logs_full_envelope(self): ), ], ) - with self.assertLogs( - "azure.monitor.opentelemetry.exporter.export._base", level="ERROR" - ) as cm: + with self.assertLogs("azure.monitor.opentelemetry.exporter.export._base", level="ERROR") as cm: result = exporter._transmit(custom_envelopes_to_export) self.assertEqual(result, ExportResult.FAILED_NOT_RETRYABLE) log_output = "\n".join(cm.output) @@ -1421,9 +1417,7 @@ def test_transmission_413_storage_disabled_dropped(self): self.assertEqual(result, ExportResult.FAILED_NOT_RETRYABLE) # The batch is dropped with the storage-disabled reason, matching every other # retryable failure when offline storage is turned off. - mock_track_dropped.assert_called_once_with( - custom_envelopes_to_export, DropCode.CLIENT_STORAGE_DISABLED - ) + mock_track_dropped.assert_called_once_with(custom_envelopes_to_export, DropCode.CLIENT_STORAGE_DISABLED) def test_transmission_413_persists_and_retries(self): """End-to-end with real on-disk storage: a 413 persists the split halves to @@ -1478,9 +1472,7 @@ def test_transmission_413_sub_batch_persist_failure_drops_that_batch_only(self): TelemetryItem(name="Test4", time=datetime.now()), ] with mock.patch.object(AzureMonitorClient, "track", side_effect=_make_http_response_error(413)): - with self.assertLogs( - "azure.monitor.opentelemetry.exporter.export._base", level="DEBUG" - ) as cm: + with self.assertLogs("azure.monitor.opentelemetry.exporter.export._base", level="DEBUG") as cm: result = exporter._transmit(custom_envelopes_to_export) self.assertEqual(result, ExportResult.FAILED_NOT_RETRYABLE) # Both sub-batches were attempted despite the first failing (loop did not abort). From 81ddc27757c2e740d16815398551c6a3fca4f731 Mon Sep 17 00:00:00 2001 From: Radhika Gupta Date: Tue, 18 Aug 2026 14:59:48 -0700 Subject: [PATCH 07/18] Address feedback --- .../opentelemetry/exporter/_constants.py | 17 ++- .../monitor/opentelemetry/exporter/_utils.py | 53 ++++++++- .../opentelemetry/exporter/export/_base.py | 107 ++++-------------- .../tests/test_base_exporter.py | 54 +++++---- 4 files changed, 122 insertions(+), 109 deletions(-) diff --git a/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_constants.py b/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_constants.py index e3754df96f7e..772a1febeba0 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_constants.py +++ b/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_constants.py @@ -76,7 +76,22 @@ 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) + +# Ingestion size limits + +# Maximum request body size (in bytes) the ingestion endpoint accepts on the wire before +# responding with 413 (Payload Too Large). +_MAX_INGESTION_PAYLOAD_SIZE_BYTES = 30 * 1000 * 1000 + +# Lowercased markers used by the ingestion endpoint when it rejects a single item for exceeding +# the per-item size limit (~1.2 MiB, surfaced as a 400 within a 206 partial response). Such an +# item cannot be split further, and the server tells us why it was dropped, so there is no need +# to re-serialize the envelope to find out. +_ITEM_TOO_LARGE_MESSAGE_MARKERS = ( + "length must not exceed", + "too large", +) # Envelope constants diff --git a/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_utils.py b/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_utils.py index 25fcb9b86782..e69829f506e5 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_utils.py +++ b/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_utils.py @@ -14,7 +14,7 @@ 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 @@ -22,6 +22,7 @@ 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 ( @@ -31,7 +32,9 @@ _DEFAULT_AAD_SCOPE, _FUNCTIONS_WORKER_RUNTIME, _INSTRUMENTATIONS_BIT_MAP, + _ITEM_TOO_LARGE_MESSAGE_MARKERS, _KUBERNETES_SERVICE_HOST, + _MAX_INGESTION_PAYLOAD_SIZE_BYTES, _MICROSOFT_OPENTELEMETRY_VERSION, _PYTHON_APPLICATIONINSIGHTS_ENABLE_TELEMETRY, _WEBSITE_SITE_NAME, @@ -574,3 +577,51 @@ def _get_retry_delay_from_headers(headers: Any) -> Optional[int]: except ValueError: return None return None + + +# Ingestion size helpers + + +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) + # Guarantee progress: 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. Only split when there is more than one envelope; a single envelope cannot be divided + # and is handled by the caller (which drops it), so this never produces an empty sub-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) diff --git a/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/export/_base.py b/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/export/_base.py index fcb093c2b698..06682fb9eff8 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/export/_base.py +++ b/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/export/_base.py @@ -1,7 +1,6 @@ # Copyright (c) Microsoft Corporation. All rights reserved. # Licensed under the MIT License. import getpass -import json import logging import os import tempfile @@ -28,9 +27,6 @@ from azure.monitor.opentelemetry.exporter._generated.exporter.models import ( TelemetryItem, ) -from azure.monitor.opentelemetry.exporter._generated.exporter._utils.model_base import ( - SdkJSONEncoder, -) from azure.monitor.opentelemetry.exporter._constants import ( _ALLOWED_REDIRECT_DOMAIN_SUFFIXES, _AZURE_MONITOR_DISTRO_VERSION_ARG, @@ -67,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 ( @@ -99,15 +97,6 @@ _TEMPDIR_PREFIX = "opentelemetry-python-" _SERVICE_API_LATEST = "2020-09-15_Preview" -# Maximum request body size (in bytes) the ingestion endpoint accepts on the wire before -# responding with 413 (Payload Too Large). -_MAX_INGESTION_PAYLOAD_SIZE_BYTES = 30 * 1000 * 1000 - -# Maximum size (in bytes) of a single telemetry item the ingestion endpoint accepts. Items -# larger than this are rejected per-item (surfaced as a 400 within a 206 partial response) and -# cannot be split further. ~1.2 MiB. -_MAX_INGESTION_ITEM_SIZE_BYTES = 1258291 - class ExportResult(Enum): SUCCESS = 0 @@ -349,8 +338,6 @@ def _handle_payload_too_large( len(envelopes), response_error.message, ) - if self._should_collect_customer_sdkstats(): - track_dropped_items(envelopes, DropCode.CLIENT_STORAGE_DISABLED) return ExportResult.FAILED_NOT_RETRYABLE if len(envelopes) <= 1: if not self._is_stats_exporter(): @@ -366,7 +353,7 @@ def _handle_payload_too_large( 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 self._should_collect_customer_sdkstats(): + 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(): @@ -509,29 +496,19 @@ def _transmit(self, envelopes: List[TelemetryItem], _skip_rate_limit: bool = Fal ): track_dropped_items([envelopes[error.index]], error.status_code) dropped_envelope = envelopes[error.index] if error.index is not None else None - # An envelope that exceeds the per-item ingestion size limit (~1.2 MiB) - # cannot be split further and is dropped here. Dumping it would flood the - # log with a huge (and potentially PII-laden) payload, so log a concise - # summary instead. For all other drops keep the existing behavior of - # logging the full envelope to aid debugging. - if ( - dropped_envelope is not None - and _get_envelope_serialized_size(dropped_envelope) > _MAX_INGESTION_ITEM_SIZE_BYTES - ): - logger.error( - "Data drop %s: %s. Envelope of type %s exceeds the per-item " - "ingestion size limit and is omitted from the log.", - error.status_code, - error.message, - dropped_envelope.name, - ) - else: - logger.error( - "Data drop %s: %s. %s", - error.status_code, - error.message, - (dropped_envelope if dropped_envelope is not None else ""), - ) + # The ingestion endpoint already tells us when an item was rejected + # for exceeding the per-item size limit, so trust that message + # instead of re-serializing every dropped envelope to measure it. + # Such an envelope is huge (and potentially PII-laden), so stop the + # log at the server message and leave the payload out. + 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, + dropped_envelope, + ) storage = self.storage if storage and resend_envelopes: envelopes_to_store = [x.as_dict() for x in resend_envelopes] @@ -891,55 +868,6 @@ def _host(netloc: str) -> str: return False -def _get_envelope_serialized_size(envelope: TelemetryItem) -> int: - """Return the serialized wire size (in bytes) of a single envelope on the wire. - - :param envelope: The telemetry envelope to measure. - :type envelope: ~azure.monitor.opentelemetry.exporter._generated.models.TelemetryItem - :return: The serialized size of the envelope in bytes. - :rtype: int - """ - 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. - - :param envelopes: The oversized batch of telemetry envelopes to split. - :type envelopes: list[~azure.monitor.opentelemetry.exporter._generated.models.TelemetryItem] - :return: A list of sub-batches, each expected to fit under the ingestion size limit. - :rtype: list[list[~azure.monitor.opentelemetry.exporter._generated.models.TelemetryItem]] - """ - 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) - # Guarantee progress: 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. Only split when there is more than one envelope; a single envelope cannot be divided - # and is handled by the caller (which drops it), so this never produces an empty sub-batch. - if len(chunks) < 2 and len(envelopes) > 1: - midpoint = len(envelopes) // 2 - chunks = [envelopes[:midpoint], envelopes[midpoint:]] - return chunks - - def _is_invalid_code(response_code: Optional[int]) -> bool: """Determine if response is a invalid response. @@ -1006,6 +934,9 @@ def _is_sampling_rejection(message: Optional[str]) -> bool: return message.lower() == "telemetry sampled out." + + + # mypy: disable-error-code="union-attr" def _get_authentication_credential( **kwargs: Any, diff --git a/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/test_base_exporter.py b/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/test_base_exporter.py index c82af2f69316..d062cdb18bff 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/test_base_exporter.py +++ b/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/test_base_exporter.py @@ -16,12 +16,13 @@ BaseExporter, ExportResult, _get_storage_directory, +) +from azure.monitor.opentelemetry.exporter._utils import ( + _get_retry_delay_from_headers, _get_envelope_serialized_size, + _is_item_too_large, _split_oversized_batch, - _MAX_INGESTION_PAYLOAD_SIZE_BYTES, - _MAX_INGESTION_ITEM_SIZE_BYTES, ) -from azure.monitor.opentelemetry.exporter._utils import _get_retry_delay_from_headers from azure.monitor.opentelemetry.exporter._storage import StorageExportResult from azure.monitor.opentelemetry.exporter._constants import DropCode from azure.monitor.opentelemetry.exporter.statsbeat._state import ( @@ -31,6 +32,7 @@ from azure.monitor.opentelemetry.exporter.export.metrics._exporter import AzureMonitorMetricExporter from azure.monitor.opentelemetry.exporter._constants import ( _DEFAULT_AAD_SCOPE, + _MAX_INGESTION_PAYLOAD_SIZE_BYTES, _REQ_DURATION_NAME, _REQ_EXCEPTION_NAME, _REQ_FAILURE_NAME, @@ -1121,8 +1123,10 @@ def test_transmission_206_no_retry(self): exporter.storage.put.assert_not_called() def test_transmission_206_oversized_item_omits_envelope_from_log(self): - """An item that exceeds the per-item ingestion size limit is dropped with a concise log - message; the full (potentially large / PII-laden) envelope must NOT be dumped to the log.""" + """An item the server reports as exceeding the per-item ingestion size limit is dropped and + the log stops at the server message; the full (potentially large / PII-laden) envelope must + NOT be dumped. The decision comes from the server's error message, so the envelope is never + re-serialized to measure it.""" exporter = BaseExporter(disable_offline_storage=True) exporter.storage = mock.Mock() oversized_envelope = TelemetryItem( @@ -1132,9 +1136,8 @@ def test_transmission_206_oversized_item_omits_envelope_from_log(self): ) custom_envelopes_to_export = [oversized_envelope] with mock.patch.object(AzureMonitorClient, "track") as post, mock.patch( - "azure.monitor.opentelemetry.exporter.export._base._get_envelope_serialized_size", - return_value=_MAX_INGESTION_ITEM_SIZE_BYTES + 1, - ): + "azure.monitor.opentelemetry.exporter._utils._get_envelope_serialized_size", + ) as mock_size: post.return_value = TrackResponse( items_received=1, items_accepted=0, @@ -1149,10 +1152,11 @@ def test_transmission_206_oversized_item_omits_envelope_from_log(self): with self.assertLogs("azure.monitor.opentelemetry.exporter.export._base", level="ERROR") as cm: result = exporter._transmit(custom_envelopes_to_export) self.assertEqual(result, ExportResult.FAILED_NOT_RETRYABLE) + # The drop path never re-serializes the envelope to decide what to log. + mock_size.assert_not_called() log_output = "\n".join(cm.output) - # Concise summary is logged (status, server message, and envelope type only). - self.assertIn("exceeds the per-item ingestion size limit and is omitted from the log", log_output) - self.assertIn("OversizedItem", log_output) + # The server message is still logged so the drop reason is visible. + self.assertIn("Telemetry item length must not exceed 1258291", log_output) # The full envelope payload (its tags/data) must NOT be dumped to the log. self.assertNotIn("SECRET_PAYLOAD_MARKER", log_output) @@ -1400,8 +1404,9 @@ def test_transmission_413_single_envelope_dropped(self): exporter.storage.put.assert_not_called() def test_transmission_413_storage_disabled_dropped(self): - """A 413 with storage disabled cannot be persisted and should be dropped with the - CLIENT_STORAGE_DISABLED reason, consistent with other retryable codes.""" + """A 413 with storage disabled cannot be persisted and should be dropped exactly once + with the CLIENT_STORAGE_DISABLED reason. Drop attribution for the storage-disabled case + is owned solely by _handle_transmit_from_storage, so _transmit must not also count it.""" exporter = BaseExporter(disable_offline_storage=True) exporter.storage = None custom_envelopes_to_export = [ @@ -1414,9 +1419,11 @@ def test_transmission_413_storage_disabled_dropped(self): "azure.monitor.opentelemetry.exporter.export._base.track_dropped_items" ) as mock_track_dropped: result = exporter._transmit(custom_envelopes_to_export) + # _transmit itself must not attribute the drop (avoids the double count). + mock_track_dropped.assert_not_called() + exporter._handle_transmit_from_storage(custom_envelopes_to_export, result) self.assertEqual(result, ExportResult.FAILED_NOT_RETRYABLE) - # The batch is dropped with the storage-disabled reason, matching every other - # retryable failure when offline storage is turned off. + # The batch is dropped exactly once, with the storage-disabled reason. mock_track_dropped.assert_called_once_with(custom_envelopes_to_export, DropCode.CLIENT_STORAGE_DISABLED) def test_transmission_413_persists_and_retries(self): @@ -1497,12 +1504,21 @@ def test_get_envelope_serialized_size_failure_returns_max(self): is isolated into its own sub-batch instead of being counted as weightless.""" envelope = TelemetryItem(name="Test", time=datetime.now()) with mock.patch( - "azure.monitor.opentelemetry.exporter.export._base.json.dumps", + "azure.monitor.opentelemetry.exporter._utils.json.dumps", side_effect=TypeError("not serializable"), ): size = _get_envelope_serialized_size(envelope) self.assertEqual(size, _MAX_INGESTION_PAYLOAD_SIZE_BYTES) + def test_is_item_too_large(self): + """The oversized-item decision is driven by the server's error message, so no envelope + needs to be re-serialized to make it.""" + self.assertTrue(_is_item_too_large("Telemetry item length must not exceed 1258291")) + self.assertTrue(_is_item_too_large("Payload too large")) + self.assertFalse(_is_item_too_large("Invalid instrumentation key")) + self.assertFalse(_is_item_too_large("")) + self.assertFalse(_is_item_too_large(None)) + def test_split_oversized_batch_empty(self): """Splitting an empty batch yields no sub-batches (and never an empty half).""" self.assertEqual(_split_oversized_batch([]), []) @@ -1524,7 +1540,7 @@ def test_split_oversized_batch_isolates_oversized_envelopes(self): TelemetryItem(name="Test3", time=datetime.now()), ] with mock.patch( - "azure.monitor.opentelemetry.exporter.export._base._get_envelope_serialized_size", + "azure.monitor.opentelemetry.exporter._utils._get_envelope_serialized_size", return_value=_MAX_INGESTION_PAYLOAD_SIZE_BYTES, ): chunks = _split_oversized_batch(envelopes) @@ -1538,7 +1554,7 @@ def test_split_oversized_batch_packs_multiple_chunks_by_size(self): # 3*0.4 exceeds), yielding chunks of [2, 2, 1]. per_item = int(_MAX_INGESTION_PAYLOAD_SIZE_BYTES * 0.4) with mock.patch( - "azure.monitor.opentelemetry.exporter.export._base._get_envelope_serialized_size", + "azure.monitor.opentelemetry.exporter._utils._get_envelope_serialized_size", return_value=per_item, ): chunks = _split_oversized_batch(envelopes) @@ -1554,7 +1570,7 @@ def test_split_oversized_batch_falls_back_to_count_halving(self): TelemetryItem(name="Test4", time=datetime.now()), ] with mock.patch( - "azure.monitor.opentelemetry.exporter.export._base._get_envelope_serialized_size", + "azure.monitor.opentelemetry.exporter._utils._get_envelope_serialized_size", return_value=1, ): chunks = _split_oversized_batch(envelopes) From aa7b1332a08251cbf3a08c45088186f346188879 Mon Sep 17 00:00:00 2001 From: Radhika Gupta Date: Wed, 19 Aug 2026 10:29:56 -0700 Subject: [PATCH 08/18] Fix black --- .../azure/monitor/opentelemetry/exporter/export/_base.py | 3 --- 1 file changed, 3 deletions(-) diff --git a/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/export/_base.py b/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/export/_base.py index 06682fb9eff8..ec49c0faaf57 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/export/_base.py +++ b/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/export/_base.py @@ -934,9 +934,6 @@ def _is_sampling_rejection(message: Optional[str]) -> bool: return message.lower() == "telemetry sampled out." - - - # mypy: disable-error-code="union-attr" def _get_authentication_credential( **kwargs: Any, From 2b6e121116f62cd55223558b66e12c2da39e76ed Mon Sep 17 00:00:00 2001 From: Radhika Gupta Date: Fri, 21 Aug 2026 08:12:17 -0700 Subject: [PATCH 09/18] Fix formatting --- .../opentelemetry/exporter/_constants.py | 14 -------------- .../monitor/opentelemetry/exporter/_utils.py | 19 +++++++++++-------- .../opentelemetry/exporter/export/_base.py | 18 ++++-------------- .../tests/test_base_exporter.py | 3 ++- 4 files changed, 17 insertions(+), 37 deletions(-) diff --git a/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_constants.py b/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_constants.py index 772a1febeba0..b35e90c19be0 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_constants.py +++ b/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_constants.py @@ -78,20 +78,6 @@ _REACHED_INGESTION_STATUS_CODES = (200, 206, 402, 408, 413, 429, 439, 500) -# Ingestion size limits - -# Maximum request body size (in bytes) the ingestion endpoint accepts on the wire before -# responding with 413 (Payload Too Large). -_MAX_INGESTION_PAYLOAD_SIZE_BYTES = 30 * 1000 * 1000 - -# Lowercased markers used by the ingestion endpoint when it rejects a single item for exceeding -# the per-item size limit (~1.2 MiB, surfaced as a 400 within a 206 partial response). Such an -# item cannot be split further, and the server tells us why it was dropped, so there is no need -# to re-serialize the envelope to find out. -_ITEM_TOO_LARGE_MESSAGE_MARKERS = ( - "length must not exceed", - "too large", -) # Envelope constants diff --git a/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_utils.py b/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_utils.py index e69829f506e5..efba03076af5 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_utils.py +++ b/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_utils.py @@ -32,9 +32,7 @@ _DEFAULT_AAD_SCOPE, _FUNCTIONS_WORKER_RUNTIME, _INSTRUMENTATIONS_BIT_MAP, - _ITEM_TOO_LARGE_MESSAGE_MARKERS, _KUBERNETES_SERVICE_HOST, - _MAX_INGESTION_PAYLOAD_SIZE_BYTES, _MICROSOFT_OPENTELEMETRY_VERSION, _PYTHON_APPLICATIONINSIGHTS_ENABLE_TELEMETRY, _WEBSITE_SITE_NAME, @@ -50,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 @@ -579,9 +586,6 @@ def _get_retry_delay_from_headers(headers: Any) -> Optional[int]: return None -# Ingestion size helpers - - def _get_envelope_serialized_size(envelope: TelemetryItem) -> int: """Return the serialized wire size (in bytes) of a single envelope on the wire.""" try: @@ -609,10 +613,9 @@ def _split_oversized_batch(envelopes: List[TelemetryItem]) -> List[List[Telemetr current_size += separator_size + envelope_size if current: chunks.append(current) - # Guarantee progress: if size-based packing produced a single chunk (e.g. because the byte + # 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. Only split when there is more than one envelope; a single envelope cannot be divided - # and is handled by the caller (which drops it), so this never produces an empty sub-batch. + # batch. if len(chunks) < 2 and len(envelopes) > 1: midpoint = len(envelopes) // 2 chunks = [envelopes[:midpoint], envelopes[midpoint:]] diff --git a/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/export/_base.py b/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/export/_base.py index ec49c0faaf57..dfb98c4be60f 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/export/_base.py +++ b/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/export/_base.py @@ -322,15 +322,7 @@ def _handle_transmit_from_storage(self, envelopes: List[TelemetryItem], result: 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. - - :param envelopes: The telemetry envelopes that were rejected as too large. - :type envelopes: list[~azure.monitor.opentelemetry.exporter._generated.models.TelemetryItem] - :param response_error: The error returned for the oversized request. - :type response_error: ~azure.core.exceptions.HttpResponseError - :return: The export result for the oversized batch. - :rtype: ~azure.monitor.opentelemetry.exporter.export._base.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( @@ -496,11 +488,9 @@ def _transmit(self, envelopes: List[TelemetryItem], _skip_rate_limit: bool = Fal ): track_dropped_items([envelopes[error.index]], error.status_code) dropped_envelope = envelopes[error.index] if error.index is not None else None - # The ingestion endpoint already tells us when an item was rejected - # for exceeding the per-item size limit, so trust that message - # instead of re-serializing every dropped envelope to measure it. - # Such an envelope is huge (and potentially PII-laden), so stop the - # log at the server message and leave the payload out. + # 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( diff --git a/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/test_base_exporter.py b/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/test_base_exporter.py index d062cdb18bff..e26c94ff4c6b 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/test_base_exporter.py +++ b/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/test_base_exporter.py @@ -32,7 +32,6 @@ from azure.monitor.opentelemetry.exporter.export.metrics._exporter import AzureMonitorMetricExporter from azure.monitor.opentelemetry.exporter._constants import ( _DEFAULT_AAD_SCOPE, - _MAX_INGESTION_PAYLOAD_SIZE_BYTES, _REQ_DURATION_NAME, _REQ_EXCEPTION_NAME, _REQ_FAILURE_NAME, @@ -56,6 +55,8 @@ TrackResponse, ) +_MAX_INGESTION_PAYLOAD_SIZE_BYTES = 30 * 1000 * 1000 + TEST_AUTH_POLICY = "TEST_AUTH_POLICY" TEST_TEMP_DIR = "TEST_TEMP_DIR" TEST_USER_DIR = "test-user" From 765909cba2b2e0b0bc66c73fb8a145493b68ad32 Mon Sep 17 00:00:00 2001 From: Radhika Gupta Date: Fri, 21 Aug 2026 09:57:35 -0700 Subject: [PATCH 10/18] Fix lint --- .../azure/monitor/opentelemetry/exporter/_utils.py | 6 +++--- .../azure/monitor/opentelemetry/exporter/export/_base.py | 2 +- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_utils.py b/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_utils.py index efba03076af5..9175712c65bd 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_utils.py +++ b/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_utils.py @@ -587,7 +587,7 @@ def _get_retry_delay_from_headers(headers: Any) -> Optional[int]: def _get_envelope_serialized_size(envelope: TelemetryItem) -> int: - """Return the serialized wire size (in bytes) of a single envelope on the wire.""" + # 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")) @@ -596,7 +596,7 @@ def _get_envelope_serialized_size(envelope: TelemetryItem) -> int: 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.""" + # 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 @@ -623,7 +623,7 @@ def _split_oversized_batch(envelopes: List[TelemetryItem]) -> List[List[Telemetr def _is_item_too_large(message: Optional[str]) -> bool: - """Determine if a per-item error indicates the envelope exceeded the ingestion size limit.""" + # Determine if a per-item error indicates the envelope exceeded the ingestion size limit. if not message: return False lowered = message.lower() diff --git a/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/export/_base.py b/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/export/_base.py index dfb98c4be60f..cbfff26e2c2f 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/export/_base.py +++ b/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/export/_base.py @@ -322,7 +322,7 @@ def _handle_transmit_from_storage(self, envelopes: List[TelemetryItem], result: 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.""" + # 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( From e6d0404ca957485e5db960ccef590022b25f225e Mon Sep 17 00:00:00 2001 From: Radhika Gupta Date: Thu, 27 Aug 2026 16:33:42 -0700 Subject: [PATCH 11/18] Address feedback --- .../monitor/opentelemetry/exporter/statsbeat/customer/_utils.py | 2 +- .../tests/customer_sdk_stats/test_utlities.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/statsbeat/customer/_utils.py b/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/statsbeat/customer/_utils.py index cba227e1a7d0..f58476145a3d 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/statsbeat/customer/_utils.py +++ b/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) diff --git a/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/customer_sdk_stats/test_utlities.py b/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/customer_sdk_stats/test_utlities.py index b1df9c005118..7ef16883c8bc 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/customer_sdk_stats/test_utlities.py +++ b/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): From 5096f0b415c115f0818fc1d862c2d644c00f10d8 Mon Sep 17 00:00:00 2001 From: Radhika Gupta Date: Fri, 28 Aug 2026 09:30:37 -0700 Subject: [PATCH 12/18] Fix flaky tests --- .../tests/test_base_exporter.py | 60 ++++++++++--------- 1 file changed, 31 insertions(+), 29 deletions(-) diff --git a/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/test_base_exporter.py b/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/test_base_exporter.py index e26c94ff4c6b..f2d2b7d07b15 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/test_base_exporter.py +++ b/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/test_base_exporter.py @@ -1430,35 +1430,37 @@ def test_transmission_413_storage_disabled_dropped(self): def test_transmission_413_persists_and_retries(self): """End-to-end with real on-disk storage: a 413 persists the split halves to disk, and a later successful drain reads them back and retries them.""" - exporter = self._base - clean_folder(exporter.storage._path) - custom_envelopes_to_export = [ - TelemetryItem(name="Test1", time=datetime.now()), - TelemetryItem(name="Test2", time=datetime.now()), - TelemetryItem(name="Test3", time=datetime.now()), - TelemetryItem(name="Test4", time=datetime.now()), - ] - # First transmit gets a 413 -> the batch is split and each half is persisted. - with mock.patch.object(AzureMonitorClient, "track", side_effect=_make_http_response_error(413)): - result = exporter._transmit(custom_envelopes_to_export) - self.assertEqual(result, ExportResult.FAILED_NOT_RETRYABLE) - # Two separate blobs were written to disk (one per half). - blobs = [f for f in os.listdir(exporter.storage._path) if ".blob" in f] - self.assertEqual(len(blobs), 2) - # Strip the retry lease so the freshly-persisted blobs are drainable now. - for name in os.listdir(exporter.storage._path): - if name.endswith(".lock"): - src = os.path.join(exporter.storage._path, name) - dst = os.path.join(exporter.storage._path, name[: name.rindex("@")]) - os.rename(src, dst) - # Now the endpoint accepts data -> draining retries the persisted halves. - with mock.patch.object(AzureMonitorClient, "track") as post: - post.return_value = TrackResponse(items_received=2, items_accepted=2, errors=[]) - exporter._transmit_from_storage() - # Both halves were re-sent and their blobs removed from disk. - self.assertEqual(post.call_count, 2) - remaining = [f for f in os.listdir(exporter.storage._path) if ".blob" in f] - self.assertEqual(len(remaining), 0) + import tempfile + + with tempfile.TemporaryDirectory() as storage_directory: + exporter = BaseExporter(storage_directory=storage_directory) + custom_envelopes_to_export = [ + TelemetryItem(name="Test1", time=datetime.now()), + TelemetryItem(name="Test2", time=datetime.now()), + TelemetryItem(name="Test3", time=datetime.now()), + TelemetryItem(name="Test4", time=datetime.now()), + ] + # First transmit gets a 413 -> the batch is split and each half is persisted. + with mock.patch.object(AzureMonitorClient, "track", side_effect=_make_http_response_error(413)): + result = exporter._transmit(custom_envelopes_to_export) + self.assertEqual(result, ExportResult.FAILED_NOT_RETRYABLE) + # Two separate blobs were written to disk (one per half). + blobs = [f for f in os.listdir(storage_directory) if ".blob" in f] + self.assertEqual(len(blobs), 2) + # Strip the retry lease so the freshly-persisted blobs are drainable now. + for name in os.listdir(storage_directory): + if name.endswith(".lock"): + src = os.path.join(storage_directory, name) + dst = os.path.join(storage_directory, name[: name.rindex("@")]) + os.rename(src, dst) + # Now the endpoint accepts data -> draining retries the persisted halves. + with mock.patch.object(AzureMonitorClient, "track") as post: + post.return_value = TrackResponse(items_received=2, items_accepted=2, errors=[]) + exporter._transmit_from_storage() + # Both halves were re-sent and their blobs removed from disk. + self.assertEqual(post.call_count, 2) + remaining = [f for f in os.listdir(storage_directory) if ".blob" in f] + self.assertEqual(len(remaining), 0) def test_transmission_413_sub_batch_persist_failure_drops_that_batch_only(self): """storage.put reports failure by return value (not by raising). If persisting one From 038d9f59a2ee8fba9eb8353990d838be3972b7b1 Mon Sep 17 00:00:00 2001 From: Radhika Gupta Date: Thu, 10 Sep 2026 08:39:23 -0700 Subject: [PATCH 13/18] Cleanup --- .../azure/monitor/opentelemetry/exporter/_constants.py | 1 - .../azure/monitor/opentelemetry/exporter/export/_base.py | 2 +- 2 files changed, 1 insertion(+), 2 deletions(-) diff --git a/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_constants.py b/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_constants.py index b35e90c19be0..bbf2c798efde 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_constants.py +++ b/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_constants.py @@ -78,7 +78,6 @@ _REACHED_INGESTION_STATUS_CODES = (200, 206, 402, 408, 413, 429, 439, 500) - # Envelope constants _METRIC_ENVELOPE_NAME = "Microsoft.ApplicationInsights.Metric" diff --git a/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/export/_base.py b/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/export/_base.py index cbfff26e2c2f..802140a1a8a9 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/export/_base.py +++ b/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/export/_base.py @@ -494,7 +494,7 @@ def _transmit(self, envelopes: List[TelemetryItem], _skip_rate_limit: bool = Fal if dropped_envelope is None or _is_item_too_large(error.message): dropped_envelope = "" # type: ignore[assignment] logger.error( - "Data drop %s: %s. %s", + "Data drop %s: %s %s.", error.status_code, error.message, dropped_envelope, From fc811bc9158d0c3ea10044abf10ac2f3eb3c492c Mon Sep 17 00:00:00 2001 From: Radhika Gupta Date: Thu, 10 Sep 2026 13:07:32 -0700 Subject: [PATCH 14/18] Update CHANGELOG --- .../azure-monitor-opentelemetry-exporter/CHANGELOG.md | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/sdk/monitor/azure-monitor-opentelemetry-exporter/CHANGELOG.md b/sdk/monitor/azure-monitor-opentelemetry-exporter/CHANGELOG.md index 173a8e5ece86..854dd77ac4fc 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-exporter/CHANGELOG.md +++ b/sdk/monitor/azure-monitor-opentelemetry-exporter/CHANGELOG.md @@ -3,8 +3,10 @@ ## 1.0.0b58 (Unreleased) ### Features Added -- Retry payloads for which request is sent successfully but no response is received, per [SPEC.](https://github.com/aep-health-and-standards/Telemetry-Collection-Spec/pull/1018) - ([#47870](https://github.com/Azure/azure-sdk-for-python/pull/47870)) +- Handle `413 Payload Too Large` responses by splitting the batch and persisting each half 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 From 03e2f28f3204bdfdd991ba1171526f38a719b445 Mon Sep 17 00:00:00 2001 From: Radhika Gupta Date: Thu, 10 Sep 2026 13:09:59 -0700 Subject: [PATCH 15/18] Update CHANGELOG --- sdk/monitor/azure-monitor-opentelemetry-exporter/CHANGELOG.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sdk/monitor/azure-monitor-opentelemetry-exporter/CHANGELOG.md b/sdk/monitor/azure-monitor-opentelemetry-exporter/CHANGELOG.md index 854dd77ac4fc..3fd3db57fa5a 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-exporter/CHANGELOG.md +++ b/sdk/monitor/azure-monitor-opentelemetry-exporter/CHANGELOG.md @@ -6,7 +6,7 @@ - Handle `413 Payload Too Large` responses by splitting the batch and persisting each half 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)) + ([#48548](https://github.com/Azure/azure-sdk-for-python/pull/48548)) ### Breaking Changes From cb66b692459b2b287b9661129c461c47976965e8 Mon Sep 17 00:00:00 2001 From: Radhika Gupta Date: Fri, 11 Sep 2026 09:31:55 -0700 Subject: [PATCH 16/18] Update tests --- .../tests/test_base_exporter.py | 150 ++++++++++++++---- 1 file changed, 118 insertions(+), 32 deletions(-) diff --git a/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/test_base_exporter.py b/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/test_base_exporter.py index f2d2b7d07b15..401a65eeef16 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/test_base_exporter.py +++ b/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/test_base_exporter.py @@ -3,6 +3,8 @@ import os import shutil +import tempfile +import threading import unittest from unittest import mock from datetime import datetime, timedelta, timezone @@ -1377,8 +1379,7 @@ def test_transmission_408(self): self.assertEqual(result, ExportResult.FAILED_RETRYABLE) def test_transmission_413_splits_batch(self): - """A 413 with more than one envelope should split the batch in half and - persist each half as a separate blob so they are retried at a smaller size.""" + """A 413 persists every envelope in ordered sub-batches, including an odd tail.""" exporter = BaseExporter(disable_offline_storage=True) exporter.storage = mock.Mock() exporter.storage.put.return_value = StorageExportResult.LOCAL_FILE_BLOB_SUCCESS @@ -1387,12 +1388,16 @@ def test_transmission_413_splits_batch(self): TelemetryItem(name="Test2", time=datetime.now()), TelemetryItem(name="Test3", time=datetime.now()), TelemetryItem(name="Test4", time=datetime.now()), + TelemetryItem(name="Test5", time=datetime.now()), ] with mock.patch.object(AzureMonitorClient, "track", side_effect=_make_http_response_error(413)): result = exporter._transmit(custom_envelopes_to_export) self.assertEqual(result, ExportResult.FAILED_NOT_RETRYABLE) - # Two separate blobs (one per half) so the halves are retried independently. - self.assertEqual(exporter.storage.put.call_count, 2) + expected_children = [ + [envelope.as_dict() for envelope in custom_envelopes_to_export[:2]], + [envelope.as_dict() for envelope in custom_envelopes_to_export[2:]], + ] + self.assertEqual(exporter.storage.put.call_args_list, [mock.call(child) for child in expected_children]) def test_transmission_413_single_envelope_dropped(self): """A 413 with a single envelope cannot be split further and should be dropped.""" @@ -1428,39 +1433,117 @@ def test_transmission_413_storage_disabled_dropped(self): mock_track_dropped.assert_called_once_with(custom_envelopes_to_export, DropCode.CLIENT_STORAGE_DISABLED) def test_transmission_413_persists_and_retries(self): - """End-to-end with real on-disk storage: a 413 persists the split halves to - disk, and a later successful drain reads them back and retries them.""" - import tempfile - + """Split children retain a future lease and retry on a later successful export cycle.""" with tempfile.TemporaryDirectory() as storage_directory: - exporter = BaseExporter(storage_directory=storage_directory) + exporter = BaseExporter( + storage_directory=storage_directory, + storage_maintenance_period=3600, + storage_min_retry_interval=60, + ) + now = datetime(2026, 9, 10, tzinfo=timezone.utc) custom_envelopes_to_export = [ TelemetryItem(name="Test1", time=datetime.now()), TelemetryItem(name="Test2", time=datetime.now()), TelemetryItem(name="Test3", time=datetime.now()), TelemetryItem(name="Test4", time=datetime.now()), + TelemetryItem(name="Test5", time=datetime.now()), ] - # First transmit gets a 413 -> the batch is split and each half is persisted. - with mock.patch.object(AzureMonitorClient, "track", side_effect=_make_http_response_error(413)): - result = exporter._transmit(custom_envelopes_to_export) - self.assertEqual(result, ExportResult.FAILED_NOT_RETRYABLE) - # Two separate blobs were written to disk (one per half). - blobs = [f for f in os.listdir(storage_directory) if ".blob" in f] - self.assertEqual(len(blobs), 2) - # Strip the retry lease so the freshly-persisted blobs are drainable now. - for name in os.listdir(storage_directory): - if name.endswith(".lock"): - src = os.path.join(storage_directory, name) - dst = os.path.join(storage_directory, name[: name.rindex("@")]) - os.rename(src, dst) - # Now the endpoint accepts data -> draining retries the persisted halves. - with mock.patch.object(AzureMonitorClient, "track") as post: - post.return_value = TrackResponse(items_received=2, items_accepted=2, errors=[]) - exporter._transmit_from_storage() - # Both halves were re-sent and their blobs removed from disk. - self.assertEqual(post.call_count, 2) - remaining = [f for f in os.listdir(storage_directory) if ".blob" in f] - self.assertEqual(len(remaining), 0) + try: + with mock.patch("azure.monitor.opentelemetry.exporter._storage._now", return_value=now): + with mock.patch.object(AzureMonitorClient, "track", side_effect=_make_http_response_error(413)): + result = exporter._transmit(custom_envelopes_to_export) + self.assertEqual(result, ExportResult.FAILED_NOT_RETRYABLE) + persisted = sorted(os.listdir(storage_directory)) + self.assertEqual(len(persisted), 2) + self.assertTrue(all(name.endswith(".lock") for name in persisted)) + self.assertEqual(list(exporter.storage.gets()), []) + + with mock.patch( + "azure.monitor.opentelemetry.exporter._storage._now", return_value=now + timedelta(seconds=61) + ): + with mock.patch.object(AzureMonitorClient, "track") as post: + post.side_effect = lambda envelopes, **_kwargs: TrackResponse( + items_received=len(envelopes), items_accepted=len(envelopes), errors=[] + ) + exporter._handle_transmit_from_storage([], ExportResult.SUCCESS) + + retried_names = [[envelope.name for envelope in call.args[0]] for call in post.call_args_list] + self.assertCountEqual(retried_names, [["Test1", "Test2"], ["Test3", "Test4", "Test5"]]) + self.assertEqual(os.listdir(storage_directory), []) + finally: + exporter.storage.close() + + def test_transmission_413_repeated_delayed_splitting_reaches_singletons(self): + """Later successful cycles repeatedly split leased children until singleton 413s are dropped.""" + with tempfile.TemporaryDirectory() as storage_directory: + exporter = BaseExporter( + storage_directory=storage_directory, + storage_maintenance_period=3600, + storage_min_retry_interval=60, + ) + now = datetime(2026, 9, 10, tzinfo=timezone.utc) + envelopes = [TelemetryItem(name=f"Test{index}", time=datetime.now()) for index in range(1, 6)] + transmitted_batches = [] + + def reject(envelopes_to_transmit, **_kwargs): + transmitted_batches.append([envelope.name for envelope in envelopes_to_transmit]) + raise _make_http_response_error(413) + + try: + with mock.patch.object(AzureMonitorClient, "track", side_effect=reject): + with mock.patch("azure.monitor.opentelemetry.exporter._storage._now", return_value=now): + exporter._transmit(envelopes) + for cycle in range(1, 4): + with mock.patch( + "azure.monitor.opentelemetry.exporter._storage._now", + return_value=now + timedelta(seconds=61 * cycle), + ): + exporter._handle_transmit_from_storage([], ExportResult.SUCCESS) + + self.assertEqual(transmitted_batches[0], ["Test1", "Test2", "Test3", "Test4", "Test5"]) + self.assertCountEqual(transmitted_batches[1:3], [["Test1", "Test2"], ["Test3", "Test4", "Test5"]]) + self.assertCountEqual(transmitted_batches[3:7], [["Test1"], ["Test2"], ["Test3"], ["Test4", "Test5"]]) + self.assertCountEqual(transmitted_batches[7:], [["Test4"], ["Test5"]]) + self.assertEqual(os.listdir(storage_directory), []) + finally: + exporter.storage.close() + + def test_concurrent_successful_cycles_do_not_transmit_stored_batch_twice(self): + """Atomic blob leasing lets only one concurrent successful cycle drain a stored batch.""" + with tempfile.TemporaryDirectory() as storage_directory: + exporter = BaseExporter(storage_directory=storage_directory, storage_maintenance_period=3600) + stored_envelopes = [ + TelemetryItem(name="Test1", time=datetime.now()), + TelemetryItem(name="Test2", time=datetime.now()), + ] + exporter.storage.put([envelope.as_dict() for envelope in stored_envelopes], lease_period=0) + start = threading.Barrier(3) + worker_errors = [] + + def drain_after_success(): + try: + start.wait(timeout=5) + exporter._handle_transmit_from_storage([], ExportResult.SUCCESS) + except Exception as error: # pylint: disable=broad-exception-caught + worker_errors.append(error) + + try: + with mock.patch.object(AzureMonitorClient, "track") as post: + post.return_value = TrackResponse(items_received=2, items_accepted=2, errors=[]) + threads = [threading.Thread(target=drain_after_success) for _ in range(2)] + for thread in threads: + thread.start() + start.wait(timeout=5) + for thread in threads: + thread.join(timeout=5) + + self.assertFalse(any(thread.is_alive() for thread in threads)) + self.assertEqual(worker_errors, []) + post.assert_called_once() + self.assertEqual([envelope.name for envelope in post.call_args.args[0]], ["Test1", "Test2"]) + self.assertEqual(os.listdir(storage_directory), []) + finally: + exporter.storage.close() def test_transmission_413_sub_batch_persist_failure_drops_that_batch_only(self): """storage.put reports failure by return value (not by raising). If persisting one @@ -1485,8 +1568,11 @@ def test_transmission_413_sub_batch_persist_failure_drops_that_batch_only(self): with self.assertLogs("azure.monitor.opentelemetry.exporter.export._base", level="DEBUG") as cm: result = exporter._transmit(custom_envelopes_to_export) self.assertEqual(result, ExportResult.FAILED_NOT_RETRYABLE) - # Both sub-batches were attempted despite the first failing (loop did not abort). - self.assertEqual(exporter.storage.put.call_count, 2) + expected_children = [ + [envelope.as_dict() for envelope in custom_envelopes_to_export[:2]], + [envelope.as_dict() for envelope in custom_envelopes_to_export[2:]], + ] + self.assertEqual(exporter.storage.put.call_args_list, [mock.call(child) for child in expected_children]) # The failed sub-batch is reported via a warning. self.assertTrue(any("Failed to persist a split sub-batch" in line for line in cm.output)) From 77613eec6745ff70017ae402fcaf78cfa47adeb5 Mon Sep 17 00:00:00 2001 From: Radhika Gupta Date: Fri, 11 Sep 2026 09:55:07 -0700 Subject: [PATCH 17/18] Update CHANGELOG --- sdk/monitor/azure-monitor-opentelemetry-exporter/CHANGELOG.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sdk/monitor/azure-monitor-opentelemetry-exporter/CHANGELOG.md b/sdk/monitor/azure-monitor-opentelemetry-exporter/CHANGELOG.md index 3fd3db57fa5a..d4b560c0bbf1 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-exporter/CHANGELOG.md +++ b/sdk/monitor/azure-monitor-opentelemetry-exporter/CHANGELOG.md @@ -3,7 +3,7 @@ ## 1.0.0b58 (Unreleased) ### Features Added -- Handle `413 Payload Too Large` responses by splitting the batch and persisting each half to +- 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)) From f112df0e1a221971b7e8b6260d5ccb38fd248850 Mon Sep 17 00:00:00 2001 From: Radhika Gupta Date: Fri, 11 Sep 2026 12:24:42 -0700 Subject: [PATCH 18/18] Fix test --- .../tests/test_base_exporter.py | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/test_base_exporter.py b/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/test_base_exporter.py index 401a65eeef16..8ae407cb15d7 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/test_base_exporter.py +++ b/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/test_base_exporter.py @@ -1508,8 +1508,8 @@ def reject(envelopes_to_transmit, **_kwargs): finally: exporter.storage.close() - def test_concurrent_successful_cycles_do_not_transmit_stored_batch_twice(self): - """Atomic blob leasing lets only one concurrent successful cycle drain a stored batch.""" + def test_concurrent_successful_cycles_drain_complete_stored_batches(self): + """Concurrent successful cycles transmit only complete stored batches and finish draining.""" with tempfile.TemporaryDirectory() as storage_directory: exporter = BaseExporter(storage_directory=storage_directory, storage_maintenance_period=3600) stored_envelopes = [ @@ -1539,8 +1539,9 @@ def drain_after_success(): self.assertFalse(any(thread.is_alive() for thread in threads)) self.assertEqual(worker_errors, []) - post.assert_called_once() - self.assertEqual([envelope.name for envelope in post.call_args.args[0]], ["Test1", "Test2"]) + self.assertGreaterEqual(post.call_count, 1) + for call in post.call_args_list: + self.assertEqual([envelope.name for envelope in call.args[0]], ["Test1", "Test2"]) self.assertEqual(os.listdir(storage_directory), []) finally: exporter.storage.close()