Create Client side metrics

memorystore/redis/client_side_metrics/main.py

@@ -0,0 +1,168 @@
+# 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.
+
+# [START memorystore_redis_client_side_metrics]
+import os
+import time
+
+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.sdk.trace import TracerProvider
+from opentelemetry.sdk.trace.export import BatchSpanProcessor
+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
+
+    # 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
+
+
+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:
+        # Check connection pool health
+        conn = pool.get_connection("PING")
+        pool.release(conn)
+    except Exception:
+        pass
+
+    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)
+            if rtt_hist:
+                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(
+                    (time.time() - app_start) * 1000,
+                    {"operation": operation_name},
+                )
+
+            return response
+
+        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 attempt >= max_retries:
+                raise e
+            time.sleep((2**attempt) * 0.1)
+
+
+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]

memorystore/redis/client_side_metrics/main_test.py

@@ -0,0 +1,141 @@
+# 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
+
+import main
+import pytest
+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."""
+    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()

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