PYCBC-1721: Wrapper SDK Observability - Tracing Baseline

Changes
-------
* Add tracing "hooks" to all KV and streaming operations
* Add tracing "hooks" to all bucket, collection and query index
  management operations
* Add appropriate observability (RequestSpan, RequestTracer, tags, etc.)
  logic to align with the Extended Observability RFC
* Add tests to confirm various tracer (legacy, modern, no-op and
  threshold) behavior across baseline operations

Change-Id: I4b1eda52324d0314f95eb6631b4ec3015cff23c2
Reviewed-on: https://review.couchbase.org/c/couchbase-python-client/+/241023
Tested-by: Build Bot <build@couchbase.com>
Reviewed-by: Brett Lawson <brett19@gmail.com>

Author: thejcfactor

acouchbase/analytics.py

@@ -81,29 +81,42 @@ def _get_next_row(self):
         if self.done_streaming is True:
             return
 
+        # this is a blocking operation
         row = next(self._streaming_result)
         if isinstance(row, PycbcCoreException):
             raise ErrorMapper.build_exception(row)
-        # should only be None one query request is complete and _no_ errors found
+
+        # should only be None onc query request is complete and _no_ errors found
         if row is None:
             raise StopAsyncIteration
+
         # this should allow the event loop to pick up something else
         return self.serializer.deserialize(row)
 
     async def __anext__(self):
         try:
-            return await self._loop.run_in_executor(self._tp_executor, self._get_next_row)
+            row = await self._loop.run_in_executor(self._tp_executor, self._get_next_row)
+            # We want to end the streaming op span once we have a response from the C++ core.
+            # Unfortunately right now, that means we need to wait until we have the first row (or we have an error).
+            # As this method is idempotent, it is safe to call for each row (it will only do work for the first call).
+            self._process_core_span()
+            return row
         except asyncio.QueueEmpty:
+            self._process_core_span(with_error=True)
             exc_cls = PYCBC_ERROR_MAP.get(ExceptionMap.InternalSDKException.value, CouchbaseException)
             excptn = exc_cls('Unexpected QueueEmpty exception caught when doing Analytics query.')
             raise excptn
         except StopAsyncIteration:
             self._done_streaming = True
+            if self._processed_core_span is False:
+                self._process_core_span()
             self._get_metadata()
             raise
         except CouchbaseException as ex:
+            self._process_core_span(with_error=True)
             raise ex
         except Exception as ex:
+            self._process_core_span(with_error=True)
             exc_cls = PYCBC_ERROR_MAP.get(ExceptionMap.InternalSDKException.value, CouchbaseException)
             excptn = exc_cls(str(ex))
             raise excptn

acouchbase/binary_collection.py

@@ -20,6 +20,8 @@
                     Any,
                     Union)
 
+from couchbase.logic.observability import ObservableRequestHandler
+from couchbase.logic.operation_types import KeyValueOperationType
 from couchbase.result import CounterResult, MutationResult
 
 if TYPE_CHECKING:
@@ -80,8 +82,10 @@ async def increment(self,
                 print(f'Counter value: {res.content}')
 
         """
-        req = self._impl.request_builder.build_increment_request(key, *opts, **kwargs)
-        return await self._impl.increment(req)
+        op_type = KeyValueOperationType.Increment
+        async with ObservableRequestHandler(op_type, self._impl.observability_instruments) as obs_handler:
+            req = self._impl.request_builder.build_increment_request(key, obs_handler, *opts, **kwargs)
+            return await self._impl.increment(req, obs_handler)
 
     async def decrement(self,
                         key,  # type: str
@@ -128,8 +132,10 @@ async def decrement(self,
                 print(f'Counter value: {res.content}')
 
         """
-        req = self._impl.request_builder.build_decrement_request(key, *opts, **kwargs)
-        return await self._impl.decrement(req)
+        op_type = KeyValueOperationType.Decrement
+        async with ObservableRequestHandler(op_type, self._impl.observability_instruments) as obs_handler:
+            req = self._impl.request_builder.build_decrement_request(key, obs_handler, *opts, **kwargs)
+            return await self._impl.decrement(req, obs_handler)
 
     async def append(self,
                      key,  # type: str
@@ -184,8 +190,10 @@ async def append(self,
                                                         AppendOptions(timeout=timedelta(seconds=2)))
 
         """
-        req = self._impl.request_builder.build_append_request(key, value, *opts, **kwargs)
-        return await self._impl.append(req)
+        op_type = KeyValueOperationType.Append
+        async with ObservableRequestHandler(op_type, self._impl.observability_instruments) as obs_handler:
+            req = self._impl.request_builder.build_append_request(key, value, obs_handler, *opts, **kwargs)
+            return await self._impl.append(req, obs_handler)
 
     async def prepend(self,
                       key,  # type: str
@@ -240,5 +248,7 @@ async def prepend(self,
                                                         PrependOptions(timeout=timedelta(seconds=2)))
 
         """
-        req = self._impl.request_builder.build_prepend_request(key, value, *opts, **kwargs)
-        return await self._impl.prepend(req)
+        op_type = KeyValueOperationType.Prepend
+        async with ObservableRequestHandler(op_type, self._impl.observability_instruments) as obs_handler:
+            req = self._impl.request_builder.build_prepend_request(key, value, obs_handler, *opts, **kwargs)
+            return await self._impl.prepend(req, obs_handler)

acouchbase/bucket.py

@@ -24,6 +24,8 @@
 from acouchbase.management.collections import CollectionManager
 from acouchbase.management.views import ViewIndexManager
 from acouchbase.scope import Scope
+from couchbase.logic.observability import ObservableRequestHandler
+from couchbase.logic.operation_types import StreamingOperationType
 from couchbase.result import PingResult, ViewResult
 
 if TYPE_CHECKING:
@@ -189,7 +191,13 @@ def view_query(self,
                     print(f'Found row: {row}')
 
         """
-        req = self._impl.request_builder.build_view_query_request(design_doc, view_name, *view_options, **kwargs)
+        op_type = StreamingOperationType.ViewQuery
+        obs_handler = ObservableRequestHandler(op_type, self._impl.observability_instruments)
+        req = self._impl.request_builder.build_view_query_request(design_doc,
+                                                                  view_name,
+                                                                  obs_handler,
+                                                                  *view_options,
+                                                                  **kwargs)
         return self._impl.view_query(req)
 
     def collections(self) -> CollectionManager:
@@ -200,7 +208,7 @@ def collections(self) -> CollectionManager:
         Returns:
             :class:`~acouchbase.management.collections.CollectionManager`: A :class:`~couchbase.management.collections.CollectionManager` instance.
         """  # noqa: E501
-        return CollectionManager(self._impl._client_adapter, self.name)
+        return CollectionManager(self._impl._client_adapter, self.name, self._impl.observability_instruments)
 
     def view_indexes(self) -> ViewIndexManager:
         """

acouchbase/cluster.py

@@ -35,6 +35,8 @@
 from couchbase.auth import (CertificateAuthenticator,
                             JwtAuthenticator,
                             PasswordAuthenticator)
+from couchbase.logic.observability import ObservableRequestHandler
+from couchbase.logic.operation_types import StreamingOperationType
 from couchbase.result import (AnalyticsResult,
                               ClusterInfoResult,
                               DiagnosticsResult,
@@ -352,7 +354,9 @@ def query(self,
                 print(f'Query metrics: {q_res.metadata().metrics()}')
 
         """
-        req = self._impl.request_builder.build_query_request(statement, *options, **kwargs)
+        op_type = StreamingOperationType.Query
+        obs_handler = ObservableRequestHandler(op_type, self._impl.observability_instruments)
+        req = self._impl.request_builder.build_query_request(statement, obs_handler, *options, **kwargs)
         return self._impl.query(req)
 
     def analytics_query(self,  # type: Cluster
@@ -428,7 +432,9 @@ def analytics_query(self,  # type: Cluster
                 print(f'Analytics query metrics: {q_res.metadata().metrics()}')
 
         """  # noqa: E501
-        req = self._impl.request_builder.build_analytics_query_request(statement, *options, **kwargs)
+        op_type = StreamingOperationType.AnalyticsQuery
+        obs_handler = ObservableRequestHandler(op_type, self._impl.observability_instruments)
+        req = self._impl.request_builder.build_analytics_query_request(statement, obs_handler, *options, **kwargs)
         return self._impl.analytics_query(req)
 
     def search_query(self,
@@ -524,7 +530,9 @@ def search_query(self,
                     print(f'Locations: {row.locations}')
 
         """
-        req = self._impl.request_builder.build_search_request(index, query, *options, **kwargs)
+        op_type = StreamingOperationType.SearchQuery
+        obs_handler = ObservableRequestHandler(op_type, self._impl.observability_instruments)
+        req = self._impl.request_builder.build_search_request(index, query, obs_handler, *options, **kwargs)
         return self._impl.search(req)
 
     def search(self,
@@ -616,7 +624,9 @@ def search(self,
                 async for row in q_res.rows():
                     print(f'Found row: {row}')
         """  # noqa: E501
-        req = self._impl.request_builder.build_search_request(index, request, *options, **kwargs)
+        op_type = StreamingOperationType.SearchQuery
+        obs_handler = ObservableRequestHandler(op_type, self._impl.observability_instruments)
+        req = self._impl.request_builder.build_search_request(index, request, obs_handler, *options, **kwargs)
         return self._impl.search(req)
 
     def buckets(self) -> BucketManager:
@@ -627,7 +637,7 @@ def buckets(self) -> BucketManager:
         Returns:
             :class:`~acouchbase.management.buckets.BucketManager`: A :class:`~acouchbase.management.buckets.BucketManager` instance.
         """  # noqa: E501
-        return BucketManager(self._impl._client_adapter)
+        return BucketManager(self._impl._client_adapter, self._impl.observability_instruments)
 
     def users(self) -> UserManager:
         """
@@ -647,7 +657,7 @@ def query_indexes(self) -> QueryIndexManager:
         Returns:
             :class:`~acouchbase.management.queries.QueryIndexManager`: A :class:`~acouchbase.management.queries.QueryIndexManager` instance.
         """  # noqa: E501
-        return QueryIndexManager(self._impl._client_adapter)
+        return QueryIndexManager(self._impl._client_adapter, self._impl.observability_instruments)
 
     def analytics_indexes(self) -> AnalyticsIndexManager:
         """