PYCBC-1750: Wrapper SDK Observability - Remaining Management APIs

Changes
-------
* Add tracing "hooks" to all remaining management operations:
  Analytics Index, Eventing Function, Search Index, User and
  View Index management
* Update request builder and implementation layers to accept and
  propagate observability_instruments and obs_handler throughout
  all management API call paths (sync, async and Twisted)
* Fix req_to_dict in management type files to propagate
  wrapper_span_name and parent_span from obs_handler into
  the operation request dictionaries
* Fix typo: rename search_index_mgmt_imply.py to
  search_index_mgmt_impl.py in both acouchbase and txcouchbase
* Update cluster, bucket and scope constructors (sync, async and
  Twisted) to propagate observability_instruments when creating
  management objects
* Add tracing support to txcouchbase management APIs using manual
  span management with addBoth/_finish_span pattern
* Add span option to all management operation option classes
  (Analytics, Eventing, Search Index, User, Views) with docstrings
  and @overload signatures where applicable
* Add op_type_toggle param to SearchIndexManager and
  ScopeSearchIndexManager control methods (pause/resume ingest,
  allow/disallow querying, freeze/unfreeze plan) across sync,
  async and Twisted managers so OpName.from_op_type() can
  resolve the correct operation name
* Add explanatory comments to analytics managers (sync, async and
  Twisted) clarifying the arbitrary choice of
  AnalyticsLinkCreateCouchbaseRemoteLink /
  AnalyticsLinkReplaceCouchbaseRemoteLink as the op_type
* Expand test suite to cover upsert, get, get_all and drop (plus
  error cases) for Search Index and User management; confirm
  parent span propagation for all management APIs

Change-Id: Ic1b406151b79994837052e9ac97b57b6deb53f8b
Reviewed-on: https://review.couchbase.org/c/couchbase-python-client/+/241426
Tested-by: Build Bot <build@couchbase.com>
Reviewed-by: Sergey Avseyev <sergey.avseyev@gmail.com>

Author: thejcfactor

acouchbase/bucket.py

@@ -224,7 +224,7 @@ def view_indexes(self) -> ViewIndexManager:
         Returns:
             :class:`~acouchbase.management.views.ViewIndexManager`: A :class:`~couchbase.management.views.ViewIndexManager` instance.
         """  # noqa: E501
-        return ViewIndexManager(self._impl._client_adapter, self.name)
+        return ViewIndexManager(self._impl._client_adapter, self.name, self._impl.observability_instruments)
 
 
 Bucket = AsyncBucket

acouchbase/cluster.py

@@ -647,7 +647,7 @@ def users(self) -> UserManager:
         Returns:
             :class:`~acouchbase.management.users.UserManager`: A :class:`~couchbase.management.users.UserManager` instance.
         """  # noqa: E501
-        return UserManager(self._impl._client_adapter)
+        return UserManager(self._impl._client_adapter, self._impl.observability_instruments)
 
     def query_indexes(self) -> QueryIndexManager:
         """
@@ -667,7 +667,7 @@ def analytics_indexes(self) -> AnalyticsIndexManager:
         Returns:
             :class:`~acouchbase.management.analytics.AnalyticsIndexManager`: An :class:`~acouchbase.management.analytics.AnalyticsIndexManager` instance.
         """  # noqa: E501
-        return AnalyticsIndexManager(self._impl._client_adapter)
+        return AnalyticsIndexManager(self._impl._client_adapter, self._impl.observability_instruments)
 
     def search_indexes(self) -> SearchIndexManager:
         """
@@ -678,7 +678,7 @@ def search_indexes(self) -> SearchIndexManager:
             :class:`~acouchbase.management.search.SearchIndexManager`: A :class:`~acouchbase.management.search.SearchIndexManager` instance.
 
         """  # noqa: E501
-        return SearchIndexManager(self._impl._client_adapter)
+        return SearchIndexManager(self._impl._client_adapter, self._impl.observability_instruments)
 
     def eventing_functions(self) -> EventingFunctionManager:
         """
@@ -693,7 +693,7 @@ def eventing_functions(self) -> EventingFunctionManager:
             :class:`~acouchbase.management.eventing.EventingFunctionManager`: An :class:`~acouchbase.management.eventing.EventingFunctionManager` instance.
 
         """  # noqa: E501
