googleapis · chalmerlowe · Jun 22, 2026 · Jun 22, 2026 · Jun 22, 2026 · gemini-code-assist
@@ -0,0 +1,22 @@
+from .options import (
+    clear_test_env_overrides,
+    is_signal_enabled,
+    set_test_env_override,
+)
+
+try:
+    # Tell flake8 that it's okay this is unused, it's just being exposed to the package namespace.
+    from .tracing import OtelSpanEnricher  # noqa: F401
+
+    __all__ = [
+        "is_signal_enabled",
+        "set_test_env_override",
+        "clear_test_env_overrides",
+        "OtelSpanEnricher",
+    ]
+except ImportError:
+    __all__ = [
+        "is_signal_enabled",
+        "set_test_env_override",
+        "clear_test_env_overrides",
+    ]
@@ -0,0 +1,133 @@
+"""Observability environment variable and client options resolution helpers."""
+
+import os
+import warnings
+from typing import Any, Dict, List, Optional, Union
+
+# Allowed truthy and falsy patterns for environment variables
+_TRUTHY_VALUES = ("y", "yes", "t", "true", "on", "1")
+_FALSY_VALUES = ("n", "no", "f", "false", "off", "0")
+
+
+def _strtobool(val: str) -> Optional[bool]:
+    """Convert a string representation of truth to a boolean."""
+    clean_val = val.lower().strip()
+    if not clean_val:
+        return None
+    if clean_val in _TRUTHY_VALUES:
+        return True
+    if clean_val in _FALSY_VALUES:
+        return False
+    raise ValueError(f"Invalid truth value: {val!r}")
+
+
+_TEST_ENV_OVERRIDES: Dict[str, bool] = {}
+
+
+def set_test_env_override(name: str, value: Optional[bool]) -> None:
+    """Sets a test-only override for a specific environment variable.
+
+    This is intended ONLY for unit/integration testing to prevent mutating
+    os.environ.
+    """
+    if value is None:
+        _TEST_ENV_OVERRIDES.pop(name, None)
+    else:
+        _TEST_ENV_OVERRIDES[name] = value
+
+
+def clear_test_env_overrides() -> None:
+    """Clears all test-only overrides."""
+    _TEST_ENV_OVERRIDES.clear()
+
+
+def _get_env_bool(name: str) -> Optional[bool]:
+    """Retrieve the boolean value of an environment variable."""
+    if name in _TEST_ENV_OVERRIDES:
+        return _TEST_ENV_OVERRIDES[name]
+
+    val = os.getenv(name)
+    if val is None:
+        return None
+    try:
+        return _strtobool(val)
+    except ValueError:
+        return None
+
+
+def _get_env_bool_with_dev_fallback(name: str) -> Optional[bool]:
+    """Retrieve the boolean value of an environment variable, checking dev/exp fallbacks first."""
+    if name.startswith("GOOGLE_CLOUD_"):
+        exp_name = name.replace("GOOGLE_CLOUD_", "GOOGLE_CLOUD_EXPERIMENTAL_", 1)
+        val = _get_env_bool(exp_name)
+        if val is not None:
+            return val
+    return _get_env_bool(name)
+
+
+def is_signal_enabled(
+    service_name: str,
+    signal_type: str,
+    client_options: Optional[Union[Dict[str, Any], Any]] = None,
+    default: bool = False,
+    legacy_vars: Optional[List[str]] = None,
+) -> bool:
+    """Determines if a telemetry signal is enabled."""
+    service_upper = service_name.upper().replace("-", "_")
+    signal_upper = signal_type.upper()
+
+    # 1. Resolve Programmatic Options First
+    if client_options is not None:
+        options_dict = (
+            client_options
+            if isinstance(client_options, dict)
+            else getattr(client_options, "__dict__", {})
+        )
+        option_key = f"enable_{signal_type.lower()}"
+        provider_key = f"{signal_type.rstrip('s').lower()}_provider"
+
+        if options_dict.get(option_key) is not None:
+            return bool(options_dict.get(option_key))
+        if options_dict.get(provider_key) is not None:
+            return True
-    if client_options is not None:
-        options_dict = (
-            client_options
-            if isinstance(client_options, dict)
-            else getattr(client_options, "__dict__", {})
-        )
-        option_key = f"enable_{signal_type.lower()}"
-        provider_key = f"{signal_type.rstrip('s').lower()}_provider"
-
-        if options_dict.get(option_key) is not None:
-            return bool(options_dict.get(option_key))
-        if options_dict.get(provider_key) is not None:
-            return True
+    if client_options is not None:
+        option_key = f"enable_{signal_type.lower()}"
+        provider_key = f"{signal_type.rstrip('s').lower()}_provider"
+
+        if isinstance(client_options, dict):
+            opt_val = client_options.get(option_key)
+            prov_val = client_options.get(provider_key)
+        else:
+            opt_val = getattr(client_options, option_key, None)
+            prov_val = getattr(client_options, provider_key, None)
+
+        if opt_val is not None:
+            return bool(opt_val)
+        if prov_val is not None:
+            return True
-    if client_options is not None:
-        options_dict = (
-            client_options
-            if isinstance(client_options, dict)
-            else getattr(client_options, "__dict__", {})
-        )
-        option_key = f"enable_{signal_type.lower()}"
-        provider_key = f"{signal_type.rstrip('s').lower()}_provider"
-
-        if options_dict.get(option_key) is not None:
-            return bool(options_dict.get(option_key))
-        if options_dict.get(provider_key) is not None:
-            return True
+    if client_options is not None:
+        option_key = f"enable_{signal_type.lower()}"
+        provider_key = f"{signal_type.rstrip('s').lower()}_provider"
+
+        if isinstance(client_options, dict):
+            opt_val = client_options.get(option_key)
+            prov_val = client_options.get(provider_key)
+        else:
+            opt_val = getattr(client_options, option_key, None)
+            prov_val = getattr(client_options, provider_key, None)
+
+        if opt_val is not None:
+            return bool(opt_val)
+        if prov_val is not None:
+            return True
+
+    # 2. Language & Service-specific
+    val = _get_env_bool_with_dev_fallback(
+        f"GOOGLE_CLOUD_PYTHON_{service_upper}_{signal_upper}_ENABLED"
+    )
+    if val is not None:
+        return val
+
+    # 3. Language-wide Global
+    val = _get_env_bool_with_dev_fallback(f"GOOGLE_CLOUD_PYTHON_{signal_upper}_ENABLED")
+    if val is not None:
+        return val
+
+    # 4. Cross-language Service-specific
+    val = _get_env_bool_with_dev_fallback(
+        f"GOOGLE_CLOUD_{service_upper}_{signal_upper}_ENABLED"
+    )
+    if val is not None:
+        return val
+
+    # 5. Cross-language Global
+    val = _get_env_bool_with_dev_fallback(f"GOOGLE_CLOUD_{signal_upper}_ENABLED")
+    if val is not None:
+        return val
+
+    # 6. Legacy Variables
+    if legacy_vars:
+        for legacy_var in legacy_vars:
+            val = _get_env_bool(legacy_var)
+            if val is not None:
+                warnings.warn(
+                    f"Environment variable {legacy_var!r} is deprecated and will be removed "
+                    "in a future release. Please migrate to the standardized "
+                    f"GOOGLE_CLOUD_PYTHON_{service_upper}_{signal_upper}_ENABLED instead.",
+                    DeprecationWarning,
+                    stacklevel=2,
+                )
+                return val
+
+    # 7. Default Fallback
+    return default
@@ -0,0 +1,75 @@
+# Copyright 2026 Google LLC
+#
+# 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.
+
+"""OpenTelemetry Tracing Enrichment Interceptors."""
+
+from typing import Any, Callable, Dict, Optional
+
+import grpc
+from opentelemetry import trace
+
+
+class OtelSpanEnricher(grpc.UnaryUnaryClientInterceptor):
+    """A gRPC client interceptor that enriches the active OpenTelemetry span.
+
+    This interceptor relies on the standard OpenTelemetry gRPC instrumentor
+    to create the baseline span. It runs in the interceptor chain to inject
+    additional Google Cloud specific domain attributes.
+    """
+
+    def __init__(
+        self,
+        static_attributes: Optional[Dict[str, Any]] = None,
+        attribute_extractor: Optional[
+            Callable[[Any, grpc.ClientCallDetails], Dict[str, Any]]
+        ] = None,
+    ):
+        """Initializes the OtelSpanEnricher.
+
+        Args:
+            static_attributes: Standard static attributes to attach to every span.
+                E.g. {"gcp.client.repo": "googleapis/google-cloud-python"}
+            attribute_extractor: A callable that extracts dynamic attributes from
+                the request and client call details.
+        """
+        self._static_attributes = static_attributes or {}
+        self._attribute_extractor = attribute_extractor
+
+    def intercept_unary_unary(
+        self,
+        continuation: Callable[[grpc.ClientCallDetails, Any], Any],
+        client_call_details: grpc.ClientCallDetails,
+        request: Any,
+    ) -> Any:
+        span = trace.get_current_span()
+
+        if span.is_recording():
+            # Inject static attributes
+            for key, val in self._static_attributes.items():
+                span.set_attribute(key, val)
-            # Inject static attributes
-            for key, val in self._static_attributes.items():
-                span.set_attribute(key, val)
+            # Inject static attributes
+            for key, val in self._static_attributes.items():
+                if val is not None:
+                    span.set_attribute(key, val)
-            # Inject static attributes
-            for key, val in self._static_attributes.items():
-                span.set_attribute(key, val)
+            # Inject static attributes
+            for key, val in self._static_attributes.items():
+                if val is not None:
+                    span.set_attribute(key, val)
+
+            # Extract and inject dynamic attributes
+            if self._attribute_extractor:
+                try:
+                    dynamic_attrs = self._attribute_extractor(
+                        request, client_call_details
+                    )
+                    for key, val in dynamic_attrs.items():
+                        if val is not None:
+                            span.set_attribute(key, val)
+                except Exception:
+                    # Prevent custom extractor exceptions from failing the RPC
+                    pass
+
+        return continuation(client_call_details, request)
@@ -49,6 +49,7 @@ dependencies = [
   "proto-plus >= 1.25.0, < 2.0.0; python_version >= '3.13'",
   "google-auth >= 2.14.1, < 3.0.0",
   "requests >= 2.33.0, < 3.0.0",
+  "opentelemetry-api >= 1.1.0, < 2.0.0",
 ]
 dynamic = ["version"]
 

