AAP-57817 Add Redis connection retry using redis-py 7.0+ built-in (#16176)

djyasin · web-flow · commit 4f41b50a09e1 · 2025-12-01T09:08:47.000-05:00
* AAP-57817 Add Redis connection retry using redis-py 7.0+ built-in mechanism

* Refactor Redis client helpers to use settings and eliminate code duplication

* Create awx/main/utils/redis.py and move Redis client functions to avoid circular imports

* Fix subsystem_metrics to share Redis connection pool between
  client and pipeline

* Cache Redis clients in RelayConsumer and RelayWebsocketStatsManager to avoid creating new connection pools on every call

* Add cap and base config

* Add Redis retry logic with exponential backoff to handle connection failures during long-running operations

* Add REDIS_BACKOFF_CAP and REDIS_BACKOFF_BASE settings to allow
  adjustment of retry timing in worst-case scenarios without code changes

* Simplify Redis retry tests by removing unnecessary reload logic
diff --git a/awx/main/analytics/broadcast_websocket.py b/awx/main/analytics/broadcast_websocket.py
@@ -1,8 +1,6 @@
 import datetime
 import asyncio
 import logging
-import redis
-import redis.asyncio
 import re
 
 from prometheus_client import (
@@ -15,6 +13,7 @@
 )
 
 from django.conf import settings
+from awx.main.utils.redis import get_redis_client, get_redis_client_async
 
 
 BROADCAST_WEBSOCKET_REDIS_KEY_NAME = 'broadcast_websocket_stats'
@@ -66,6 +65,8 @@ def render(self, ts=None):
 
 
 class RelayWebsocketStatsManager:
+    _redis_client = None  # Cached Redis client for get_stats_sync()
+
     def __init__(self, local_hostname):
         self._local_hostname = local_hostname
         self._stats = dict()
@@ -80,7 +81,7 @@ def delete_remote_host_stats(self, remote_hostname):
 
     async def run_loop(self):
         try:
-            redis_conn = await redis.asyncio.Redis.from_url(settings.BROKER_URL)
+            redis_conn = get_redis_client_async()
             while True:
                 stats_data_str = ''.join(stat.serialize() for stat in self._stats.values())
                 await redis_conn.set(self._redis_key, stats_data_str)
@@ -103,8 +104,10 @@ def get_stats_sync(cls):
         """
         Stringified verion of all the stats
         """
-        redis_conn = redis.Redis.from_url(settings.BROKER_URL)
-        stats_str = redis_conn.get(BROADCAST_WEBSOCKET_REDIS_KEY_NAME) or b''
+        # Reuse cached Redis client to avoid creating new connection pools on every call
+        if cls._redis_client is None:
+            cls._redis_client = get_redis_client()
+        stats_str = cls._redis_client.get(BROADCAST_WEBSOCKET_REDIS_KEY_NAME) or b''
         return parser.text_string_to_metric_families(stats_str.decode('UTF-8'))
 
 
diff --git a/awx/main/analytics/subsystem_metrics.py b/awx/main/analytics/subsystem_metrics.py
@@ -14,6 +14,7 @@
 
 from awx.main.consumers import emit_channel_notification
 from awx.main.utils import is_testing
+from awx.main.utils.redis import get_redis_client
 
 root_key = settings.SUBSYSTEM_METRICS_REDIS_KEY_PREFIX
 logger = logging.getLogger('awx.main.analytics')
@@ -198,8 +199,8 @@ class Metrics(MetricsNamespace):
     def __init__(self, namespace, auto_pipe_execute=False, instance_name=None, metrics_have_changed=True, **kwargs):
         MetricsNamespace.__init__(self, namespace)
 
-        self.pipe = redis.Redis.from_url(settings.BROKER_URL).pipeline()
-        self.conn = redis.Redis.from_url(settings.BROKER_URL)
+        self.conn = get_redis_client()
+        self.pipe = self.conn.pipeline()
         self.last_pipe_execute = time.time()
         # track if metrics have been modified since last saved to redis
         # start with True so that we get an initial save to redis
diff --git a/awx/main/consumers.py b/awx/main/consumers.py
@@ -3,7 +3,6 @@
 import time
 import hmac
 import asyncio
-import redis
 
 from django.core.serializers.json import DjangoJSONEncoder
 from django.conf import settings
@@ -14,6 +13,8 @@
 from channels.layers import get_channel_layer
 from channels.db import database_sync_to_async
 
+from awx.main.utils.redis import get_redis_client_async
+
 logger = logging.getLogger('awx.main.consumers')
 XRF_KEY = '_auth_user_xrf'
 
@@ -94,6 +95,9 @@ async def connect(self):
         await self.channel_layer.group_add(settings.BROADCAST_WEBSOCKET_GROUP_NAME, self.channel_name)
         logger.info(f"client '{self.channel_name}' joined the broadcast group.")
 
+        # Initialize Redis client once for reuse across all message handling
+        self._redis_conn = get_redis_client_async()
+
     async def disconnect(self, code):
         logger.info(f"client '{self.channel_name}' disconnected from the broadcast group.")
         await self.channel_layer.group_discard(settings.BROADCAST_WEBSOCKET_GROUP_NAME, self.channel_name)
@@ -105,8 +109,9 @@ async def receive_json(self, data):
         (group, message) = unwrap_broadcast_msg(data)
         if group == "metrics":
             message = json.loads(message['text'])
-            conn = redis.Redis.from_url(settings.BROKER_URL)
-            conn.set(settings.SUBSYSTEM_METRICS_REDIS_KEY_PREFIX + "-" + message['metrics_namespace'] + "_instance_" + message['instance'], message['metrics'])
+            await self._redis_conn.set(
+                settings.SUBSYSTEM_METRICS_REDIS_KEY_PREFIX + "-" + message['metrics_namespace'] + "_instance_" + message['instance'], message['metrics']
+            )
         else:
             await self.channel_layer.group_send(group, message)
 
diff --git a/awx/main/dispatch/control.py b/awx/main/dispatch/control.py
@@ -2,11 +2,10 @@
 import uuid
 import json
 
-from django.conf import settings
 from django.db import connection
-import redis
 
 from awx.main.dispatch import get_task_queuename
+from awx.main.utils.redis import get_redis_client
 
 from . import pg_bus_conn
 
@@ -24,7 +23,7 @@ def __init__(self, service, host=None):
         self.queuename = host or get_task_queuename()
 
     def status(self, *args, **kwargs):
-        r = redis.Redis.from_url(settings.BROKER_URL)
+        r = get_redis_client()
         if self.service == 'dispatcher':
             stats = r.get(f'awx_{self.service}_statistics') or b''
             return stats.decode('utf-8')
diff --git a/awx/main/dispatch/worker/base.py b/awx/main/dispatch/worker/base.py
@@ -19,6 +19,7 @@
 
 from ansible_base.lib.logging.runtime import log_excess_runtime
 
+from awx.main.utils.redis import get_redis_client
 from awx.main.dispatch.pool import WorkerPool
 from awx.main.dispatch.periodic import Scheduler
 from awx.main.dispatch import pg_bus_conn
@@ -59,7 +60,7 @@ def __init__(self, name, worker, queues=[], pool=None):
         if pool is None:
             self.pool = WorkerPool()
         self.pool.init_workers(self.worker.work_loop)
-        self.redis = redis.Redis.from_url(settings.BROKER_URL)
+        self.redis = get_redis_client()
 
     @property
     def listening_on(self):
diff --git a/awx/main/dispatch/worker/callback.py b/awx/main/dispatch/worker/callback.py
@@ -15,6 +15,7 @@
 
 import redis
 
+from awx.main.utils.redis import get_redis_client
 from awx.main.consumers import emit_channel_notification
 from awx.main.models import JobEvent, AdHocCommandEvent, ProjectUpdateEvent, InventoryUpdateEvent, SystemJobEvent, UnifiedJob
 from awx.main.constants import ACTIVE_STATES
@@ -72,7 +73,7 @@ class CallbackBrokerWorker(BaseWorker):
 
     def __init__(self):
         self.buff = {}
-        self.redis = redis.Redis.from_url(settings.BROKER_URL)
+        self.redis = get_redis_client()
         self.subsystem_metrics = s_metrics.CallbackReceiverMetrics(auto_pipe_execute=False)
         self.queue_pop = 0
         self.queue_name = settings.CALLBACK_QUEUE
diff --git a/awx/main/models/ha.py b/awx/main/models/ha.py
@@ -33,6 +33,7 @@
 )
 from awx.main.models.unified_jobs import UnifiedJob
 from awx.main.utils.common import get_corrected_cpu, get_cpu_effective_capacity, get_corrected_memory, get_mem_effective_capacity
