From 549eee690708f67181b8c6b815f603f4194026ac Mon Sep 17 00:00:00 2001 From: Paul Van Eck Date: Wed, 9 Nov 2022 15:26:46 -0800 Subject: [PATCH 1/4] [Monitor][Ingestion] Adjust upload error handling This uses a callback approach to error handling where the user can optionally pass in an error callback function for custom error handling. If no callback function is provided, the default behavior is to raise the first error encountered. Signed-off-by: Paul Van Eck --- .../azure-monitor-ingestion/CHANGELOG.md | 1 + sdk/monitor/azure-monitor-ingestion/README.md | 24 ++++++++-- .../azure-monitor-ingestion/assets.json | 2 +- .../azure/monitor/ingestion/_helpers.py | 33 ++++++------- .../monitor/ingestion/_operations/_patch.py | 38 ++++++++++----- .../ingestion/aio/_operations/_patch.py | 40 +++++++++++----- .../sample_custom_error_callback_async.py | 48 +++++++++++++++++++ .../sample_send_small_logs_async.py | 14 ++---- .../samples/sample_custom_error_callback.py | 41 ++++++++++++++++ .../samples/sample_send_small_logs.py | 9 +--- .../tests/test_logs_ingestion.py | 27 +++++++++++ .../tests/test_logs_ingestion_async.py | 29 +++++++++++ 12 files changed, 242 insertions(+), 64 deletions(-) create mode 100644 sdk/monitor/azure-monitor-ingestion/samples/async_samples/sample_custom_error_callback_async.py create mode 100644 sdk/monitor/azure-monitor-ingestion/samples/sample_custom_error_callback.py diff --git a/sdk/monitor/azure-monitor-ingestion/CHANGELOG.md b/sdk/monitor/azure-monitor-ingestion/CHANGELOG.md index 1e76605be31c..8d9f17e7cb4d 100644 --- a/sdk/monitor/azure-monitor-ingestion/CHANGELOG.md +++ b/sdk/monitor/azure-monitor-ingestion/CHANGELOG.md @@ -3,6 +3,7 @@ ## 1.0.0b2 (Unreleased) ### Features Added + - Added new `on_error` parameter to the `upload` method to allow users to handle errors in their own way. ### Breaking Changes - Removed support for max_concurrency diff --git a/sdk/monitor/azure-monitor-ingestion/README.md b/sdk/monitor/azure-monitor-ingestion/README.md index fef340fb3d13..2d6df96335a3 100644 --- a/sdk/monitor/azure-monitor-ingestion/README.md +++ b/sdk/monitor/azure-monitor-ingestion/README.md @@ -93,6 +93,7 @@ The logs that were uploaded using this library can be queried using the [Azure M ## Examples - [Upload custom logs](#upload-custom-logs) +- [Upload with custom error handling](#upload-with-custom-error-handling) ### Upload custom logs @@ -100,7 +101,7 @@ This example shows uploading logs to Azure Monitor. ```python import os -from azure.monitor.ingestion import LogsIngestionClient, UploadLogsStatus +from azure.monitor.ingestion import LogsIngestionClient from azure.identity import DefaultAzureCredential endpoint = os.environ['DATA_COLLECTION_ENDPOINT'] @@ -122,10 +123,23 @@ body = [ } ] -response = client.upload(rule_id=rule_id, stream_name=os.environ['LOGS_DCR_STREAM_NAME'], logs=body) -if response.status != UploadLogsStatus.SUCCESS: - failed_logs = response.failed_logs_index - print(failed_logs) +client.upload(rule_id=rule_id, stream_name=os.environ['LOGS_DCR_STREAM_NAME'], logs=body) +``` + +### Upload with custom error handling + +To upload logs with custom error handling, you can pass a callback function to the `on_error` parameter of the `upload` method. +The callback function will be called for each error that occurs during the upload and should expect two keyword arguments: `error` and `logs`. +These arguments correspond to the error encountered and the list of logs that failed to upload. + +```python +failed_logs = [] +def on_error(**kwargs): + print("Log chunk failed to upload with error: ", kwargs.get("error")) + # Collect all logs that failed to upload. + failed_logs.extend(kwargs.get("logs", [])) + +client.upload(rule_id=rule_id, stream_name=os.environ['LOGS_DCR_STREAM_NAME'], logs=body, on_error=on_error) ``` ## Troubleshooting diff --git a/sdk/monitor/azure-monitor-ingestion/assets.json b/sdk/monitor/azure-monitor-ingestion/assets.json index 01aca6c8dd2e..96fd358e9ce6 100644 --- a/sdk/monitor/azure-monitor-ingestion/assets.json +++ b/sdk/monitor/azure-monitor-ingestion/assets.json @@ -2,5 +2,5 @@ "AssetsRepo": "Azure/azure-sdk-assets", "AssetsRepoPrefixPath": "python", "TagPrefix": "python/monitor/azure-monitor-ingestion", - "Tag": "python/monitor/azure-monitor-ingestion_de97e8025d" + "Tag": "python/monitor/azure-monitor-ingestion_7c256dddd2" } diff --git a/sdk/monitor/azure-monitor-ingestion/azure/monitor/ingestion/_helpers.py b/sdk/monitor/azure-monitor-ingestion/azure/monitor/ingestion/_helpers.py index 490260425dff..e13fbfcb9ea3 100644 --- a/sdk/monitor/azure-monitor-ingestion/azure/monitor/ingestion/_helpers.py +++ b/sdk/monitor/azure-monitor-ingestion/azure/monitor/ingestion/_helpers.py @@ -2,40 +2,41 @@ # Copyright (c) Microsoft Corporation. # Licensed under the MIT License. # ------------------------------------ - import json +import logging +from typing import Generator, List import zlib -MAX_CHUNK_SIZE = 1000000 -CHAR_SIZE = 4 +_LOGGER = logging.getLogger(__name__) + +MAX_CHUNK_SIZE_BYTES = 1024 * 1024 # 1 MiB +CHAR_SIZE_BYTES = 4 -def _split_chunks(logs): - chunks = [] +def _split_chunks(logs: List) -> Generator: chunk_size = 0 curr_chunk = [] for log in logs: # each char is 4 bytes - size = len(json.dumps(log)) * CHAR_SIZE - if chunk_size + size < MAX_CHUNK_SIZE: + size = len(json.dumps(log)) * CHAR_SIZE_BYTES + if chunk_size + size < MAX_CHUNK_SIZE_BYTES: curr_chunk.append(log) chunk_size += size else: - chunks.append(curr_chunk) + _LOGGER.debug('Yielding chunk with size: %d', chunk_size) + yield curr_chunk curr_chunk = [log] chunk_size = size if len(curr_chunk) > 0: - chunks.append(curr_chunk) - return chunks + _LOGGER.debug('Yielding chunk with size: %d', chunk_size) + yield curr_chunk -def _create_gzip_requests(logs): - requests = [] - chunks = _split_chunks(logs) - for chunk in chunks: +def _create_gzip_requests(logs: List) -> Generator: + for chunk in _split_chunks(logs): zlib_mode = 16 + zlib.MAX_WBITS # for gzip encoding _compress = zlib.compressobj(wbits=zlib_mode) data = _compress.compress(bytes(json.dumps(chunk), encoding="utf-8")) data += _compress.flush() - requests.append(data) - return requests + _LOGGER.debug('Yielding gzip compressed data, Length: %d', len(data)) + yield data, chunk diff --git a/sdk/monitor/azure-monitor-ingestion/azure/monitor/ingestion/_operations/_patch.py b/sdk/monitor/azure-monitor-ingestion/azure/monitor/ingestion/_operations/_patch.py index 951e08fdcf91..911b643a6a40 100644 --- a/sdk/monitor/azure-monitor-ingestion/azure/monitor/ingestion/_operations/_patch.py +++ b/sdk/monitor/azure-monitor-ingestion/azure/monitor/ingestion/_operations/_patch.py @@ -6,9 +6,10 @@ Follow our quickstart for examples: https://aka.ms/azsdk/python/dpcodegen/python/customize """ +import logging import sys -from typing import List, Any, Union, IO, Iterable, Tuple -from azure.core.exceptions import HttpResponseError +from typing import Callable, List, Any, Union, IO, Optional + from ._operations import LogsIngestionClientOperationsMixin as GeneratedOps from .._helpers import _create_gzip_requests @@ -16,6 +17,9 @@ from collections.abc import MutableMapping else: from typing import MutableMapping # type: ignore # pylint: disable=ungrouped-imports + + +_LOGGER = logging.getLogger(__name__) JSON = MutableMapping[str, Any] # pylint: disable=unsubscriptable-object @@ -25,32 +29,40 @@ def upload( # pylint: disable=arguments-renamed, arguments-differ rule_id: str, stream_name: str, logs: Union[List[JSON], IO], + on_error: Optional[Callable[..., None]] = None, **kwargs: Any - ) -> Iterable[Tuple[HttpResponseError, List[JSON]]]: + ) -> None: """Ingestion API used to directly ingest data using Data Collection Rules. - See error response code and error response message for more detail. + Logs are divided into chunks of 1MB or less, then each chunk is gzip-compressed and uploaded. - :param rule_id: The immutable Id of the Data Collection Rule resource. + :param rule_id: The immutable ID of the Data Collection Rule resource. :type rule_id: str :param stream_name: The streamDeclaration name as defined in the Data Collection Rule. :type stream_name: str :param logs: An array of objects matching the schema defined by the provided stream. :type logs: list[JSON] or IO - :return: Iterable[Tuple[HttpResponseError, List[JSON]]] - :rtype: Iterable[Tuple[HttpResponseError, List[JSON]]] + :param on_error: The callback function that is called when a chunk of logs fail to upload. + This function should expect two keyword arguments: `error` and `logs`. + These arguments correspond to the error encountered and the list of logs that failed to upload. + If no function is provided, then the first exception encountered will be raised. + :type on_error: Optional[Callable[..., None]] + :return: None + :rtype: None :raises: ~azure.core.exceptions.HttpResponseError """ - requests = _create_gzip_requests(logs) - results = [] - for request in requests: + for gzip_data, log_chunk in _create_gzip_requests(logs): try: super().upload( - rule_id, stream=stream_name, body=request, content_encoding="gzip", **kwargs + rule_id, stream=stream_name, body=gzip_data, content_encoding="gzip", **kwargs ) except Exception as err: # pylint: disable=broad-except - results.append((err, request)) - return results + if on_error: + on_error(error=err, logs=log_chunk) + else: + _LOGGER.error( "Failed to upload chunk containing %d log entries", len(log_chunk)) + raise err + __all__: List[str] = [ "LogsIngestionClientOperationsMixin" diff --git a/sdk/monitor/azure-monitor-ingestion/azure/monitor/ingestion/aio/_operations/_patch.py b/sdk/monitor/azure-monitor-ingestion/azure/monitor/ingestion/aio/_operations/_patch.py index 46c9155b9ce2..efb56b7d0768 100644 --- a/sdk/monitor/azure-monitor-ingestion/azure/monitor/ingestion/aio/_operations/_patch.py +++ b/sdk/monitor/azure-monitor-ingestion/azure/monitor/ingestion/aio/_operations/_patch.py @@ -6,9 +6,10 @@ Follow our quickstart for examples: https://aka.ms/azsdk/python/dpcodegen/python/customize """ +import logging import sys -from typing import List, Any, Union, IO, Iterable, Tuple -from azure.core.exceptions import HttpResponseError +from typing import Callable, List, Any, Awaitable, Union, IO, Optional + from ._operations import LogsIngestionClientOperationsMixin as GeneratedOps from ..._helpers import _create_gzip_requests @@ -16,6 +17,9 @@ from collections.abc import MutableMapping else: from typing import MutableMapping # type: ignore # pylint: disable=ungrouped-imports + + +_LOGGER = logging.getLogger(__name__) JSON = MutableMapping[str, Any] # pylint: disable=unsubscriptable-object @@ -25,30 +29,40 @@ async def upload( # pylint: disable=arguments-renamed, arguments-differ rule_id: str, stream_name: str, logs: Union[List[JSON], IO], + on_error: Optional[Callable[..., Awaitable[None]]] = None, **kwargs: Any - ) -> Iterable[Tuple[HttpResponseError, List[JSON]]]: + ) -> None: """Ingestion API used to directly ingest data using Data Collection Rules. - See error response code and error response message for more detail. + Logs are divided into chunks of 1MB or less, then each chunk is gzip-compressed and uploaded. - :param rule_id: The immutable Id of the Data Collection Rule resource. + :param rule_id: The immutable ID of the Data Collection Rule resource. :type rule_id: str :param stream: The streamDeclaration name as defined in the Data Collection Rule. :type stream: str :param logs: An array of objects matching the schema defined by the provided stream. :type logs: list[JSON] or IO - :return: Iterable[Tuple[HttpResponseError, List[JSON]]] - :rtype: Iterable[Tuple[HttpResponseError, List[JSON]]] + :param on_error: The asynchronous callback function that is called when a set of logs fail to upload. + This function should expect two keyword arguments: `error` and `logs`. + These arguments correspond to the error encountered and the list of logs that failed to upload. + If no function is provided, then the first exception encountered will be raised. + :type on_error: Optional[Callable[..., Awaitable[None]]] + :return: None + :rtype: None :raises: ~azure.core.exceptions.HttpResponseError """ - requests = _create_gzip_requests(logs) - results = [] - for request in requests: + for gzip_data, log_chunk in _create_gzip_requests(logs): try: - await super().upload(rule_id, stream=stream_name, body=request, content_encoding="gzip", **kwargs) + await super().upload( + rule_id, stream=stream_name, body=gzip_data, content_encoding="gzip", **kwargs + ) except Exception as err: # pylint: disable=broad-except - results.append((err, request)) - return results + if on_error: + await on_error(error=err, logs=log_chunk) + else: + _LOGGER.error( "Failed to upload chunk containing %d log entries", len(log_chunk)) + raise err + __all__: List[str] = [ "LogsIngestionClientOperationsMixin" diff --git a/sdk/monitor/azure-monitor-ingestion/samples/async_samples/sample_custom_error_callback_async.py b/sdk/monitor/azure-monitor-ingestion/samples/async_samples/sample_custom_error_callback_async.py new file mode 100644 index 000000000000..ee418da197ad --- /dev/null +++ b/sdk/monitor/azure-monitor-ingestion/samples/async_samples/sample_custom_error_callback_async.py @@ -0,0 +1,48 @@ +""" +Usage: python sample_custom_error_callback_async.py +""" + +import os +import asyncio +from azure.monitor.ingestion.aio import LogsIngestionClient +from azure.identity.aio import DefaultAzureCredential + + +async def send_logs(): + endpoint = os.environ['DATA_COLLECTION_ENDPOINT'] + rule_id = os.environ['LOGS_DCR_RULE_ID'] + body = [ + { + "Time": "2021-12-08T23:51:14.1104269Z", + "Computer": "Computer1", + "AdditionalContext": "sabhyrav-2" + }, + { + "Time": "2021-12-08T23:51:14.1104269Z", + "Computer": "Computer2", + "AdditionalContext": "sabhyrav" + } + ] + credential = DefaultAzureCredential() + + failed_logs = [] + async def on_error(**kwargs): + print("Log chunk failed to upload with error: ", kwargs.get("error")) + failed_logs.extend(kwargs.get("logs", [])) + + async def on_error_pass(**kwargs): + return + + client = LogsIngestionClient(endpoint=endpoint, credential=credential, logging_enable=True) + async with client: + await client.upload(rule_id=rule_id, stream_name=os.environ['LOGS_DCR_STREAM_NAME'], logs=body, on_error=on_error) + + # Retry once with any failed logs, and this time ignore any errors. + print("Retrying logs that failed to upload...") + if failed_logs: + await client.upload(rule_id=rule_id, stream_name=os.environ['LOGS_DCR_STREAM_NAME'], logs=failed_logs, on_error=on_error_pass) + await credential.close() + + +if __name__ == '__main__': + asyncio.run(send_logs()) diff --git a/sdk/monitor/azure-monitor-ingestion/samples/async_samples/sample_send_small_logs_async.py b/sdk/monitor/azure-monitor-ingestion/samples/async_samples/sample_send_small_logs_async.py index dbaa625c6044..abb9b5661330 100644 --- a/sdk/monitor/azure-monitor-ingestion/samples/async_samples/sample_send_small_logs_async.py +++ b/sdk/monitor/azure-monitor-ingestion/samples/async_samples/sample_send_small_logs_async.py @@ -5,15 +5,10 @@ import os import asyncio from azure.monitor.ingestion.aio import LogsIngestionClient -from azure.monitor.ingestion import UploadLogsStatus from azure.identity.aio import DefaultAzureCredential async def send_logs(): endpoint = os.environ['DATA_COLLECTION_ENDPOINT'] - credential = DefaultAzureCredential() - - client = LogsIngestionClient(endpoint=endpoint, credential=credential, logging_enable=True) - rule_id = os.environ['LOGS_DCR_RULE_ID'] body = [ { @@ -27,11 +22,12 @@ async def send_logs(): "AdditionalContext": "sabhyrav" } ] + credential = DefaultAzureCredential() - response = await client.upload(rule_id=rule_id, stream_name=os.environ['LOGS_DCR_STREAM_NAME'], logs=body) - if response.status != UploadLogsStatus.SUCCESS: - failed_logs = response.failed_logs_index - print(failed_logs) + client = LogsIngestionClient(endpoint=endpoint, credential=credential, logging_enable=True) + async with client: + await client.upload(rule_id=rule_id, stream_name=os.environ['LOGS_DCR_STREAM_NAME'], logs=body) + await credential.close() if __name__ == '__main__': diff --git a/sdk/monitor/azure-monitor-ingestion/samples/sample_custom_error_callback.py b/sdk/monitor/azure-monitor-ingestion/samples/sample_custom_error_callback.py new file mode 100644 index 000000000000..2f1d1d7eeec2 --- /dev/null +++ b/sdk/monitor/azure-monitor-ingestion/samples/sample_custom_error_callback.py @@ -0,0 +1,41 @@ +""" +Usage: python sample_custom_error_callback.py +""" + +import os +from azure.monitor.ingestion import LogsIngestionClient +from azure.identity import DefaultAzureCredential + +endpoint = os.environ['DATA_COLLECTION_ENDPOINT'] +credential = DefaultAzureCredential() + +client = LogsIngestionClient(endpoint=endpoint, credential=credential, logging_enable=True) + +rule_id = os.environ['LOGS_DCR_RULE_ID'] +body = [ + { + "Time": "2021-12-08T23:51:14.1104269Z", + "Computer": "Computer1", + "AdditionalContext": "context-2" + }, + { + "Time": "2021-12-08T23:51:14.1104269Z", + "Computer": "Computer2", + "AdditionalContext": "context" + } + ] + +failed_logs = [] +def on_error(**kwargs): + print("Log chunk failed to upload with error: ", kwargs.get("error")) + failed_logs.extend(kwargs.get("logs", [])) + +def on_error_pass(**kwargs): + pass + +client.upload(rule_id=rule_id, stream_name=os.environ['LOGS_DCR_STREAM_NAME'], logs=body, on_error=on_error) + +# Retry once with any failed logs, and this time ignore any errors. +print("Retrying logs that failed to upload...") +if failed_logs: + client.upload(rule_id=rule_id, stream_name=os.environ['LOGS_DCR_STREAM_NAME'], logs=failed_logs, on_error=on_error_pass) diff --git a/sdk/monitor/azure-monitor-ingestion/samples/sample_send_small_logs.py b/sdk/monitor/azure-monitor-ingestion/samples/sample_send_small_logs.py index 7070a5daf019..34a83fafa6e0 100644 --- a/sdk/monitor/azure-monitor-ingestion/samples/sample_send_small_logs.py +++ b/sdk/monitor/azure-monitor-ingestion/samples/sample_send_small_logs.py @@ -3,7 +3,7 @@ """ import os -from azure.monitor.ingestion import LogsIngestionClient, UploadLogsStatus +from azure.monitor.ingestion import LogsIngestionClient from azure.identity import DefaultAzureCredential endpoint = os.environ['DATA_COLLECTION_ENDPOINT'] @@ -25,9 +25,4 @@ } ] -response = client.upload(rule_id=rule_id, stream_name=os.environ['LOGS_DCR_STREAM_NAME'], logs=body) - -## iterates directly over Tuple[HttpResponseError, JSON]: -for error, failed_logs in response: - # prints nothing if there are no failed_logs (i/e the status is success) - print(failed_logs, error) +client.upload(rule_id=rule_id, stream_name=os.environ['LOGS_DCR_STREAM_NAME'], logs=body) diff --git a/sdk/monitor/azure-monitor-ingestion/tests/test_logs_ingestion.py b/sdk/monitor/azure-monitor-ingestion/tests/test_logs_ingestion.py index d9a56ce87cbb..a9681cd500dd 100644 --- a/sdk/monitor/azure-monitor-ingestion/tests/test_logs_ingestion.py +++ b/sdk/monitor/azure-monitor-ingestion/tests/test_logs_ingestion.py @@ -3,6 +3,9 @@ # Licensed under the MIT License. See LICENSE.txt in the project root for # license information. # ------------------------------------------------------------------------- +import pytest + +from azure.core.exceptions import HttpResponseError from azure.monitor.ingestion import LogsIngestionClient from devtools_testutils import AzureRecordedTestCase @@ -33,3 +36,27 @@ def test_send_logs(self, recorded_test, monitor_info): ] client.upload(rule_id=monitor_info['dcr_id'], stream_name=monitor_info['stream_name'], logs=body) + + def test_send_logs_error(self, recorded_test, monitor_info): + client = self.create_client_from_credential( + LogsIngestionClient, self.get_credential(LogsIngestionClient), endpoint=monitor_info['dce']) + body = [{"foo": "bar"}] + + with pytest.raises(HttpResponseError) as ex: + client.upload(rule_id='bad-rule', stream_name=monitor_info['stream_name'], logs=body) + + def test_send_logs_error_custom(self, recorded_test, monitor_info): + client = self.create_client_from_credential( + LogsIngestionClient, self.get_credential(LogsIngestionClient), endpoint=monitor_info['dce']) + body = [{"foo": "bar"}] + + def on_error(**kwargs): + on_error.called = True + assert isinstance(kwargs.get("error"), HttpResponseError) + assert kwargs.get("logs") == body + + on_error.called = False + + client.upload( + rule_id='bad-rule', stream_name=monitor_info['stream_name'], logs=body, on_error=on_error) + assert on_error.called diff --git a/sdk/monitor/azure-monitor-ingestion/tests/test_logs_ingestion_async.py b/sdk/monitor/azure-monitor-ingestion/tests/test_logs_ingestion_async.py index bb6763f236c3..f389a3faa212 100644 --- a/sdk/monitor/azure-monitor-ingestion/tests/test_logs_ingestion_async.py +++ b/sdk/monitor/azure-monitor-ingestion/tests/test_logs_ingestion_async.py @@ -5,6 +5,7 @@ # ------------------------------------------------------------------------- import pytest +from azure.core.exceptions import HttpResponseError from azure.monitor.ingestion.aio import LogsIngestionClient from devtools_testutils import AzureRecordedTestCase @@ -35,3 +36,31 @@ async def test_send_logs_async(self, recorded_test, monitor_info): } ] await client.upload(rule_id=monitor_info['dcr_id'], stream_name=monitor_info['stream_name'], logs=body) + + @pytest.mark.asyncio + async def test_send_logs_error(self, recorded_test, monitor_info): + client = self.create_client_from_credential( + LogsIngestionClient, self.get_credential(LogsIngestionClient, is_async=True), endpoint=monitor_info['dce']) + body = [{"foo": "bar"}] + + with pytest.raises(HttpResponseError) as ex: + async with client: + await client.upload(rule_id='bad-rule', stream_name=monitor_info['stream_name'], logs=body) + + @pytest.mark.asyncio + async def test_send_logs_error_custom(self, recorded_test, monitor_info): + client = self.create_client_from_credential( + LogsIngestionClient, self.get_credential(LogsIngestionClient, is_async=True), endpoint=monitor_info['dce']) + body = [{"foo": "bar"}] + + async def on_error(**kwargs): + on_error.called = True + assert isinstance(kwargs.get("error"), HttpResponseError) + assert kwargs.get("logs") == body + + on_error.called = False + + async with client: + await client.upload( + rule_id='bad-rule', stream_name=monitor_info['stream_name'], logs=body, on_error=on_error) + assert on_error.called From 399be1c6f8225d078d745df17533b116794588dd Mon Sep 17 00:00:00 2001 From: Paul Van Eck Date: Tue, 27 Dec 2022 12:35:59 -0800 Subject: [PATCH 2/4] Use positional args Seems the type hints are more feasible with positional args. Signed-off-by: Paul Van Eck --- sdk/monitor/azure-monitor-ingestion/README.md | 10 ++++------ .../azure/monitor/ingestion/_operations/_patch.py | 14 +++++++------- .../monitor/ingestion/aio/_operations/_patch.py | 14 +++++++------- .../sample_custom_error_callback_async.py | 15 +++++++++------ .../samples/sample_custom_error_callback.py | 13 ++++++++----- .../tests/test_logs_ingestion.py | 6 +++--- .../tests/test_logs_ingestion_async.py | 6 +++--- 7 files changed, 41 insertions(+), 37 deletions(-) diff --git a/sdk/monitor/azure-monitor-ingestion/README.md b/sdk/monitor/azure-monitor-ingestion/README.md index 2d6df96335a3..b4164bb7fc2b 100644 --- a/sdk/monitor/azure-monitor-ingestion/README.md +++ b/sdk/monitor/azure-monitor-ingestion/README.md @@ -128,16 +128,14 @@ client.upload(rule_id=rule_id, stream_name=os.environ['LOGS_DCR_STREAM_NAME'], l ### Upload with custom error handling -To upload logs with custom error handling, you can pass a callback function to the `on_error` parameter of the `upload` method. -The callback function will be called for each error that occurs during the upload and should expect two keyword arguments: `error` and `logs`. -These arguments correspond to the error encountered and the list of logs that failed to upload. +To upload logs with custom error handling, you can pass a callback function to the `on_error` parameter of the `upload` method. The callback function will be called for each error that occurs during the upload and should expect two arguments which correspond to the error encountered and the list of logs that failed to upload. ```python failed_logs = [] -def on_error(**kwargs): - print("Log chunk failed to upload with error: ", kwargs.get("error")) +def on_error(error, logs): + print("Log chunk failed to upload with error: ", error) # Collect all logs that failed to upload. - failed_logs.extend(kwargs.get("logs", [])) + failed_logs.extend(logs) client.upload(rule_id=rule_id, stream_name=os.environ['LOGS_DCR_STREAM_NAME'], logs=body, on_error=on_error) ``` diff --git a/sdk/monitor/azure-monitor-ingestion/azure/monitor/ingestion/_operations/_patch.py b/sdk/monitor/azure-monitor-ingestion/azure/monitor/ingestion/_operations/_patch.py index 911b643a6a40..72a27673ecd2 100644 --- a/sdk/monitor/azure-monitor-ingestion/azure/monitor/ingestion/_operations/_patch.py +++ b/sdk/monitor/azure-monitor-ingestion/azure/monitor/ingestion/_operations/_patch.py @@ -29,7 +29,7 @@ def upload( # pylint: disable=arguments-renamed, arguments-differ rule_id: str, stream_name: str, logs: Union[List[JSON], IO], - on_error: Optional[Callable[..., None]] = None, + on_error: Optional[Callable[[Exception, List[JSON]], None]] = None, **kwargs: Any ) -> None: """Ingestion API used to directly ingest data using Data Collection Rules. @@ -42,11 +42,11 @@ def upload( # pylint: disable=arguments-renamed, arguments-differ :type stream_name: str :param logs: An array of objects matching the schema defined by the provided stream. :type logs: list[JSON] or IO - :param on_error: The callback function that is called when a chunk of logs fail to upload. - This function should expect two keyword arguments: `error` and `logs`. - These arguments correspond to the error encountered and the list of logs that failed to upload. - If no function is provided, then the first exception encountered will be raised. - :type on_error: Optional[Callable[..., None]] + :param on_error: The callback function that is called when a chunk of logs fails to upload. + This function should expect two arguments that correspond to the error encountered and + the list of logs that failed to upload. If no function is provided, then the first exception + encountered will be raised. + :type on_error: Optional[Callable[[Exception, List[JSON]], None]] :return: None :rtype: None :raises: ~azure.core.exceptions.HttpResponseError @@ -58,7 +58,7 @@ def upload( # pylint: disable=arguments-renamed, arguments-differ ) except Exception as err: # pylint: disable=broad-except if on_error: - on_error(error=err, logs=log_chunk) + on_error(err, log_chunk) else: _LOGGER.error( "Failed to upload chunk containing %d log entries", len(log_chunk)) raise err diff --git a/sdk/monitor/azure-monitor-ingestion/azure/monitor/ingestion/aio/_operations/_patch.py b/sdk/monitor/azure-monitor-ingestion/azure/monitor/ingestion/aio/_operations/_patch.py index efb56b7d0768..60eda208efdf 100644 --- a/sdk/monitor/azure-monitor-ingestion/azure/monitor/ingestion/aio/_operations/_patch.py +++ b/sdk/monitor/azure-monitor-ingestion/azure/monitor/ingestion/aio/_operations/_patch.py @@ -29,7 +29,7 @@ async def upload( # pylint: disable=arguments-renamed, arguments-differ rule_id: str, stream_name: str, logs: Union[List[JSON], IO], - on_error: Optional[Callable[..., Awaitable[None]]] = None, + on_error: Optional[Callable[[Exception, List[JSON]], Awaitable[None]]] = None, **kwargs: Any ) -> None: """Ingestion API used to directly ingest data using Data Collection Rules. @@ -42,11 +42,11 @@ async def upload( # pylint: disable=arguments-renamed, arguments-differ :type stream: str :param logs: An array of objects matching the schema defined by the provided stream. :type logs: list[JSON] or IO - :param on_error: The asynchronous callback function that is called when a set of logs fail to upload. - This function should expect two keyword arguments: `error` and `logs`. - These arguments correspond to the error encountered and the list of logs that failed to upload. - If no function is provided, then the first exception encountered will be raised. - :type on_error: Optional[Callable[..., Awaitable[None]]] + :param on_error: The asynchronous callback function that is called when a chunk of logs fails to upload. + This function should expect two arguments that correspond to the error encountered and + the list of logs that failed to upload. If no function is provided, then the first exception + encountered will be raised. + :type on_error: Optional[Callable[[Exception, List[JSON]], None]] :return: None :rtype: None :raises: ~azure.core.exceptions.HttpResponseError @@ -58,7 +58,7 @@ async def upload( # pylint: disable=arguments-renamed, arguments-differ ) except Exception as err: # pylint: disable=broad-except if on_error: - await on_error(error=err, logs=log_chunk) + await on_error(err, log_chunk) else: _LOGGER.error( "Failed to upload chunk containing %d log entries", len(log_chunk)) raise err diff --git a/sdk/monitor/azure-monitor-ingestion/samples/async_samples/sample_custom_error_callback_async.py b/sdk/monitor/azure-monitor-ingestion/samples/async_samples/sample_custom_error_callback_async.py index ee418da197ad..a0e35d2fc717 100644 --- a/sdk/monitor/azure-monitor-ingestion/samples/async_samples/sample_custom_error_callback_async.py +++ b/sdk/monitor/azure-monitor-ingestion/samples/async_samples/sample_custom_error_callback_async.py @@ -26,20 +26,23 @@ async def send_logs(): credential = DefaultAzureCredential() failed_logs = [] - async def on_error(**kwargs): - print("Log chunk failed to upload with error: ", kwargs.get("error")) - failed_logs.extend(kwargs.get("logs", [])) - async def on_error_pass(**kwargs): - return + # Sample callback that stores the logs that failed to upload. + async def on_error(error, logs): + print("Log chunk failed to upload with error: ", error) + failed_logs.extend(logs) + + # Sample callback that just ignores the error. + async def on_error_pass(*_): + pass client = LogsIngestionClient(endpoint=endpoint, credential=credential, logging_enable=True) async with client: await client.upload(rule_id=rule_id, stream_name=os.environ['LOGS_DCR_STREAM_NAME'], logs=body, on_error=on_error) # Retry once with any failed logs, and this time ignore any errors. - print("Retrying logs that failed to upload...") if failed_logs: + print("Retrying logs that failed to upload...") await client.upload(rule_id=rule_id, stream_name=os.environ['LOGS_DCR_STREAM_NAME'], logs=failed_logs, on_error=on_error_pass) await credential.close() diff --git a/sdk/monitor/azure-monitor-ingestion/samples/sample_custom_error_callback.py b/sdk/monitor/azure-monitor-ingestion/samples/sample_custom_error_callback.py index 2f1d1d7eeec2..15c2302047ec 100644 --- a/sdk/monitor/azure-monitor-ingestion/samples/sample_custom_error_callback.py +++ b/sdk/monitor/azure-monitor-ingestion/samples/sample_custom_error_callback.py @@ -26,16 +26,19 @@ ] failed_logs = [] -def on_error(**kwargs): - print("Log chunk failed to upload with error: ", kwargs.get("error")) - failed_logs.extend(kwargs.get("logs", [])) -def on_error_pass(**kwargs): +# Sample callback that stores the logs that failed to upload. +def on_error(error, logs): + print("Log chunk failed to upload with error: ", error) + failed_logs.extend(logs) + +# Sample callback that just ignores the error. +def on_error_pass(*_): pass client.upload(rule_id=rule_id, stream_name=os.environ['LOGS_DCR_STREAM_NAME'], logs=body, on_error=on_error) # Retry once with any failed logs, and this time ignore any errors. -print("Retrying logs that failed to upload...") if failed_logs: + print("Retrying logs that failed to upload...") client.upload(rule_id=rule_id, stream_name=os.environ['LOGS_DCR_STREAM_NAME'], logs=failed_logs, on_error=on_error_pass) diff --git a/sdk/monitor/azure-monitor-ingestion/tests/test_logs_ingestion.py b/sdk/monitor/azure-monitor-ingestion/tests/test_logs_ingestion.py index a9681cd500dd..719d013e5a40 100644 --- a/sdk/monitor/azure-monitor-ingestion/tests/test_logs_ingestion.py +++ b/sdk/monitor/azure-monitor-ingestion/tests/test_logs_ingestion.py @@ -50,10 +50,10 @@ def test_send_logs_error_custom(self, recorded_test, monitor_info): LogsIngestionClient, self.get_credential(LogsIngestionClient), endpoint=monitor_info['dce']) body = [{"foo": "bar"}] - def on_error(**kwargs): + def on_error(error, logs): on_error.called = True - assert isinstance(kwargs.get("error"), HttpResponseError) - assert kwargs.get("logs") == body + assert isinstance(error, HttpResponseError) + assert logs == body on_error.called = False diff --git a/sdk/monitor/azure-monitor-ingestion/tests/test_logs_ingestion_async.py b/sdk/monitor/azure-monitor-ingestion/tests/test_logs_ingestion_async.py index f389a3faa212..f7a863b3b691 100644 --- a/sdk/monitor/azure-monitor-ingestion/tests/test_logs_ingestion_async.py +++ b/sdk/monitor/azure-monitor-ingestion/tests/test_logs_ingestion_async.py @@ -53,10 +53,10 @@ async def test_send_logs_error_custom(self, recorded_test, monitor_info): LogsIngestionClient, self.get_credential(LogsIngestionClient, is_async=True), endpoint=monitor_info['dce']) body = [{"foo": "bar"}] - async def on_error(**kwargs): + async def on_error(error, logs): on_error.called = True - assert isinstance(kwargs.get("error"), HttpResponseError) - assert kwargs.get("logs") == body + assert isinstance(error, HttpResponseError) + assert logs == body on_error.called = False From 70438663dca53d1b3d9d1437a6a6cd23f25b26c5 Mon Sep 17 00:00:00 2001 From: Paul Van Eck Date: Tue, 3 Jan 2023 12:22:21 -0800 Subject: [PATCH 3/4] Address review comments Signed-off-by: Paul Van Eck --- sdk/monitor/azure-monitor-ingestion/README.md | 9 +++++++-- .../azure/monitor/ingestion/_helpers.py | 16 ++++++++++++---- .../monitor/ingestion/_operations/_patch.py | 7 ++++--- .../monitor/ingestion/aio/_operations/_patch.py | 7 ++++--- .../sample_custom_error_callback_async.py | 6 +++--- .../sample_send_small_logs_async.py | 13 +++++++++---- .../samples/sample_custom_error_callback.py | 5 +++-- .../samples/sample_send_small_logs.py | 11 ++++++++--- 8 files changed, 50 insertions(+), 24 deletions(-) diff --git a/sdk/monitor/azure-monitor-ingestion/README.md b/sdk/monitor/azure-monitor-ingestion/README.md index b4164bb7fc2b..54b71c366e46 100644 --- a/sdk/monitor/azure-monitor-ingestion/README.md +++ b/sdk/monitor/azure-monitor-ingestion/README.md @@ -101,8 +101,10 @@ This example shows uploading logs to Azure Monitor. ```python import os -from azure.monitor.ingestion import LogsIngestionClient + +from azure.core.exceptions import HttpResponseError from azure.identity import DefaultAzureCredential +from azure.monitor.ingestion import LogsIngestionClient endpoint = os.environ['DATA_COLLECTION_ENDPOINT'] credential = DefaultAzureCredential() @@ -123,7 +125,10 @@ body = [ } ] -client.upload(rule_id=rule_id, stream_name=os.environ['LOGS_DCR_STREAM_NAME'], logs=body) +try: + client.upload(rule_id=rule_id, stream_name=os.environ['LOGS_DCR_STREAM_NAME'], logs=body) +except HttpResponseError as e: + print(f"Upload failed: {e}") ``` ### Upload with custom error handling diff --git a/sdk/monitor/azure-monitor-ingestion/azure/monitor/ingestion/_helpers.py b/sdk/monitor/azure-monitor-ingestion/azure/monitor/ingestion/_helpers.py index e13fbfcb9ea3..d27ca8cf6d69 100644 --- a/sdk/monitor/azure-monitor-ingestion/azure/monitor/ingestion/_helpers.py +++ b/sdk/monitor/azure-monitor-ingestion/azure/monitor/ingestion/_helpers.py @@ -4,22 +4,30 @@ # ------------------------------------ import json import logging -from typing import Generator, List +import sys +from typing import Any, Generator, List, Tuple import zlib +if sys.version_info >= (3, 9): + from collections.abc import MutableMapping +else: + from typing import MutableMapping # type: ignore # pylint: disable=ungrouped-imports + + _LOGGER = logging.getLogger(__name__) +JSON = MutableMapping[str, Any] # pylint: disable=unsubscriptable-object MAX_CHUNK_SIZE_BYTES = 1024 * 1024 # 1 MiB CHAR_SIZE_BYTES = 4 -def _split_chunks(logs: List) -> Generator: +def _split_chunks(logs: List[JSON]) -> Generator[List[JSON], None, None]: chunk_size = 0 curr_chunk = [] for log in logs: # each char is 4 bytes size = len(json.dumps(log)) * CHAR_SIZE_BYTES - if chunk_size + size < MAX_CHUNK_SIZE_BYTES: + if chunk_size + size <= MAX_CHUNK_SIZE_BYTES: curr_chunk.append(log) chunk_size += size else: @@ -32,7 +40,7 @@ def _split_chunks(logs: List) -> Generator: yield curr_chunk -def _create_gzip_requests(logs: List) -> Generator: +def _create_gzip_requests(logs: List[JSON]) -> Generator[Tuple[bytes, List[JSON]], None, None]: for chunk in _split_chunks(logs): zlib_mode = 16 + zlib.MAX_WBITS # for gzip encoding _compress = zlib.compressobj(wbits=zlib_mode) diff --git a/sdk/monitor/azure-monitor-ingestion/azure/monitor/ingestion/_operations/_patch.py b/sdk/monitor/azure-monitor-ingestion/azure/monitor/ingestion/_operations/_patch.py index 72a27673ecd2..a8f60b4653e4 100644 --- a/sdk/monitor/azure-monitor-ingestion/azure/monitor/ingestion/_operations/_patch.py +++ b/sdk/monitor/azure-monitor-ingestion/azure/monitor/ingestion/_operations/_patch.py @@ -8,7 +8,7 @@ """ import logging import sys -from typing import Callable, List, Any, Union, IO, Optional +from typing import Callable, List, Any, Optional from ._operations import LogsIngestionClientOperationsMixin as GeneratedOps from .._helpers import _create_gzip_requests @@ -28,7 +28,8 @@ def upload( # pylint: disable=arguments-renamed, arguments-differ self, rule_id: str, stream_name: str, - logs: Union[List[JSON], IO], + logs: List[JSON], + *, on_error: Optional[Callable[[Exception, List[JSON]], None]] = None, **kwargs: Any ) -> None: @@ -41,7 +42,7 @@ def upload( # pylint: disable=arguments-renamed, arguments-differ :param stream_name: The streamDeclaration name as defined in the Data Collection Rule. :type stream_name: str :param logs: An array of objects matching the schema defined by the provided stream. - :type logs: list[JSON] or IO + :type logs: list[JSON] :param on_error: The callback function that is called when a chunk of logs fails to upload. This function should expect two arguments that correspond to the error encountered and the list of logs that failed to upload. If no function is provided, then the first exception diff --git a/sdk/monitor/azure-monitor-ingestion/azure/monitor/ingestion/aio/_operations/_patch.py b/sdk/monitor/azure-monitor-ingestion/azure/monitor/ingestion/aio/_operations/_patch.py index 60eda208efdf..c6297aecd74a 100644 --- a/sdk/monitor/azure-monitor-ingestion/azure/monitor/ingestion/aio/_operations/_patch.py +++ b/sdk/monitor/azure-monitor-ingestion/azure/monitor/ingestion/aio/_operations/_patch.py @@ -8,7 +8,7 @@ """ import logging import sys -from typing import Callable, List, Any, Awaitable, Union, IO, Optional +from typing import Callable, List, Any, Awaitable, Optional from ._operations import LogsIngestionClientOperationsMixin as GeneratedOps from ..._helpers import _create_gzip_requests @@ -28,7 +28,8 @@ async def upload( # pylint: disable=arguments-renamed, arguments-differ self, rule_id: str, stream_name: str, - logs: Union[List[JSON], IO], + logs: List[JSON], + *, on_error: Optional[Callable[[Exception, List[JSON]], Awaitable[None]]] = None, **kwargs: Any ) -> None: @@ -41,7 +42,7 @@ async def upload( # pylint: disable=arguments-renamed, arguments-differ :param stream: The streamDeclaration name as defined in the Data Collection Rule. :type stream: str :param logs: An array of objects matching the schema defined by the provided stream. - :type logs: list[JSON] or IO + :type logs: list[JSON] :param on_error: The asynchronous callback function that is called when a chunk of logs fails to upload. This function should expect two arguments that correspond to the error encountered and the list of logs that failed to upload. If no function is provided, then the first exception diff --git a/sdk/monitor/azure-monitor-ingestion/samples/async_samples/sample_custom_error_callback_async.py b/sdk/monitor/azure-monitor-ingestion/samples/async_samples/sample_custom_error_callback_async.py index a0e35d2fc717..39d78f454542 100644 --- a/sdk/monitor/azure-monitor-ingestion/samples/async_samples/sample_custom_error_callback_async.py +++ b/sdk/monitor/azure-monitor-ingestion/samples/async_samples/sample_custom_error_callback_async.py @@ -1,11 +1,11 @@ """ Usage: python sample_custom_error_callback_async.py """ - -import os import asyncio -from azure.monitor.ingestion.aio import LogsIngestionClient +import os + from azure.identity.aio import DefaultAzureCredential +from azure.monitor.ingestion.aio import LogsIngestionClient async def send_logs(): diff --git a/sdk/monitor/azure-monitor-ingestion/samples/async_samples/sample_send_small_logs_async.py b/sdk/monitor/azure-monitor-ingestion/samples/async_samples/sample_send_small_logs_async.py index abb9b5661330..172179c71621 100644 --- a/sdk/monitor/azure-monitor-ingestion/samples/async_samples/sample_send_small_logs_async.py +++ b/sdk/monitor/azure-monitor-ingestion/samples/async_samples/sample_send_small_logs_async.py @@ -1,11 +1,13 @@ """ Usage: python sample_send_small_logs_async.py """ - -import os import asyncio -from azure.monitor.ingestion.aio import LogsIngestionClient +import os + +from azure.core.exceptions import HttpResponseError from azure.identity.aio import DefaultAzureCredential +from azure.monitor.ingestion.aio import LogsIngestionClient + async def send_logs(): endpoint = os.environ['DATA_COLLECTION_ENDPOINT'] @@ -26,7 +28,10 @@ async def send_logs(): client = LogsIngestionClient(endpoint=endpoint, credential=credential, logging_enable=True) async with client: - await client.upload(rule_id=rule_id, stream_name=os.environ['LOGS_DCR_STREAM_NAME'], logs=body) + try: + await client.upload(rule_id=rule_id, stream_name=os.environ['LOGS_DCR_STREAM_NAME'], logs=body) + except HttpResponseError as e: + print(f"Upload failed: {e}") await credential.close() diff --git a/sdk/monitor/azure-monitor-ingestion/samples/sample_custom_error_callback.py b/sdk/monitor/azure-monitor-ingestion/samples/sample_custom_error_callback.py index 15c2302047ec..8bdfa4ca6546 100644 --- a/sdk/monitor/azure-monitor-ingestion/samples/sample_custom_error_callback.py +++ b/sdk/monitor/azure-monitor-ingestion/samples/sample_custom_error_callback.py @@ -1,10 +1,11 @@ """ Usage: python sample_custom_error_callback.py """ - import os -from azure.monitor.ingestion import LogsIngestionClient + from azure.identity import DefaultAzureCredential +from azure.monitor.ingestion import LogsIngestionClient + endpoint = os.environ['DATA_COLLECTION_ENDPOINT'] credential = DefaultAzureCredential() diff --git a/sdk/monitor/azure-monitor-ingestion/samples/sample_send_small_logs.py b/sdk/monitor/azure-monitor-ingestion/samples/sample_send_small_logs.py index 34a83fafa6e0..4f550da43413 100644 --- a/sdk/monitor/azure-monitor-ingestion/samples/sample_send_small_logs.py +++ b/sdk/monitor/azure-monitor-ingestion/samples/sample_send_small_logs.py @@ -1,10 +1,12 @@ """ Usage: python sample_send_small_logs.py """ - import os -from azure.monitor.ingestion import LogsIngestionClient + +from azure.core.exceptions import HttpResponseError from azure.identity import DefaultAzureCredential +from azure.monitor.ingestion import LogsIngestionClient + endpoint = os.environ['DATA_COLLECTION_ENDPOINT'] credential = DefaultAzureCredential() @@ -25,4 +27,7 @@ } ] -client.upload(rule_id=rule_id, stream_name=os.environ['LOGS_DCR_STREAM_NAME'], logs=body) +try: + client.upload(rule_id=rule_id, stream_name=os.environ['LOGS_DCR_STREAM_NAME'], logs=body) +except HttpResponseError as e: + print(f"Upload failed: {e}") From 99e5a0a54d1f774d4029b28ea76bd4a43b88ae7e Mon Sep 17 00:00:00 2001 From: Paul Van Eck Date: Fri, 6 Jan 2023 11:13:32 -0800 Subject: [PATCH 4/4] Apply suggestions from code review Co-authored-by: Anna Tisch --- .../azure/monitor/ingestion/_operations/_patch.py | 4 ++-- .../azure/monitor/ingestion/aio/_operations/_patch.py | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/sdk/monitor/azure-monitor-ingestion/azure/monitor/ingestion/_operations/_patch.py b/sdk/monitor/azure-monitor-ingestion/azure/monitor/ingestion/_operations/_patch.py index a8f60b4653e4..66e91649872b 100644 --- a/sdk/monitor/azure-monitor-ingestion/azure/monitor/ingestion/_operations/_patch.py +++ b/sdk/monitor/azure-monitor-ingestion/azure/monitor/ingestion/_operations/_patch.py @@ -43,11 +43,11 @@ def upload( # pylint: disable=arguments-renamed, arguments-differ :type stream_name: str :param logs: An array of objects matching the schema defined by the provided stream. :type logs: list[JSON] - :param on_error: The callback function that is called when a chunk of logs fails to upload. + :keyword on_error: The callback function that is called when a chunk of logs fails to upload. This function should expect two arguments that correspond to the error encountered and the list of logs that failed to upload. If no function is provided, then the first exception encountered will be raised. - :type on_error: Optional[Callable[[Exception, List[JSON]], None]] + :paramtype on_error: Optional[Callable[[Exception, List[JSON]], None]] :return: None :rtype: None :raises: ~azure.core.exceptions.HttpResponseError diff --git a/sdk/monitor/azure-monitor-ingestion/azure/monitor/ingestion/aio/_operations/_patch.py b/sdk/monitor/azure-monitor-ingestion/azure/monitor/ingestion/aio/_operations/_patch.py index c6297aecd74a..d0b68f0d727b 100644 --- a/sdk/monitor/azure-monitor-ingestion/azure/monitor/ingestion/aio/_operations/_patch.py +++ b/sdk/monitor/azure-monitor-ingestion/azure/monitor/ingestion/aio/_operations/_patch.py @@ -43,11 +43,11 @@ async def upload( # pylint: disable=arguments-renamed, arguments-differ :type stream: str :param logs: An array of objects matching the schema defined by the provided stream. :type logs: list[JSON] - :param on_error: The asynchronous callback function that is called when a chunk of logs fails to upload. + :keyword on_error: The asynchronous callback function that is called when a chunk of logs fails to upload. This function should expect two arguments that correspond to the error encountered and the list of logs that failed to upload. If no function is provided, then the first exception encountered will be raised. - :type on_error: Optional[Callable[[Exception, List[JSON]], None]] + :paramtype on_error: Optional[Callable[[Exception, List[JSON]], None]] :return: None :rtype: None :raises: ~azure.core.exceptions.HttpResponseError