feat: Add OpenTelemetry environment variable and options configuration helpers.

packages/google-api-core/google/api_core/observability/__init__.py

@@ -0,0 +1,11 @@
+from .options import (
+    clear_test_env_overrides,
+    is_signal_enabled,
+    set_test_env_override,
+)
+
+__all__ = [
+    "is_signal_enabled",
+    "set_test_env_override",
+    "clear_test_env_overrides",
+]

packages/google-api-core/google/api_core/observability/options.py

@@ -0,0 +1,128 @@
+# -*- coding: utf-8 -*-
+# 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.
+#
+
+"""Observability environment variable and client options resolution helpers."""
+
+import os
+from typing import Any, Dict, 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")
+
+# Test-only overrides for environment variables.
+# This is intended ONLY for unit/integration testing to prevent mutating
+# os.environ.
+_TEST_ENV_OVERRIDES: Dict[str, bool] = {}
+
+
+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}")
+
+
+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 experimental 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(
+    signal_type: str,
+    client_options: Optional[Union[Dict[str, Any], Any]] = None,
+    default: bool = False,
+) -> bool:
+    """Determines if a telemetry signal is enabled.
+
+    Resolves settings in the following order of precedence:
+    1. Programmatic overrides in client_options (checks tracer_provider)
+    2. Language-wide Environment Variable: GOOGLE_CLOUD_PYTHON_TRACING_ENABLED
+       (natively checks for a variant with an "EXPERIMENTAL" token first)
+    3. Default fallback
+
+    Args:
+        signal_type: The signal type: must be 'tracing'.
+        client_options: A dictionary or object containing client configuration.
+        default: Fallback boolean if no options or env variables match.
+
+    Returns:
+        bool: True if the signal is resolved to enabled, False otherwise.
+    """
+    if signal_type != "tracing":
+        raise ValueError(
+            f"Invalid signal_type: {signal_type!r}. Only 'tracing' is supported at this time."
+        )
+
+    # 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__", {})
+        )
+
+        if options_dict.get("tracer_provider") is not None:
+            return True
+
+    # 2. Check Language-Wide Environment Variable
+    val = _get_env_bool_with_dev_fallback("GOOGLE_CLOUD_PYTHON_TRACING_ENABLED")
+    if val is not None:
+        return val
+
+    # 3. Default Fallback
+    return default

packages/google-api-core/pyproject.toml

@@ -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.27.0, < 2.0.0",
 ]
 dynamic = ["version"]
 
@@ -94,4 +95,6 @@ filterwarnings = [
   "ignore:.*custom tp_new.*in Python 3.14:DeprecationWarning",
   # Remove once https://github.com/grpc/grpc/issues/35086 is fixed (and version newer than 1.60.0 is published)
   "ignore:There is no current event loop:DeprecationWarning",
+  # Ignore external OpenTelemetry/importlib.metadata SelectableGroups warning
+  "ignore:.*SelectableGroups dict interface is deprecated:DeprecationWarning",
 ]

packages/google-api-core/testing/constraints-3.10.txt

@@ -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.27.0

packages/google-api-core/testing/constraints-async-rest-3.10.txt

@@ -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.27.0

packages/google-api-core/tests/asyncio/retry/test_retry_streaming_async.py

@@ -293,7 +293,7 @@ async def test___call___generator_send_retry(self, sleep):
         generator = await retry_(self._generator_mock)(error_on=3)
         with pytest.raises(TypeError) as exc_info:
             await generator.asend("cannot send to fresh generator")
-            assert exc_info.match("can't send non-None value")
+        assert exc_info.match("can't send non-None value")
         await generator.aclose()
 
         # error thrown on 3

packages/google-api-core/tests/unit/observability/test_options.py