+from awx.main.utils.redis import get_redis_client
 from awx.main.models.mixins import RelatedJobsMixin, ResourceMixin
 from awx.main.models.receptor_address import ReceptorAddress
 
@@ -397,7 +398,7 @@ def local_health_check(self):
         try:
             # if redis is down for some reason, that means we can't persist
             # playbook event data; we should consider this a zero capacity event
-            redis.Redis.from_url(settings.BROKER_URL).ping()
+            get_redis_client().ping()
         except redis.ConnectionError:
             errors = _('Failed to connect to Redis')
 
diff --git a/awx/main/queue.py b/awx/main/queue.py
@@ -4,11 +4,13 @@
 # Python
 import json
 import logging
-import redis
 
 # Django
 from django.conf import settings
 
+# AWX
+from awx.main.utils.redis import get_redis_client
+
 __all__ = ['CallbackQueueDispatcher']
 
 
@@ -26,7 +28,7 @@ class CallbackQueueDispatcher(object):
     def __init__(self):
         self.queue = getattr(settings, 'CALLBACK_QUEUE', '')
         self.logger = logging.getLogger('awx.main.queue.CallbackQueueDispatcher')
-        self.connection = redis.Redis.from_url(settings.BROKER_URL)
+        self.connection = get_redis_client()
 
     def dispatch(self, obj):
         self.connection.rpush(self.queue, json.dumps(obj, cls=AnsibleJSONEncoder))