-        return EventingFunctionManager(self._impl._client_adapter)
+        return EventingFunctionManager(self._impl._client_adapter, self._impl.observability_instruments)
 
     @staticmethod
     async def connect(connstr,  # type: str

acouchbase/management/analytics.py

@@ -22,13 +22,16 @@
                     Optional)
 
 from acouchbase.management.logic.analytics_mgmt_impl import AsyncAnalyticsMgmtImpl
+from couchbase.logic.observability import ObservableRequestHandler
+from couchbase.logic.operation_types import AnalyticsMgmtOperationType
 from couchbase.management.logic.analytics_mgmt_types import (AnalyticsDataset,
                                                              AnalyticsDataType,
                                                              AnalyticsIndex,
                                                              AnalyticsLink)
 
 if TYPE_CHECKING:
     from acouchbase.logic.client_adapter import AsyncClientAdapter
+    from couchbase.logic.observability import ObservabilityInstruments
     from couchbase.management.options import (ConnectLinkOptions,
                                               CreateAnalyticsIndexOptions,
                                               CreateDatasetOptions,
@@ -48,48 +51,61 @@
 
 class AnalyticsIndexManager:
 
-    def __init__(self, client_adapter: AsyncClientAdapter) -> None:
-        self._impl = AsyncAnalyticsMgmtImpl(client_adapter)
+    def __init__(self, client_adapter: AsyncClientAdapter, observability_instruments: ObservabilityInstruments) -> None:
+        self._impl = AsyncAnalyticsMgmtImpl(client_adapter, observability_instruments)
 
     async def create_dataverse(self,
                                dataverse_name,    # type: str
                                options=None,      # type: Optional[CreateDataverseOptions]
                                **kwargs           # type: Any
                                ) -> None:
-        req = self._impl.request_builder.build_create_dataverse_request(dataverse_name, options, **kwargs)
-        await self._impl.create_dataverse(req)
+        op_type = AnalyticsMgmtOperationType.AnalyticsDataverseCreate
+        async with ObservableRequestHandler(op_type, self._impl.observability_instruments) as obs_handler:
+            req = self._impl.request_builder.build_create_dataverse_request(
+                dataverse_name, obs_handler, options, **kwargs)
+            await self._impl.create_dataverse(req, obs_handler)
 
     async def drop_dataverse(self,
                              dataverse_name,    # type: str
                              options=None,      # type: Optional[DropDataverseOptions]
                              **kwargs           # type: Any
                              ) -> None:
-        req = self._impl.request_builder.build_drop_dataverse_request(dataverse_name, options, **kwargs)
-        await self._impl.drop_dataverse(req)
+        op_type = AnalyticsMgmtOperationType.AnalyticsDataverseDrop
+        async with ObservableRequestHandler(op_type, self._impl.observability_instruments) as obs_handler:
+            req = self._impl.request_builder.build_drop_dataverse_request(
+                dataverse_name, obs_handler, options, **kwargs)
+            await self._impl.drop_dataverse(req, obs_handler)
 
     async def create_dataset(self,
                              dataset_name,    # type: str
                              bucket_name,     # type: str
                              options=None,    # type: Optional[CreateDatasetOptions]
                              **kwargs         # type: Any
                              ) -> None:
-        req = self._impl.request_builder.build_create_dataset_request(dataset_name, bucket_name, options, **kwargs)
-        await self._impl.create_dataset(req)
+        op_type = AnalyticsMgmtOperationType.AnalyticsDatasetCreate
+        async with ObservableRequestHandler(op_type, self._impl.observability_instruments) as obs_handler:
+            req = self._impl.request_builder.build_create_dataset_request(
+                dataset_name, bucket_name, obs_handler, options, **kwargs)
+            await self._impl.create_dataset(req, obs_handler)
 
     async def drop_dataset(self,
                            dataset_name,  # type: str
                            options=None,  # type: Optional[DropDatasetOptions]
                            **kwargs       # type: Any
                            ) -> None:
-        req = self._impl.request_builder.build_drop_dataset_request(dataset_name, options, **kwargs)
-        await self._impl.drop_dataset(req)
+        op_type = AnalyticsMgmtOperationType.AnalyticsDatasetDrop
+        async with ObservableRequestHandler(op_type, self._impl.observability_instruments) as obs_handler:
+            req = self._impl.request_builder.build_drop_dataset_request(dataset_name, obs_handler, options, **kwargs)
+            await self._impl.drop_dataset(req, obs_handler)
 
     async def get_all_datasets(self,
                                options=None,   # type: Optional[GetAllDatasetOptions]
                                **kwargs   # type: Any
                                ) -> Iterable[AnalyticsDataset]:
-        req = self._impl.request_builder.build_get_all_datasets_request(options, **kwargs)
-        return await self._impl.get_all_datasets(req)
+        op_type = AnalyticsMgmtOperationType.AnalyticsDatasetGetAll
+        async with ObservableRequestHandler(op_type, self._impl.observability_instruments) as obs_handler:
+            req = self._impl.request_builder.build_get_all_datasets_request(obs_handler, options, **kwargs)
+            return await self._impl.get_all_datasets(req, obs_handler)
 
     async def create_index(self,
                            index_name,    # type: str
@@ -98,81 +114,110 @@ async def create_index(self,
                            options=None,  # type: Optional[CreateAnalyticsIndexOptions]
                            **kwargs       # type: Any
                            ) -> None:
-        req = self._impl.request_builder.build_create_index_request(index_name,
-                                                                    dataset_name,
-                                                                    fields,
-                                                                    options,
-                                                                    **kwargs)
-        await self._impl.create_index(req)
+        op_type = AnalyticsMgmtOperationType.AnalyticsIndexCreate
+        async with ObservableRequestHandler(op_type, self._impl.observability_instruments) as obs_handler:
+            req = self._impl.request_builder.build_create_index_request(index_name,
+                                                                        dataset_name,
+                                                                        fields,
+                                                                        obs_handler,
+                                                                        options,
+                                                                        **kwargs)
+            await self._impl.create_index(req, obs_handler)
 
     async def drop_index(self,
                          index_name,    # type: str
                          dataset_name,  # type: str
                          options=None,  # type: Optional[DropAnalyticsIndexOptions]
                          **kwargs       # type: Any
                          ) -> None:
-        req = self._impl.request_builder.build_drop_index_request(index_name,
-                                                                  dataset_name,
-                                                                  options,
-                                                                  **kwargs)
-        await self._impl.drop_index(req)
+        op_type = AnalyticsMgmtOperationType.AnalyticsIndexDrop
+        async with ObservableRequestHandler(op_type, self._impl.observability_instruments) as obs_handler:
+            req = self._impl.request_builder.build_drop_index_request(index_name,
+                                                                      dataset_name,
+                                                                      obs_handler,
+                                                                      options,
+                                                                      **kwargs)
+            await self._impl.drop_index(req, obs_handler)
 
     async def get_all_indexes(self,
                               options=None,   # type: Optional[GetAllAnalyticsIndexesOptions]
                               **kwargs   # type: Any
                               ) -> Iterable[AnalyticsIndex]:
-        req = self._impl.request_builder.build_get_all_indexes_request(options, **kwargs)
-        return await self._impl.get_all_indexes(req)
+        op_type = AnalyticsMgmtOperationType.AnalyticsIndexGetAll
+        async with ObservableRequestHandler(op_type, self._impl.observability_instruments) as obs_handler:
+            req = self._impl.request_builder.build_get_all_indexes_request(obs_handler, options, **kwargs)
+            return await self._impl.get_all_indexes(req, obs_handler)
 
     async def connect_link(self,
                            options=None,  # type: Optional[ConnectLinkOptions]
                            **kwargs   # type: Any
                            ) -> None:
-        req = self._impl.request_builder.build_connect_link_request(options, **kwargs)
-        await self._impl.connect_link(req)
+        op_type = AnalyticsMgmtOperationType.AnalyticsLinkConnect
+        async with ObservableRequestHandler(op_type, self._impl.observability_instruments) as obs_handler:
+            req = self._impl.request_builder.build_connect_link_request(obs_handler, options, **kwargs)
+            await self._impl.connect_link(req, obs_handler)
 
     async def disconnect_link(self,
                               options=None,  # type: Optional[DisconnectLinkOptions]
                               **kwargs   # type: Any
                               ) -> None:
-        req = self._impl.request_builder.build_disconnect_link_request(options, **kwargs)
-        await self._impl.disconnect_link(req)
+        op_type = AnalyticsMgmtOperationType.AnalyticsLinkDisconnect
+        async with ObservableRequestHandler(op_type, self._impl.observability_instruments) as obs_handler:
+            req = self._impl.request_builder.build_disconnect_link_request(obs_handler, options, **kwargs)
+            await self._impl.disconnect_link(req, obs_handler)
 
     async def get_pending_mutations(self,
                                     options=None,     # type: Optional[GetPendingMutationsOptions]
                                     **kwargs     # type: Any
                                     ) -> Dict[str, int]:
-        req = self._impl.request_builder.build_get_pending_mutations_request(options, **kwargs)
-        return await self._impl.get_pending_mutations(req)
+        op_type = AnalyticsMgmtOperationType.AnalyticsGetPendingMutations
+        async with ObservableRequestHandler(op_type, self._impl.observability_instruments) as obs_handler:
+            req = self._impl.request_builder.build_get_pending_mutations_request(obs_handler, options, **kwargs)
+            return await self._impl.get_pending_mutations(req, obs_handler)
 
     async def create_link(self,
                           link,  # type: AnalyticsLink
                           options=None,     # type: Optional[CreateLinkAnalyticsOptions]
                           **kwargs          # type: Any
                           ) -> None:
-        req = self._impl.request_builder.build_create_link_request(link, options, **kwargs)
-        await self._impl.create_link(req)
+        # We choose AnalyticsLinkCreateCouchbaseRemoteLink arbitrarily b/c the ObservableRequestHandler will
+        # translate all LinkCreate options in AnalyticsMgmtOperationType to the appropriate op name and
+        # the request builder will build the appropriate request type based on the link passed in.
+        op_type = AnalyticsMgmtOperationType.AnalyticsLinkCreateCouchbaseRemoteLink
+        async with ObservableRequestHandler(op_type, self._impl.observability_instruments) as obs_handler:
+            req = self._impl.request_builder.build_create_link_request(link, obs_handler, options, **kwargs)
+            await self._impl.create_link(req, obs_handler)
 
     async def replace_link(self,
                            link,  # type: AnalyticsLink
                            options=None,     # type: Optional[ReplaceLinkAnalyticsOptions]
                            **kwargs          # type: Any
                            ) -> None:
-        req = self._impl.request_builder.build_replace_link_request(link, options, **kwargs)
-        await self._impl.replace_link(req)
+        # We choose AnalyticsLinkReplaceCouchbaseRemoteLink arbitrarily b/c the ObservableRequestHandler will
+        # translate all LinkReplace options in AnalyticsMgmtOperationType to the appropriate op name and
+        # the request builder will build the appropriate request type based on the link passed in.
+        op_type = AnalyticsMgmtOperationType.AnalyticsLinkReplaceCouchbaseRemoteLink
+        async with ObservableRequestHandler(op_type, self._impl.observability_instruments) as obs_handler:
+            req = self._impl.request_builder.build_replace_link_request(link, obs_handler, options, **kwargs)
+            await self._impl.replace_link(req, obs_handler)
 
     async def drop_link(self,
                         link_name,  # type: str
                         dataverse_name,  # type: str
                         options=None,     # type: Optional[DropLinkAnalyticsOptions]
                         **kwargs          # type: Any
                         ) -> None:
-        req = self._impl.request_builder.build_drop_link_request(link_name, dataverse_name, options, **kwargs)
-        await self._impl.drop_link(req)
+        op_type = AnalyticsMgmtOperationType.AnalyticsLinkDrop
+        async with ObservableRequestHandler(op_type, self._impl.observability_instruments) as obs_handler:
+            req = self._impl.request_builder.build_drop_link_request(
+                link_name, dataverse_name, obs_handler, options, **kwargs)
+            await self._impl.drop_link(req, obs_handler)
 
     async def get_links(self,
                         options=None,  # type: Optional[GetLinksAnalyticsOptions]
                         **kwargs       # type: Any
                         ) -> Iterable[AnalyticsLink]:
-        req = self._impl.request_builder.build_get_links_request(options, **kwargs)
-        return await self._impl.get_links(req)
+        op_type = AnalyticsMgmtOperationType.AnalyticsLinkGetAll
+        async with ObservableRequestHandler(op_type, self._impl.observability_instruments) as obs_handler:
+            req = self._impl.request_builder.build_get_links_request(obs_handler, options, **kwargs)
+            return await self._impl.get_links(req, obs_handler)