diff --git a/sdk/monitor/azure-monitor-opentelemetry-exporter/CHANGELOG.md b/sdk/monitor/azure-monitor-opentelemetry-exporter/CHANGELOG.md index b408f3b9d37c..d4b560c0bbf1 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-exporter/CHANGELOG.md +++ b/sdk/monitor/azure-monitor-opentelemetry-exporter/CHANGELOG.md @@ -3,6 +3,10 @@ ## 1.0.0b58 (Unreleased) ### Features Added +- Handle `413 Payload Too Large` responses by splitting the batch and persisting each batch to + local storage for retry at a smaller size, instead of dropping the entire batch. A single + envelope that is too large to split is dropped. + ([#48548](https://github.com/Azure/azure-sdk-for-python/pull/48548)) ### Breaking Changes 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..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 @@ -76,7 +76,7 @@ 439, # Quota, too Many Requests over extended time (legacy) ) -_REACHED_INGESTION_STATUS_CODES = (200, 206, 402, 408, 429, 439, 500) +_REACHED_INGESTION_STATUS_CODES = (200, 206, 402, 408, 413, 429, 439, 500) # Envelope constants 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..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 @@ -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 ( @@ -47,6 +48,15 @@ # Workaround for missing version file opentelemetry_version = version("opentelemetry-sdk") +# Maximum ingestion limit over wire +_MAX_INGESTION_PAYLOAD_SIZE_BYTES = 30 * 1000 * 1000 + +# Marker to check the per-item size limit messages +_ITEM_TOO_LARGE_MESSAGE_MARKERS = ( + "length must not exceed", + "too large", +) + # Azure App Service @@ -574,3 +584,47 @@ def _get_retry_delay_from_headers(headers: Any) -> Optional[int]: except ValueError: return None return None + + +def _get_envelope_serialized_size(envelope: TelemetryItem) -> int: + # Return the serialized wire size (in bytes) of a single envelope on the wire. + try: + serialized = json.dumps(envelope, cls=SdkJSONEncoder, exclude_readonly=True) + return len(serialized.encode("utf-8")) + except Exception: # pylint: disable=broad-except + return _MAX_INGESTION_PAYLOAD_SIZE_BYTES + + +def _split_oversized_batch(envelopes: List[TelemetryItem]) -> List[List[TelemetryItem]]: + # Split an oversized batch into sub-batches that each fit under the ingestion size limit. + chunks: List[List[TelemetryItem]] = [] + current: List[TelemetryItem] = [] + current_size = 2 # account for the enclosing "[" and "]" of the JSON array + for envelope in envelopes: + envelope_size = _get_envelope_serialized_size(envelope) + # Default json.dumps (used by the transport) separates array elements with ", " (2 bytes). + separator_size = 2 if current else 0 + if current and current_size + separator_size + envelope_size > _MAX_INGESTION_PAYLOAD_SIZE_BYTES: + chunks.append(current) + current = [] + current_size = 2 + separator_size = 0 + current.append(envelope) + current_size += separator_size + envelope_size + if current: + chunks.append(current) + # If size-based packing produced a single chunk (e.g. because the byte + # estimate under-reported), fall back to halving by count so repeated 413s keep shrinking the + # batch. + if len(chunks) < 2 and len(envelopes) > 1: + midpoint = len(envelopes) // 2 + chunks = [envelopes[:midpoint], envelopes[midpoint:]] + return chunks + + +def _is_item_too_large(message: Optional[str]) -> bool: + # Determine if a per-item error indicates the envelope exceeded the ingestion size limit. + if not message: + return False + lowered = message.lower() + return any(marker in lowered for marker in _ITEM_TOO_LARGE_MESSAGE_MARKERS) 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..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 @@ -49,7 +49,7 @@ from azure.monitor.opentelemetry.exporter._connection_string_parser import ( ConnectionStringParser, ) -from azure.monitor.opentelemetry.exporter._storage import LocalFileStorage +from azure.monitor.opentelemetry.exporter._storage import LocalFileStorage, StorageExportResult from azure.monitor.opentelemetry.exporter._configuration._state import ( get_configuration_manager, ) @@ -63,6 +63,8 @@ _get_os_name, _get_rp_name, _get_attach_type_name, + _is_item_too_large, + _split_oversized_batch, ) from azure.monitor.opentelemetry.exporter._version import VERSION as ext_version from azure.monitor.opentelemetry.exporter.export._rate_limiter import ( @@ -317,6 +319,45 @@ def _handle_transmit_from_storage(self, envelopes: List[TelemetryItem], result: if self._should_collect_customer_sdkstats(): track_dropped_items(envelopes, DropCode.CLIENT_STORAGE_DISABLED) + def _handle_payload_too_large( + self, envelopes: List[TelemetryItem], response_error: HttpResponseError + ) -> ExportResult: + # Handle a 413 (Payload Too Large) response by splitting and persisting for retry. + if not self.storage: + if not self._is_stats_exporter(): + logger.debug( + "Payload too large but offline storage is disabled; dropping %d envelope(s): %s.", + len(envelopes), + response_error.message, + ) + return ExportResult.FAILED_NOT_RETRYABLE + if len(envelopes) <= 1: + if not self._is_stats_exporter(): + logger.warning( + "Payload too large and cannot be split further; dropping %d envelope(s): %s.", + len(envelopes), + response_error.message, + ) + if self._should_collect_customer_sdkstats() and isinstance(response_error.status_code, int): + track_dropped_items(envelopes, response_error.status_code) + return ExportResult.FAILED_NOT_RETRYABLE + + for sub_batch in _split_oversized_batch(envelopes): + result_from_storage_put = self.storage.put([x.as_dict() for x in sub_batch]) + if result_from_storage_put == StorageExportResult.LOCAL_FILE_BLOB_SUCCESS: + if not self._is_stats_exporter() and self._should_collect_customer_sdkstats(): + track_retry_items(sub_batch, response_error) + else: + if not self._is_stats_exporter(): + logger.debug( + "Failed to persist a split sub-batch of %d envelope(s) after a 413; " + "these envelopes are dropped.", + len(sub_batch), + ) + if self._should_collect_customer_sdkstats(): + track_dropped_items_from_storage(result_from_storage_put, sub_batch) + return ExportResult.FAILED_NOT_RETRYABLE + # pylint: disable=too-many-branches # pylint: disable=too-many-nested-blocks # pylint: disable=too-many-statements @@ -446,11 +487,17 @@ def _transmit(self, envelopes: List[TelemetryItem], _skip_rate_limit: bool = Fal and isinstance(error.status_code, int) ): track_dropped_items([envelopes[error.index]], error.status_code) + dropped_envelope = envelopes[error.index] if error.index is not None else None + # When the server rejects an item for exceeding the per-item size limit + # (detected via _is_item_too_large), omit its payload from the log: it is + # too large to dump and may contain PII, which would leak and flood the logs. + if dropped_envelope is None or _is_item_too_large(error.message): + dropped_envelope = "" # type: ignore[assignment] logger.error( "Data drop %s: %s %s.", error.status_code, error.message, - (envelopes[error.index] if error.index is not None else ""), + dropped_envelope, ) storage = self.storage if storage and resend_envelopes: @@ -577,6 +624,11 @@ def _transmit(self, envelopes: List[TelemetryItem], _skip_rate_limit: bool = Fal if self._should_collect_stats(): _update_requests_map(_REQ_EXCEPTION_NAME[1], value="Circular Redirect") result = ExportResult.FAILED_NOT_RETRYABLE + elif response_error.status_code == 413: + if self._should_collect_stats(): + _update_requests_map(_REQ_FAILURE_NAME[1], value=response_error.status_code) + result = self._handle_payload_too_large(envelopes, response_error) + else: # Any other status code counts as failure (non-retryable) # 400 - Invalid - The server cannot or will not process the request due to the invalid telemetry (invalid data, iKey, etc.) 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): 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..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 @@ -3,6 +3,8 @@ import os import shutil +import tempfile +import threading import unittest from unittest import mock from datetime import datetime, timedelta, timezone @@ -17,8 +19,14 @@ ExportResult, _get_storage_directory, ) -from azure.monitor.opentelemetry.exporter._utils import _get_retry_delay_from_headers +from azure.monitor.opentelemetry.exporter._utils import ( + _get_retry_delay_from_headers, + _get_envelope_serialized_size, + _is_item_too_large, + _split_oversized_batch, +) 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, @@ -49,6 +57,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" @@ -1115,6 +1125,75 @@ 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 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( + 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._utils._get_envelope_serialized_size", + ) as mock_size: + 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) + # The drop path never re-serializes the envelope to decide what to log. + mock_size.assert_not_called() + log_output = "\n".join(cm.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) + + 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) @@ -1299,6 +1378,294 @@ 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 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 + 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()), + ] + 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) + 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.""" + 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 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 = [ + 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)): + 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) + # _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 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): + """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, + 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()), + ] + 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_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 = [ + 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, []) + 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() + + 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 + 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 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()), + 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="DEBUG") as cm: + result = exporter._transmit(custom_envelopes_to_export) + self.assertEqual(result, ExportResult.FAILED_NOT_RETRYABLE) + 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)) + + 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._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([]), []) + + 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._utils._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._utils._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._utils._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)