diff --git a/awx/main/routing.py b/awx/main/routing.py
@@ -8,6 +8,7 @@
 
 from ansible_base.lib.channels.middleware import DrfAuthMiddlewareStack
 
+from awx.main.utils.redis import get_redis_client
 from . import consumers
 
 
@@ -18,7 +19,7 @@
 class AWXProtocolTypeRouter(ProtocolTypeRouter):
     def __init__(self, *args, **kwargs):
         try:
-            r = redis.Redis.from_url(settings.BROKER_URL)
+            r = get_redis_client()
             for k in r.scan_iter('asgi:*', 500):
                 logger.debug(f"cleaning up Redis key {k}")
                 r.delete(k)
diff --git a/awx/main/tests/functional/conftest.py b/awx/main/tests/functional/conftest.py
@@ -77,15 +77,34 @@ def swagger_autogen(requests=__SWAGGER_REQUESTS__):
 
 
 class FakeRedis:
+    def __init__(self, *args, **kwargs):
+        # Accept and ignore all arguments to match redis.Redis signature
+        pass
+
     def keys(self, *args, **kwargs):
         return []
 
-    def set(self):
+    def set(self, *args, **kwargs):
         pass
 
-    def get(self):
+    def get(self, *args, **kwargs):
+        return None
+
+    def rpush(self, *args, **kwargs):
+        return 1
+
+    def blpop(self, *args, **kwargs):
         return None
 
+    def delete(self, *args, **kwargs):
+        pass
+
+    def llen(self, *args, **kwargs):
+        return 0
+
+    def scan_iter(self, *args, **kwargs):
+        return iter([])
+
     @classmethod
     def from_url(cls, *args, **kwargs):
         return cls()
diff --git a/awx/main/tests/unit/analytics/test_broadcast_websocket.py b/awx/main/tests/unit/analytics/test_broadcast_websocket.py
@@ -1,6 +1,7 @@
 import datetime
+from unittest.mock import Mock, patch
 
-from awx.main.analytics.broadcast_websocket import FixedSlidingWindow
+from awx.main.analytics.broadcast_websocket import FixedSlidingWindow, RelayWebsocketStatsManager
 from awx.main.analytics.broadcast_websocket import dt_to_seconds
 
 
@@ -59,3 +60,70 @@ def test_record_same_minute_render_diff_minute(self):
             assert 20 - i == fsw.render(self.ts(minute=1, second=i, microsecond=0)), "E. Sliding window where 1 record() should drop from the results each time"
 
         assert 0 == fsw.render(self.ts(minute=1, second=20, microsecond=0)), "F. First second one minute after all record() calls"