@@ -12,3 +12,4 @@ requests==2.33.0
 grpcio==1.41.0
 grpcio-status==1.41.0
 proto-plus==1.24.0
+opentelemetry-api==1.1.0
@@ -13,3 +13,4 @@ grpcio==1.41.0
 grpcio-status==1.41.0
 proto-plus==1.24.0
 aiohttp==3.13.4
+opentelemetry-api==1.1.0
@@ -0,0 +1,134 @@
+# Copyright 2026 Google LLC
+#
+# 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.
+
+import pytest
+
+from google.api_core.observability import options
+from google.api_core.observability.options import (
+    _get_env_bool,
+    _strtobool,
+    clear_test_env_overrides,
+    set_test_env_override,
+)
+
+
+@pytest.fixture(autouse=True)
+def clean_overrides():
+    yield
+    clear_test_env_overrides()
+
+
+@pytest.mark.parametrize(
+    "value,expected",
+    [
+        ("y", True),
+        ("yes", True),
+        ("t", True),
+        ("true", True),
+        ("on", True),
+        ("1", True),
+        ("n", False),
+        ("no", False),
+        ("f", False),
+        ("false", False),
+        ("off", False),
+        ("0", False),
+        ("  True  ", True),
+        (" FALSE ", False),
+        ("", None),
+    ],
+)
+def test_strtobool(value, expected):
+    assert _strtobool(value) is expected
+
+
+def test_strtobool_invalid():
+    with pytest.raises(ValueError):
+        _strtobool("invalid")
+
+
+def test_get_env_bool(monkeypatch):
+    monkeypatch.setenv("TEST_VAR", "true")
+    assert _get_env_bool("TEST_VAR") is True
+
+    monkeypatch.setenv("TEST_VAR", "invalid")
+    assert _get_env_bool("TEST_VAR") is None
+
+    monkeypatch.delenv("TEST_VAR", raising=False)
+    assert _get_env_bool("TEST_VAR") is None
+
+
+@pytest.mark.parametrize(
+    "env_vars, client_options, default_val, expected",
+    [
+        # Default fallback tests
+        ({}, None, False, False),
+        ({}, None, True, True),
+        # Service-specific env var
+        ({"GOOGLE_CLOUD_PYTHON_TRANSLATE_TRACES_ENABLED": True}, None, False, True),
+        ({"GOOGLE_CLOUD_PYTHON_TRANSLATE_TRACES_ENABLED": False}, None, True, False),
+        # Experimental fallback
+        (
+            {"GOOGLE_CLOUD_EXPERIMENTAL_PYTHON_TRANSLATE_TRACES_ENABLED": True},
+            None,
+            False,
+            True,
+        ),
+        # Precedence: Service specific overrides global
+        (
+            {
+                "GOOGLE_CLOUD_PYTHON_TRACES_ENABLED": True,
+                "GOOGLE_CLOUD_PYTHON_TRANSLATE_TRACES_ENABLED": False,
+            },
+            None,
+            False,
+            False,
+        ),
+        (
+            {
+                "GOOGLE_CLOUD_PYTHON_TRACES_ENABLED": False,
+                "GOOGLE_CLOUD_PYTHON_TRANSLATE_TRACES_ENABLED": True,
+            },
+            None,
+            False,
+            True,
+        ),
+        # Precedence: Client options override env vars
+        (
+            {"GOOGLE_CLOUD_PYTHON_TRANSLATE_TRACES_ENABLED": False},
+            {"enable_traces": True},
+            False,
+            True,
+        ),
+    ],
+)
+def test_is_signal_enabled(env_vars, client_options, default_val, expected):
+    # Setup environment variables using our test overrides
+    for k, v in env_vars.items():
+        set_test_env_override(k, v)
+
+    result = options.is_signal_enabled(
+        "translate", "traces", client_options=client_options, default=default_val
+    )
+    assert result is expected
+
+
+def test_legacy_var_with_warning():
+    set_test_env_override("LEGACY_TRACE_VAR", True)
+
+    with pytest.warns(DeprecationWarning, match="LEGACY_TRACE_VAR"):
+        result = options.is_signal_enabled(
+            "translate", "traces", legacy_vars=["LEGACY_TRACE_VAR"]
+        )
+        assert result is True