From 78cf95e4b9f2b061d07b14b98ed4198c2c2d083b Mon Sep 17 00:00:00 2001 From: fosky94 Date: Tue, 19 May 2026 17:23:48 -0500 Subject: [PATCH 01/11] Create Client side metrics --- memorystore/redis/client_side_metrics/main.py | 88 +++++++++++++++++++ 1 file changed, 88 insertions(+) create mode 100644 memorystore/redis/client_side_metrics/main.py diff --git a/memorystore/redis/client_side_metrics/main.py b/memorystore/redis/client_side_metrics/main.py new file mode 100644 index 00000000000..6d038f93fe9 --- /dev/null +++ b/memorystore/redis/client_side_metrics/main.py @@ -0,0 +1,88 @@ +import os +import time +import redis +from redis.exceptions import ConnectionError, TimeoutError +from opentelemetry import trace, metrics +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import BatchSpanProcessor +from opentelemetry.exporter.cloud_trace import CloudTraceSpanExporter +from opentelemetry.sdk.metrics import MeterProvider +from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader +from opentelemetry.exporter.cloud_monitoring import CloudMonitoringMetricsExporter +from opentelemetry.instrumentation.redis import RedisInstrumentor + +# 1. Initialize Tracing +tracer_provider = TracerProvider() +tracer_provider.add_span_processor(BatchSpanProcessor(CloudTraceSpanExporter())) +trace.set_tracer_provider(tracer_provider) +tracer = trace.get_tracer("redis.client") + +# 2. Initialize Metrics +metrics_exporter = CloudMonitoringMetricsExporter() +metric_reader = PeriodicExportingMetricReader(metrics_exporter, export_interval_millis=10000) +meter_provider = MeterProvider(metric_readers=[metric_reader]) +metrics.set_meter_provider(meter_provider) +meter = metrics.get_meter("redis.metrics") + +rtt_hist = meter.create_histogram("redis_client_rtt", unit="ms") +client_block_hist = meter.create_histogram("redis_client_blocking_latency", unit="ms") +app_block_hist = meter.create_histogram("redis_application_blocking_latency", unit="ms") +retry_counter = meter.create_counter("redis_retry_count") +conn_error_counter = meter.create_counter("redis_connectivity_error_count") + +retry_counter.add(0, {"operation": "startup"}) +conn_error_counter.add(0, {"operation": "startup"}) + +# 3. Setup Redis +RedisInstrumentor().instrument() +redis_host = os.environ.get("REDISHOST", "localhost") +redis_port = int(os.environ.get("REDISPORT", 6379)) + +pool = redis.ConnectionPool(host=redis_host, port=redis_port, max_connections=10, decode_responses=True) +r = redis.Redis(connection_pool=pool) + +def smart_redis_call(operation_name, func, *args, **kwargs): + max_retries = 3 + attempt = 0 + + pool_start = time.time() + try: + conn = pool.get_connection('PING') + pool.release(conn) + except Exception: + pass + client_block_hist.record((time.time() - pool_start) * 1000, {"operation": operation_name}) + + while attempt < max_retries: + try: + req_start = time.time() + response = func(*args, **kwargs) + rtt_hist.record((time.time() - req_start) * 1000, {"operation": operation_name}) + + app_start = time.time() + _ = str(response) + app_block_hist.record((time.time() - app_start) * 1000, {"operation": operation_name}) + + return response + + except (ConnectionError, TimeoutError) as e: + attempt += 1 + conn_error_counter.add(1, {"operation": operation_name}) + retry_counter.add(1, {"operation": operation_name}) + if attempt >= max_retries: + raise e + time.sleep((2 ** attempt) * 0.1) + +if __name__ == "__main__": + with tracer.start_as_current_span("process_user_span"): + try: + # Simple write and read operations + smart_redis_call("set_user", r.set, "user:123", "active") + + result = smart_redis_call("get_user", r.get, "user:123") + print(f"Retrieved: {result}") + except Exception as e: + print(f"Error: {e}") + + tracer_provider.force_flush() + meter_provider.force_flush() From f0ec85e81d5a7804cba192e8f717ff2600f9efe3 Mon Sep 17 00:00:00 2001 From: fosky94 Date: Tue, 19 May 2026 17:59:20 -0500 Subject: [PATCH 02/11] Update license --- memorystore/redis/client_side_metrics/main.py | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/memorystore/redis/client_side_metrics/main.py b/memorystore/redis/client_side_metrics/main.py index 6d038f93fe9..1b42e9e5eb4 100644 --- a/memorystore/redis/client_side_metrics/main.py +++ b/memorystore/redis/client_side_metrics/main.py @@ -1,3 +1,17 @@ +# 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 os import time import redis From 8498cd373d5d733c8350ee4fbd61235f61d005bb Mon Sep 17 00:00:00 2001 From: fosky94 Date: Tue, 9 Jun 2026 17:11:06 -0500 Subject: [PATCH 03/11] Update main.py --- memorystore/redis/client_side_metrics/main.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/memorystore/redis/client_side_metrics/main.py b/memorystore/redis/client_side_metrics/main.py index 1b42e9e5eb4..b850fa65667 100644 --- a/memorystore/redis/client_side_metrics/main.py +++ b/memorystore/redis/client_side_metrics/main.py @@ -12,6 +12,7 @@ # See the License for the specific language governing permissions and # limitations under the License. +# [START memorystore_redis_client_side_metrics] import os import time import redis @@ -100,3 +101,4 @@ def smart_redis_call(operation_name, func, *args, **kwargs): tracer_provider.force_flush() meter_provider.force_flush() +# [END memorystore_redis_client_side_metrics] From 8acd7bb1619ee0db1e1b7a32fed907bf9baa2203 Mon Sep 17 00:00:00 2001 From: fosky94 Date: Thu, 25 Jun 2026 18:23:48 -0500 Subject: [PATCH 04/11] Update main.py --- memorystore/redis/client_side_metrics/main.py | 15 ++++++++------- 1 file changed, 8 insertions(+), 7 deletions(-) diff --git a/memorystore/redis/client_side_metrics/main.py b/memorystore/redis/client_side_metrics/main.py index b850fa65667..efebb43916e 100644 --- a/memorystore/redis/client_side_metrics/main.py +++ b/memorystore/redis/client_side_metrics/main.py @@ -15,16 +15,17 @@ # [START memorystore_redis_client_side_metrics] import os import time -import redis -from redis.exceptions import ConnectionError, TimeoutError -from opentelemetry import trace, metrics -from opentelemetry.sdk.trace import TracerProvider -from opentelemetry.sdk.trace.export import BatchSpanProcessor + +from opentelemetry import metrics, trace +from opentelemetry.exporter.cloud_monitoring import CloudMonitoringMetricsExporter from opentelemetry.exporter.cloud_trace import CloudTraceSpanExporter +from opentelemetry.instrumentation.redis import RedisInstrumentor from opentelemetry.sdk.metrics import MeterProvider from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader -from opentelemetry.exporter.cloud_monitoring import CloudMonitoringMetricsExporter -from opentelemetry.instrumentation.redis import RedisInstrumentor +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import BatchSpanProcessor +import redis +from redis.exceptions import ConnectionError, TimeoutError # 1. Initialize Tracing tracer_provider = TracerProvider() From 2a64545673cd889620b10220f3fcaaa289cd9804 Mon Sep 17 00:00:00 2001 From: fosky94 Date: Thu, 9 Jul 2026 20:47:55 +0000 Subject: [PATCH 05/11] fix:add tests --- memorystore/redis/client_side_metrics/main.py | 159 ++++++++++++------ .../redis/client_side_metrics/main_test.py | 144 ++++++++++++++++ .../client_side_metrics/requirements.txt | 7 + 3 files changed, 262 insertions(+), 48 deletions(-) create mode 100644 memorystore/redis/client_side_metrics/main_test.py create mode 100644 memorystore/redis/client_side_metrics/requirements.txt diff --git a/memorystore/redis/client_side_metrics/main.py b/memorystore/redis/client_side_metrics/main.py index efebb43916e..050a1c88956 100644 --- a/memorystore/redis/client_side_metrics/main.py +++ b/memorystore/redis/client_side_metrics/main.py @@ -17,7 +17,9 @@ import time from opentelemetry import metrics, trace -from opentelemetry.exporter.cloud_monitoring import CloudMonitoringMetricsExporter +from opentelemetry.exporter.cloud_monitoring import ( + CloudMonitoringMetricsExporter, +) from opentelemetry.exporter.cloud_trace import CloudTraceSpanExporter from opentelemetry.instrumentation.redis import RedisInstrumentor from opentelemetry.sdk.metrics import MeterProvider @@ -27,79 +29,140 @@ import redis from redis.exceptions import ConnectionError, TimeoutError -# 1. Initialize Tracing -tracer_provider = TracerProvider() -tracer_provider.add_span_processor(BatchSpanProcessor(CloudTraceSpanExporter())) -trace.set_tracer_provider(tracer_provider) -tracer = trace.get_tracer("redis.client") +# Telemetry and Redis Globals (initialized dynamically in production or mocked in tests) +tracer = None +rtt_hist = None +client_block_hist = None +app_block_hist = None +retry_counter = None +conn_error_counter = None +pool = None +r = None + + +def init_telemetry(): + """Initializes the OpenTelemetry SDK with Google Cloud Exporters.""" + global tracer, rtt_hist, client_block_hist, app_block_hist + global retry_counter, conn_error_counter + + # 1. Initialize Tracing + tracer_provider = TracerProvider() + tracer_provider.add_span_processor( + BatchSpanProcessor(CloudTraceSpanExporter()) + ) + trace.set_tracer_provider(tracer_provider) + tracer = trace.get_tracer("redis.client") + + # 2. Initialize Metrics + metrics_exporter = CloudMonitoringMetricsExporter() + metric_reader = PeriodicExportingMetricReader( + metrics_exporter, export_interval_millis=10000 + ) + meter_provider = MeterProvider(metric_readers=[metric_reader]) + metrics.set_meter_provider(meter_provider) + meter = metrics.get_meter("redis.metrics") + + rtt_hist = meter.create_histogram("redis_client_rtt", unit="ms") + client_block_hist = meter.create_histogram( + "redis_client_blocking_latency", unit="ms" + ) + app_block_hist = meter.create_histogram( + "redis_application_blocking_latency", unit="ms" + ) + retry_counter = meter.create_counter("redis_retry_count") + conn_error_counter = meter.create_counter( + "redis_connectivity_error_count" + ) + + retry_counter.add(0, {"operation": "startup"}) + conn_error_counter.add(0, {"operation": "startup"}) + + # Setup Redis Instrumentation + RedisInstrumentor().instrument() + + return tracer_provider, meter_provider + + +def init_redis_pool(): + """Initializes the Redis connection pool.""" + global pool, r + redis_host = os.environ.get("REDISHOST", "localhost") + redis_port = int(os.environ.get("REDISPORT", 6379)) + + pool = redis.ConnectionPool( + host=redis_host, + port=redis_port, + max_connections=10, + decode_responses=True, + ) + r = redis.Redis(connection_pool=pool) + return pool, r -# 2. Initialize Metrics -metrics_exporter = CloudMonitoringMetricsExporter() -metric_reader = PeriodicExportingMetricReader(metrics_exporter, export_interval_millis=10000) -meter_provider = MeterProvider(metric_readers=[metric_reader]) -metrics.set_meter_provider(meter_provider) -meter = metrics.get_meter("redis.metrics") - -rtt_hist = meter.create_histogram("redis_client_rtt", unit="ms") -client_block_hist = meter.create_histogram("redis_client_blocking_latency", unit="ms") -app_block_hist = meter.create_histogram("redis_application_blocking_latency", unit="ms") -retry_counter = meter.create_counter("redis_retry_count") -conn_error_counter = meter.create_counter("redis_connectivity_error_count") - -retry_counter.add(0, {"operation": "startup"}) -conn_error_counter.add(0, {"operation": "startup"}) - -# 3. Setup Redis -RedisInstrumentor().instrument() -redis_host = os.environ.get("REDISHOST", "localhost") -redis_port = int(os.environ.get("REDISPORT", 6379)) - -pool = redis.ConnectionPool(host=redis_host, port=redis_port, max_connections=10, decode_responses=True) -r = redis.Redis(connection_pool=pool) def smart_redis_call(operation_name, func, *args, **kwargs): + """Executes a Redis operation with latency metrics and retry handling.""" max_retries = 3 attempt = 0 pool_start = time.time() try: - conn = pool.get_connection('PING') + # Check connection pool health + conn = pool.get_connection("PING") pool.release(conn) except Exception: pass - client_block_hist.record((time.time() - pool_start) * 1000, {"operation": operation_name}) + + if client_block_hist: + client_block_hist.record( + (time.time() - pool_start) * 1000, {"operation": operation_name} + ) while attempt < max_retries: try: req_start = time.time() response = func(*args, **kwargs) - rtt_hist.record((time.time() - req_start) * 1000, {"operation": operation_name}) + if rtt_hist: + rtt_hist.record( + (time.time() - req_start) * 1000, + {"operation": operation_name}, + ) app_start = time.time() _ = str(response) - app_block_hist.record((time.time() - app_start) * 1000, {"operation": operation_name}) + if app_block_hist: + app_block_hist.record( + (time.time() - app_start) * 1000, + {"operation": operation_name}, + ) return response except (ConnectionError, TimeoutError) as e: attempt += 1 - conn_error_counter.add(1, {"operation": operation_name}) - retry_counter.add(1, {"operation": operation_name}) + if conn_error_counter: + conn_error_counter.add(1, {"operation": operation_name}) + if retry_counter: + retry_counter.add(1, {"operation": operation_name}) if attempt >= max_retries: raise e - time.sleep((2 ** attempt) * 0.1) + time.sleep((2**attempt) * 0.1) -if __name__ == "__main__": - with tracer.start_as_current_span("process_user_span"): - try: - # Simple write and read operations - smart_redis_call("set_user", r.set, "user:123", "active") - result = smart_redis_call("get_user", r.get, "user:123") - print(f"Retrieved: {result}") - except Exception as e: - print(f"Error: {e}") - - tracer_provider.force_flush() - meter_provider.force_flush() +if __name__ == "__main__": + tracer_provider, meter_provider = init_telemetry() + init_redis_pool() + + if tracer: + with tracer.start_as_current_span("process_user_span"): + try: + # Simple write and read operations + smart_redis_call("set_user", r.set, "user:123", "active") + + result = smart_redis_call("get_user", r.get, "user:123") + print(f"Retrieved: {result}") + except Exception as e: + print(f"Error: {e}") + + tracer_provider.force_flush() + meter_provider.force_flush() # [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 new file mode 100644 index 00000000000..11c5a924e95 --- /dev/null +++ b/memorystore/redis/client_side_metrics/main_test.py @@ -0,0 +1,144 @@ +# 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. + +from unittest import mock + +from opentelemetry.instrumentation.redis import RedisInstrumentor +import pytest +from redis.exceptions import ConnectionError + +import main + + +@pytest.fixture +def mock_telemetry(monkeypatch): + """Hermetically binds mock OpenTelemetry and Pool globals for the duration of the test.""" + mock_tracer = mock.MagicMock() + mock_rtt = mock.MagicMock() + mock_client = mock.MagicMock() + mock_app = mock.MagicMock() + mock_retry = mock.MagicMock() + mock_conn_err = mock.MagicMock() + mock_pool = mock.MagicMock() + + # Prevent real time.sleep calls during retry handling to make tests lightning fast + monkeypatch.setattr(main.time, "sleep", lambda x: None) + + monkeypatch.setattr(main, "tracer", mock_tracer) + monkeypatch.setattr(main, "rtt_hist", mock_rtt) + monkeypatch.setattr(main, "client_block_hist", mock_client) + monkeypatch.setattr(main, "app_block_hist", mock_app) + monkeypatch.setattr(main, "retry_counter", mock_retry) + monkeypatch.setattr(main, "conn_error_counter", mock_conn_err) + monkeypatch.setattr(main, "pool", mock_pool) + + return { + "rtt_hist": mock_rtt, + "client_block_hist": mock_client, + "app_block_hist": mock_app, + "retry_counter": mock_retry, + "conn_error_counter": mock_conn_err, + "pool": mock_pool, + } + + +def test_smart_redis_call_success(mock_telemetry): + """Verifies standard successful SET and GET execution paths and latency recording.""" + mock_func = mock.MagicMock(return_value="active") + + result = main.smart_redis_call("set_user", mock_func, "user:123", "active") + + assert result == "active" + mock_func.assert_called_once_with("user:123", "active") + mock_telemetry["pool"].get_connection.assert_called_once_with("PING") + mock_telemetry["pool"].release.assert_called_once() + + # Verify all metrics were cleanly invoked + mock_telemetry["client_block_hist"].record.assert_called_once() + mock_telemetry["rtt_hist"].record.assert_called_once() + mock_telemetry["app_block_hist"].record.assert_called_once() + + # Verify retry counters were untouched + mock_telemetry["retry_counter"].add.assert_not_called() + mock_telemetry["conn_error_counter"].add.assert_not_called() + + +def test_smart_redis_call_retry_success(mock_telemetry): + """Simulates 2 sequential connectivity failures followed by a successful 3rd retry.""" + mock_func = mock.MagicMock( + side_effect=[ + ConnectionError("cluster unreachable attempt 1"), + ConnectionError("cluster unreachable attempt 2"), + "success_response", + ] + ) + + result = main.smart_redis_call("get_user", mock_func, "user:123") + + assert result == "success_response" + assert mock_func.call_count == 3 + + # Verify counters captured exactly 2 retry events + assert mock_telemetry["retry_counter"].add.call_count == 2 + assert mock_telemetry["conn_error_counter"].add.call_count == 2 + + # Expect calls with increment of 1 and proper operation attribute + mock_telemetry["retry_counter"].add.assert_called_with( + 1, {"operation": "get_user"} + ) + mock_telemetry["conn_error_counter"].add.assert_called_with( + 1, {"operation": "get_user"} + ) + + # Success on 3rd attempt means round-trip latency was recorded + mock_telemetry["rtt_hist"].record.assert_called_once() + + +def test_smart_redis_call_permanent_failure(mock_telemetry): + """Verifies that permanent connectivity failures bubble up and trigger exactly 3 attempts.""" + mock_func = mock.MagicMock( + side_effect=ConnectionError("permanent DNS error") + ) + + with pytest.raises(ConnectionError) as exc_info: + main.smart_redis_call("set_user", mock_func, "user:123", "active") + + assert "permanent DNS error" in str(exc_info.value) + assert mock_func.call_count == 3 + + # Verify all 3 attempts were tracked + assert mock_telemetry["retry_counter"].add.call_count == 3 + assert mock_telemetry["conn_error_counter"].add.call_count == 3 + + +@mock.patch("main.BatchSpanProcessor") +@mock.patch("main.PeriodicExportingMetricReader") +@mock.patch("main.CloudTraceSpanExporter") +@mock.patch("main.CloudMonitoringMetricsExporter") +@mock.patch("main.RedisInstrumentor") +def test_init_telemetry_mocks( + mock_redis_instrumentor, + mock_cloud_metrics, + mock_cloud_trace, + mock_metric_reader, + mock_span_processor, +): + """Verifies the init_telemetry function cleanly instruments Redis and the OTel SDK.""" + tp, mp = main.init_telemetry() + + assert tp is not None + assert mp is not None + 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 diff --git a/memorystore/redis/client_side_metrics/requirements.txt b/memorystore/redis/client_side_metrics/requirements.txt new file mode 100644 index 00000000000..d04d5f087e7 --- /dev/null +++ b/memorystore/redis/client_side_metrics/requirements.txt @@ -0,0 +1,7 @@ +redis +opentelemetry-api +opentelemetry-sdk +opentelemetry-instrumentation-redis +opentelemetry-exporter-gcp-trace +opentelemetry-exporter-gcp-monitoring +pytest \ No newline at end of file From 9fd6c9332257a08d8f8f0e4aff6a6ede9c6aa0b1 Mon Sep 17 00:00:00 2001 From: fosky94 Date: Thu, 9 Jul 2026 20:51:11 +0000 Subject: [PATCH 06/11] style: fix style violations --- memorystore/redis/client_side_metrics/main_test.py | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/memorystore/redis/client_side_metrics/main_test.py b/memorystore/redis/client_side_metrics/main_test.py index 11c5a924e95..24993178f94 100644 --- a/memorystore/redis/client_side_metrics/main_test.py +++ b/memorystore/redis/client_side_metrics/main_test.py @@ -14,13 +14,10 @@ from unittest import mock -from opentelemetry.instrumentation.redis import RedisInstrumentor +import main import pytest from redis.exceptions import ConnectionError -import main - - @pytest.fixture def mock_telemetry(monkeypatch): """Hermetically binds mock OpenTelemetry and Pool globals for the duration of the test.""" From 0eb3c9034acbd178a9bf12169ca4b48bce0c58e9 Mon Sep 17 00:00:00 2001 From: fosky94 Date: Thu, 9 Jul 2026 23:17:29 +0000 Subject: [PATCH 07/11] fix:fix comments from reviewer --- memorystore/redis/client_side_metrics/main.py | 118 +++++----- .../redis/client_side_metrics/main_test.py | 221 +++++++++++++----- .../client_side_metrics/requirements-test.txt | 1 + .../client_side_metrics/requirements.txt | 13 +- 4 files changed, 230 insertions(+), 123 deletions(-) create mode 100644 memorystore/redis/client_side_metrics/requirements-test.txt diff --git a/memorystore/redis/client_side_metrics/main.py b/memorystore/redis/client_side_metrics/main.py index 050a1c88956..10914bbfe7b 100644 --- a/memorystore/redis/client_side_metrics/main.py +++ b/memorystore/redis/client_side_metrics/main.py @@ -29,22 +29,11 @@ import redis from redis.exceptions import ConnectionError, TimeoutError -# Telemetry and Redis Globals (initialized dynamically in production or mocked in tests) -tracer = None -rtt_hist = None -client_block_hist = None -app_block_hist = None -retry_counter = None -conn_error_counter = None -pool = None -r = None -def init_telemetry(): - """Initializes the OpenTelemetry SDK with Google Cloud Exporters.""" - global tracer, rtt_hist, client_block_hist, app_block_hist - global retry_counter, conn_error_counter +def init_telemetry(): + """Initializes OpenTelemetry with GCP Exporters and returns the SDK objects.""" # 1. Initialize Tracing tracer_provider = TracerProvider() tracer_provider.add_span_processor( @@ -62,58 +51,61 @@ def init_telemetry(): metrics.set_meter_provider(meter_provider) meter = metrics.get_meter("redis.metrics") - rtt_hist = meter.create_histogram("redis_client_rtt", unit="ms") - client_block_hist = meter.create_histogram( - "redis_client_blocking_latency", unit="ms" - ) - app_block_hist = meter.create_histogram( - "redis_application_blocking_latency", unit="ms" - ) - retry_counter = meter.create_counter("redis_retry_count") - conn_error_counter = meter.create_counter( - "redis_connectivity_error_count" - ) - - retry_counter.add(0, {"operation": "startup"}) - conn_error_counter.add(0, {"operation": "startup"}) - - # Setup Redis Instrumentation + # Bundle all metric handlers safely into a dictionary + redis_metrics = { + "rtt_hist": meter.create_histogram("redis_client_rtt", unit="ms"), + "client_block_hist": meter.create_histogram( + "redis_client_blocking_latency", unit="ms" + ), + "app_block_hist": meter.create_histogram( + "redis_application_blocking_latency", unit="ms" + ), + "retry_counter": meter.create_counter("redis_retry_count"), + "conn_error_counter": meter.create_counter( + "redis_connectivity_error_count" + ), + } + + redis_metrics["retry_counter"].add(0, {"operation": "startup"}) + redis_metrics["conn_error_counter"].add(0, {"operation": "startup"}) + + # 3. Setup Redis Auto-Instrumentation RedisInstrumentor().instrument() - return tracer_provider, meter_provider + return tracer, redis_metrics, tracer_provider, meter_provider def init_redis_pool(): - """Initializes the Redis connection pool.""" - global pool, r + """Initializes and returns the Redis ConnectionPool and Client.""" redis_host = os.environ.get("REDISHOST", "localhost") redis_port = int(os.environ.get("REDISPORT", 6379)) - pool = redis.ConnectionPool( + redis_pool = redis.ConnectionPool( host=redis_host, port=redis_port, max_connections=10, decode_responses=True, ) - r = redis.Redis(connection_pool=pool) - return pool, r + redis_client = redis.Redis(connection_pool=redis_pool) + return redis_pool, redis_client -def smart_redis_call(operation_name, func, *args, **kwargs): - """Executes a Redis operation with latency metrics and retry handling.""" +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 pool_start = time.time() try: - # Check connection pool health - conn = pool.get_connection("PING") - pool.release(conn) + conn = redis_pool.get_connection("PING") + redis_pool.release(conn) except Exception: pass - if client_block_hist: - client_block_hist.record( + if metrics and metrics.get("client_block_hist"): + metrics["client_block_hist"].record( (time.time() - pool_start) * 1000, {"operation": operation_name} ) @@ -121,16 +113,18 @@ def smart_redis_call(operation_name, func, *args, **kwargs): try: req_start = time.time() response = func(*args, **kwargs) - if rtt_hist: - rtt_hist.record( + + if metrics and metrics.get("rtt_hist"): + metrics["rtt_hist"].record( (time.time() - req_start) * 1000, {"operation": operation_name}, ) app_start = time.time() _ = str(response) - if app_block_hist: - app_block_hist.record( + + if metrics and metrics.get("app_block_hist"): + metrics["app_block_hist"].record( (time.time() - app_start) * 1000, {"operation": operation_name}, ) @@ -139,26 +133,40 @@ def smart_redis_call(operation_name, func, *args, **kwargs): except (ConnectionError, TimeoutError) as e: attempt += 1 - if conn_error_counter: - conn_error_counter.add(1, {"operation": operation_name}) - if retry_counter: - retry_counter.add(1, {"operation": operation_name}) + if metrics and metrics.get("conn_error_counter"): + 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_provider, meter_provider = init_telemetry() - init_redis_pool() + tracer, redis_metrics, tracer_provider, meter_provider = init_telemetry() + redis_pool, redis_client = init_redis_pool() if tracer: with tracer.start_as_current_span("process_user_span"): try: # Simple write and read operations - smart_redis_call("set_user", r.set, "user:123", "active") + smart_redis_call( + "set_user", + redis_client.set, + redis_pool, + redis_metrics, + "user:123", + "active", + ) - result = smart_redis_call("get_user", r.get, "user:123") + result = smart_redis_call( + "get_user", + redis_client.get, + redis_pool, + redis_metrics, + "user:123", + ) print(f"Retrieved: {result}") except Exception as e: print(f"Error: {e}") diff --git a/memorystore/redis/client_side_metrics/main_test.py b/memorystore/redis/client_side_metrics/main_test.py index 24993178f94..139ece3e68c 100644 --- a/memorystore/redis/client_side_metrics/main_test.py +++ b/memorystore/redis/client_side_metrics/main_test.py @@ -15,64 +15,49 @@ 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 -@pytest.fixture -def mock_telemetry(monkeypatch): - """Hermetically binds mock OpenTelemetry and Pool globals for the duration of the test.""" - mock_tracer = mock.MagicMock() - mock_rtt = mock.MagicMock() - mock_client = mock.MagicMock() - mock_app = mock.MagicMock() - mock_retry = mock.MagicMock() - mock_conn_err = mock.MagicMock() - mock_pool = mock.MagicMock() - - # Prevent real time.sleep calls during retry handling to make tests lightning fast - monkeypatch.setattr(main.time, "sleep", lambda x: None) - - monkeypatch.setattr(main, "tracer", mock_tracer) - monkeypatch.setattr(main, "rtt_hist", mock_rtt) - monkeypatch.setattr(main, "client_block_hist", mock_client) - monkeypatch.setattr(main, "app_block_hist", mock_app) - monkeypatch.setattr(main, "retry_counter", mock_retry) - monkeypatch.setattr(main, "conn_error_counter", mock_conn_err) - monkeypatch.setattr(main, "pool", mock_pool) - - return { - "rtt_hist": mock_rtt, - "client_block_hist": mock_client, - "app_block_hist": mock_app, - "retry_counter": mock_retry, - "conn_error_counter": mock_conn_err, - "pool": mock_pool, - } - -def test_smart_redis_call_success(mock_telemetry): - """Verifies standard successful SET and GET execution paths and latency recording.""" +def test_smart_redis_call_success(): + """Verifies successful SET and GET execution paths and latency recording.""" mock_func = mock.MagicMock(return_value="active") + mock_pool = mock.MagicMock() + mock_metrics = { + "client_block_hist": mock.MagicMock(), + "rtt_hist": mock.MagicMock(), + "app_block_hist": mock.MagicMock(), + "retry_counter": mock.MagicMock(), + "conn_error_counter": mock.MagicMock(), + } - result = main.smart_redis_call("set_user", mock_func, "user:123", "active") + result = main.smart_redis_call( + "set_user", + mock_func, + mock_pool, + mock_metrics, + "user:123", + "active", + ) assert result == "active" mock_func.assert_called_once_with("user:123", "active") - mock_telemetry["pool"].get_connection.assert_called_once_with("PING") - mock_telemetry["pool"].release.assert_called_once() - # Verify all metrics were cleanly invoked - mock_telemetry["client_block_hist"].record.assert_called_once() - mock_telemetry["rtt_hist"].record.assert_called_once() - mock_telemetry["app_block_hist"].record.assert_called_once() + # Verify all 3 latency histograms were neatly invoked + mock_metrics["client_block_hist"].record.assert_called_once() + mock_metrics["rtt_hist"].record.assert_called_once() + mock_metrics["app_block_hist"].record.assert_called_once() # Verify retry counters were untouched - mock_telemetry["retry_counter"].add.assert_not_called() - mock_telemetry["conn_error_counter"].add.assert_not_called() + mock_metrics["retry_counter"].add.assert_not_called() + mock_metrics["conn_error_counter"].add.assert_not_called() -def test_smart_redis_call_retry_success(mock_telemetry): - """Simulates 2 sequential connectivity failures followed by a successful 3rd retry.""" +def test_smart_redis_call_retry_success(): + """Simulates 2 sequential connectivity failures followed by 3rd attempt success.""" mock_func = mock.MagicMock( side_effect=[ ConnectionError("cluster unreachable attempt 1"), @@ -80,43 +65,151 @@ def test_smart_redis_call_retry_success(mock_telemetry): "success_response", ] ) + mock_pool = mock.MagicMock() + mock_metrics = { + "client_block_hist": mock.MagicMock(), + "rtt_hist": mock.MagicMock(), + "app_block_hist": mock.MagicMock(), + "retry_counter": mock.MagicMock(), + "conn_error_counter": mock.MagicMock(), + } - result = main.smart_redis_call("get_user", mock_func, "user:123") + # Wrap in a sleep patch so the test suite runs in milliseconds + with mock.patch("time.sleep", return_value=None): + result = main.smart_redis_call( + "get_user", + mock_func, + mock_pool, + mock_metrics, + "user:123", + ) assert result == "success_response" assert mock_func.call_count == 3 # Verify counters captured exactly 2 retry events - assert mock_telemetry["retry_counter"].add.call_count == 2 - assert mock_telemetry["conn_error_counter"].add.call_count == 2 - - # Expect calls with increment of 1 and proper operation attribute - mock_telemetry["retry_counter"].add.assert_called_with( + 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_telemetry["conn_error_counter"].add.assert_called_with( + mock_metrics["conn_error_counter"].add.assert_called_with( 1, {"operation": "get_user"} ) - # Success on 3rd attempt means round-trip latency was recorded - mock_telemetry["rtt_hist"].record.assert_called_once() + # Success on 3rd attempt means latency was recorded + mock_metrics["app_block_hist"].record.assert_called_once() -def test_smart_redis_call_permanent_failure(mock_telemetry): - """Verifies that permanent connectivity failures bubble up and trigger exactly 3 attempts.""" +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_pool = mock.MagicMock() + mock_metrics = { + "client_block_hist": mock.MagicMock(), + "rtt_hist": mock.MagicMock(), + "app_block_hist": mock.MagicMock(), + "retry_counter": mock.MagicMock(), + "conn_error_counter": mock.MagicMock(), + } - with pytest.raises(ConnectionError) as exc_info: - main.smart_redis_call("set_user", mock_func, "user:123", "active") + with mock.patch("time.sleep", return_value=None): + with pytest.raises(ConnectionError) as exc_info: + main.smart_redis_call( + "set_user", + mock_func, + mock_pool, + mock_metrics, + "user:123", + "active", + ) assert "permanent DNS error" in str(exc_info.value) assert mock_func.call_count == 3 # Verify all 3 attempts were tracked - assert mock_telemetry["retry_counter"].add.call_count == 3 - assert mock_telemetry["conn_error_counter"].add.call_count == 3 + assert mock_metrics["retry_counter"].add.call_count == 3 + assert mock_metrics["conn_error_counter"].add.call_count == 3 + + # Verify latency was NOT recorded on a permanent failure + mock_metrics["app_block_hist"].record.assert_not_called() + + +def test_init_redis_pool_live(monkeypatch): + """Verifies init_redis_pool correctly parses environment variables without network calls.""" + monkeypatch.setenv("REDISHOST", "10.0.0.1") + monkeypatch.setenv("REDISPORT", "7000") + + redis_pool, redis_client = main.init_redis_pool() + + assert redis_pool is not None + assert redis_client is not None + assert redis_pool.connection_kwargs["host"] == "10.0.0.1" + assert redis_pool.connection_kwargs["port"] == 7000 + + # Test default fallback + monkeypatch.delenv("REDISHOST", raising=False) + monkeypatch.delenv("REDISPORT", raising=False) + + pool_def, r_def = main.init_redis_pool() + assert pool_def.connection_kwargs["host"] == "localhost" + assert pool_def.connection_kwargs["port"] == 6379 + + +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 + + # Initialize REAL No-op OpenTelemetry SDK providers + tp = TracerProvider() + mp = MeterProvider() + + try: + # Exercise actual pool creation and connectivity ping + redis_pool, redis_client = live_main.init_redis_pool() + redis_client.ping() + except (redis.exceptions.ConnectionError, redis.exceptions.TimeoutError): + pytest.skip( + "Skipping live Redis test: No running server found at localhost:6379" + ) + + # Bind all 5 real OpenTelemetry No-op metrics for this live execution block + meter = mp.get_meter("test_redis_live") + live_metrics = { + "client_block_hist": meter.create_histogram("client_block"), + "rtt_hist": meter.create_histogram("rtt_hist"), + "app_block_hist": meter.create_histogram("app_block"), + "retry_counter": meter.create_counter("retry_count"), + "conn_error_counter": meter.create_counter("conn_error"), + } + + try: + # Run a real Redis write and read hitting the ACTUAL endpoint! + result_set = live_main.smart_redis_call( + "set_test", + redis_client.set, + redis_pool, + live_metrics, + "test:key", + "live_val", + ) + result_get = live_main.smart_redis_call( + "get_test", + redis_client.get, + redis_pool, + live_metrics, + "test:key", + ) + + assert result_get == "live_val" + + # Safe DB Cleanup + redis_client.delete("test:key") + + finally: + redis_pool.disconnect() @mock.patch("main.BatchSpanProcessor") @@ -131,11 +224,17 @@ def test_init_telemetry_mocks( mock_metric_reader, mock_span_processor, ): - """Verifies the init_telemetry function cleanly instruments Redis and the OTel SDK.""" - tp, mp = main.init_telemetry() + """Verifies the init_telemetry function correctly instruments Redis.""" + tracer, redis_metrics, tp, mp = main.init_telemetry() + assert tracer is not None + assert redis_metrics is not None assert tp is not None assert mp is not None + assert "rtt_hist" in redis_metrics + assert "app_block_hist" in redis_metrics + 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 diff --git a/memorystore/redis/client_side_metrics/requirements-test.txt b/memorystore/redis/client_side_metrics/requirements-test.txt new file mode 100644 index 00000000000..d9ad1f716cc --- /dev/null +++ b/memorystore/redis/client_side_metrics/requirements-test.txt @@ -0,0 +1 @@ +pytest==9.0.3 \ No newline at end of file diff --git a/memorystore/redis/client_side_metrics/requirements.txt b/memorystore/redis/client_side_metrics/requirements.txt index d04d5f087e7..a4cf9c14472 100644 --- a/memorystore/redis/client_side_metrics/requirements.txt +++ b/memorystore/redis/client_side_metrics/requirements.txt @@ -1,7 +1,6 @@ -redis -opentelemetry-api -opentelemetry-sdk -opentelemetry-instrumentation-redis -opentelemetry-exporter-gcp-trace -opentelemetry-exporter-gcp-monitoring -pytest \ No newline at end of file +redis==5.3.1 +opentelemetry-api==1.37.0 +opentelemetry-sdk==1.37.0 +opentelemetry-instrumentation-redis==0.46b0 +opentelemetry-exporter-gcp-trace==1.12.0 +opentelemetry-exporter-gcp-monitoring==1.12.0a0 \ No newline at end of file From eee128ec26a44526bf68924f2b132552476d9ad2 Mon Sep 17 00:00:00 2001 From: fosky94 Date: Thu, 9 Jul 2026 23:20:36 +0000 Subject: [PATCH 08/11] style:remove unused variable lint --- memorystore/redis/client_side_metrics/main_test.py | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) diff --git a/memorystore/redis/client_side_metrics/main_test.py b/memorystore/redis/client_side_metrics/main_test.py index 139ece3e68c..e90dc2c7407 100644 --- a/memorystore/redis/client_side_metrics/main_test.py +++ b/memorystore/redis/client_side_metrics/main_test.py @@ -161,9 +161,13 @@ 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 - # Initialize REAL No-op OpenTelemetry SDK providers - tp = TracerProvider() + # Initialize REAL No-op OpenTelemetry SDK providers (invoked directly) + TracerProvider() mp = MeterProvider() try: @@ -186,8 +190,8 @@ def test_smart_redis_call_live_or_skip(): } try: - # Run a real Redis write and read hitting the ACTUAL endpoint! - result_set = live_main.smart_redis_call( + # Run a real Redis write hitting the ACTUAL endpoint (invoked directly) + live_main.smart_redis_call( "set_test", redis_client.set, redis_pool, @@ -195,6 +199,7 @@ def test_smart_redis_call_live_or_skip(): "test:key", "live_val", ) + result_get = live_main.smart_redis_call( "get_test", redis_client.get, From 453ddc9f326e003a4fec4debe3dc4e93c8257f3c Mon Sep 17 00:00:00 2001 From: fosky94 Date: Thu, 9 Jul 2026 23:23:13 +0000 Subject: [PATCH 09/11] style:remove global imports lint --- memorystore/redis/client_side_metrics/main_test.py | 3 --- 1 file changed, 3 deletions(-) diff --git a/memorystore/redis/client_side_metrics/main_test.py b/memorystore/redis/client_side_metrics/main_test.py index e90dc2c7407..81c95552c83 100644 --- a/memorystore/redis/client_side_metrics/main_test.py +++ b/memorystore/redis/client_side_metrics/main_test.py @@ -15,10 +15,7 @@ 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 From 9a4d35371f7264efa23c79752a768df4cd81d5bf Mon Sep 17 00:00:00 2001 From: fosky94 Date: Fri, 10 Jul 2026 02:07:26 +0000 Subject: [PATCH 10/11] fix:change version requirements --- memorystore/redis/client_side_metrics/main.py | 2 +- memorystore/redis/client_side_metrics/main_test.py | 3 ++- .../redis/client_side_metrics/requirements-test.txt | 2 +- .../redis/client_side_metrics/requirements.txt | 12 ++++++------ 4 files changed, 10 insertions(+), 9 deletions(-) diff --git a/memorystore/redis/client_side_metrics/main.py b/memorystore/redis/client_side_metrics/main.py index 10914bbfe7b..eb47d7002d5 100644 --- a/memorystore/redis/client_side_metrics/main.py +++ b/memorystore/redis/client_side_metrics/main.py @@ -99,7 +99,7 @@ def smart_redis_call( pool_start = time.time() try: - conn = redis_pool.get_connection("PING") + conn = redis_pool.get_connection() redis_pool.release(conn) except Exception: pass diff --git a/memorystore/redis/client_side_metrics/main_test.py b/memorystore/redis/client_side_metrics/main_test.py index 81c95552c83..196d758a54d 100644 --- a/memorystore/redis/client_side_metrics/main_test.py +++ b/memorystore/redis/client_side_metrics/main_test.py @@ -239,4 +239,5 @@ def test_init_telemetry_mocks( 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_cloud_metrics.assert_called_once() + \ No newline at end of file diff --git a/memorystore/redis/client_side_metrics/requirements-test.txt b/memorystore/redis/client_side_metrics/requirements-test.txt index d9ad1f716cc..9471b3d92fb 100644 --- a/memorystore/redis/client_side_metrics/requirements-test.txt +++ b/memorystore/redis/client_side_metrics/requirements-test.txt @@ -1 +1 @@ -pytest==9.0.3 \ No newline at end of file +pytest==8.4.2 diff --git a/memorystore/redis/client_side_metrics/requirements.txt b/memorystore/redis/client_side_metrics/requirements.txt index a4cf9c14472..e8caf96a29a 100644 --- a/memorystore/redis/client_side_metrics/requirements.txt +++ b/memorystore/redis/client_side_metrics/requirements.txt @@ -1,6 +1,6 @@ -redis==5.3.1 -opentelemetry-api==1.37.0 -opentelemetry-sdk==1.37.0 -opentelemetry-instrumentation-redis==0.46b0 -opentelemetry-exporter-gcp-trace==1.12.0 -opentelemetry-exporter-gcp-monitoring==1.12.0a0 \ No newline at end of file +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 From 690eaa060448f4bfed1bde3d2ad44e5d5d96d8ec Mon Sep 17 00:00:00 2001 From: fosky94 Date: Fri, 10 Jul 2026 16:32:50 +0000 Subject: [PATCH 11/11] fix:use pytest version 9.0.3 --- memorystore/redis/client_side_metrics/requirements-test.txt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/memorystore/redis/client_side_metrics/requirements-test.txt b/memorystore/redis/client_side_metrics/requirements-test.txt index 9471b3d92fb..b3b2b3ca4cc 100644 --- a/memorystore/redis/client_side_metrics/requirements-test.txt +++ b/memorystore/redis/client_side_metrics/requirements-test.txt @@ -1 +1 @@ -pytest==8.4.2 +pytest==9.0.3