add telemetry to plugin

Author: joyc-bq

Diff for: aws_advanced_python_wrapper/fastest_response_strategy_plugin.py

@@ -34,6 +34,9 @@
                                                           WrapperProperties)
 from aws_advanced_python_wrapper.utils.sliding_expiration_cache import \
     SlidingExpirationCacheWithCleanupThread
+from aws_advanced_python_wrapper.utils.telemetry.telemetry import (
+    TelemetryContext, TelemetryCounter, TelemetryFactory, TelemetryGauge,
+    TelemetryTraceLevel)
 
 if TYPE_CHECKING:
     from aws_advanced_python_wrapper.driver_dialect import DriverDialect
@@ -160,6 +163,7 @@ def __init__(self, plugin_service: PluginService, host_info: HostInfo, props: Pr
         self._properties = props
         self._interval_ms = interval_ms
 
+        self._telemetry_factory = self._plugin_service.get_telemetry_factory()
         self._response_time: int = 0
         self._check_timestamp = datetime.now()
         self._lock: Lock = Lock()
@@ -172,6 +176,11 @@ def __init__(self, plugin_service: PluginService, host_info: HostInfo, props: Pr
 
         self._executor: Executor = ThreadPoolExecutor(thread_name_prefix="HostResponseTimeMonitorExecutor")
 
+        # Report current response time (in milliseconds) to telemetry engine.
+        # Report -1 if response time couldn't be measured.
+        self._response_time_gauge = self._telemetry_factory.create_gauge("frt.response.time." + self._host_id,
+                                                                         lambda: self._response_time if self._response_time != 2 ^ 31 else -1)
+
     @property
     def response_time(self):
         return self._response_time
@@ -206,6 +215,9 @@ def _get_current_time(self):
         return time()
 
     def run(self):
+        context = self._telemetry_factory.open_telemetry_context(
+            "node response time thread", TelemetryTraceLevel.TOP_LEVEL)
+        context.set_attribute("url", self._host_info.url)
         try:
             while not self.is_stopped:
                 try:
@@ -249,6 +261,8 @@ def run(self):
                 except Exception:
                     # Do nothing
                     pass
+            if context is not None:
+                context.close_context()
 
     def _open_connection(self):
         try:
@@ -282,8 +296,8 @@ def get_response_time(self, host_info: HostInfo) -> int:
         Return a response time in milliseconds to the host.
         Return 2 ^ 31 if response time is not available.
 
-        @param hostSpec the host details
-        @return response time in milliseconds for a desired host. It should return 2 ^ 31
+        :param host_info: the host details
+        :return: response time in milliseconds for a desired host. It should return 2 ^ 31
         if response time couldn't be measured.
         """
         ...
@@ -307,6 +321,8 @@ def __init__(self, plugin_service: PluginService, props: Properties, interval_ms
         self._properties = props
         self._interval_ms = interval_ms
         self._hosts: Tuple[HostInfo, ...] = ()
+        self._telemetry_factory = self._plugin_service.get_telemetry_factory()
+        self._node_count_gauge = self._telemetry_factory.create_gauge("frt.nodes.count", lambda: len(self._monitoring_nodes))
 
     def get_response_time(self, host_info: HostInfo) -> int:
         monitor: Optional[HostResponseTimeMonitor] = HostResponseTimeServiceImpl._monitoring_nodes.get(host_info.url)