PYCBC-1720: Wrapper SDK Observability Improvments - OTel Integration

Changes
-------
* Add entry point for OTel tracing and metrics
* Update SDK to set start and end time for spans/metrics when it is in
  control of the start/end time
* Update ObservabilityHandler to take into account clock skew between
  Node and underlying C++ core when handling start/end of core spans and
  handling end of request span
* Add Wrapper classes for OTel integration with the Couchbase
  observability APIs
* Add optional dependency group "otel" to enable `python -m pip install
  couchbase[otel]` which will also install:
  * opentelemetry-api~=1.22
  * opentelemetry-sdk~=1.22
* Add observability examples

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

Author: thejcfactor

couchbase/logic/n1ql.py

@@ -295,8 +295,9 @@ def metrics(self) -> Optional[QueryMetrics]:
         Returns:
             Optional[:class:`.QueryMetrics`]: A :class:`.QueryMetrics` instance.
         """
-        if "metrics" in self._raw:
-            return QueryMetrics(self._raw.get("metrics", {}))
+        raw_metrics = self._raw.get('metrics', None)
+        if raw_metrics:
+            return QueryMetrics(raw_metrics)
         return None
 
     def profile(self) -> Optional[JSONType]:

couchbase/logic/observability/handler.py

@@ -124,6 +124,10 @@ def get_attributes_for_http_op(op_name: str,
     return attrs
 
 
+def get_latest_time(time_a_ns: int, time_b_ns: int) -> int:
+    return max(time_a_ns, time_b_ns)
+
+
 class ObservableRequestHandler:
 
     def __init__(self,
@@ -486,7 +490,7 @@ def _process_end(self,
         if exc_val is not None:
             self._handle_error(attrs, exc_val)
 
-        self._meter.value_recorder(OpAttributeName.MeterOperationDuration,
+        self._meter.value_recorder(OpAttributeName.MeterOperationDuration.value,
                                    tags=attrs).record_value(self._to_microseconds(req_duration_ns))
 
     def _to_microseconds(self, duration_ns: int) -> int:
@@ -642,6 +646,7 @@ def __init__(self,
         self._op_name = op_name
         self._wrapped_tracer = wrapped_tracer
         self._collection_details = options.get('collection_details', None)
+        self._start_time = start_time or time.time_ns()
         # The request_span should _only_ take a SpanProtocol for the parent
         self._parent_span = options.get('parent_span', None)
         if isinstance(self._parent_span, WrappedSpan):
@@ -650,13 +655,14 @@ def __init__(self,
             p_span = self._parent_span
         self._request_span = self._create_request_span(self._op_name.value,
                                                        parent_span=p_span,
-                                                       start_time=start_time)
+                                                       start_time=self._start_time)
         self._has_multiple_encoding_spans = (self._op_name.is_multi_op() or self._op_name is OpName.MutateIn)
         self._encoding_spans_ended = False
         self._encoding_spans: Optional[Union[WrappedEncodingSpan, List[WrappedEncodingSpan]]] = None
         self._set_span_attrs(**options)
         self._cluster_name: Optional[str] = None
         self._cluster_uuid: Optional[str] = None
+        self._end_time_watermark = time.time_ns()
 
     @property
     def cluster_name(self) -> Optional[str]:
@@ -717,6 +723,9 @@ def maybe_create_encoding_span(self, encoding_fn: Callable[..., Tuple[bytes, int
             # core span so that we can add the cluster_[name|uuid] attributes
             self._encoding_spans = WrappedEncodingSpan(encoding_span, time.time_ns())
 
+    def maybe_update_end_time_watermark(self, end_time: int) -> None:
+        self._end_time_watermark = get_latest_time(self._end_time_watermark, end_time)
+
     def process_core_span(self, core_span: CppWrapperSdkSpan) -> None:
         self._maybe_set_attribute_from_core_span(core_span, CppOpAttributeName.ClusterName)
         self._maybe_set_attribute_from_core_span(core_span, CppOpAttributeName.ClusterUUID)
@@ -749,12 +758,16 @@ def set_status(self, status: SpanStatusCode) -> None:
         self._status = status
         self._request_span.set_status(status)
 
-    def end(self, end_time: Optional[int]) -> None:
+    def end(self, end_time: int) -> None:
         self._end_encoding_spans()
         if self._wrapped_tracer.is_legacy:
             self._request_span.finish()
-        else:
-            self._request_span.end(end_time)
+            return
+
+        self._end_time_watermark = get_latest_time(self._end_time_watermark, end_time)
+        if self._parent_span and isinstance(self._parent_span, WrappedSpan):
+            self._parent_span.maybe_update_end_time_watermark(self._end_time_watermark)
+        self._request_span.end(self._end_time_watermark)
 
     def _build_core_spans(self,
                           core_spans: List[CppWrapperSdkChildSpan],
@@ -768,44 +781,48 @@ def _build_core_spans(self,
     def _build_dispatch_core_span(self,
                                   core_span: CppWrapperSdkChildSpan,
                                   parent_span: Optional[Union[SpanProtocol, WrappedSpan]] = None) -> None:
-        # TODO: handle parent span as WrappedSpan
         if isinstance(parent_span, WrappedSpan):
             p_span = parent_span.request_span
         else:
             p_span = parent_span
-        new_span = self._create_request_span(core_span['name'], parent_span=p_span, start_time=core_span['start'])
+        latest_start_time = get_latest_time(self._start_time, core_span['start'])
+        new_span = self._create_request_span(core_span['name'], parent_span=p_span, start_time=latest_start_time)
         children = core_span.get('children', None)
         if children:
             self._build_core_spans(children, parent_span=new_span)
 
         for attr_name, attr_val in core_span.get('attributes', {}).items():
             new_span.set_attribute(attr_name, attr_val)
 
-        new_span.end(core_span['end'])
+        core_span_end_time = core_span['end']
+        self._end_time_watermark = get_latest_time(self._end_time_watermark, core_span_end_time)
+        new_span.end(core_span_end_time)
 
     def _build_non_dispatch_core_span(self,
                                       core_span: CppWrapperSdkChildSpan,
                                       parent_span: Optional[Union[SpanProtocol, WrappedSpan]] = None) -> None:
-        # TODO: handle parent span as WrappedSpan
         if isinstance(parent_span, WrappedSpan):
             p_span = parent_span.request_span
         else:
             p_span = parent_span
 
+        latest_start_time = get_latest_time(self._start_time, core_span['start'])
         new_span = WrappedSpan(self._service_type,
                                OpName(core_span['name']),
                                self._wrapped_tracer,
                                collection_details=self._collection_details,
                                parent_span=p_span,
-                               start_time=core_span['start'])
+                               start_time=latest_start_time)
         children = core_span.get('children', None)
         if children:
             self._build_core_spans(children, parent_span=new_span)
 
         for attr_name, attr_val in core_span.get('attributes', {}).items():
             new_span.set_attribute(attr_name, attr_val)
 
-        new_span.end(core_span['end'])
+        core_span_end_time = core_span['end']
+        self._end_time_watermark = get_latest_time(self._end_time_watermark, core_span_end_time)
+        new_span.end(core_span_end_time)
 
     def _create_request_span(self,
                              name: str,

couchbase/observability/__init__.py

@@ -12,3 +12,26 @@
 #  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
+
+"""
+Observability integrations for the Couchbase Python SDK.
+
+This package provides integrations with observability platforms:
+- OpenTelemetry tracing (otel_tracing)
+- OpenTelemetry metrics (otel_metrics)
+"""
+
+__all__ = []
+
+# OpenTelemetry integrations (optional - only available if otel installed)
+try:
+    from couchbase.observability.otel_tracing import get_otel_tracer  # noqa: F401
+    __all__.append('get_otel_tracer')
+except ImportError:
+    pass  # OTel not installed
+
+try:
+    from couchbase.observability.otel_metrics import get_otel_meter  # noqa: F401
+    __all__.append('get_otel_meter')
+except ImportError:
+    pass  # OTel not installed

couchbase/observability/otel_metrics.py

@@ -0,0 +1,103 @@
+#  Copyright 2016-2026. Couchbase, Inc.
+#  All Rights Reserved.
+#
+#  Licensed under the Apache License, Version 2.0 (the "License")
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+
+from __future__ import annotations
+
+import typing
+from typing import (Any,
+                    Dict,
+                    Mapping,
+                    Optional)
+
+from couchbase._version import __version__
+from couchbase.observability.metrics import Meter, ValueRecorder
+
+if typing.TYPE_CHECKING:
+    from opentelemetry import metrics as otel_metrics
+    from opentelemetry.metrics import MeterProvider
+
+try:
+    from opentelemetry import metrics as otel_metrics  # noqa: F811
+    HAS_OTEL = True
+except ImportError:
+    HAS_OTEL = False
+
+
+if HAS_OTEL:
+
+    class OtelWrapperValueRecorder(ValueRecorder):
+        """
+        Wraps an OpenTelemetry Histogram to implement SDK's ValueRecorder interface.
+        """
+
+        def __init__(self, histogram: otel_metrics.Histogram, tags: Dict[str, Any], unit: str) -> None:
+            self._histogram = histogram
+            self._attributes = tags if tags else {}
+            self._unit = unit
+
+        def record_value(self, value: int) -> None:
+            # The SDK records value in micros and will place a special __unit tag in the tags set to 's'
+            # So self._unit == 's' is indication that we should convert from micros to seconds
+            value = value / 1_000_000 if self._unit == 's' else value
+            self._histogram.record(float(value), attributes=self._attributes)
+
+    class OtelWrapperMeter(Meter):
+        """
+        Wraps an OpenTelemetry Meter to implement SDK's Meter interface.
+        """
+
+        def __init__(self, otel_meter: otel_metrics.Meter) -> None:
+            self._otel_meter = otel_meter
+            self._histograms: Dict[str, otel_metrics.Histogram] = {}
+
+        def value_recorder(self, name: str, tags: Mapping[str, Any]) -> ValueRecorder:
+            local_tags = dict(tags) if tags else {}
+            unit = local_tags.pop('__unit', '')
+            if name not in self._histograms:
+                self._histograms[name] = self._otel_meter.create_histogram(name=name, unit=unit)
+
+            histogram = self._histograms[name]
+            return OtelWrapperValueRecorder(histogram, local_tags, unit=unit)
+
+
+def get_otel_meter(provider: Optional[MeterProvider] = None) -> Meter:
+    """
+    Creates an OpenTelemetry wrapper meter.
+
+    Args:
+        provider: Optional OpenTelemetry MeterProvider. If not provided,
+                  falls back to the global OTel meter provider.
+
+    Returns:
+        OtelWrapperMeter instance implementing SDK's Meter interface
+
+    Raises:
+        ImportError: If OpenTelemetry is not installed
+    """
+    if not HAS_OTEL:
+        raise ImportError(
+            "OpenTelemetry is not installed. Please install with: "
+            "pip install couchbase[otel]"
+        )
+
+    pkg_name = "com.couchbase.client/python"
+    pkg_version = __version__
+
+    if provider:
+        otel_meter = provider.get_meter(pkg_name, pkg_version)
+    else:
+        otel_meter = otel_metrics.get_meter(pkg_name, pkg_version)
+
+    return OtelWrapperMeter(otel_meter)