@@ -0,0 +1,171 @@
+# 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",
+    [
+        # Truthy values
+        ("y", True),
+        ("yes", True),
+        ("t", True),
+        ("true", True),
+        ("on", True),
+        ("1", True),
+        ("  True  ", True),
+        # Falsy values
+        ("n", False),
+        ("no", False),
+        ("f", False),
+        ("false", False),
+        ("off", False),
+        ("0", False),
+        (" FALSE ", False),
+        # Empty string
+        ("", 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
+
+
+def test_set_test_env_override_clear_specific():
+    """Verify that setting an override to None clears that specific override.
+
+    This is important to ensure tests can reset individual environment overrides
+    without affecting other overrides that might be set for other tests running
+    concurrently or subsequently.
+    """
+    set_test_env_override("TEST_A", True)
+    set_test_env_override("TEST_B", True)
+    assert _get_env_bool("TEST_A") is True
+    assert _get_env_bool("TEST_B") is True
+
+    # Clear only TEST_A
+    set_test_env_override("TEST_A", None)
+
+    # Verify TEST_A is cleared but TEST_B remains
+    assert _get_env_bool("TEST_A") is None
+    assert _get_env_bool("TEST_B") is True
+
+
+def test_get_env_bool_with_dev_fallback_other_prefix(monkeypatch):
+    """Verify that environment variables without the 'GOOGLE_CLOUD_' prefix fall back directly.
+
+    This is important to ensure that generic or non-GCP environment variables
+    are handled correctly by the fallback logic without triggering GCP-specific
+    replacement logic.
+    """
+    monkeypatch.setenv("OTHER_PREFIX_VAR", "true")
+    assert options._get_env_bool_with_dev_fallback("OTHER_PREFIX_VAR") is True
+
+
+@pytest.mark.parametrize(
+    "signal_type, env_vars, client_options, default_val, expected",
+    [
+        # Default fallback tests
+        ("tracing", {}, None, False, False),
+        ("tracing", {}, None, True, True),
+        # Global env var
+        ("tracing", {"GOOGLE_CLOUD_PYTHON_TRACING_ENABLED": True}, None, False, True),
+        ("tracing", {"GOOGLE_CLOUD_PYTHON_TRACING_ENABLED": False}, None, True, False),
+        # Experimental fallback
+        (
+            "tracing",
+            {"GOOGLE_CLOUD_EXPERIMENTAL_PYTHON_TRACING_ENABLED": True},
+            None,
+            False,
+            True,
+        ),
+        (
+            "tracing",
+            {"GOOGLE_CLOUD_EXPERIMENTAL_PYTHON_TRACING_ENABLED": False},
+            None,
+            True,
+            False,
+        ),
+        # Implicit opt-in with provider
+        (
+            "tracing",
+            {"GOOGLE_CLOUD_PYTHON_TRACING_ENABLED": False},
+            {"tracer_provider": object()},
+            False,
+            True,
+        ),
+        # Programmatic boolean flags are NOT supported in client_options
+        # (should default/fallback to False)
+        (
+            "tracing",
+            {"GOOGLE_CLOUD_PYTHON_TRACING_ENABLED": False},
+            {"enable_metrics": True},
+            False,
+            False,
+        ),
+        (
+            "tracing",
+            {"GOOGLE_CLOUD_PYTHON_TRACING_ENABLED": False},
+            {"enable_tracing": True},
+            False,
+            False,
+        ),
+    ],
+)
+def test_is_signal_enabled(
+    signal_type, 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(
+        signal_type, client_options=client_options, default=default_val
+    )
+    assert result is expected
+
+
+def test_is_signal_enabled_invalid_signal():
+    with pytest.raises(ValueError, match="Only 'tracing' is supported"):
+        options.is_signal_enabled("metrics")

packages/google-api-core/tests/unit/retry/test_retry_streaming.py

@@ -263,7 +263,7 @@ def test___call___with_generator_send_retry(self, sleep):
         with pytest.raises(TypeError) as exc_info:
             # calling first send with non-None input should raise a TypeError
             result.send("can not send to fresh generator")
-            assert exc_info.match("can't send non-None value")
+        assert exc_info.match("can't send non-None value")
         # initiate iteration with None
         result = retry_(self._generator_mock)(error_on=3)
         assert result.send(None) == 0