diff --git a/sdk/cosmos/azure-cosmos/CHANGELOG.md b/sdk/cosmos/azure-cosmos/CHANGELOG.md index 41f540bf6b14..ecca691082e9 100644 --- a/sdk/cosmos/azure-cosmos/CHANGELOG.md +++ b/sdk/cosmos/azure-cosmos/CHANGELOG.md @@ -3,6 +3,7 @@ ### 4.10.0b5 (Unreleased) #### Features Added +* Added ability to use Filters from Logging module on Diagnostics Logging based on Http request/response related attributes. See [PR 39897](https://github.com/Azure/azure-sdk-for-python/pull/39897) #### Breaking Changes @@ -11,8 +12,10 @@ * Fixed health check to check the first write region when it is not specified in the preferred regions. See [PR 40588](https://github.com/Azure/azure-sdk-for-python/pull/40588). #### Other Changes +* Optimized Diagnostics Logging by reducing time spent on logging. Logged Errors are more readable and formatted. See [PR 39897](https://github.com/Azure/azure-sdk-for-python/pull/39897) * Health checks are now done concurrently and for all regions for async apis. See [PR 40588](https://github.com/Azure/azure-sdk-for-python/pull/40588). + ### 4.10.0b4 (2025-04-01) #### Bugs Fixed diff --git a/sdk/cosmos/azure-cosmos/README.md b/sdk/cosmos/azure-cosmos/README.md index 90a6b8bf9119..eb1d2ca70509 100644 --- a/sdk/cosmos/azure-cosmos/README.md +++ b/sdk/cosmos/azure-cosmos/README.md @@ -950,53 +950,39 @@ However, if you desire to use the CosmosHttpLoggingPolicy to obtain additional i client = CosmosClient(URL, credential=KEY, enable_diagnostics_logging=True) database = client.create_database(DATABASE_NAME, logger=logger) ``` -**NOTICE: The Following is a Preview Feature that is subject to significant change.** -To further customize what gets logged, you can use a **PREVIEW** diagnostics handler to filter out the logs you don't want to see. -There are several ways to use the diagnostics handler, those include the following: -- Using the "CosmosDiagnosticsHandler" class, which has default behaviour that can be modified. - **NOTE: The diagnostics handler will only be used if the `enable_diagnostics_logging` argument is passed in at the client constructor. - The CosmosDiagnosticsHandler is also a special type of dictionary that is callable and that has preset keys. The values it expects are functions related to it's relevant diagnostic data. (e.g. ```diagnostics_handler["duration"]``` expects a function that takes in an int and returns a boolean as it relates to the duration of an operation to complete).** - ```python - from azure.cosmos import CosmosClient, CosmosDiagnosticsHandler - import logging - # Initialize the logger - logger = logging.getLogger('azure.cosmos') - logger.setLevel(logging.INFO) - file_handler = logging.FileHandler('diagnostics1.output') - logger.addHandler(file_handler) - diagnostics_handler = cosmos_diagnostics_handler.CosmosDiagnosticsHandler() - diagnostics_handler["duration"] = lambda x: x > 2000 - client = CosmosClient(URL, credential=KEY,logger=logger, diagnostics_handler=diagnostics_handler, enable_diagnostics_logging=True) - - ``` -- Using a dictionary with the relevant functions to filter out the logs you don't want to see. - ```python - # Initialize the logger - logger = logging.getLogger('azure.cosmos') - logger.setLevel(logging.INFO) - file_handler = logging.FileHandler('diagnostics2.output') - logger.addHandler(file_handler) - diagnostics_handler = { - "duration": lambda x: x > 2000 - } - client = CosmosClient(URL, credential=KEY,logger=logger, diagnostics_handler=diagnostics_handler, enable_diagnostics_logging=True) - ``` -- Using a function that will replace the should_log function in the CosmosHttpLoggingPolicy which expects certain paramameters and returns a boolean. **Note: the parameters of the custom should_log must match the parameters of the original should_log function as shown in the sample.** - ```python - # Custom should_log method - def should_log(self, **kwargs): - return kwargs.get('duration') and kwargs['duration'] > 2000 - - # Initialize the logger +**NOTICE: The Following is a Preview Feature.** +To further customize what gets logged, you can use logger filters to filter out the logs you don't want to see. You are able to filter based on the following attributes in the log record of cosmos diagnostics logs: +- `status_code` +- `sub_status_code` +- `duration` +- `verb` +- `database_name` +- `collection_name` +- `operation_type` +- `url` +- `resource_type` +- `is_request` + +You can take a look at the samples [here][cosmos_diagnostics_filter_sample] or take a quick look at this snippet: +- Using **filters** from the **logging** library, it is possible to filter the diagnostics logs. Several filterable attributes are made available to the log record of the diagnostics logs when using logging filters. +```python + import logging + from azure.cosmos import CosmosClient logger = logging.getLogger('azure.cosmos') logger.setLevel(logging.INFO) - file_handler = logging.FileHandler('diagnostics3.output') + file_handler = logging.FileHandler('diagnostics.output') logger.addHandler(file_handler) - - # Initialize the Cosmos client with custom diagnostics handler - client = CosmosClient(endpoint, key,logger=logger, diagnostics_handler=should_log, enable_diagnostics_logging=True) - ``` - + # Create a filter to filter out logs + class CustomFilter(logging.Filter): + def filter(self, record): + ret = (hasattr(record, 'status_code') and record.status_code > 400 + and not (record.status_code in [404, 409, 412] and getattr(record, 'sub_status_code', None) in [0, None]) + and hasattr(record, 'duration') and record.duration > 1000) + return ret + # Add the filter to the logger + logger.addFilter(CustomFilter()) + client = CosmosClient(endpoint, key,logger=logger, enable_diagnostics_logging=True) +``` ### Telemetry Azure Core provides the ability for our Python SDKs to use OpenTelemetry with them. The only packages that need to be installed to use this functionality are the following: @@ -1060,6 +1046,7 @@ For more extensive documentation on the Cosmos DB service, see the [Azure Cosmos [BM25]: https://learn.microsoft.com/azure/search/index-similarity-and-scoring [cosmos_fts]: https://aka.ms/cosmosfulltextsearch [cosmos_index_policy_change]: https://learn.microsoft.com/azure/cosmos-db/index-policy#modifying-the-indexing-policy +[cosmos_diagnostics_filter_sample]: https://github.com/Azure/azure-sdk-for-python/blob/main/sdk/cosmos/azure-cosmos/samples/diagnostics_filter_sample.py ## Contributing diff --git a/sdk/cosmos/azure-cosmos/azure/cosmos/_cosmos_client_connection.py b/sdk/cosmos/azure-cosmos/azure/cosmos/_cosmos_client_connection.py index 2de1dedf58e7..6b65afd6c848 100644 --- a/sdk/cosmos/azure-cosmos/azure/cosmos/_cosmos_client_connection.py +++ b/sdk/cosmos/azure-cosmos/azure/cosmos/_cosmos_client_connection.py @@ -214,7 +214,6 @@ def __init__( # pylint: disable=too-many-statements logger=kwargs.pop("logger", None), enable_diagnostics_logging=kwargs.pop("enable_diagnostics_logging", False), global_endpoint_manager=self._global_endpoint_manager, - diagnostics_handler=kwargs.pop("diagnostics_handler", None), **kwargs ), ] diff --git a/sdk/cosmos/azure-cosmos/azure/cosmos/_cosmos_http_logging_policy.py b/sdk/cosmos/azure-cosmos/azure/cosmos/_cosmos_http_logging_policy.py index 29d2ccee8452..6139857ea98a 100644 --- a/sdk/cosmos/azure-cosmos/azure/cosmos/_cosmos_http_logging_policy.py +++ b/sdk/cosmos/azure-cosmos/azure/cosmos/_cosmos_http_logging_policy.py @@ -26,7 +26,9 @@ import json import logging import time -from typing import Optional, Union, Dict, Any, TYPE_CHECKING, Callable, Mapping +import os +import urllib.parse +from typing import Optional, Union, Dict, Any, TYPE_CHECKING, Set, List import types from azure.core.pipeline import PipelineRequest, PipelineResponse @@ -41,213 +43,302 @@ from azure.core.pipeline.transport import ( # pylint: disable=no-legacy-azure-core-http-response-import HttpRequest as LegacyHttpRequest, HttpResponse as LegacyHttpResponse, - AsyncHttpResponse as LegacyAsyncHttpResponse + AsyncHttpResponse as LegacyAsyncHttpResponse, ) - + from azure.core.pipeline.transport._base import _HttpResponseBase as LegacySansIOHttpResponse + from azure.core.rest._rest_py3 import _HttpResponseBase as SansIOHttpResponse HTTPRequestType = Union["LegacyHttpRequest", "HttpRequest"] -HTTPResponseType = Union["LegacyHttpResponse", "HttpResponse", "LegacyAsyncHttpResponse", "AsyncHttpResponse"] +HTTPResponseType = Union["LegacyHttpResponse", "HttpResponse", "LegacyAsyncHttpResponse", +"AsyncHttpResponse", "SansIOHttpResponse", "LegacySansIOHttpResponse"] def _format_error(payload: str) -> str: output = json.loads(payload) - return output['message'].replace("\r", " ") + ret_str = "\n\t" + "Code: " + output['code'] + "\n" + message = output["message"].replace("\r\n", "\n\t\t").replace(",", ",\n\t\t") + ret_str += "\t" + message + "\n" + return ret_str class CosmosHttpLoggingPolicy(HttpLoggingPolicy): def __init__( - self, - logger: Optional[logging.Logger] = None, - global_endpoint_manager: Optional[_GlobalEndpointManager] = None, - database_account: Optional[DatabaseAccount] = None, - *, - enable_diagnostics_logging: bool = False, - diagnostics_handler: Optional[Union[Callable, Mapping]] = None, - **kwargs + self, + logger: Optional[logging.Logger] = None, + global_endpoint_manager: Optional[_GlobalEndpointManager] = None, + database_account: Optional[DatabaseAccount] = None, + *, + enable_diagnostics_logging: bool = False, + **kwargs ): + super().__init__(logger, **kwargs) + self.logger: logging.Logger = logger or logging.getLogger("azure.cosmos._cosmos_http_logging_policy") self._enable_diagnostics_logging = enable_diagnostics_logging - self.diagnostics_handler = diagnostics_handler - self.__request_already_logged = False - if self.diagnostics_handler and callable(self.diagnostics_handler): - if hasattr(self.diagnostics_handler, '__get__'): - self._should_log = types.MethodType(diagnostics_handler, self) # type: ignore - else: - self._should_log = self.diagnostics_handler # type: ignore - elif isinstance(self.diagnostics_handler, Mapping): - self._should_log = self._dict_should_log # type: ignore - else: - self._should_log = self._default_should_log # type: ignore self.__global_endpoint_manager = global_endpoint_manager - self.__client_settings = self.__get_client_settings() self.__database_account_settings: Optional[DatabaseAccount] = (database_account or self.__get_database_account_settings()) - self._resource_map = { - 'docs': 'document', - 'colls': 'container', - 'dbs': 'database' - } - super().__init__(logger, **kwargs) - if self._enable_diagnostics_logging: - cosmos_disallow_list = ["Authorization", "ProxyAuthorization"] - cosmos_allow_list = [ - v for k, v in HttpHeaders.__dict__.items() if not k.startswith("_") and k not in cosmos_disallow_list - ] - self.allowed_header_names = set(cosmos_allow_list) - - def on_request(self, request: PipelineRequest[HTTPRequestType]) -> None: - verb = request.http_request.method + # The list of headers we do not want to log, it needs to be updated if any new headers should not be logged + cosmos_disallow_list = ["Authorization", "ProxyAuthorization", "TransferEncoding"] + cosmos_allow_list = [ + v for k, v in HttpHeaders.__dict__.items() if not k.startswith("_") and k not in cosmos_disallow_list + ] + self.allowed_header_names = set(cosmos_allow_list) + # For optimizing header redaction. We create the set with lower case allowed headers + self.lower_case_allowed_header_names: Set[str] = {header.lower() for header in self.allowed_header_names} + self.lower_case_allowed_query_params: Set[str] = {param.lower() for param in self.allowed_query_params} + + def _redact_query_param(self, key: str, value: str) -> str: + return value if key.lower() in self.lower_case_allowed_query_params else HttpLoggingPolicy.REDACTED_PLACEHOLDER + + def _redact_header(self, key: str, value: str) -> str: + if key.lower() in self.lower_case_allowed_header_names: + return value + return HttpLoggingPolicy.REDACTED_PLACEHOLDER + + def on_request( + # pylint: disable=too-many-return-statements, too-many-statements, too-many-nested-blocks, too-many-branches + self, request: PipelineRequest[HTTPRequestType] + ) -> None: + """Logs HTTP method, url and headers. + :param request: The PipelineRequest object. + :type request: ~azure.core.pipeline.PipelineRequest + """ if self._enable_diagnostics_logging: + + http_request = request.http_request request.context["start_time"] = time.time() - url = None - if self.diagnostics_handler: + operation_type = http_request.headers.get('x-ms-thinclient-proxy-operation-type') + try: url = request.http_request.url + except AttributeError: + url = None database_name = None collection_name = None - resource_type = None + resource_type = http_request.headers.get('x-ms-thinclient-proxy-resource-type') if url: url_parts = url.split('/') if 'dbs' in url_parts: dbs_index = url_parts.index('dbs') if dbs_index + 1 < len(url_parts): database_name = url_parts[url_parts.index('dbs') + 1] - resource_type = self._resource_map['dbs'] if 'colls' in url_parts: colls_index = url_parts.index('colls') if colls_index + 1 < len(url_parts): collection_name = url_parts[url_parts.index('colls') + 1] - resource_type = self._resource_map['colls'] - if 'docs' in url_parts: - resource_type = self._resource_map['docs'] - if self._should_log(verb=verb,database_name=database_name,collection_name=collection_name, - resource_type=resource_type, is_request=True): - self._log_client_settings() - self._log_database_account_settings() - super().on_request(request) - self.__request_already_logged = True - - # pylint: disable=too-many-statements - def on_response( - self, - request: PipelineRequest[HTTPRequestType], - response: PipelineResponse[HTTPRequestType, HTTPResponseType], # type: ignore[override] + options = request.context.options + # Get logger in my context first (request has been retried) + # then read from kwargs (pop if that's the case) + # then use my instance logger + logger = request.context.setdefault("logger", options.pop("logger", self.logger)) + if not logger.isEnabledFor(logging.INFO): + return + try: + parsed_url = list(urllib.parse.urlparse(http_request.url)) + parsed_qp = urllib.parse.parse_qsl(parsed_url[4], keep_blank_values=True) + filtered_qp = [(key, self._redact_query_param(key, value)) for key, value in parsed_qp] + # 4 is query + parsed_url[4] = "&".join(["=".join(part) for part in filtered_qp]) + redacted_url = urllib.parse.urlunparse(parsed_url) + + multi_record = os.environ.get(HttpLoggingPolicy.MULTI_RECORD_LOG, False) + + if 'logger_attributes' in request.context: + cosmos_logger_attributes = request.context['logger_attributes'] + cosmos_logger_attributes['is_request'] = True + elif logger.filters: + return + else: + cosmos_logger_attributes = { + 'duration': None, + 'status_code': None, + 'sub_status_code': None, + 'verb': http_request.method, + 'url': redacted_url, + 'database_name': database_name, + 'collection_name': collection_name, + 'resource_type': resource_type, + 'operation_type': operation_type, + 'is_request': True} + + client_settings = self._log_client_settings() + db_settings = self._log_database_account_settings() + if multi_record: + logger.info(client_settings, extra=cosmos_logger_attributes) + logger.info(db_settings, extra=cosmos_logger_attributes) + logger.info("Request URL: %r", redacted_url, extra=cosmos_logger_attributes) + logger.info("Request method: %r", http_request.method, extra=cosmos_logger_attributes) + logger.info("Request headers:", extra=cosmos_logger_attributes) + for header, value in http_request.headers.items(): + value = self._redact_header(header, value) + if value and value != HttpLoggingPolicy.REDACTED_PLACEHOLDER: + logger.info(" %r: %r", header, value, extra=cosmos_logger_attributes) + if isinstance(http_request.body, types.GeneratorType): + logger.info("File upload", extra=cosmos_logger_attributes) + return + try: + if isinstance(http_request.body, types.AsyncGeneratorType): + logger.info("File upload", extra=cosmos_logger_attributes) + return + except AttributeError: + pass + if http_request.body: + logger.info("A body is sent with the request", extra=cosmos_logger_attributes) + return + logger.info("No body was attached to the request", extra=cosmos_logger_attributes) + return + log_string = client_settings + log_string += db_settings + log_string += "\nRequest URL: '{}'".format(redacted_url) + log_string += "\nRequest method: '{}'".format(http_request.method) + log_string += "\nRequest headers:" + for header, value in http_request.headers.items(): + value = self._redact_header(header, value) + if value and value != HttpLoggingPolicy.REDACTED_PLACEHOLDER: + log_string += "\n '{}': '{}'".format(header, value) + if isinstance(http_request.body, types.GeneratorType): + log_string += "\nFile upload" + logger.info(log_string, extra=cosmos_logger_attributes) + return + try: + if isinstance(http_request.body, types.AsyncGeneratorType): + log_string += "\nFile upload" + logger.info(log_string, extra=cosmos_logger_attributes) + return + except AttributeError: + pass + if http_request.body: + log_string += "\nA body is sent with the request" + logger.info(log_string, extra=cosmos_logger_attributes) + return + log_string += "\nNo body was attached to the request" + logger.info(log_string, extra=cosmos_logger_attributes) + + except Exception as err: # pylint: disable=broad-except + logger.warning("Failed to log request: %s", repr(err)) + return + super().on_request(request) + + def on_response( # pylint: disable=too-many-statements, too-many-branches + self, + request: PipelineRequest[HTTPRequestType], + response: PipelineResponse[HTTPRequestType, HTTPResponseType], ) -> None: - duration = time.time() - request.context["start_time"] if "start_time" in request.context else None - status_code = response.http_response.status_code - sub_status_str = response.http_response.headers.get("x-ms-substatus") - sub_status_code = int(sub_status_str) if sub_status_str else None - verb = request.http_request.method - http_version_obj = None - url = None - if self.diagnostics_handler: + + if self._enable_diagnostics_logging: + context = request.context + http_response = response.http_response + headers = request.http_request.headers + sub_status_str = http_response.headers.get("x-ms-substatus") + sub_status_code: Optional[int] = int(sub_status_str) if sub_status_str else None + url_obj = request.http_request.url # type: ignore[attr-defined, union-attr] try: - major = response.http_response.internal_response.version.major # type: ignore[attr-defined, union-attr] - minor = response.http_response.internal_response.version.minor # type: ignore[attr-defined, union-attr] - http_version_obj = f"{major}." - http_version_obj += f"{minor}" - except (AttributeError, TypeError): - http_version_obj = None + duration: Optional[float] = float(http_response.headers.get("x-ms-request-duration-ms")) # type: ignore[union-attr, arg-type] # pylint: disable=line-too-long + except (ValueError, TypeError): + duration = (time.time() - context["start_time"]) * 1000 \ + if "start_time" in context else None # type: ignore[union-attr, arg-type] + + log_data = {"duration": duration, + "status_code": http_response.status_code, "sub_status_code": sub_status_code, + "verb": request.http_request.method, + "operation_type": headers.get('x-ms-thinclient-proxy-operation-type'), + "url": str(url_obj), "database_name": None, "collection_name": None, + "resource_type": headers.get('x-ms-thinclient-proxy-resource-type'), "is_request": False} # type: ignore[assignment] # pylint: disable=line-too-long + + if log_data["url"]: + url_parts: List[str] = log_data["url"].split('/') # type: ignore[union-attr] + if 'dbs' in url_parts: + dbs_index = url_parts.index('dbs') + if dbs_index + 1 < len(url_parts): + log_data["database_name"] = url_parts[dbs_index + 1] + if 'colls' in url_parts: + colls_index = url_parts.index('colls') + if colls_index + 1 < len(url_parts): + log_data["collection_name"] = url_parts[colls_index + 1] + + options = context.options + logger = context.setdefault("logger", options.pop("logger", self.logger)) + + context["logger_attributes"] = log_data.copy() + self.on_request(request) + try: - url = response.http_response.internal_response.url.geturl() # type: ignore[attr-defined, union-attr] - except AttributeError: - url = str(response.http_response.internal_response.url) # type: ignore[attr-defined, union-attr] - database_name = None - collection_name = None - resource_type = None - if url: - url_parts = url.split('/') - if 'dbs' in url_parts: - dbs_index = url_parts.index('dbs') - if dbs_index + 1 < len(url_parts): - database_name = url_parts[url_parts.index('dbs') + 1] - resource_type = self._resource_map['dbs'] - if 'colls' in url_parts: - colls_index = url_parts.index('colls') - if colls_index + 1 < len(url_parts): - collection_name = url_parts[url_parts.index('colls') + 1] - resource_type = self._resource_map['colls'] - if 'docs' in url_parts: - resource_type = self._resource_map['docs'] - - - if self._should_log(duration=duration, status_code=status_code, sub_status_code=sub_status_code, - verb=verb, http_version=http_version_obj, database_name=database_name, - collection_name=collection_name, resource_type=resource_type, is_request=False): - if not self.__request_already_logged: - self._log_client_settings() - self._log_database_account_settings() - super().on_request(request) - else: - self.__request_already_logged = False - super().on_response(request, response) - if self._enable_diagnostics_logging: - http_response = response.http_response - options = response.context.options - logger = request.context.setdefault("logger", options.pop("logger", self.logger)) - try: - if "start_time" in request.context: - logger.info("Elapsed time in seconds: {}".format(duration)) + if not logger.isEnabledFor(logging.INFO): + return + + multi_record = os.environ.get(HttpLoggingPolicy.MULTI_RECORD_LOG, False) + if multi_record: + logger.info("Response status: %r", log_data["status_code"], extra=log_data) + logger.info("Response headers:", extra=log_data) + for res_header, value in http_response.headers.items(): + value = self._redact_header(res_header, value) + if value and value != HttpLoggingPolicy.REDACTED_PLACEHOLDER: + logger.info(" %r: %r", res_header, value, extra=log_data) + if "start_time" in context and duration: + seconds = duration / 1000 # type: ignore[operator] + logger.info(f"Elapsed time in seconds: {seconds:.6f}".rstrip('0').rstrip('.'), + extra=log_data) else: - logger.info("Elapsed time in seconds: unknown") - if http_response.status_code >= 400: - logger.info("Response error message: %r", _format_error(http_response.text())) - except Exception as err: # pylint: disable=broad-except - logger.warning("Failed to log request: %s", repr(err)) # pylint: disable=do-not-log-exceptions - - # pylint: disable=unused-argument - def _default_should_log( - self, - **kwargs - ) -> bool: - return True - - def _dict_should_log(self, **kwargs) -> bool: - params = { - 'duration': kwargs.get('duration', None), - 'status code': kwargs.get('status_code', None), - 'verb': kwargs.get('verb', None), - 'http version': kwargs.get('http_version', None), - 'database name': kwargs.get('database_name', None), - 'collection name': kwargs.get('collection_name', None), - 'resource type': kwargs.get('resource_type', None) - } - for key, param in params.items(): - if (param and isinstance(self.diagnostics_handler, Mapping) and key in self.diagnostics_handler - and self.diagnostics_handler[key] is not None): - if self.diagnostics_handler[key](param): - return True - return False + logger.info("Elapsed time in seconds: unknown", extra=log_data) + if isinstance(log_data["status_code"], int) and log_data["status_code"] >= 400: + logger.info("\nResponse error message: %r", _format_error(http_response.text()), + extra=log_data) + return + log_string = "\nResponse status: {}".format(log_data["status_code"]) + log_string += "\nResponse headers:" + for res_header, value in http_response.headers.items(): + value = self._redact_header(res_header, value) + if value and value != HttpLoggingPolicy.REDACTED_PLACEHOLDER: + log_string += "\n '{}': '{}'".format(res_header, value) + if "start_time" in context and duration: + seconds = duration / 1000 # type: ignore[operator] + log_string += f"\nElapsed time in seconds: {seconds:.6f}".rstrip('0').rstrip('.') + else: + log_string += "\nElapsed time in seconds: unknown" + if isinstance(log_data["status_code"], int) and log_data["status_code"] >= 400: + log_string += "\nResponse error message: {}".format(_format_error(http_response.text())) + logger.info(log_string, extra=log_data) + except Exception as err: # pylint: disable=broad-except + logger.warning("Failed to log response: %s", repr(err), extra=log_data) + return + super().on_response(request, response) def __get_client_settings(self) -> Optional[Dict[str, Any]]: # Place any client settings we want to log here + client_info: Dict[str, Any] = {"Client Preferred Regions": [], "Client Available Read Regions": [], + "Client Available Write Regions": []} if self.__global_endpoint_manager: - if hasattr(self.__global_endpoint_manager, 'PreferredLocations'): - return {"Client Preferred Regions": self.__global_endpoint_manager.PreferredLocations} - return {"Client Preferred Regions": []} - return None + client_info['Client Preferred Regions'] = self.__global_endpoint_manager.PreferredLocations \ + if hasattr(self.__global_endpoint_manager, "PreferredLocations") else [] + try: + location_cache = self.__global_endpoint_manager.location_cache + client_info["Client Available Read Regions"] = location_cache.available_read_locations + client_info["Client Available Write Regions"] = location_cache.available_write_locations + except AttributeError: + client_info["Client Available Read Regions"] = [] + client_info["Client Available Write Regions"] = [] + return client_info def __get_database_account_settings(self) -> Optional[DatabaseAccount]: if self.__global_endpoint_manager and hasattr(self.__global_endpoint_manager, '_database_account_cache'): return self.__global_endpoint_manager._database_account_cache # pylint: disable=protected-access return None - def _log_client_settings(self) -> None: - self.logger.info("Client Settings:", exc_info=False) - if self.__client_settings and isinstance(self.__client_settings, dict): - self.logger.info("\tClient Preferred Regions: %s", self.__client_settings["Client Preferred Regions"], - exc_info=False) + def _log_client_settings(self) -> str: + logger_str = "\nClient Settings: \n" + client_settings = self.__get_client_settings() + if client_settings and isinstance(client_settings, dict): + logger_str += ''.join([f"\t{k}: {v}\n" for k, v in client_settings.items()]) + return logger_str # pylint: disable=protected-access - def _log_database_account_settings(self) -> None: - self.logger.info("Database Account Settings:", exc_info=False) + def _log_database_account_settings(self) -> str: + logger_str = "\nDatabase Account Settings: \n" self.__database_account_settings = self.__get_database_account_settings() if self.__database_account_settings and self.__database_account_settings.ConsistencyPolicy: - self.logger.info("\tConsistency Level: %s", - self.__database_account_settings.ConsistencyPolicy.get("defaultConsistencyLevel"), - exc_info=False) - self.logger.info("\tWritable Locations: %s", self.__database_account_settings.WritableLocations, - exc_info=False) - self.logger.info("\tReadable Locations: %s", self.__database_account_settings.ReadableLocations, - exc_info=False) - self.logger.info("\tMulti-Region Writes: %s", - self.__database_account_settings._EnableMultipleWritableLocations, exc_info=False) + logger_str += f"\tConsistency Level: {self.__database_account_settings.ConsistencyPolicy.get('defaultConsistencyLevel')}\n" # pylint: disable=line-too-long + logger_str += f"\tWritable Locations: {self.__database_account_settings.WritableLocations}\n" + logger_str += f"\tReadable Locations: {self.__database_account_settings.ReadableLocations}\n" + logger_str += f"\tMulti-Region Writes: {self.__database_account_settings._EnableMultipleWritableLocations}\n" # pylint: disable=protected-access, line-too-long + + return logger_str diff --git a/sdk/cosmos/azure-cosmos/azure/cosmos/aio/_cosmos_client_connection_async.py b/sdk/cosmos/azure-cosmos/azure/cosmos/aio/_cosmos_client_connection_async.py index b9384103e2cd..7dd0a3ff4f77 100644 --- a/sdk/cosmos/azure-cosmos/azure/cosmos/aio/_cosmos_client_connection_async.py +++ b/sdk/cosmos/azure-cosmos/azure/cosmos/aio/_cosmos_client_connection_async.py @@ -219,7 +219,6 @@ def __init__( # pylint: disable=too-many-statements logger=kwargs.pop("logger", None), enable_diagnostics_logging=kwargs.pop("enable_diagnostics_logging", False), global_endpoint_manager=self._global_endpoint_manager, - diagnostics_handler=kwargs.pop("diagnostics_handler", None), **kwargs ), ] diff --git a/sdk/cosmos/azure-cosmos/azure/cosmos/cosmos_diagnostics_handler.py b/sdk/cosmos/azure-cosmos/azure/cosmos/cosmos_diagnostics_handler.py deleted file mode 100644 index 7d777843427d..000000000000 --- a/sdk/cosmos/azure-cosmos/azure/cosmos/cosmos_diagnostics_handler.py +++ /dev/null @@ -1,95 +0,0 @@ -# -*- coding: utf-8 -*- -# The MIT License (MIT) -# Copyright (c) 2014 Microsoft Corporation - -# Permission is hereby granted, free of charge, to any person obtaining a copy -# of this software and associated documentation files (the "Software"), to deal -# in the Software without restriction, including without limitation the rights -# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell -# copies of the Software, and to permit persons to whom the Software is -# furnished to do so, subject to the following conditions: - -# The above copyright notice and this permission notice shall be included in all -# copies or substantial portions of the Software. - -# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR -# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, -# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE -# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER -# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, -# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE -# SOFTWARE. - - -from multidict import CIMultiDict - - -class CosmosDiagnosticsHandler(CIMultiDict): - # To customize the diagnostic handler, you can modify the conditions in the _preset_keys dictionary. - # For example, to log if the duration is greater than 500 ms, you can set: - # self._preset_keys['duration'] = lambda x: x > 500 - # Similarly, you can add or modify conditions for other keys as needed. - def __init__(self) -> None: - """ - A Preview Feature That is subject to significant changes. - Initializes the CosmosDiagnosticsHandler with preset keys and their corresponding conditions. - To customize the diagnostic handler, you can modify the conditions in the _preset_keys dictionary. - For example, to log if the duration is greater than 500 ms, you can set: - self._preset_keys['duration'] = lambda x: x > 500 - or - def check_duration(duration: int) -> bool: - return duration > 500 - self._preset_keys['duration'] = check_duration - Similarly, you can add or modify conditions for other keys as needed. - When Using this as an instance, you can modify the values of the keys in the dictionary. - Example: - handler = CosmosDiagnosticsHandler() - handler['duration'] = lambda x: x > 500 - or with a function. - def check_duration(duration: int) -> bool: - return duration > 500 - handler['duration'] = check_duration - """ - super().__init__() - self._preset_keys = { - 'duration': lambda x: x > 1000, # Log if duration is greater than 1000 ms - 'status code': (lambda x: ( - isinstance(x, (list, tuple)) and x[0] != 200 and (x[1] is None or x[1] != 0) - ) if isinstance(x, (list, tuple)) else x != 200), # Log if status code is not 200 - 'verb': None, # No condition for verb - 'http version': None, # No condition for http_version - 'database name': None, # No condition for database_name - 'collection name': None, # No condition for collection_name - 'resource type': None # No condition for resource_type - } - for key, value in self._preset_keys.items(): - self[key] = value - - def __setitem__(self, key, value): - if key.lower() in self._preset_keys: - super().__setitem__(key, value) - else: - raise KeyError(f"Cannot add new key: {key}") - - def __delitem__(self, key): - raise KeyError(f"Cannot delete key: {key}") - - def __call__(self, **kwargs) -> bool: - status_code = kwargs.get('status_code') - params = { - 'duration': kwargs.get('duration'), - 'status code': (status_code, kwargs.get('sub_status_code')) if status_code else None, - 'verb': kwargs.get('verb'), - 'http version': kwargs.get('http_version'), - 'database name': kwargs.get('database_name'), - 'collection name': kwargs.get('collection_name'), - 'resource type': kwargs.get('resource_type') - } - for key, param in params.items(): - if param is not None and self[key] is not None: - if self[key](param): - return True - return False - - def get_preset_keys(self): - return self._preset_keys.keys() diff --git a/sdk/cosmos/azure-cosmos/samples/diagnostics_filter_sample.py b/sdk/cosmos/azure-cosmos/samples/diagnostics_filter_sample.py new file mode 100644 index 000000000000..f0d2c32a34a5 --- /dev/null +++ b/sdk/cosmos/azure-cosmos/samples/diagnostics_filter_sample.py @@ -0,0 +1,93 @@ +# ------------------------------------------------------------------------- +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. See LICENSE.txt in the project root for +# license information. +# ------------------------------------------------------------------------- +import logging, os +from typing import Dict, Callable, Any +from azure.cosmos import CosmosClient, PartitionKey, exceptions + +endpoint = os.environ["ACCOUNT_URI"] +key = os.environ["ACCOUNT_KEY"] +# Sample usage of using logging filters for diagnostics filtering +# You can filter based on request and response related attributes that are added to the log record +class CosmosStatusCodeFilter(logging.Filter): + def filter(self, record): + ret = (hasattr(record, 'status_code') and record.status_code > 400 + and not (record.status_code in [404, 409, 412] and getattr(record, 'sub_status_code', None) in [0, None]) + and hasattr(record, 'duration') and record.duration > 1000) + return ret +# Initialize the logger +logger = logging.getLogger('azure.cosmos') +logger.setLevel(logging.INFO) +file_handler = logging.FileHandler('diagnostics1.output') +logger.addHandler(file_handler) +# When using the logging filter, you can set the filter directly on the logger +logger.addFilter(CosmosStatusCodeFilter()) +# Initialize the Cosmos client with diagnostics enabled, no need to pass a diagnostics handler +client = CosmosClient(endpoint, key, logger=logger, enable_diagnostics_logging=True) +# Create a database and container +database_name = 'SD' +database = client.create_database_if_not_exists(id=database_name) +container_name = 'SampleContainer' +partition_key = PartitionKey(path=['/State', '/City']) +container = database.create_container_if_not_exists(id=container_name, partition_key=partition_key, + offer_throughput=400) +items = [ + {'id': '1', 'State': 'California', 'City': 'Los Angeles', 'city_level': 1}, + {'id': '2', 'State': 'Texas', 'City': 'Houston', 'city_level': 2}, + {'id': '3', 'State': 'New York', 'City': 'New York City', 'city_level': 3} +] +# Attempt to read nonexistent items to cause a 404 error +for item in items: + try: + container.read_item(item=str(item['id']), partition_key=[str(item['State']), str(item['City'])]) + except exceptions.CosmosHttpResponseError: + pass + +# When using the async client it can also be possible to use the logging filter with a queue logger handler +import asyncio +import queue +from queue import Queue +import logging.handlers +from azure.cosmos.aio import CosmosClient as CosmosAsyncClient +async def log_cosmos_operations(): + # Initialize the logger + logger = logging.getLogger('azure.cosmos') + logger.setLevel(logging.INFO) + # Create a queue + log_queue: Queue = queue.Queue(-1) + # Set up the QueueHandler + queue_handler = logging.handlers.QueueHandler(log_queue) + # Set up the QueueListener with a FileHandler + file_handler = logging.FileHandler('diagnostics2.output') + file_handler.setLevel(logging.INFO) + queue_listener = logging.handlers.QueueListener(log_queue, file_handler) + # Configure the root logger + logging.basicConfig(level=logging.INFO, handlers=[queue_handler]) + # Add the filter to the logger + logger.addFilter(CosmosStatusCodeFilter()) + # Start the QueueListener + queue_listener.start() + # Initialize the Cosmos client with diagnostics enabled, no need to pass a diagnostics handler + async with CosmosAsyncClient(endpoint, key, logger=logger, enable_diagnostics_logging=True) as client: + # Create a database and container + database_name = 'SD' + database = await client.create_database_if_not_exists(id=database_name) + container_name = 'SampleContainer' + partition_key = PartitionKey(path=['/State', '/City']) + container = await database.create_container_if_not_exists(id=container_name, partition_key=partition_key, + offer_throughput=400) + # Attempt to read nonexistent items to cause a 404 error + for item in items: + try: + await container.read_item(item=str(item['id']), partition_key=[str(item['State']), str(item['City'])]) + except exceptions.CosmosHttpResponseError: + pass + # Stop the QueueListener + queue_listener.stop() + +# Run the async method +asyncio.run(log_cosmos_operations()) + + diff --git a/sdk/cosmos/azure-cosmos/samples/diagnostics_handler_sample.py b/sdk/cosmos/azure-cosmos/samples/diagnostics_handler_sample.py deleted file mode 100644 index d1ce875e0f57..000000000000 --- a/sdk/cosmos/azure-cosmos/samples/diagnostics_handler_sample.py +++ /dev/null @@ -1,111 +0,0 @@ -import logging, os -from typing import Dict, Callable, Any -from azure.cosmos import CosmosClient, PartitionKey, cosmos_diagnostics_handler - -# Initialize the logger -logger = logging.getLogger('azure.cosmos') -logger.setLevel(logging.INFO) -file_handler = logging.FileHandler('diagnostics1.output') -logger.addHandler(file_handler) -diagnostics_handler = cosmos_diagnostics_handler.CosmosDiagnosticsHandler() -diagnostics_handler["duration"] = lambda x: x > 20 - -# Initialize the Cosmos client with diagnostics handler -endpoint = os.environ["ACCOUNT_URI"] -key = os.environ["ACCOUNT_KEY"] -client = CosmosClient(endpoint, key,logger=logger, diagnostics_handler=diagnostics_handler, enable_diagnostics_logging=True) - -# Create a database and container -database_name = 'SD' -database = client.create_database_if_not_exists(id=database_name) -container_name = 'SampleContainer' -partition_key = PartitionKey(path=['/State', '/City']) -container = database.create_container_if_not_exists(id=container_name, partition_key=partition_key, offer_throughput=400) - -# Add 3 items to the container -items = [ - {'id': '1', 'State': 'California', 'City': 'Los Angeles', 'city_level': 1}, - {'id': '2', 'State': 'Texas', 'City': 'Houston', 'city_level': 2}, - {'id': '3', 'State': 'New York', 'City': 'New York City', 'city_level': 3} -] -for item in items: - container.create_item(body=item) - -# Delete the items -for item in items: - container.delete_item(item=str(item['id']), partition_key=[str(item['State']), str(item['City'])]) - - -# Custom should_log method -def should_log(self, **kwargs): - duration = kwargs.get('duration') - return duration and duration > 100 - -# Initialize the logger -logger = logging.getLogger('azure.cosmos') -logger.setLevel(logging.INFO) -file_handler = logging.FileHandler('diagnostics2.output') -logger.addHandler(file_handler) - -# Initialize the Cosmos client with custom diagnostics handler -client = CosmosClient(endpoint, key,logger=logger, diagnostics_handler=should_log, enable_diagnostics_logging=True) - -# Create a database and container -database_name = 'SD' -database = client.create_database_if_not_exists(id=database_name) -container_name = 'SampleContainer' -partition_key = PartitionKey(path=['/State', '/City']) -container = database.create_container_if_not_exists(id=container_name, partition_key=partition_key, offer_throughput=400) - -# Add 3 items to the container -items = [ - {'id': '1', 'State': 'California', 'City': 'Los Angeles', 'city_level': 1}, - {'id': '2', 'State': 'Texas', 'City': 'Houston', 'city_level': 2}, - {'id': '3', 'State': 'New York', 'City': 'New York City', 'city_level': 3} -] -for item in items: - container.create_item(body=item) - -# Delete the items -for item in items: - container.delete_item(item=str(item['id']), partition_key=[str(item['State']), str(item['City'])]) - - -# Diagnostics handler dictionary -diagnostics_handler_dict: Dict[str, Callable[[Any], Any]] = { - 'duration': lambda duration: duration > 2000, - 'status code': (lambda x: ( - isinstance(x, (list, tuple)) and x[0] >= 400) if isinstance(x, (list, tuple)) else x >= 400), - 'verb': lambda verb: verb in ['POST', 'DELETE'], - 'resource type': lambda resource_type: resource_type == 'item' -} - -# Initialize the logger -logger = logging.getLogger('azure.cosmos') -logger.setLevel(logging.INFO) -file_handler = logging.FileHandler('diagnostics3.output') -logger.addHandler(file_handler) - -# Initialize the Cosmos client with diagnostics handler dictionary -client = CosmosClient(endpoint, key,logger=logger, diagnostics_handler=diagnostics_handler_dict, enable_diagnostics_logging=True) - -# Create a database and container -database_name = 'SD' -database = client.create_database_if_not_exists(id=database_name) -container_name = 'SampleContainer' -partition_key = PartitionKey(path=['/State', '/City']) -container = database.create_container_if_not_exists(id=container_name, partition_key=partition_key, offer_throughput=400) - -# Add 3 items to the container -items = [ - {'id': '1', 'State': 'California', 'City': 'Los Angeles', 'city_level': 1}, - {'id': '2', 'State': 'Texas', 'City': 'Houston', 'city_level': 2}, - {'id': '3', 'State': 'New York', 'City': 'New York City', 'city_level': 3} -] -for item in items: - container.create_item(body=item) - -# Delete the items -for item in items: - container.delete_item(item=str(item['id']), partition_key=[str(item['State']), str(item['City'])]) - diff --git a/sdk/cosmos/azure-cosmos/tests/test_cosmos_http_logging_policy.py b/sdk/cosmos/azure-cosmos/tests/test_cosmos_http_logging_policy.py index 5ce7418892a5..2e43d13c09e7 100644 --- a/sdk/cosmos/azure-cosmos/tests/test_cosmos_http_logging_policy.py +++ b/sdk/cosmos/azure-cosmos/tests/test_cosmos_http_logging_policy.py @@ -31,6 +31,8 @@ def emit(self, record): self.messages.append(record) + + @pytest.mark.cosmosEmulator class TestCosmosHttpLogger(unittest.TestCase): mock_handler_diagnostic = None @@ -54,10 +56,10 @@ def setUpClass(cls): cls.mock_handler_diagnostic = MockHandler() cls.logger_default = logging.getLogger("testloggerdefault") cls.logger_default.addHandler(cls.mock_handler_default) - cls.logger_default.setLevel(logging.DEBUG) + cls.logger_default.setLevel(logging.INFO) cls.logger_diagnostic = logging.getLogger("testloggerdiagnostic") cls.logger_diagnostic.addHandler(cls.mock_handler_diagnostic) - cls.logger_diagnostic.setLevel(logging.DEBUG) + cls.logger_diagnostic.setLevel(logging.INFO) cls.client_default = cosmos_client.CosmosClient(cls.host, cls.masterKey, consistency_level="Session", connection_policy=cls.connectionPolicy, @@ -73,8 +75,8 @@ def test_default_http_logging_policy(self): database_id = "database_test-" + str(uuid.uuid4()) self.client_default.create_database(id=database_id) assert all(m.levelname == 'INFO' for m in self.mock_handler_default.messages) - messages_request = self.mock_handler_default.messages[3].message.split("\n") - messages_response = self.mock_handler_default.messages[4].message.split("\n") + messages_request = self.mock_handler_default.messages[0].message.split("\n") + messages_response = self.mock_handler_default.messages[1].message.split("\n") assert messages_request[1] == "Request method: 'GET'" assert 'Request headers:' in messages_request[2] assert messages_request[14] == 'No body was attached to the request' @@ -87,20 +89,29 @@ def test_default_http_logging_policy(self): self.client_default.delete_database(database_id) def test_cosmos_http_logging_policy(self): - # Test if we can log into from creating a database + # Test if we can log into from reading a database database_id = "database_test-" + str(uuid.uuid4()) self.client_diagnostic.create_database(id=database_id) assert all(m.levelname == 'INFO' for m in self.mock_handler_diagnostic.messages) - messages_request = self.mock_handler_diagnostic.messages[13].message.split("\n") - messages_response = self.mock_handler_diagnostic.messages[14].message.split("\n") - elapsed_time = self.mock_handler_diagnostic.messages[5].message.split("\n") - assert "/dbs" in messages_request[0] - assert messages_request[1] == "Request method: 'POST'" - assert 'Request headers:' in messages_request[2] - assert messages_request[15] == 'A body is sent with the request' - assert messages_response[0] == 'Response status: 201' - assert "Elapsed time in seconds:" in elapsed_time[0] - assert "Response headers" in messages_response[1] + messages_request = self.mock_handler_diagnostic.messages[1] + messages_response = self.mock_handler_diagnostic.messages[2] + elapsed_time = messages_request.duration + assert "databaseaccount" == messages_request.resource_type + assert messages_request.verb == "GET" + assert 200 == messages_request.status_code + assert "Read" == messages_request.operation_type + assert elapsed_time is not None + assert "Response headers" in messages_response.message + # Test if we can log into from creating a database + messages_request = self.mock_handler_diagnostic.messages[4] + messages_response = self.mock_handler_diagnostic.messages[5] + elapsed_time = messages_request.duration + assert "dbs" == messages_request.resource_type + assert messages_request.verb == "POST" + assert 201 == messages_request.status_code + assert messages_request.operation_type == "Create" + assert elapsed_time is not None + assert "Response headers" in messages_response.message self.mock_handler_diagnostic.reset() # now test in case of an error @@ -109,16 +120,16 @@ def test_cosmos_http_logging_policy(self): except: pass assert all(m.levelname == 'INFO' for m in self.mock_handler_diagnostic.messages) - messages_request = self.mock_handler_diagnostic.messages[7].message.split("\n") - messages_response = self.mock_handler_diagnostic.messages[8].message.split("\n") - elapsed_time = self.mock_handler_diagnostic.messages[9].message.split("\n") - assert "/dbs" in messages_request[0] - assert messages_request[1] == "Request method: 'POST'" - assert 'Request headers:' in messages_request[2] - assert messages_request[15] == 'A body is sent with the request' - assert messages_response[0] == 'Response status: 409' - assert "Elapsed time in seconds:" in elapsed_time[0] - assert "Response headers" in messages_response[1] + messages_request = self.mock_handler_diagnostic.messages[1] + messages_response = self.mock_handler_diagnostic.messages[2] + elapsed_time = messages_request.duration + assert "dbs" == messages_request.resource_type + assert messages_request.operation_type == "Create" + assert 'Request headers:' in messages_request.msg + assert 'A body is sent with the request' in messages_request.msg + assert messages_request.status_code == 409 + assert elapsed_time is not None + assert "Response headers" in messages_response.msg # delete database self.client_diagnostic.delete_database(database_id)