feat(memorystore): migrate Redis client-side metrics sample to OTLP

memorystore/redis/client_side_metrics/main.py

@@ -16,38 +16,68 @@
 import os
 import time
 
+import google.auth
+from google.auth.transport.grpc import AuthMetadataPlugin
+import google.auth.transport.requests
+import grpc
 from opentelemetry import metrics, trace
-from opentelemetry.exporter.cloud_monitoring import (
-    CloudMonitoringMetricsExporter,
+from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import (
+    OTLPMetricExporter,
+)
+from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import (
+    OTLPSpanExporter,
 )
-from opentelemetry.exporter.cloud_trace import CloudTraceSpanExporter
 from opentelemetry.instrumentation.redis import RedisInstrumentor
+from opentelemetry.resourcedetector.gcp_resource_detector import (
+    GoogleCloudResourceDetector,
+)
 from opentelemetry.sdk.metrics import MeterProvider
 from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
+from opentelemetry.sdk.resources import get_aggregated_resources, Resource
 from opentelemetry.sdk.trace import TracerProvider
 from opentelemetry.sdk.trace.export import BatchSpanProcessor
 import redis
 from redis.exceptions import ConnectionError, TimeoutError
 
 
+def init_telemetry():
+    """Initializes OpenTelemetry with OTLP Exporters and returns the SDK objects."""
+    credentials, project_id = google.auth.default(
+        scopes=["https://www.googleapis.com/auth/cloud-platform"]
+    )
+    request = google.auth.transport.requests.Request()
+    auth_metadata_plugin = AuthMetadataPlugin(credentials=credentials, request=request)
+    channel_creds = grpc.composite_channel_credentials(
+        grpc.ssl_channel_credentials(),
+        grpc.metadata_call_credentials(auth_metadata_plugin),
+    )
 
+    resource = get_aggregated_resources([GoogleCloudResourceDetector()])
+    if project_id:
+        resource = resource.merge(Resource.create({"gcp.project_id": project_id}))
 
-def init_telemetry():
-    """Initializes OpenTelemetry with GCP Exporters and returns the SDK objects."""
     # 1. Initialize Tracing
-    tracer_provider = TracerProvider()
+    tracer_provider = TracerProvider(resource=resource)
     tracer_provider.add_span_processor(
-        BatchSpanProcessor(CloudTraceSpanExporter())
+        BatchSpanProcessor(
+            OTLPSpanExporter(
+                endpoint="telemetry.googleapis.com:443",
+                credentials=channel_creds,
+            )
+        )
     )
     trace.set_tracer_provider(tracer_provider)
     tracer = trace.get_tracer("redis.client")
 
     # 2. Initialize Metrics
-    metrics_exporter = CloudMonitoringMetricsExporter()
+    metrics_exporter = OTLPMetricExporter(
+        endpoint="telemetry.googleapis.com:443",
+        credentials=channel_creds,
+    )
     metric_reader = PeriodicExportingMetricReader(
         metrics_exporter, export_interval_millis=10000
     )
-    meter_provider = MeterProvider(metric_readers=[metric_reader])
+    meter_provider = MeterProvider(resource=resource, metric_readers=[metric_reader])
     metrics.set_meter_provider(meter_provider)
     meter = metrics.get_meter("redis.metrics")
 
@@ -61,9 +91,7 @@ def init_telemetry():
             "redis_application_blocking_latency", unit="ms"
         ),
         "retry_counter": meter.create_counter("redis_retry_count"),
-        "conn_error_counter": meter.create_counter(
-            "redis_connectivity_error_count"
-        ),
+        "conn_error_counter": meter.create_counter("redis_connectivity_error_count"),
     }
 
     redis_metrics["retry_counter"].add(0, {"operation": "startup"})
@@ -90,9 +118,7 @@ def init_redis_pool():
     return redis_pool, redis_client
 
 
-def smart_redis_call(
-    operation_name, func, redis_pool, metrics, *args, **kwargs
-):
+def smart_redis_call(operation_name, func, redis_pool, metrics, *args, **kwargs):
     """Executes a Redis operation with metrics and retry handling (No Globals!)."""
     max_retries = 3
     attempt = 0
