Implement live-metrics skeleton (#33983)

Author: lzchen

sdk/monitor/azure-monitor-opentelemetry-exporter/CHANGELOG.md

@@ -4,6 +4,9 @@
 
 ### Features Added
 
+- Add live metrics skeleton + swagger definitions
+    ([#33983](https://github.com/Azure/azure-sdk-for-python/pull/33983))
+
 ### Breaking Changes
 
 ### Bugs Fixed
@@ -34,7 +37,7 @@
 - Implement distro detection for statsbeat feature
     ([#33761](https://github.com/Azure/azure-sdk-for-python/pull/33761))
 - Use empty resource for statsbeat `MeterProvider`
-    ([#33761](https://github.com/Azure/azure-sdk-for-python/pull/33761))
+    ([#33768](https://github.com/Azure/azure-sdk-for-python/pull/33768))
 
 ## 1.0.0b19 (2023-11-20)
 

sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_connection_string_parser.py

@@ -4,6 +4,7 @@
 import re
 import typing
 
+LIVE_ENDPOINT = "liveendpoint"
 INGESTION_ENDPOINT = "ingestionendpoint"
 INSTRUMENTATION_KEY = "instrumentationkey"
 
@@ -32,6 +33,7 @@ def __init__(
     ) -> None:
         self.instrumentation_key = None
         self.endpoint = ""
+        self.live_endpoint = ""
         self._connection_string = connection_string
         self._initialize()
         self._validate_instrumentation_key()
@@ -56,7 +58,7 @@ def _initialize(self) -> None:
             or env_cs.get(INSTRUMENTATION_KEY)
             or env_ikey
         )
-        # The priority of the ingestion endpoint is as follows:
+        # The priority of the endpoints is as follows:
         # 1. The endpoint explicitly passed in connection string
         # 2. The endpoint from the connection string in environment variable
         # 3. The default breeze endpoint
@@ -65,6 +67,12 @@ def _initialize(self) -> None:
             or env_cs.get(INGESTION_ENDPOINT)
             or "https://dc.services.visualstudio.com"
         )
+        self.live_endpoint = (
+            code_cs.get(LIVE_ENDPOINT)
+            or env_cs.get(LIVE_ENDPOINT)
+            or "https://rt.services.visualstudio.com"
+        )
+
 
     def _validate_instrumentation_key(self) -> None:
         """Validates the instrumentation key used for Azure Monitor.
@@ -93,22 +101,34 @@ def _parse_connection_string(self, connection_string) -> typing.Dict:
         auth = result.get("authorization")
         if auth is not None and auth.lower() != "ikey":
             raise ValueError("Invalid authorization mechanism")
-        # Construct the ingestion endpoint if not passed in explicitly
+
+        # Construct the endpoints if not passed in explicitly
+        endpoint_suffix = ""
+        location_prefix = ""
+        suffix = result.get("endpointsuffix")
+        # Get regional information if provided
+        prefix = result.get("location")
+        if suffix is not None:
+            endpoint_suffix = suffix
+            # Get regional information if provided
+            prefix = result.get("location")
+            if prefix is not None:
+                location_prefix = prefix + "."
+        # Construct the endpoints if not passed in explicitly
         if result.get(INGESTION_ENDPOINT) is None:
-            endpoint_suffix = ""
-            location_prefix = ""
-            suffix = result.get("endpointsuffix")
-            if suffix is not None:
-                endpoint_suffix = suffix
-                # Get regional information if provided
-                prefix = result.get("location")
-                if prefix is not None:
-                    location_prefix = prefix + "."
-                endpoint = "https://{0}dc.{1}".format(
+            if endpoint_suffix:
+                result[INGESTION_ENDPOINT] = "https://{0}dc.{1}".format(
                     location_prefix, endpoint_suffix
                 )
-                result[INGESTION_ENDPOINT] = endpoint
             else:
                 # Default to None if cannot construct
                 result[INGESTION_ENDPOINT] = None
+        if result.get(LIVE_ENDPOINT) is None:
+            if endpoint_suffix:
+                result[LIVE_ENDPOINT] = "https://{0}live.{1}".format(
+                    location_prefix, endpoint_suffix
+                )
+            else:
+                result[LIVE_ENDPOINT] = None
+
         return result

sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_quickpulse/__init__.py

@@ -0,0 +1,11 @@
+# -------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License in the project root for
+# license information.
+# -------------------------------------------------------------------------
+
+from azure.monitor.opentelemetry.exporter._quickpulse._exporter import QuickpulseExporter
+
+__all__ = [
+    "QuickpulseExporter",
+]

sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_quickpulse/_exporter.py

@@ -0,0 +1,106 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+import logging
+
+from typing import Any
+
+from opentelemetry.sdk.metrics import (
+    Counter,
+    Histogram,
+    ObservableCounter,
+    ObservableGauge,
+    ObservableUpDownCounter,
+    UpDownCounter,
+)
+from opentelemetry.sdk.metrics.export import (
+    AggregationTemporality,
+    MetricExporter,
+    MetricExportResult,
+    MetricsData as OTMetricsData,
+)
+from azure.monitor.opentelemetry.exporter._quickpulse._generated._client import QuickpulseClient
+from azure.monitor.opentelemetry.exporter._connection_string_parser import ConnectionStringParser
+
+_logger = logging.getLogger(__name__)
+
+__all__ = ["QuickpulseExporter"]
+
+
+APPLICATION_INSIGHTS_METRIC_TEMPORALITIES = {
+    Counter: AggregationTemporality.DELTA,
+    Histogram: AggregationTemporality.DELTA,
+    ObservableCounter: AggregationTemporality.DELTA,
+    ObservableGauge: AggregationTemporality.CUMULATIVE,
+    ObservableUpDownCounter: AggregationTemporality.CUMULATIVE,
+    UpDownCounter: AggregationTemporality.CUMULATIVE,
+}
+
+
+class QuickpulseExporter(MetricExporter):
+
+    def __init__(self, **kwargs: Any) -> None:
+        """Metric exporter for Quickpulse.
+
+        :keyword str connection_string: The connection string used for your Application Insights resource.
+        :rtype: None
+        """
+        parsed_connection_string = ConnectionStringParser(kwargs.get('connection_string'))
+
+        self._endpoint = parsed_connection_string.endpoint
+        # TODO: Support AADaudience (scope)/credentials
+
+        self.client = QuickpulseClient(host=self._endpoint, **kwargs)
+        # TODO: Support redirect
+
+        MetricExporter.__init__(
+            self,
+            preferred_temporality=APPLICATION_INSIGHTS_METRIC_TEMPORALITIES, # type: ignore
+            preferred_aggregation=kwargs.get("preferred_aggregation"), # type: ignore
+        )
+
+    def export(
+        self,
+        metrics_data: OTMetricsData,
+        timeout_millis: float = 10_000,  # pylint: disable=unused-argument
+        **kwargs: Any,  # pylint: disable=unused-argument
+    ) -> MetricExportResult:
+        """Exports a batch of metric data
+
+        :param metrics_data: OpenTelemetry Metric(s) to export.
+        :type metrics_data: Sequence[~opentelemetry.sdk.metrics._internal.point.MetricsData]
+        :param timeout_millis: The maximum amount of time to wait for each export. Not currently used.
+        :type timeout_millis: float
+        :return: The result of the export.
+        :rtype: ~opentelemetry.sdk.metrics.export.MetricExportResult
+        """
+        # TODO
+        return MetricExportResult.SUCCESS
+
+    def force_flush(
+        self,
+        timeout_millis: float = 10_000,
+    ) -> bool:
+        """
+        Ensure that export of any metrics currently received by the exporter
+        are completed as soon as possible. Called when SDK is flushed.
+
+        :param timeout_millis: The maximum amount of time to wait for shutdown. Not currently used.
+        :type timeout_millis: float
+        :return: The result of the export.
+        :rtype: bool
+        """
+        return True
+
+
+    def shutdown(
+        self,
+        timeout_millis: float = 30_000,  # pylint: disable=unused-argument
+        **kwargs: Any,  # pylint: disable=unused-argument
+    ) -> None:
+        """Shuts down the exporter.
+
+        Called when the SDK is shut down.
+
+        :param timeout_millis: The maximum amount of time to wait for shutdown. Not currently used.
+        :type timeout_millis: float
+        """

sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_quickpulse/_generated/__init__.py

@@ -0,0 +1,23 @@
+# coding=utf-8
+# --------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# Code generated by Microsoft (R) AutoRest Code Generator.
+# Changes may cause incorrect behavior and will be lost if the code is regenerated.
+# --------------------------------------------------------------------------
+
+from ._client import QuickpulseClient
+
+try:
+    from ._patch import __all__ as _patch_all
+    from ._patch import *  # pylint: disable=unused-wildcard-import
+except ImportError:
+    _patch_all = []
+from ._patch import patch_sdk as _patch_sdk
+
+__all__ = [
+    "QuickpulseClient",
+]
+__all__.extend([p for p in _patch_all if p not in __all__])
+
+_patch_sdk()

sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_quickpulse/_generated/_client.py

@@ -0,0 +1,93 @@
+# coding=utf-8
+# --------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# Code generated by Microsoft (R) AutoRest Code Generator.
+# Changes may cause incorrect behavior and will be lost if the code is regenerated.
+# --------------------------------------------------------------------------
+
+from copy import deepcopy
+from typing import Any
+
+from azure.core import PipelineClient
+from azure.core.pipeline import policies
+from azure.core.rest import HttpRequest, HttpResponse
+
+from . import models as _models
+from ._configuration import QuickpulseClientConfiguration
+from ._operations import QuickpulseClientOperationsMixin
+from ._serialization import Deserializer, Serializer
+
+
+class QuickpulseClient(QuickpulseClientOperationsMixin):  # pylint: disable=client-accepts-api-version-keyword
+    """Quickpulse Client.
+
+    :param host: QuickPulse endpoint: https://rt.services.visualstudio.com. Default value is
+     "https://rt.services.visualstudio.com".
+    :type host: str
+    """
+
+    def __init__(  # pylint: disable=missing-client-constructor-parameter-credential
+        self, host: str = "https://rt.services.visualstudio.com", **kwargs: Any
+    ) -> None:
+        _endpoint = "{Host}"
+        self._config = QuickpulseClientConfiguration(host=host, **kwargs)
+        _policies = kwargs.pop("policies", None)
+        if _policies is None:
+            _policies = [
+                policies.RequestIdPolicy(**kwargs),
+                self._config.headers_policy,
+                self._config.user_agent_policy,
+                self._config.proxy_policy,
+                policies.ContentDecodePolicy(**kwargs),
+                self._config.redirect_policy,
+                self._config.retry_policy,
+                self._config.authentication_policy,
+                self._config.custom_hook_policy,
+                self._config.logging_policy,
+                policies.DistributedTracingPolicy(**kwargs),
+                policies.SensitiveHeaderCleanupPolicy(**kwargs) if self._config.redirect_policy else None,
+                self._config.http_logging_policy,
+            ]
+        self._client: PipelineClient = PipelineClient(base_url=_endpoint, policies=_policies, **kwargs)
+
+        client_models = {k: v for k, v in _models.__dict__.items() if isinstance(v, type)}
+        self._serialize = Serializer(client_models)
+        self._deserialize = Deserializer(client_models)
+        self._serialize.client_side_validation = False
+
+    def send_request(self, request: HttpRequest, *, stream: bool = False, **kwargs: Any) -> HttpResponse:
+        """Runs the network request through the client's chained policies.
+
+        >>> from azure.core.rest import HttpRequest
+        >>> request = HttpRequest("GET", "https://www.example.org/")
+        <HttpRequest [GET], url: 'https://www.example.org/'>
+        >>> response = client.send_request(request)
+        <HttpResponse: 200 OK>
+
+        For more information on this code flow, see https://aka.ms/azsdk/dpcodegen/python/send_request
+
+        :param request: The network request you want to make. Required.
+        :type request: ~azure.core.rest.HttpRequest
+        :keyword bool stream: Whether the response payload will be streamed. Defaults to False.
+        :return: The response of your network call. Does not do error handling on your response.
+        :rtype: ~azure.core.rest.HttpResponse
+        """
+
+        request_copy = deepcopy(request)
+        path_format_arguments = {
+            "Host": self._serialize.url("self._config.host", self._config.host, "str", skip_quote=True),
+        }
+
+        request_copy.url = self._client.format_url(request_copy.url, **path_format_arguments)
+        return self._client.send_request(request_copy, stream=stream, **kwargs)  # type: ignore
+
+    def close(self) -> None:
+        self._client.close()
+
+    def __enter__(self) -> "QuickpulseClient":
+        self._client.__enter__()
+        return self
+
+    def __exit__(self, *exc_details: Any) -> None:
+        self._client.__exit__(*exc_details)

sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_quickpulse/_generated/_configuration.py

@@ -0,0 +1,45 @@
+# coding=utf-8
+# --------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# Code generated by Microsoft (R) AutoRest Code Generator.
+# Changes may cause incorrect behavior and will be lost if the code is regenerated.
+# --------------------------------------------------------------------------
+
+from typing import Any
+
+from azure.core.pipeline import policies
+
+VERSION = "unknown"
+
+
+class QuickpulseClientConfiguration:  # pylint: disable=too-many-instance-attributes,name-too-long
+    """Configuration for QuickpulseClient.
+
+    Note that all parameters used to create this instance are saved as instance
+    attributes.
+
+    :param host: QuickPulse endpoint: https://rt.services.visualstudio.com. Default value is
+     "https://rt.services.visualstudio.com".
+    :type host: str
+    """
+
+    def __init__(self, host: str = "https://rt.services.visualstudio.com", **kwargs: Any) -> None:
+        if host is None:
+            raise ValueError("Parameter 'host' must not be None.")
+
+        self.host = host
+        kwargs.setdefault("sdk_moniker", "quickpulseclient/{}".format(VERSION))
+        self.polling_interval = kwargs.get("polling_interval", 30)
+        self._configure(**kwargs)
+
+    def _configure(self, **kwargs: Any) -> None:
+        self.user_agent_policy = kwargs.get("user_agent_policy") or policies.UserAgentPolicy(**kwargs)
+        self.headers_policy = kwargs.get("headers_policy") or policies.HeadersPolicy(**kwargs)
+        self.proxy_policy = kwargs.get("proxy_policy") or policies.ProxyPolicy(**kwargs)
+        self.logging_policy = kwargs.get("logging_policy") or policies.NetworkTraceLoggingPolicy(**kwargs)
+        self.http_logging_policy = kwargs.get("http_logging_policy") or policies.HttpLoggingPolicy(**kwargs)
+        self.custom_hook_policy = kwargs.get("custom_hook_policy") or policies.CustomHookPolicy(**kwargs)
+        self.redirect_policy = kwargs.get("redirect_policy") or policies.RedirectPolicy(**kwargs)
+        self.retry_policy = kwargs.get("retry_policy") or policies.RetryPolicy(**kwargs)
+        self.authentication_policy = kwargs.get("authentication_policy")