diff --git a/memorystore/redis/client_side_metrics/main.py b/memorystore/redis/client_side_metrics/main.py index eb47d7002d..53695978e4 100644 --- a/memorystore/redis/client_side_metrics/main.py +++ b/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] diff --git a/memorystore/redis/client_side_metrics/main_test.py b/memorystore/redis/client_side_metrics/main_test.py index 196d758a54..ea21e3caf2 100644 --- a/memorystore/redis/client_side_metrics/main_test.py +++ b/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() - \ No newline at end of file + 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" diff --git a/memorystore/redis/client_side_metrics/requirements.txt b/memorystore/redis/client_side_metrics/requirements.txt index e8caf96a29..db0de12603 100644 --- a/memorystore/redis/client_side_metrics/requirements.txt +++ b/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