[Monitor Query] Add batch metrics query integration (#31049)

This adds the `MetricsBatchQueryClient` for querying resource URIs in batches.

Signed-off-by: Paul Van Eck <paulvaneck@microsoft.com>

Author: pvaneck

sdk/monitor/azure-monitor-query/CHANGELOG.md

@@ -1,9 +1,11 @@
 # Release History
 
-## 1.2.1 (Unreleased)
+## 1.3.0b1 (Unreleased)
 
 ### Features Added
 
+- Added `MetricsBatchQueryClient` to support batch querying metrics from Azure resources. ([#31049](https://github.com/Azure/azure-sdk-for-python/pull/31049))
+
 ### Breaking Changes
 
 ### Bugs Fixed

sdk/monitor/azure-monitor-query/README.md

@@ -37,7 +37,7 @@ pip install azure-monitor-query
 
 ### Create the client
 
-An authenticated client is required to query Logs or Metrics. The library includes both synchronous and asynchronous forms of the clients. To authenticate, create an instance of a token credential. Use that instance when creating a `LogsQueryClient` or `MetricsQueryClient`. The following examples use `DefaultAzureCredential` from the [azure-identity](https://pypi.org/project/azure-identity/) package.
+An authenticated client is required to query Logs or Metrics. The library includes both synchronous and asynchronous forms of the clients. To authenticate, create an instance of a token credential. Use that instance when creating a `LogsQueryClient`, `MetricsQueryClient`, or `MetricsBatchQueryClient`. The following examples use `DefaultAzureCredential` from the [azure-identity](https://pypi.org/project/azure-identity/) package.
 
 #### Synchronous clients
 
@@ -524,6 +524,43 @@ for metric in response.metrics:
                 )
 ```
 
+### Metrics batch query
+
+A user can also query metrics from multiple resources at once using the `query_batch` method of `MetricsBatchQueryClient`. This uses a different API than the `MetricsQueryClient` and requires that a user pass in a regional endpoint when instantiating the client (for example, "https://westus3.metrics.monitor.azure.com").
+
+Note, each resource must be in the same region as the endpoint passed in when instantiating the client, and each resource must be in the same Azure subscription. Furthermore, the metric namespace that contains the metrics to be queried must also be passed. A list of metric namespaces can be found [here][metric_namespaces].
+
+
+```python
+from datetime import timedelta
+import os
+
+from azure.core.exceptions import HttpResponseError
+from azure.identity import DefaultAzureCredential
+from azure.monitor.query import MetricsBatchQueryClient, MetricAggregationType
+
+
+credential = DefaultAzureCredential()
+client = MetricsBatchQueryClient(endpoint, credential)
+
+resource_uris = [
+    "/subscriptions/<id>/resourceGroups/<rg-name>/providers/<source>/storageAccounts/<resource-name-1>",
+    "/subscriptions/<id>/resourceGroups/<rg-name>/providers/<source>/storageAccounts/<resource-name-2>"
+]
+
+response = client.query_batch(
+    resource_uris,
+    metric_namespace="Microsoft.Storage/storageAccounts",
+    metric_names=["Ingress"],
+    timespan=timedelta(hours=2),
+    granularity=timedelta(minutes=5),
+    aggregations=[MetricAggregationType.AVERAGE],
+)
+
+for metrics_query_result in response:
+    print(metrics_query_result.timespan)
+```
+
 ## Troubleshooting
 
 See our [troubleshooting guide][troubleshooting_guide] for details on how to diagnose various failure scenarios.
@@ -568,6 +605,7 @@ This project has adopted the [Microsoft Open Source Code of Conduct][code_of_con
 [azure_subscription]: https://azure.microsoft.com/free/python/
 [changelog]: https://github.com/Azure/azure-sdk-for-python/tree/main/sdk/monitor/azure-monitor-query/CHANGELOG.md
 [kusto_query_language]: https://learn.microsoft.com/azure/data-explorer/kusto/query/
+[metric_namespaces]: https://learn.microsoft.com/azure/azure-monitor/reference/supported-metrics/metrics-index#metrics-by-resource-provider
 [package]: https://aka.ms/azsdk-python-monitor-query-pypi
 [pip]: https://pypi.org/project/pip/
 [python_logging]: https://docs.python.org/3/library/logging.html

sdk/monitor/azure-monitor-query/assets.json

@@ -2,5 +2,5 @@
   "AssetsRepo": "Azure/azure-sdk-assets",
   "AssetsRepoPrefixPath": "python",
   "TagPrefix": "python/monitor/azure-monitor-query",
-  "Tag": "python/monitor/azure-monitor-query_8b3197a327"
+  "Tag": "python/monitor/azure-monitor-query_70a241fcfc"
 }

sdk/monitor/azure-monitor-query/azure/monitor/query/__init__.py

@@ -6,6 +6,7 @@
 
 from ._logs_query_client import LogsQueryClient
 from ._metrics_query_client import MetricsQueryClient
+from ._metrics_batch_query_client import MetricsBatchQueryClient
 
 from ._enums import (
     LogsQueryStatus,
@@ -45,6 +46,7 @@
     "LogsTableRow",
     "LogsBatchQuery",
     "MetricsQueryClient",
+    "MetricsBatchQueryClient",
     "MetricNamespace",
     "MetricNamespaceClassification",
     "MetricDefinition",

sdk/monitor/azure-monitor-query/azure/monitor/query/_generated/_serialization.py

@@ -662,8 +662,9 @@ def _serialize(self, target_obj, data_type=None, **kwargs):
                                 _serialized.update(_new_attr)  # type: ignore
                             _new_attr = _new_attr[k]  # type: ignore
                             _serialized = _serialized[k]
-                except ValueError:
-                    continue
+                except ValueError as err:
+                    if isinstance(err, SerializationError):
+                        raise
 
         except (AttributeError, KeyError, TypeError) as err:
             msg = "Attribute {} in object {} cannot be serialized.\n{}".format(attr_name, class_name, str(target_obj))
@@ -741,6 +742,8 @@ def query(self, name, data, data_type, **kwargs):
 
         :param data: The data to be serialized.
         :param str data_type: The type to be serialized from.
+        :keyword bool skip_quote: Whether to skip quote the serialized result.
+        Defaults to False.
         :rtype: str
         :raises: TypeError if serialization fails.
         :raises: ValueError if data is None
@@ -749,10 +752,8 @@ def query(self, name, data, data_type, **kwargs):
             # Treat the list aside, since we don't want to encode the div separator
             if data_type.startswith("["):
                 internal_data_type = data_type[1:-1]
-                data = [self.serialize_data(d, internal_data_type, **kwargs) if d is not None else "" for d in data]
-                if not kwargs.get("skip_quote", False):
-                    data = [quote(str(d), safe="") for d in data]
-                return str(self.serialize_iter(data, internal_data_type, **kwargs))
+                do_quote = not kwargs.get("skip_quote", False)
+                return str(self.serialize_iter(data, internal_data_type, do_quote=do_quote, **kwargs))
 
             # Not a list, regular serialization
             output = self.serialize_data(data, data_type, **kwargs)
@@ -891,6 +892,8 @@ def serialize_iter(self, data, iter_type, div=None, **kwargs):
          not be None or empty.
         :param str div: If set, this str will be used to combine the elements
          in the iterable into a combined string. Default is 'None'.
+        :keyword bool do_quote: Whether to quote the serialized result of each iterable element.
+        Defaults to False.
         :rtype: list, str
         """
         if isinstance(data, str):
@@ -903,9 +906,14 @@ def serialize_iter(self, data, iter_type, div=None, **kwargs):
         for d in data:
             try:
                 serialized.append(self.serialize_data(d, iter_type, **kwargs))
-            except ValueError:
+            except ValueError as err:
+                if isinstance(err, SerializationError):
+                    raise
                 serialized.append(None)
 
+        if kwargs.get("do_quote", False):
+            serialized = ["" if s is None else quote(str(s), safe="") for s in serialized]
+
         if div:
             serialized = ["" if s is None else str(s) for s in serialized]
             serialized = div.join(serialized)
@@ -950,7 +958,9 @@ def serialize_dict(self, attr, dict_type, **kwargs):
         for key, value in attr.items():
             try:
                 serialized[self.serialize_unicode(key)] = self.serialize_data(value, dict_type, **kwargs)
-            except ValueError:
+            except ValueError as err:
+                if isinstance(err, SerializationError):
+                    raise
                 serialized[self.serialize_unicode(key)] = None
 
         if "xml" in serialization_ctxt:

sdk/monitor/azure-monitor-query/azure/monitor/query/_generated/_vendor.py

@@ -7,6 +7,7 @@
 # Changes may cause incorrect behavior and will be lost if the code is regenerated.
 # --------------------------------------------------------------------------
 import datetime
+from io import IOBase
 import sys
 from typing import Any, Callable, Dict, IO, Optional, TypeVar, Union, cast, overload
 
@@ -19,8 +20,7 @@
     map_error,
 )
 from azure.core.pipeline import PipelineResponse
-from azure.core.pipeline.transport import AsyncHttpResponse
-from azure.core.rest import HttpRequest
+from azure.core.rest import AsyncHttpResponse, HttpRequest
 from azure.core.tracing.decorator_async import distributed_trace_async
 from azure.core.utils import case_insensitive_dict
 
@@ -165,6 +165,8 @@ async def get(
         response = pipeline_response.http_response
 
         if response.status_code not in [200]:
+            if _stream:
+                await response.read()  # Load the body in memory and close the socket
             map_error(status_code=response.status_code, response=response, error_map=error_map)
             raise HttpResponseError(response=response)
 
@@ -464,7 +466,7 @@ async def execute(
         content_type = content_type or "application/json"
         _json = None
         _content = None
-        if isinstance(body, (IO, bytes)):
+        if isinstance(body, (IOBase, bytes)):
             _content = body
         else:
             _json = body
@@ -488,6 +490,8 @@ async def execute(
         response = pipeline_response.http_response
 
         if response.status_code not in [200]:
+            if _stream:
+                await response.read()  # Load the body in memory and close the socket
             map_error(status_code=response.status_code, response=response, error_map=error_map)
             raise HttpResponseError(response=response)
 
@@ -605,6 +609,8 @@ async def resource_get(
         response = pipeline_response.http_response
 
         if response.status_code not in [200]:
+            if _stream:
+                await response.read()  # Load the body in memory and close the socket
             map_error(status_code=response.status_code, response=response, error_map=error_map)
             raise HttpResponseError(response=response)
 
@@ -901,7 +907,7 @@ async def resource_execute(
         content_type = content_type or "application/json"
         _json = None
         _content = None
-        if isinstance(body, (IO, bytes)):
+        if isinstance(body, (IOBase, bytes)):
             _content = body
         else:
             _json = body
@@ -925,6 +931,8 @@ async def resource_execute(
         response = pipeline_response.http_response
 
         if response.status_code not in [200]:
+            if _stream:
+                await response.read()  # Load the body in memory and close the socket
             map_error(status_code=response.status_code, response=response, error_map=error_map)
             raise HttpResponseError(response=response)
 
@@ -1287,7 +1295,7 @@ async def batch(self, body: Union[JSON, IO], **kwargs: Any) -> JSON:
         content_type = content_type or "application/json"
         _json = None
         _content = None
-        if isinstance(body, (IO, bytes)):
+        if isinstance(body, (IOBase, bytes)):
             _content = body
         else:
             _json = body
@@ -1309,6 +1317,8 @@ async def batch(self, body: Union[JSON, IO], **kwargs: Any) -> JSON:
         response = pipeline_response.http_response
 
         if response.status_code not in [200]:
+            if _stream:
+                await response.read()  # Load the body in memory and close the socket
             map_error(status_code=response.status_code, response=response, error_map=error_map)
             raise HttpResponseError(response=response)
 
@@ -1426,6 +1436,8 @@ async def resource_get_xms(
         response = pipeline_response.http_response
 
         if response.status_code not in [200]:
+            if _stream:
+                await response.read()  # Load the body in memory and close the socket
             map_error(status_code=response.status_code, response=response, error_map=error_map)
             raise HttpResponseError(response=response)
 
@@ -1722,7 +1734,7 @@ async def resource_execute_xms(
         content_type = content_type or "application/json"
         _json = None
         _content = None
-        if isinstance(body, (IO, bytes)):
+        if isinstance(body, (IOBase, bytes)):
             _content = body
         else:
             _json = body
@@ -1746,6 +1758,8 @@ async def resource_execute_xms(
         response = pipeline_response.http_response
 
         if response.status_code not in [200]:
+            if _stream:
+                await response.read()  # Load the body in memory and close the socket
             map_error(status_code=response.status_code, response=response, error_map=error_map)
             raise HttpResponseError(response=response)
 
@@ -2163,6 +2177,8 @@ async def get(self, workspace_id: str, **kwargs: Any) -> JSON:
         response = pipeline_response.http_response
 
         if response.status_code not in [200]:
+            if _stream:
+                await response.read()  # Load the body in memory and close the socket
             map_error(status_code=response.status_code, response=response, error_map=error_map)
             raise HttpResponseError(response=response)
 
@@ -2562,6 +2578,8 @@ async def post(self, workspace_id: str, **kwargs: Any) -> JSON:
         response = pipeline_response.http_response
 
         if response.status_code not in [200]:
+            if _stream:
+                await response.read()  # Load the body in memory and close the socket
             map_error(status_code=response.status_code, response=response, error_map=error_map)
             raise HttpResponseError(response=response)
 

sdk/monitor/azure-monitor-query/azure/monitor/query/_generated/aio/operations/_operations.py

@@ -662,8 +662,9 @@ def _serialize(self, target_obj, data_type=None, **kwargs):
                                 _serialized.update(_new_attr)  # type: ignore
                             _new_attr = _new_attr[k]  # type: ignore
                             _serialized = _serialized[k]
-                except ValueError:
-                    continue
+                except ValueError as err:
+                    if isinstance(err, SerializationError):
+                        raise
 
         except (AttributeError, KeyError, TypeError) as err:
             msg = "Attribute {} in object {} cannot be serialized.\n{}".format(attr_name, class_name, str(target_obj))
@@ -741,6 +742,8 @@ def query(self, name, data, data_type, **kwargs):
 
         :param data: The data to be serialized.
         :param str data_type: The type to be serialized from.
+        :keyword bool skip_quote: Whether to skip quote the serialized result.
+        Defaults to False.
         :rtype: str
         :raises: TypeError if serialization fails.
         :raises: ValueError if data is None
@@ -749,10 +752,8 @@ def query(self, name, data, data_type, **kwargs):
             # Treat the list aside, since we don't want to encode the div separator
             if data_type.startswith("["):
                 internal_data_type = data_type[1:-1]
-                data = [self.serialize_data(d, internal_data_type, **kwargs) if d is not None else "" for d in data]
-                if not kwargs.get("skip_quote", False):
-                    data = [quote(str(d), safe="") for d in data]
-                return str(self.serialize_iter(data, internal_data_type, **kwargs))
+                do_quote = not kwargs.get("skip_quote", False)
+                return str(self.serialize_iter(data, internal_data_type, do_quote=do_quote, **kwargs))
 
             # Not a list, regular serialization
             output = self.serialize_data(data, data_type, **kwargs)
@@ -891,6 +892,8 @@ def serialize_iter(self, data, iter_type, div=None, **kwargs):
          not be None or empty.
         :param str div: If set, this str will be used to combine the elements
          in the iterable into a combined string. Default is 'None'.
+        :keyword bool do_quote: Whether to quote the serialized result of each iterable element.
+        Defaults to False.
         :rtype: list, str
         """
         if isinstance(data, str):
@@ -903,9 +906,14 @@ def serialize_iter(self, data, iter_type, div=None, **kwargs):
         for d in data:
             try:
                 serialized.append(self.serialize_data(d, iter_type, **kwargs))
-            except ValueError:
+            except ValueError as err:
+                if isinstance(err, SerializationError):
+                    raise
                 serialized.append(None)
 
+        if kwargs.get("do_quote", False):
+            serialized = ["" if s is None else quote(str(s), safe="") for s in serialized]
+
         if div:
             serialized = ["" if s is None else str(s) for s in serialized]
             serialized = div.join(serialized)
@@ -950,7 +958,9 @@ def serialize_dict(self, attr, dict_type, **kwargs):
         for key, value in attr.items():
             try:
                 serialized[self.serialize_unicode(key)] = self.serialize_data(value, dict_type, **kwargs)
-            except ValueError:
+            except ValueError as err:
+                if isinstance(err, SerializationError):
+                    raise
                 serialized[self.serialize_unicode(key)] = None
 
         if "xml" in serialization_ctxt: