Repository: redis/redis-py (Python)

Issue:
Getting ReadOnly error when writing data to AWS elastic cache while the cluster is vertically scaled
I have a python application that writes some data to AWS elastic cache cluster at a regular interval.

Here's an example script that simulates the functionality of my application. It also replicates the error that I have been facing.

```py
from datetime import datetime
import redis
import time
import sys
import random

class Config:
    default_host = "localhost"
    master_host = "xxx.xx.0001.xxxx.cache.amazonaws.com"
    replica_host = "xxx.xx.0001.xxxx.cache.amazonaws.com"
    redis_db = 8
    socket_conn_timeout = 10
    request_delay_sec = 0.1

def get_redis_client():
    return redis.Redis(
        host=Config.master_host,
        db=Config.redis_db,
        socket_connect_timeout=Config.socket_conn_timeout,
    )

def get_random_key_value():
    val = time.time()
    key = "test_key_" + str(random.randint(0, 100))
    return key, val

r = get_redis_client()

r.flushdb()

flag = False

while True:
    try:
        if flag:
            print("beat:", time.time())
        r.set(*get_random_key_value())
        time.sleep(Config.request_delay_sec)
    except redis.RedisError as re:
        print(datetime.now(), "Error:", type(re), re)
        flag = True
        # sys.exit()
    except KeyboardInterrupt:
        print("Stopping loop execution")
        sys.exit()
```

Here are the environment details of my application

 - OS(Ubuntu 18.04)
 - Python(v 3.7.0)
 - Redis-py(v 3.5.3)
 - AWS elastic cache(cluster mode disabled, 1 master node, 1 read replica)

When I scale my AWS elastic cache cluster vertically and the above script is running, I get the following error for few seconds while the cluster scaling-up is in process and then it goes away.

```bash
<class 'redis.exceptions.ReadOnlyError'> You can't write against a read only replica.
```

AWS docs also states that during vertical scaling process some inconsistencies may occur because of data syncing(https://docs.aws.amazon.com/AmazonElastiCache/latest/red-ug/redis-cluster-vertical-scaling.html)

why this error occurs during the scale up process? how can it be fixed?

I tried the same thing with a golang script and it works perfectly fine.


Hints:
@Akshit8 This looks like an issue in ElastiCache as opposed to redis-py. Would you mind having a look at redis-py 4? This a release back - and to be honest, I'd expect the same behaviour.

If you're feeling particularly testworthy I'd *love* a hand testing the cluster code we pulled in from @barshaul in redis-py 4.1.0rc1.
I'll try with redis-py 4.

The above version of the script in Golang works fine, could it be related DNS resolution?
I don't think so. Ultimately it's (Redis) ElastiCache returning that error.  That phrase doesn't appear anywhere in the codebase, so we're surfacing the alert you receive.
Hey @Akshit8, I will try to reproduce it this week
Could you share the golang script you're using too?
@barshaul

```go
package main

import (
	"context"
	"fmt"
	"log"
	"os"
	"time"

	"github.com/go-redis/redis/v8"
)

const (
	DEFAULT_HOST              = "localhost"
	MASTER_HOST               = "xxx.ng.0001.xxx.cache.amazonaws.com"
	REPLICA_HOST              = "xxx.xx.ng.0001.xxx.cache.amazonaws.com"
	REDIS_DB_KEY              = 8
	SOCKET_CONNECTION_TIMEOUT = 5
	REQUEST_DELAY_SEC         = 100
)

func getRedisClient() *redis.Client {
	return redis.NewClient(&redis.Options{
		Addr:        fmt.Sprintf("%s:%d", DEFAULT_HOST, 6379),
		DB:          REDIS_DB_KEY,
		DialTimeout: time.Duration(SOCKET_CONNECTION_TIMEOUT * time.Second),
	})
}

func getRandomKeyValue() (string, string) {
	key := fmt.Sprintf("TEST_KEY_%d", time.Now().Unix())
	value := "THIS IS A TEST MESSAGE"
	return key, value
}

func main() {
	log.Println("starting execution")
	ctx := context.Background()
	flag := false
	rc := getRedisClient()

	err := rc.FlushDB(ctx).Err()
	if err != nil {
		log.Println("ERROR:", err)
		os.Exit(1)
	}

	for {
		key, value := getRandomKeyValue()
		err := rc.Set(ctx, key, value, time.Hour).Err()
		if flag {
			fmt.Println("beat:", time.Now())
		}
		if err != nil {
			flag = true
			log.Println("ERROR:", err)
		}

		time.Sleep(time.Duration(REQUEST_DELAY_SEC * time.Millisecond))
	}
}
```

Patch:
diff --git a/redis/client.py b/redis/client.py
--- a/redis/client.py
+++ b/redis/client.py
@@ -868,6 +868,7 @@ def __init__(
         errors=None,
         decode_responses=False,
         retry_on_timeout=False,
+        retry_on_error=[],
         ssl=False,
         ssl_keyfile=None,
         ssl_certfile=None,
@@ -886,8 +887,10 @@ def __init__(
     ):
         """
         Initialize a new Redis client.
-        To specify a retry policy, first set `retry_on_timeout` to `True`
-        then set `retry` to a valid `Retry` object
+        To specify a retry policy for specific errors, first set
+        `retry_on_error` to a list of the error/s to retry on, then set
+        `retry` to a valid `Retry` object.
+        To retry on TimeoutError, `retry_on_timeout` can also be set to `True`.
         """
         if not connection_pool:
             if charset is not None:
@@ -904,7 +907,8 @@ def __init__(
                     )
                 )
                 encoding_errors = errors
-
+            if retry_on_timeout is True:
+                retry_on_error.append(TimeoutError)
             kwargs = {
                 "db": db,
                 "username": username,
@@ -913,7 +917,7 @@ def __init__(
                 "encoding": encoding,
                 "encoding_errors": encoding_errors,
                 "decode_responses": decode_responses,
-                "retry_on_timeout": retry_on_timeout,
+                "retry_on_error": retry_on_error,
                 "retry": copy.deepcopy(retry),
                 "max_connections": max_connections,
                 "health_check_interval": health_check_interval,
@@ -1144,11 +1148,14 @@ def _send_command_parse_response(self, conn, command_name, *args, **options):
     def _disconnect_raise(self, conn, error):
         """
         Close the connection and raise an exception
-        if retry_on_timeout is not set or the error
-        is not a TimeoutError
+        if retry_on_error is not set or the error
+        is not one of the specified error types
         """
         conn.disconnect()
-        if not (conn.retry_on_timeout and isinstance(error, TimeoutError)):
+        if (
+            conn.retry_on_error is None
+            or isinstance(error, tuple(conn.retry_on_error)) is False
+        ):
             raise error
 
     # COMMAND EXECUTION AND PROTOCOL PARSING
diff --git a/redis/cluster.py b/redis/cluster.py
--- a/redis/cluster.py
+++ b/redis/cluster.py
@@ -996,9 +996,11 @@ def _execute_command(self, target_node, *args, **kwargs):
                 self.reinitialize_counter += 1
                 if self._should_reinitialized():
                     self.nodes_manager.initialize()
+                    # Reset the counter
+                    self.reinitialize_counter = 0
                 else:
                     self.nodes_manager.update_moved_exception(e)
-                    moved = True
+                moved = True
             except TryAgainError:
                 log.exception("TryAgainError")
 
@@ -1320,6 +1322,7 @@ def initialize(self):
         tmp_slots = {}
         disagreements = []
         startup_nodes_reachable = False
+        fully_covered = False
         kwargs = self.connection_kwargs
         for startup_node in self.startup_nodes.values():
             try:
@@ -1431,6 +1434,12 @@ def initialize(self):
                                     f'slots cache: {", ".join(disagreements)}'
                                 )
 
+            fully_covered = self.check_slots_coverage(tmp_slots)
+            if fully_covered:
+                # Don't need to continue to the next startup node if all
+                # slots are covered
+                break
+
         if not startup_nodes_reachable:
             raise RedisClusterException(
                 "Redis Cluster cannot be connected. Please provide at least "
@@ -1440,7 +1449,6 @@ def initialize(self):
         # Create Redis connections to all nodes
         self.create_redis_connections(list(tmp_nodes_cache.values()))
 
-        fully_covered = self.check_slots_coverage(tmp_slots)
         # Check if the slots are not fully covered
         if not fully_covered and self._require_full_coverage:
             # Despite the requirement that the slots be covered, there
@@ -1478,6 +1486,8 @@ def initialize(self):
         self.default_node = self.get_nodes_by_server_type(PRIMARY)[0]
         # Populate the startup nodes with all discovered nodes
         self.populate_startup_nodes(self.nodes_cache.values())
+        # If initialize was called after a MovedError, clear it
+        self._moved_exception = None
 
     def close(self):
         self.default_node = None
diff --git a/redis/connection.py b/redis/connection.py
--- a/redis/connection.py
+++ b/redis/connection.py
@@ -513,6 +513,7 @@ def __init__(
         socket_keepalive_options=None,
         socket_type=0,
         retry_on_timeout=False,
+        retry_on_error=[],
         encoding="utf-8",
         encoding_errors="strict",
         decode_responses=False,
@@ -526,8 +527,10 @@ def __init__(
     ):
         """
         Initialize a new Connection.
-        To specify a retry policy, first set `retry_on_timeout` to `True`
-        then set `retry` to a valid `Retry` object
+        To specify a retry policy for specific errors, first set
+        `retry_on_error` to a list of the error/s to retry on, then set
+        `retry` to a valid `Retry` object.
+        To retry on TimeoutError, `retry_on_timeout` can also be set to `True`.
         """
         self.pid = os.getpid()
         self.host = host
@@ -543,11 +546,17 @@ def __init__(
         self.socket_type = socket_type
         self.retry_on_timeout = retry_on_timeout
         if retry_on_timeout:
+            # Add TimeoutError to the errors list to retry on
+            retry_on_error.append(TimeoutError)
+        self.retry_on_error = retry_on_error
+        if retry_on_error:
             if retry is None:
                 self.retry = Retry(NoBackoff(), 1)
             else:
                 # deep-copy the Retry object as it is mutable
                 self.retry = copy.deepcopy(retry)
+            # Update the retry's supported errors with the specified errors
+            self.retry.update_supported_erros(retry_on_error)
         else:
             self.retry = Retry(NoBackoff(), 0)
         self.health_check_interval = health_check_interval
@@ -969,6 +978,7 @@ def __init__(
         encoding_errors="strict",
         decode_responses=False,
         retry_on_timeout=False,
+        retry_on_error=[],
         parser_class=DefaultParser,
         socket_read_size=65536,
         health_check_interval=0,
@@ -978,8 +988,10 @@ def __init__(
     ):
         """
         Initialize a new UnixDomainSocketConnection.
-        To specify a retry policy, first set `retry_on_timeout` to `True`
-        then set `retry` to a valid `Retry` object
+        To specify a retry policy for specific errors, first set
+        `retry_on_error` to a list of the error/s to retry on, then set
+        `retry` to a valid `Retry` object.
+        To retry on TimeoutError, `retry_on_timeout` can also be set to `True`.
         """
         self.pid = os.getpid()
         self.path = path
@@ -990,11 +1002,17 @@ def __init__(
         self.socket_timeout = socket_timeout
         self.retry_on_timeout = retry_on_timeout
         if retry_on_timeout:
+            # Add TimeoutError to the errors list to retry on
+            retry_on_error.append(TimeoutError)
+        self.retry_on_error = retry_on_error
+        if self.retry_on_error:
             if retry is None:
                 self.retry = Retry(NoBackoff(), 1)
             else:
                 # deep-copy the Retry object as it is mutable
                 self.retry = copy.deepcopy(retry)
+            # Update the retry's supported errors with the specified errors
+            self.retry.update_supported_erros(retry_on_error)
         else:
             self.retry = Retry(NoBackoff(), 0)
         self.health_check_interval = health_check_interval
@@ -1052,6 +1070,7 @@ def to_bool(value):
     "socket_connect_timeout": float,
     "socket_keepalive": to_bool,
     "retry_on_timeout": to_bool,
+    "retry_on_error": list,
     "max_connections": int,
     "health_check_interval": int,
     "ssl_check_hostname": to_bool,
diff --git a/redis/retry.py b/redis/retry.py
--- a/redis/retry.py
+++ b/redis/retry.py
@@ -19,6 +19,14 @@ def __init__(
         self._retries = retries
         self._supported_errors = supported_errors
 
+    def update_supported_erros(self, specified_errors: list):
+        """
+        Updates the supported errors with the specified error types
+        """
+        self._supported_errors = tuple(
+            set(self._supported_errors + tuple(specified_errors))
+        )
+
     def call_with_retry(self, do, fail):
         """
         Execute an operation that might fail and returns its result, or


Test patch:
diff --git a/tests/test_retry.py b/tests/test_retry.py
--- a/tests/test_retry.py
+++ b/tests/test_retry.py
@@ -1,10 +1,20 @@
+from unittest.mock import patch
+
 import pytest
 
 from redis.backoff import NoBackoff
+from redis.client import Redis
 from redis.connection import Connection, UnixDomainSocketConnection
-from redis.exceptions import ConnectionError
+from redis.exceptions import (
+    BusyLoadingError,
+    ConnectionError,
+    ReadOnlyError,
+    TimeoutError,
+)
 from redis.retry import Retry
 
+from .conftest import _get_client
+
 
 class BackoffMock:
     def __init__(self):
@@ -39,6 +49,37 @@ def test_retry_on_timeout_retry(self, Class, retries):
         assert isinstance(c.retry, Retry)
         assert c.retry._retries == retries
 
+    @pytest.mark.parametrize("Class", [Connection, UnixDomainSocketConnection])
+    def test_retry_on_error(self, Class):
+        c = Class(retry_on_error=[ReadOnlyError])
+        assert c.retry_on_error == [ReadOnlyError]
+        assert isinstance(c.retry, Retry)
+        assert c.retry._retries == 1
+
+    @pytest.mark.parametrize("Class", [Connection, UnixDomainSocketConnection])
+    def test_retry_on_error_empty_value(self, Class):
+        c = Class(retry_on_error=[])
+        assert c.retry_on_error == []
+        assert isinstance(c.retry, Retry)
+        assert c.retry._retries == 0
+
+    @pytest.mark.parametrize("Class", [Connection, UnixDomainSocketConnection])
+    def test_retry_on_error_and_timeout(self, Class):
+        c = Class(
+            retry_on_error=[ReadOnlyError, BusyLoadingError], retry_on_timeout=True
+        )
+        assert c.retry_on_error == [ReadOnlyError, BusyLoadingError, TimeoutError]
+        assert isinstance(c.retry, Retry)
+        assert c.retry._retries == 1
+
+    @pytest.mark.parametrize("retries", range(10))
+    @pytest.mark.parametrize("Class", [Connection, UnixDomainSocketConnection])
+    def test_retry_on_error_retry(self, Class, retries):
+        c = Class(retry_on_error=[ReadOnlyError], retry=Retry(NoBackoff(), retries))
+        assert c.retry_on_error == [ReadOnlyError]
+        assert isinstance(c.retry, Retry)
+        assert c.retry._retries == retries
+
 
 class TestRetry:
     "Test that Retry calls backoff and retries the expected number of times"
@@ -65,3 +106,85 @@ def test_retry(self, retries):
         assert self.actual_failures == 1 + retries
         assert backoff.reset_calls == 1
         assert backoff.calls == retries
+
+
+@pytest.mark.onlynoncluster
+class TestRedisClientRetry:
+    "Test the standalone Redis client behavior with retries"
+
+    def test_client_retry_on_error_with_success(self, request):
+        with patch.object(Redis, "parse_response") as parse_response:
+
+            def mock_parse_response(connection, *args, **options):
+                def ok_response(connection, *args, **options):
+                    return "MOCK_OK"
+
+                parse_response.side_effect = ok_response
+                raise ReadOnlyError()
+
+            parse_response.side_effect = mock_parse_response
+            r = _get_client(Redis, request, retry_on_error=[ReadOnlyError])
+            assert r.get("foo") == "MOCK_OK"
+            assert parse_response.call_count == 2
+
+    def test_client_retry_on_error_raise(self, request):
+        with patch.object(Redis, "parse_response") as parse_response:
+            parse_response.side_effect = BusyLoadingError()
+            retries = 3
+            r = _get_client(
+                Redis,
+                request,
+                retry_on_error=[ReadOnlyError, BusyLoadingError],
+                retry=Retry(NoBackoff(), retries),
+            )
+            with pytest.raises(BusyLoadingError):
+                try:
+                    r.get("foo")
+                finally:
+                    assert parse_response.call_count == retries + 1
+
+    def test_client_retry_on_error_different_error_raised(self, request):
+        with patch.object(Redis, "parse_response") as parse_response:
+            parse_response.side_effect = TimeoutError()
+            retries = 3
+            r = _get_client(
+                Redis,
+                request,
+                retry_on_error=[ReadOnlyError],
+                retry=Retry(NoBackoff(), retries),
+            )
+            with pytest.raises(TimeoutError):
+                try:
+                    r.get("foo")
+                finally:
+                    assert parse_response.call_count == 1
+
+    def test_client_retry_on_error_and_timeout(self, request):
+        with patch.object(Redis, "parse_response") as parse_response:
+            parse_response.side_effect = TimeoutError()
+            retries = 3
+            r = _get_client(
+                Redis,
+                request,
+                retry_on_error=[ReadOnlyError],
+                retry_on_timeout=True,
+                retry=Retry(NoBackoff(), retries),
+            )
+            with pytest.raises(TimeoutError):
+                try:
+                    r.get("foo")
+                finally:
+                    assert parse_response.call_count == retries + 1
+
+    def test_client_retry_on_timeout(self, request):
+        with patch.object(Redis, "parse_response") as parse_response:
+            parse_response.side_effect = TimeoutError()
+            retries = 3
+            r = _get_client(
+                Redis, request, retry_on_timeout=True, retry=Retry(NoBackoff(), retries)
+            )
+            with pytest.raises(TimeoutError):
+                try:
+                    r.get("foo")
+                finally:
+                    assert parse_response.call_count == retries + 1