feat(storage): add OpenTelemetry metrics gating and configuration

[shradhakatyal]

Introduce OpenTelemetry client metrics infrastructure and gating for Google Cloud Storage.

• Add _opentelemetry_metrics module with a hidden development gate (_ENABLE_METRICS_DEV_GATE) and environment variable handling (GCP_STORAGE_PYTHON_ENABLE_OTEL_METRICS, GCP_STORAGE_PYTHON_ENABLE_DEBUG_METRICS).
• Add enable_metrics and enable_advanced_metrics options and properties to Client.
• Add unit tests for gating resolution, environment variable overrides, and Client configuration.

[gemini-code-assist]

Code Review

This pull request introduces OpenTelemetry metrics support to the Google Cloud Storage Python client by adding a new internal module _opentelemetry_metrics.py and updating the Client class to support enable_metrics and enable_advanced_metrics configurations. The review feedback highlights a correctness bug where advanced metrics could be incorrectly enabled even if base metrics are disabled. To resolve this, the reviewer suggests updating is_advanced_metrics_enabled to accept and evaluate both settings, adjusting the client properties accordingly, and adding a unit test to verify this edge case.

[gemini-code-assist]

The current implementation of is_advanced_metrics_enabled evaluates the base metrics enablement by passing its own client_setting to is_metrics_enabled. This causes a correctness bug: if a client is initialized with enable_metrics=False and enable_advanced_metrics=True, is_advanced_metrics_enabled will incorrectly evaluate to True. Update the function signature to accept both client_setting and base_setting, and perform the None check within the function body to improve encapsulation.

def is_advanced_metrics_enabled(
    client_setting: Optional[bool] = None,
    base_setting: Optional[bool] = None,
) -> bool:
    """Evaluates whether high-frequency debug metrics should be recorded.

    Args:
        client_setting: Optional boolean configured on the client instance.
            Takes precedence over the environment variable if specified.
        base_setting: Optional boolean configured on the client instance for base metrics.

    Returns:
        bool: True if advanced metrics recording is enabled, False otherwise.
    """
    if not is_metrics_enabled(base_setting):
        return False

    if client_setting is not None:
        return bool(client_setting)

    return _parse_bool_env(
        ENABLE_DEBUG_METRICS_ENV_VAR, _DEFAULT_ENABLE_DEBUG_METRICS
    )

References

1. When processing optional parameters with a helper function, prefer placing the null or None check within the function body rather than at the call site to simplify the calling code and improve encapsulation.

[gemini-code-assist]

Update the call to is_advanced_metrics_enabled to pass both the advanced metrics setting and the base metrics setting, ensuring correct evaluation when base metrics are disabled.

Suggested change

	@property
	def advanced_metrics_enabled(self) -> bool:
	"""Returns True if advanced metrics recording is active for this client."""
	from google.cloud.storage import _opentelemetry_metrics
	return _opentelemetry_metrics.is_advanced_metrics_enabled(
	self._enable_advanced_metrics
	)
	@property
	def advanced_metrics_enabled(self) -> bool:
	"""Returns True if advanced metrics recording is active for this client."""
	from google.cloud.storage import _opentelemetry_metrics
	return _opentelemetry_metrics.is_advanced_metrics_enabled(
	self._enable_advanced_metrics,
	self._enable_metrics,
	)

[gemini-code-assist]

Add a test case to verify that advanced metrics are correctly disabled when base metrics are disabled, even if enable_advanced_metrics is set to True.

    def test_client_advanced_metrics_enabled_property(self):
        from google.cloud.storage import _opentelemetry_metrics

        PROJECT = "PROJECT"
        client = self._make_one(
            project=PROJECT, enable_metrics=True, enable_advanced_metrics=True
        )

        with mock.patch.object(
            _opentelemetry_metrics, "_ENABLE_METRICS_DEV_GATE", True
        ):
            self.assertTrue(client.advanced_metrics_enabled)

        with mock.patch.object(
            _opentelemetry_metrics, "_ENABLE_METRICS_DEV_GATE", False
        ):
            self.assertFalse(client.advanced_metrics_enabled)

        # Verify that advanced metrics are disabled if base metrics are disabled
        client_disabled_base = self._make_one(
            project=PROJECT, enable_metrics=False, enable_advanced_metrics=True
        )
        with mock.patch.object(
            _opentelemetry_metrics, "_ENABLE_METRICS_DEV_GATE", True
        ):
            self.assertFalse(client_disabled_base.advanced_metrics_enabled)