Introduced service time metrics to OpenSearch-Py client.

CHANGELOG.md

@@ -26,6 +26,7 @@ Inspired from [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
 - Enhance generator to generate plugins ([#700](https://github.com/opensearch-project/opensearch-py/pull/700))
 - Enhance generator to update changelog only if generated code differs from existing ([#684](https://github.com/opensearch-project/opensearch-py/pull/684))
 - Added guide for configuring ssl_assert_hostname ([#694](https://github.com/opensearch-project/opensearch-py/pull/694))
+- Introduced `service time` metrics to opensearch-py client ([#716](https://github.com/opensearch-project/opensearch-py/pull/716))
 ### Changed
 - Updated the `get_policy` API in the index_management plugin to allow the policy_id argument as optional ([#633](https://github.com/opensearch-project/opensearch-py/pull/633))
 - Updated the `point_in_time.md` guide with examples demonstrating the usage of the new APIs as alternatives to the deprecated ones. ([#661](https://github.com/opensearch-project/opensearch-py/pull/661))

dev-requirements.txt

@@ -8,6 +8,7 @@ sphinx_rtd_theme
 jinja2
 pytz
 deepmerge
+Events
 
 # No wheels for Python 3.10 yet!
 numpy; python_version<"3.10"

docs/source/api-ref/metrics.md

@@ -0,0 +1,10 @@
+# metrics
+
+```{eval-rst}
+.. autoclass:: opensearchpy.metrics.metrics.Metrics
+```
+
+```{eval-rst}
+.. autoclass:: opensearchpy.metrics.metrics_events.MetricsEvents
+```
+

opensearchpy/__init__.py

@@ -133,6 +133,7 @@
 from .helpers.update_by_query import UpdateByQuery
 from .helpers.utils import AttrDict, AttrList, DslBase
 from .helpers.wrappers import Range
+from .metrics import Metrics, MetricsEvents
 from .serializer import JSONSerializer
 from .transport import Transport
 
@@ -240,6 +241,8 @@
     "token_filter",
     "tokenizer",
     "__versionstr__",
+    "Metrics",
+    "MetricsEvents",
 ]
 
 try:

opensearchpy/client/client.py

@@ -10,6 +10,7 @@
 from typing import Any, Optional, Type
 
 from opensearchpy.client.utils import _normalize_hosts
+from opensearchpy.metrics.metrics import Metrics
 from opensearchpy.transport import Transport
 
 
@@ -22,6 +23,7 @@ def __init__(
         self,
         hosts: Optional[str] = None,
         transport_class: Type[Transport] = Transport,
+        metrics: Optional[Metrics] = None,
         **kwargs: Any
     ) -> None:
         """
@@ -38,4 +40,6 @@ class as kwargs, or a string in the format of ``host[:port]`` which will be
             :class:`~opensearchpy.Transport` class and, subsequently, to the
             :class:`~opensearchpy.Connection` instances.
         """
-        self.transport = transport_class(_normalize_hosts(hosts), **kwargs)
+        self.transport = transport_class(
+            _normalize_hosts(hosts), metrics=metrics, **kwargs
+        )

opensearchpy/connection/http_requests.py

@@ -36,6 +36,8 @@
 except ImportError:
     REQUESTS_AVAILABLE = False
 
+from opensearchpy.metrics.metrics import Metrics
+
 from ..compat import reraise_exceptions, string_types, urlencode
 from ..exceptions import (
     ConnectionError,
@@ -86,8 +88,10 @@
         http_compress: Any = None,
         opaque_id: Any = None,
         pool_maxsize: Any = None,
+        metrics: Optional[Metrics] = None,
         **kwargs: Any
     ) -> None:
+        self.metrics = metrics
         if not REQUESTS_AVAILABLE:
             raise ImproperlyConfigured(
                 "Please install requests to use RequestsHttpConnection."
@@ -188,6 +192,8 @@
         }
         send_kwargs.update(settings)
         try:
+            if self.metrics is not None:
+                self.metrics.request_start()
             response = self.session.send(prepared_request, **send_kwargs)
             duration = time.time() - start
             raw_data = response.content.decode("utf-8", "surrogatepass")
@@ -207,6 +213,9 @@
             if isinstance(e, requests.Timeout):
                 raise ConnectionTimeout("TIMEOUT", str(e), e)
             raise ConnectionError("N/A", str(e), e)
+        finally:
+            if self.metrics is not None:
+                self.metrics.request_end()
 
         # raise warnings if any from the 'Warnings' header.
         warnings_headers = (

opensearchpy/connection/http_urllib3.py

@@ -34,6 +34,8 @@
 from urllib3.exceptions import SSLError as UrllibSSLError
 from urllib3.util.retry import Retry
 
+from opensearchpy.metrics.metrics import Metrics
+
 from ..compat import reraise_exceptions, urlencode
 from ..exceptions import (
     ConnectionError,
@@ -115,8 +117,10 @@
         ssl_context: Any = None,
         http_compress: Any = None,
         opaque_id: Any = None,
+        metrics: Optional[Metrics] = None,
         **kwargs: Any
     ) -> None:
+        self.metrics = metrics
         # Initialize headers before calling super().__init__().
         self.headers = urllib3.make_headers(keep_alive=True)
 
@@ -268,6 +272,9 @@
                 if isinstance(self.http_auth, Callable):  # type: ignore
                     request_headers.update(self.http_auth(method, full_url, body))
 
+            if self.metrics is not None:
+                self.metrics.request_start()
+
             response = self.pool.urlopen(
                 method, url, body, retries=Retry(False), headers=request_headers, **kw
             )
@@ -284,6 +291,9 @@
             if isinstance(e, ReadTimeoutError):
                 raise ConnectionTimeout("TIMEOUT", str(e), e)
             raise ConnectionError("N/A", str(e), e)
+        finally:
+            if self.metrics is not None:
+                self.metrics.request_end()
 
         # raise warnings if any from the 'Warnings' header.
         warning_headers = response.headers.get_all("warning", ())

opensearchpy/metrics/__init__.py

@@ -0,0 +1,16 @@
+# SPDX-License-Identifier: Apache-2.0
+#
+# The OpenSearch Contributors require contributions made to
+# this file be licensed under the Apache-2.0 license or a
+# compatible open source license.
+#
+# Modifications Copyright OpenSearch Contributors. See
+# GitHub history for details.
+
+from .metrics import Metrics
+from .metrics_events import MetricsEvents
+
+__all__ = [
+    "Metrics",
+    "MetricsEvents",
+]

opensearchpy/metrics/metrics.py

@@ -0,0 +1,42 @@
+# SPDX-License-Identifier: Apache-2.0
+#
+# The OpenSearch Contributors require contributions made to
+# this file be licensed under the Apache-2.0 license or a
+# compatible open source license.
+#
+# Modifications Copyright OpenSearch Contributors. See
+# GitHub history for details.
+
+from abc import ABC, abstractmethod
+from typing import Optional
+
+
+class Metrics(ABC):
+    """
+    The Metrics class defines methods and properties for managing
+    request metrics, including start time, end time, and service time,
+    serving as a blueprint for concrete implementations.
+    """
+
+    @abstractmethod
+    def request_start(self) -> None:
+        pass
+
+    @abstractmethod
+    def request_end(self) -> None:
+        pass
+
+    @property
+    @abstractmethod
+    def start_time(self) -> Optional[float]:
+        pass
+
+    @property
+    @abstractmethod
+    def end_time(self) -> Optional[float]:
+        pass
+
+    @property
+    @abstractmethod
+    def service_time(self) -> Optional[float]:
+        pass

opensearchpy/metrics/metrics_events.py

@@ -0,0 +1,61 @@
+# SPDX-License-Identifier: Apache-2.0
+#
+# The OpenSearch Contributors require contributions made to
+# this file be licensed under the Apache-2.0 license or a
+# compatible open source license.
+#
+# Modifications Copyright OpenSearch Contributors. See
+# GitHub history for details.
+
+import time
+from typing import Optional
+
+from events import Events
+
+from opensearchpy.metrics.metrics import Metrics
+
+
+class MetricsEvents(Metrics):
+    """
+    The MetricsEvents class implements the Metrics abstract base class
+    and tracks metrics such as start time, end time, and service time
+    during request processing.
+    """
+
+    @property
+    def start_time(self) -> Optional[float]:
+        return self._start_time
+
+    @property
+    def end_time(self) -> Optional[float]:
+        return self._end_time
+
+    @property
+    def service_time(self) -> Optional[float]:
+        return self._service_time
+
+    def __init__(self) -> None:
+        self.events = Events()
+        self._start_time: Optional[float] = None
+        self._end_time: Optional[float] = None
+        self._service_time: Optional[float] = None
+
+        # Subscribe to the request_start and request_end events
+        self.events.request_start += self._on_request_start
+        self.events.request_end += self._on_request_end
+
+    def request_start(self) -> None:
+        self.events.request_start()
+
+    def _on_request_start(self) -> None:
+        self._start_time = time.perf_counter()
+        self._end_time = None
+        self._service_time = None
+
+    def request_end(self) -> None:
+        self.events.request_end()
+
+    def _on_request_end(self) -> None:
+        self._end_time = time.perf_counter()
+        if self._start_time is not None:
+            self._service_time = self._end_time - self._start_time

opensearchpy/transport.py

@@ -29,6 +29,8 @@
 from itertools import chain
 from typing import Any, Callable, Collection, Dict, List, Mapping, Optional, Type, Union
 
+from opensearchpy.metrics.metrics import Metrics
+
 from .connection import Connection, Urllib3HttpConnection
 from .connection_pool import ConnectionPool, DummyConnectionPool, EmptyConnectionPool
 from .exceptions import (
@@ -91,6 +93,7 @@ class Transport(object):
     last_sniff: float
     sniff_timeout: Optional[float]
     host_info_callback: Any
+    metrics: Optional[Metrics]
 
     def __init__(
         self,
@@ -112,6 +115,7 @@ def __init__(
         retry_on_status: Collection[int] = (502, 503, 504),
         retry_on_timeout: bool = False,
         send_get_body_as: str = "GET",
+        metrics: Optional[Metrics] = None,
         **kwargs: Any
     ) -> None:
         """
@@ -153,6 +157,7 @@ def __init__(
         when creating and instance unless overridden by that connection's
         options provided as part of the hosts parameter.
         """
+        self.metrics = metrics
         if connection_class is None:
             connection_class = self.DEFAULT_CONNECTION_CLASS
 
@@ -242,7 +247,7 @@ def _create_connection(host: Any) -> Any:
             kwargs.update(host)
             if self.pool_maxsize and isinstance(self.pool_maxsize, int):
                 kwargs["pool_maxsize"] = self.pool_maxsize
-            return self.connection_class(**kwargs)
+            return self.connection_class(metrics=self.metrics, **kwargs)
 
         connections = list(zip(map(_create_connection, hosts), hosts))
         if len(connections) == 1:

setup.py

@@ -59,6 +59,7 @@
     "six",
     "python-dateutil",
     "certifi>=2022.12.07",
+    "Events",
 ]
 tests_require = [
     "requests>=2.0.0, <3.0.0",