Update live metrics to use typespec generated swagger (#34840)

Author: lzchen

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

@@ -1,5 +1,6 @@
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # Licensed under the MIT License.
+import logging
 from typing import Any, Optional
 
 from opentelemetry.context import (
@@ -43,6 +44,9 @@
 from azure.monitor.opentelemetry.exporter._utils import _ticks_since_dot_net_epoch, PeriodicTask
 
 
+_logger = logging.getLogger(__name__)
+
+
 _QUICKPULSE_METRIC_TEMPORALITIES = {
     # Use DELTA temporalities because we want to reset the counts every collection interval
     Counter: AggregationTemporality.DELTA,
@@ -77,8 +81,8 @@ def __init__(self, connection_string: Optional[str]) -> None:
         self._live_endpoint = parsed_connection_string.live_endpoint
         self._instrumentation_key = parsed_connection_string.instrumentation_key
         # TODO: Support AADaudience (scope)/credentials
-
-        self._client = QuickpulseClient(host=self._live_endpoint)
+        # Pass `None` for now until swagger definition is fixed
+        self._client = QuickpulseClient(credential=None, endpoint=self._live_endpoint)  # type: ignore
         # TODO: Support redirect
 
         MetricExporter.__init__(
@@ -113,10 +117,11 @@ def export(
 
         token = attach(set_value(_SUPPRESS_INSTRUMENTATION_KEY, True))
         try:
-            post_response = self._client.post(  # type: ignore
+            post_response = self._client.publish(  # type: ignore
+                endpoint=self._live_endpoint,
                 monitoring_data_points=data_points,
                 ikey=self._instrumentation_key,
-                x_ms_qps_transmission_time=_ticks_since_dot_net_epoch(),
+                transmission_time=_ticks_since_dot_net_epoch(),
                 cls=_Response,
             )
             if not post_response:
@@ -128,7 +133,7 @@ def export(
                     # User leaving the live metrics page will be treated as an unsuccessful
                     result = MetricExportResult.FAILURE
         except Exception:  # pylint: disable=broad-except,invalid-name
-            # Errors are not reported and assumed as unsuccessful
+            _logger.exception("Exception occurred while publishing live metrics.")
             result = MetricExportResult.FAILURE
         finally:
             detach(token)
@@ -164,20 +169,25 @@ def shutdown(
         """
 
 
-    def _ping(self, monitoring_data_point) -> Optional[_Response]:
+    def _ping(self, monitoring_data_point: MonitoringDataPoint) -> Optional[_Response]:
         ping_response = None
         token = attach(set_value(_SUPPRESS_INSTRUMENTATION_KEY, True))
         try:
-            ping_response = self._client.ping(  # type: ignore
+            ping_response = self._client.is_subscribed(  # type: ignore
+                endpoint=self._live_endpoint,
                 monitoring_data_point=monitoring_data_point,
                 ikey=self._instrumentation_key,
-                x_ms_qps_transmission_time=_ticks_since_dot_net_epoch(),
+                transmission_time=_ticks_since_dot_net_epoch(),
+                machine_name=monitoring_data_point.machine_name,
+                instance_name=monitoring_data_point.instance,
+                stream_id=monitoring_data_point.stream_id,
+                role_name=monitoring_data_point.role_name,
+                invariant_version=monitoring_data_point.invariant_version,
                 cls=_Response,
             )
             return ping_response  # type: ignore
         except HttpResponseError:
-            # Errors are not reported
-            pass
+            _logger.exception("Exception occurred while pinging live metrics.")
         detach(token)
         return ping_response
 
@@ -208,41 +218,35 @@ def _ticker(self) -> None:
         if _is_ping_state():
             # Send a ping if elapsed number of request meets the threshold
             if self._elapsed_num_seconds % _get_global_quickpulse_state().value == 0:
-                print("pinging...")
                 ping_response = self._exporter._ping(  # pylint: disable=protected-access
                     self._base_monitoring_data_point,
                 )
                 if ping_response:
                     header = ping_response._response_headers.get("x-ms-qps-subscribed")  # pylint: disable=protected-access
                     if header and header == "true":
-                        print("ping succeeded: switching to post")
                         # Switch state to post if subscribed
                         _set_global_quickpulse_state(_QuickpulseState.POST_SHORT)
                         self._elapsed_num_seconds = 0
                     else:
                         # Backoff after _LONG_PING_INTERVAL_SECONDS (60s) of no successful requests
                         if _get_global_quickpulse_state() is _QuickpulseState.PING_SHORT and \
                             self._elapsed_num_seconds >= _LONG_PING_INTERVAL_SECONDS:
-                            print("ping failed for 60s, switching to pinging every 60s")
                             _set_global_quickpulse_state(_QuickpulseState.PING_LONG)
                 # TODO: Implement redirect
                 else:
                     # Erroneous ping responses instigate backoff logic
                     # Backoff after _LONG_PING_INTERVAL_SECONDS (60s) of no successful requests
                     if _get_global_quickpulse_state() is _QuickpulseState.PING_SHORT and \
                         self._elapsed_num_seconds >= _LONG_PING_INTERVAL_SECONDS:
-                        print("ping failed for 60s, switching to pinging every 60s")
                         _set_global_quickpulse_state(_QuickpulseState.PING_LONG)
         else:
-            print("posting...")
             try:
                 self.collect()
             except _UnsuccessfulQuickPulsePostError:
                 # Unsuccessful posts instigate backoff logic
                 # Backoff after _POST_CANCEL_INTERVAL_SECONDS (20s) of no successful requests
                 # And resume pinging
                 if self._elapsed_num_seconds >= _POST_CANCEL_INTERVAL_SECONDS:
-                    print("post failed for 20s, switching to pinging")
                     _set_global_quickpulse_state(_QuickpulseState.PING_SHORT)
                     self._elapsed_num_seconds = 0
 

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

@@ -7,7 +7,7 @@
 # --------------------------------------------------------------------------
 
 from copy import deepcopy
-from typing import Any
+from typing import Any, TYPE_CHECKING
 
 from azure.core import PipelineClient
 from azure.core.pipeline import policies
@@ -18,20 +18,24 @@
 from ._operations import QuickpulseClientOperationsMixin
 from ._serialization import Deserializer, Serializer
 
+if TYPE_CHECKING:
+    # pylint: disable=unused-import,ungrouped-imports
+    from azure.core.credentials import TokenCredential
+
 
 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
+    :param credential: Credential needed for the client to connect to Azure. Required.
+    :type credential: ~azure.core.credentials.TokenCredential
+    :keyword api_version: Api Version. Default value is "2024-04-01-preview". Note that overriding
+     this default value may result in unsupported behavior.
+    :paramtype api_version: 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)
+    def __init__(self, credential: "TokenCredential", **kwargs: Any) -> None:
+        _endpoint = "{endpoint}"
+        self._config = QuickpulseClientConfiguration(credential=credential, **kwargs)
         _policies = kwargs.pop("policies", None)
         if _policies is None:
             _policies = [
@@ -75,11 +79,7 @@ def send_request(self, request: HttpRequest, *, stream: bool = False, **kwargs:
         """
 
         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)
+        request_copy.url = self._client.format_url(request_copy.url)
         return self._client.send_request(request_copy, stream=stream, **kwargs)  # type: ignore
 
     def close(self) -> None:

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

@@ -6,10 +6,14 @@
 # Changes may cause incorrect behavior and will be lost if the code is regenerated.
 # --------------------------------------------------------------------------
 
-from typing import Any
+from typing import Any, TYPE_CHECKING
 
 from azure.core.pipeline import policies
 
+if TYPE_CHECKING:
+    # pylint: disable=unused-import,ungrouped-imports
+    from azure.core.credentials import TokenCredential
+
 VERSION = "unknown"
 
 
@@ -19,16 +23,22 @@ class QuickpulseClientConfiguration:  # pylint: disable=too-many-instance-attrib
     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
+    :param credential: Credential needed for the client to connect to Azure. Required.
+    :type credential: ~azure.core.credentials.TokenCredential
+    :keyword api_version: Api Version. Default value is "2024-04-01-preview". Note that overriding
+     this default value may result in unsupported behavior.
+    :paramtype api_version: 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.")
+    def __init__(self, credential: "TokenCredential", **kwargs: Any) -> None:
+        api_version: str = kwargs.pop("api_version", "2024-04-01-preview")
+
+        # if credential is None:
+        #     raise ValueError("Parameter 'credential' must not be None.")
 
-        self.host = host
+        self.credential = credential
+        self.api_version = api_version
+        self.credential_scopes = kwargs.pop("credential_scopes", ["https://monitor.azure.com/.default"])
         kwargs.setdefault("sdk_moniker", "quickpulseclient/{}".format(VERSION))
         self.polling_interval = kwargs.get("polling_interval", 30)
         self._configure(**kwargs)
@@ -43,3 +53,7 @@ def _configure(self, **kwargs: Any) -> None:
         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")
+        # if self.credential and not self.authentication_policy:
+        #     self.authentication_policy = policies.BearerTokenCredentialPolicy(
+        #         self.credential, *self.credential_scopes, **kwargs
+        #     )