@@ -134,15 +160,14 @@ def smart_redis_call(
         except (ConnectionError, TimeoutError) as e:
             attempt += 1
             if metrics and metrics.get("conn_error_counter"):
-                metrics["conn_error_counter"].add(
-                    1, {"operation": operation_name}
-                )
+                metrics["conn_error_counter"].add(1, {"operation": operation_name})
             if metrics and metrics.get("retry_counter"):
                 metrics["retry_counter"].add(1, {"operation": operation_name})
             if attempt >= max_retries:
                 raise e
             time.sleep((2**attempt) * 0.1)
 
+
 if __name__ == "__main__":
     tracer, redis_metrics, tracer_provider, meter_provider = init_telemetry()
     redis_pool, redis_client = init_redis_pool()
@@ -171,6 +196,6 @@ def smart_redis_call(
             except Exception as e:
                 print(f"Error: {e}")
 
-        tracer_provider.force_flush()
-        meter_provider.force_flush()
+        tracer_provider.shutdown()
+        meter_provider.shutdown()
 # [END memorystore_redis_client_side_metrics]

memorystore/redis/client_side_metrics/main_test.py

@@ -15,7 +15,11 @@
 from unittest import mock
 
 import main
+
+from opentelemetry.sdk.metrics import MeterProvider
+from opentelemetry.sdk.trace import TracerProvider
 import pytest
+import redis
 from redis.exceptions import ConnectionError
 
 
@@ -87,9 +91,7 @@ def test_smart_redis_call_retry_success():
     # Verify counters captured exactly 2 retry events
     assert mock_metrics["retry_counter"].add.call_count == 2
     assert mock_metrics["conn_error_counter"].add.call_count == 2
-    mock_metrics["retry_counter"].add.assert_called_with(
-        1, {"operation": "get_user"}
-    )
+    mock_metrics["retry_counter"].add.assert_called_with(1, {"operation": "get_user"})
     mock_metrics["conn_error_counter"].add.assert_called_with(
         1, {"operation": "get_user"}
     )
@@ -100,9 +102,7 @@ def test_smart_redis_call_retry_success():
 
 def test_smart_redis_call_permanent_failure():
     """Verifies that permanent connectivity failures bubble up after 3 attempts."""
-    mock_func = mock.MagicMock(
-        side_effect=ConnectionError("permanent DNS error")
-    )
+    mock_func = mock.MagicMock(side_effect=ConnectionError("permanent DNS error"))
     mock_pool = mock.MagicMock()
     mock_metrics = {
         "client_block_hist": mock.MagicMock(),
@@ -135,7 +135,7 @@ def test_smart_redis_call_permanent_failure():
 
 
 def test_init_redis_pool_live(monkeypatch):
-    """Verifies init_redis_pool correctly parses environment variables without network calls."""
+    """Verifies init_redis_pool correctly parses env vars without network calls."""
     monkeypatch.setenv("REDISHOST", "10.0.0.1")
     monkeypatch.setenv("REDISPORT", "7000")
 
@@ -156,20 +156,14 @@ def test_init_redis_pool_live(monkeypatch):
 
 
 def test_smart_redis_call_live_or_skip():
-    """Integration Test: Hits a real Redis endpoint if running locally at localhost:6379."""
-    import main as live_main
-    from opentelemetry.sdk.metrics import MeterProvider
-    from opentelemetry.sdk.trace import TracerProvider
-    import pytest
-    import redis
-
+    """Integration Test: Hits real Redis endpoint if running at localhost:6379."""
     # Initialize REAL No-op OpenTelemetry SDK providers (invoked directly)
     TracerProvider()
     mp = MeterProvider()
 
     try:
         # Exercise actual pool creation and connectivity ping
-        redis_pool, redis_client = live_main.init_redis_pool()
+        redis_pool, redis_client = main.init_redis_pool()
         redis_client.ping()
     except (redis.exceptions.ConnectionError, redis.exceptions.TimeoutError):
         pytest.skip(
@@ -188,7 +182,7 @@ def test_smart_redis_call_live_or_skip():
 
     try:
         # Run a real Redis write hitting the ACTUAL endpoint (invoked directly)
-        live_main.smart_redis_call(
+        main.smart_redis_call(
             "set_test",
             redis_client.set,
             redis_pool,
@@ -197,7 +191,7 @@ def test_smart_redis_call_live_or_skip():
             "live_val",
         )
 
-        result_get = live_main.smart_redis_call(
+        result_get = main.smart_redis_call(
             "get_test",
             redis_client.get,
             redis_pool,
@@ -214,17 +208,24 @@ def test_smart_redis_call_live_or_skip():
         redis_pool.disconnect()
 
 
+@mock.patch(
+    "main.google.auth.default",
+    return_value=(mock.MagicMock(), "test-project"),
+)
+@mock.patch("main.GoogleCloudResourceDetector")
 @mock.patch("main.BatchSpanProcessor")
 @mock.patch("main.PeriodicExportingMetricReader")
-@mock.patch("main.CloudTraceSpanExporter")
-@mock.patch("main.CloudMonitoringMetricsExporter")
+@mock.patch("main.OTLPSpanExporter")
+@mock.patch("main.OTLPMetricExporter")
 @mock.patch("main.RedisInstrumentor")
 def test_init_telemetry_mocks(
     mock_redis_instrumentor,
-    mock_cloud_metrics,
-    mock_cloud_trace,
+    mock_otlp_metrics,
+    mock_otlp_trace,
     mock_metric_reader,
     mock_span_processor,
+    mock_gcp_detector,
+    mock_google_auth,
 ):
     """Verifies the init_telemetry function correctly instruments Redis."""
     tracer, redis_metrics, tp, mp = main.init_telemetry()
@@ -238,6 +239,9 @@ def test_init_telemetry_mocks(
     assert "client_block_hist" in redis_metrics
 
     mock_redis_instrumentor.return_value.instrument.assert_called_once()
-    mock_cloud_trace.assert_called_once()
-    mock_cloud_metrics.assert_called_once()
-
+    mock_otlp_trace.assert_called_once()
+    mock_otlp_metrics.assert_called_once()
+    _, kwargs_trace = mock_otlp_trace.call_args
+    assert kwargs_trace.get("endpoint") == "telemetry.googleapis.com:443"
+    _, kwargs_metrics = mock_otlp_metrics.call_args
+    assert kwargs_metrics.get("endpoint") == "telemetry.googleapis.com:443"

memorystore/redis/client_side_metrics/requirements.txt

@@ -2,5 +2,8 @@ redis==7.0.1
 opentelemetry-api==1.39.1
 opentelemetry-sdk==1.39.1
 opentelemetry-instrumentation-redis==0.60b1
-opentelemetry-exporter-gcp-trace==1.11.0
-opentelemetry-exporter-gcp-monitoring==1.11.0a0
+opentelemetry-exporter-otlp-proto-grpc==1.39.1
+opentelemetry-resourcedetector-gcp==1.11.0a0
+google-auth==2.38.0
+grpcio==1.70.0
+requests==2.32.3