+
+
+class TestRelayWebsocketStatsManager:
+    """Test Redis client caching in RelayWebsocketStatsManager."""
+
+    def test_get_stats_sync_caches_redis_client(self):
+        """Verify get_stats_sync caches Redis client to avoid creating new connection pools."""
+        # Reset class variable
+        RelayWebsocketStatsManager._redis_client = None
+
+        mock_redis = Mock()
+        mock_redis.get.return_value = b''
+
+        with patch('awx.main.analytics.broadcast_websocket.get_redis_client', return_value=mock_redis) as mock_get_client:
+            # First call should create client
+            RelayWebsocketStatsManager.get_stats_sync()
+            assert mock_get_client.call_count == 1
+
+            # Second call should reuse cached client
+            RelayWebsocketStatsManager.get_stats_sync()
+            assert mock_get_client.call_count == 1  # Still 1, not called again
+
+            # Third call should still reuse cached client
+            RelayWebsocketStatsManager.get_stats_sync()
+            assert mock_get_client.call_count == 1
+
+        # Cleanup
+        RelayWebsocketStatsManager._redis_client = None
+
+    def test_get_stats_sync_returns_parsed_metrics(self):
+        """Verify get_stats_sync returns parsed metric families from Redis."""
+        # Reset class variable
+        RelayWebsocketStatsManager._redis_client = None
+
+        # Sample Prometheus metrics format
+        sample_metrics = b'# HELP test_metric A test metric\n# TYPE test_metric gauge\ntest_metric 42\n'
+
+        mock_redis = Mock()
+        mock_redis.get.return_value = sample_metrics
+
+        with patch('awx.main.analytics.broadcast_websocket.get_redis_client', return_value=mock_redis):
+            result = list(RelayWebsocketStatsManager.get_stats_sync())
+
+            # Should return parsed metric families
+            assert len(result) > 0
+            assert mock_redis.get.called
+
+        # Cleanup
+        RelayWebsocketStatsManager._redis_client = None
+
+    def test_get_stats_sync_handles_empty_redis_data(self):
+        """Verify get_stats_sync handles empty data from Redis gracefully."""
+        # Reset class variable
+        RelayWebsocketStatsManager._redis_client = None
+
+        mock_redis = Mock()
+        mock_redis.get.return_value = None  # Redis returns None when key doesn't exist
+
+        with patch('awx.main.analytics.broadcast_websocket.get_redis_client', return_value=mock_redis):
+            result = list(RelayWebsocketStatsManager.get_stats_sync())
+
+            # Should handle empty data gracefully
+            assert result == []
+            assert mock_redis.get.called
+
+        # Cleanup
+        RelayWebsocketStatsManager._redis_client = None
diff --git a/awx/main/tests/unit/utils/test_redis.py b/awx/main/tests/unit/utils/test_redis.py
@@ -0,0 +1,61 @@
+# -*- coding: utf-8 -*-
+
+# Copyright (c) 2025 Ansible, Inc.
+# All Rights Reserved
+
+from django.test.utils import override_settings
+
+from awx.main.utils.redis import get_redis_client, get_redis_client_async
+from redis.exceptions import BusyLoadingError, ConnectionError, TimeoutError
+from redis.backoff import ExponentialBackoff
+
+
+class TestRedisRetryConfiguration:
+    """Verify Redis retry configuration is applied to connection objects."""
+
+    def test_retry_configuration_applied_to_client(self, settings):
+        """Verify all retry settings are applied to the connection pool."""
+        # Test sync client
+        client = get_redis_client()
+        retry = client.connection_pool.connection_kwargs['retry']
+        backoff = retry._backoff
+        retry_errors = client.connection_pool.connection_kwargs['retry_on_error']
+
+        # Assert provided values match values on the object
+        assert retry._retries == settings.REDIS_RETRY_COUNT == 3
+        assert isinstance(backoff, ExponentialBackoff)
+        assert backoff._base == settings.REDIS_BACKOFF_BASE == 0.5
+        assert backoff._cap == settings.REDIS_BACKOFF_CAP == 1.0
+        assert BusyLoadingError in retry_errors
+        assert ConnectionError in retry_errors
+        assert TimeoutError in retry_errors
+
+        # Test async client has same config
+        client_async = get_redis_client_async()
+        retry_async = client_async.connection_pool.connection_kwargs['retry']
+        backoff_async = retry_async._backoff
+        retry_errors_async = client_async.connection_pool.connection_kwargs['retry_on_error']
+
+        assert retry_async._retries == settings.REDIS_RETRY_COUNT
+        assert backoff_async._base == settings.REDIS_BACKOFF_BASE
+        assert backoff_async._cap == settings.REDIS_BACKOFF_CAP
+        assert ConnectionError in retry_errors_async
+
+    @override_settings(REDIS_RETRY_COUNT=5)
+    def test_override_settings_applied_to_client(self):
+        """Verify override_settings changes are applied to client object."""
+        client = get_redis_client()
+        retry = client.connection_pool.connection_kwargs['retry']
+
+        assert retry._retries == 5
+
+    @override_settings(REDIS_BACKOFF_CAP=2.0, REDIS_BACKOFF_BASE=1.0)
+    def test_override_backoff_settings_applied_to_client(self):
+        """Verify override_settings for backoff parameters are applied to client object."""
+        client = get_redis_client()
+        retry = client.connection_pool.connection_kwargs['retry']
+        backoff = retry._backoff
+
+        # Assert provided values match values on object
+        assert backoff._cap == 2.0
+        assert backoff._base == 1.0
diff --git a/awx/main/utils/__init__.py b/awx/main/utils/__init__.py
diff --git a/awx/main/utils/redis.py b/awx/main/utils/redis.py
diff --git a/awx/settings/defaults.py b/awx/settings/defaults.py
diff --git a/requirements/requirements.in b/requirements/requirements.in
diff --git a/requirements/requirements.txt b/requirements/requirements.txt
