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

Author: fosky94

memorystore/redis/client_side_metrics/main.py

@@ -16,38 +16,74 @@
 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 Resource, get_aggregated_resources
 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")
 
@@ -143,6 +179,7 @@ def smart_redis_call(
                 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 +208,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,10 @@
 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
 
 
@@ -135,7 +138,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 +159,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 +185,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 +194,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 +211,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 +242,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