From 75824b32aa5793c068a5d5d236494687439677c6 Mon Sep 17 00:00:00 2001
From: Aaron Kuehler <aaron.kuehler@powerhrg.com>
Date: Tue, 13 Feb 2024 08:45:32 -0500
Subject: [PATCH] Monitor and heal master replica connections (#43)

When a RedisFailover's "master" Redis node is being replicated to a
"slave" Redis node that is NOT part of the RedisFailover the
redis-operator resets the sentinels indefinitely.

Consider this scenario, the RedisFailover is being replicated
asynchronously to a warm standby Redis cluster in a different data
center to handle primary data site outages. Usually we'd configure the
secondary site to replicate from the "slave" nodes of the Primary site's
RedisFailover. However, if a failover occurs in the primary data site,
it's possible that the "slave" to which the secondary site is connected
to and replicating from is promoated to the Primary site's new master.
When this happens, sentinel picks up the secondary site's replication
connections and adds them to the list of replicas to consider for leader
election. Thankfully, the operator prevents the sentinels from
communicating with any pods that it ought NOT consider for leader
election, so failovers still behave as expected. However, this causes
the redis-operator to detect that the sentinels are trying to monitor
replicas that they shouldn't and calls `SENTINEL RESET` to clear any
stale replica entries form the sentinel. The secondary site is still
replicating from the newly promoted master so the secondary site's
replication connections are added back to the sentinel replicas list
when the sentinel calls `INFO` on the primary site's "master"; repeating
the reset cycle indefinitely.

This change assumes that any replication not immediately meant to be
managed by the RedisFailover should connect via the RedisFailover's
"slave" nodes; the operator provides services to reach these nodes. When
the operator detects that the master node has replication connections
that would otherwise confuse the sentinel's leader election, it attempts
to clean stale replication connections by resetting them; forcing
replication clients to re-establish connections to a "slave" node in the
primary site rather than the master.
---
 CHANGELOG.md                                  |  6 +-
 metrics/metrics.go                            |  2 +
 .../service/RedisFailoverCheck.go             | 14 ++++
 .../service/RedisFailoverHeal.go              | 14 ++++
 mocks/service/redis/Client.go                 | 38 +++++++++
 operator/redisfailover/checker.go             |  9 +++
 operator/redisfailover/checker_test.go        |  3 +
 operator/redisfailover/service/check.go       | 13 ++++
 operator/redisfailover/service/check_test.go  | 44 +++++++++++
 operator/redisfailover/service/heal.go        | 16 ++++
 operator/redisfailover/service/heal_test.go   | 28 +++++++
 service/redis/client.go                       | 78 +++++++++++++++----
 12 files changed, 250 insertions(+), 15 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 2eebb6d76..1ca5019ba 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -9,6 +9,10 @@ Also check this project's [releases](https://github.com/powerhome/redis-operator
 
 ## Unreleased
 
+### Fixed
+
+- Operator detects and attempts to heal excessive replication connections on the master node. This prevents excessive sentinel resets from the operator when extra-RedisFailvoer replication connnections are present on the "slave" nodes. #43
+
 ## [v2.0.1] - 2024-02-09
 
 ### Fixed
@@ -25,7 +29,7 @@ This update modifies how the operator generates network policies. In version v2.
 
 Update notes:
 
-This release will change the labels of the HAProxy deployment resource. 
+This release will change the labels of the HAProxy deployment resource.
 It's important to note that in API version apps/v1, a Deployment's label selector [cannot be changed once it's created](https://kubernetes.io/docs/concepts/workloads/controllers/deployment/#label-selector-updates). Therefore, any existing HAProxy deployment placed by an <v2.0.0 version of the redis-operator MUST be deleted so the new deployment with the correct labels and selectors can be recreated by redis-operator v2.0.0+
 
 ## [v1.8.0] - 2024-01-16
diff --git a/metrics/metrics.go b/metrics/metrics.go
index 626f12aad..716cd02f8 100644
--- a/metrics/metrics.go
+++ b/metrics/metrics.go
@@ -41,6 +41,7 @@ const (
 	MISC                                   = "MISC_ERROR"
 	SENTINEL_NUMBER_IN_MEMORY_MISMATCH     = "SENTINEL_NUMBER_IN_MEMORY_MISMATCH"
 	REDIS_SLAVES_NUMBER_IN_MEMORY_MISMATCH = "REDIS_SLAVES_NUMBER_IN_MEMORY_MISMATCH"
+	REDIS_SLAVES_NUMBER_CONNECTED_MISMATCH = "REDIS_SLAVES_NUMBER_CONNECTED_MISMATCH"
 	// redis connection related errors
 	WRONG_PASSWORD_USED = "WRONG_PASSWORD_USED"
 	NOAUTH              = "AUTH_CREDENTIALS_NOT_PROVIDED"
@@ -69,6 +70,7 @@ const (
 	GET_SENTINEL_MONITOR        = "SENTINEL_GET_MASTER_INSTANCE"
 	CHECK_SENTINEL_QUORUM       = "SENTINEL_CKQUORUM"
 	SLAVE_IS_READY              = "CHECK_IF_SLAVE_IS_READY"
+	RESET_REPLICA_CONNECTIONS   = "RESET_REPLICA_CONNECTIONS"
 )
 
 var ( // used for grabage collection of metrics
diff --git a/mocks/operator/redisfailover/service/RedisFailoverCheck.go b/mocks/operator/redisfailover/service/RedisFailoverCheck.go
index c65790088..f7b318c34 100644
--- a/mocks/operator/redisfailover/service/RedisFailoverCheck.go
+++ b/mocks/operator/redisfailover/service/RedisFailoverCheck.go
@@ -53,6 +53,20 @@ func (_m *RedisFailoverCheck) CheckIfMasterLocalhost(rFailover *v1.RedisFailover
 	return r0, r1
 }
 
+// CheckNumberRedisConnectedSlaves provides a mock function with given fields: masterIP, rFailover
+func (_m *RedisFailoverCheck) CheckNumberRedisConnectedSlaves(masterIP string, rFailover *v1.RedisFailover) error {
+	ret := _m.Called(masterIP, rFailover)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(string, *v1.RedisFailover) error); ok {
+		r0 = rf(masterIP, rFailover)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
 // CheckRedisNumber provides a mock function with given fields: rFailover
 func (_m *RedisFailoverCheck) CheckRedisNumber(rFailover *v1.RedisFailover) error {
 	ret := _m.Called(rFailover)
diff --git a/mocks/operator/redisfailover/service/RedisFailoverHeal.go b/mocks/operator/redisfailover/service/RedisFailoverHeal.go
index 3cc10a21d..ea2af1260 100644
--- a/mocks/operator/redisfailover/service/RedisFailoverHeal.go
+++ b/mocks/operator/redisfailover/service/RedisFailoverHeal.go
@@ -69,6 +69,20 @@ func (_m *RedisFailoverHeal) NewSentinelMonitorWithPort(ip string, monitor strin
 	return r0
 }
 
+// ResetReplicaConnections provides a mock function with given fields: ip, rFailover
+func (_m *RedisFailoverHeal) ResetReplicaConnections(ip string, rFailover *v1.RedisFailover) error {
+	ret := _m.Called(ip, rFailover)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(string, *v1.RedisFailover) error); ok {
+		r0 = rf(ip, rFailover)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
 // RestoreSentinel provides a mock function with given fields: ip, port
 func (_m *RedisFailoverHeal) RestoreSentinel(ip string, port string) error {
 	ret := _m.Called(ip, port)
diff --git a/mocks/service/redis/Client.go b/mocks/service/redis/Client.go
index 8392f0d9e..7e20e34d7 100644
--- a/mocks/service/redis/Client.go
+++ b/mocks/service/redis/Client.go
@@ -9,6 +9,30 @@ type Client struct {
 	mock.Mock
 }
 
+// GetNumberRedisConnectedSlaves provides a mock function with given fields: ip, port
+func (_m *Client) GetNumberRedisConnectedSlaves(ip string, port string) (int32, error) {
+	ret := _m.Called(ip, port)
+
+	var r0 int32
+	var r1 error
+	if rf, ok := ret.Get(0).(func(string, string) (int32, error)); ok {
+		return rf(ip, port)
+	}
+	if rf, ok := ret.Get(0).(func(string, string) int32); ok {
+		r0 = rf(ip, port)
+	} else {
+		r0 = ret.Get(0).(int32)
+	}
+
+	if rf, ok := ret.Get(1).(func(string, string) error); ok {
+		r1 = rf(ip, port)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
 // GetNumberSentinelSlavesInMemory provides a mock function with given fields: ip, port
 func (_m *Client) GetNumberSentinelSlavesInMemory(ip string, port string) (int32, error) {
 	ret := _m.Called(ip, port)
@@ -206,6 +230,20 @@ func (_m *Client) MonitorRedisWithPort(ip string, monitor string, port string, q
 	return r0
 }
 
+// ResetReplicaConnections provides a mock function with given fields: ip, port, password
+func (_m *Client) ResetReplicaConnections(ip string, port string, password string) error {
+	ret := _m.Called(ip, port, password)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(string, string, string) error); ok {
+		r0 = rf(ip, port, password)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
 // ResetSentinel provides a mock function with given fields: ip, port
 func (_m *Client) ResetSentinel(ip string, port string) error {
 	ret := _m.Called(ip, port)
diff --git a/operator/redisfailover/checker.go b/operator/redisfailover/checker.go
index 20d93212e..d876a3e07 100644
--- a/operator/redisfailover/checker.go
+++ b/operator/redisfailover/checker.go
@@ -199,6 +199,15 @@ func (r *RedisFailoverHandler) CheckAndHeal(rf *redisfailoverv1.RedisFailover) e
 		}
 	}
 
+	err = r.rfChecker.CheckNumberRedisConnectedSlaves(master, rf)
+	setRedisCheckerMetrics(r.mClient, "redis", rf.Namespace, rf.Name, metrics.REDIS_SLAVES_NUMBER_CONNECTED_MISMATCH, metrics.NOT_APPLICABLE, err)
+	if err != nil {
+		r.logger.WithField("redisfailover", rf.ObjectMeta.Name).WithField("namespace", rf.ObjectMeta.Namespace).Warningf("Master has wrong number of slaves: %s", err.Error())
+		if err = r.rfHealer.ResetReplicaConnections(master, rf); err != nil {
+			return err
+		}
+	}
+
 	err = r.applyRedisCustomConfig(rf)
 	setRedisCheckerMetrics(r.mClient, "redis", rf.Namespace, rf.Name, metrics.APPLY_REDIS_CONFIG, metrics.NOT_APPLICABLE, err)
 	if err != nil {
diff --git a/operator/redisfailover/checker_test.go b/operator/redisfailover/checker_test.go
index c116057fc..e156a110c 100644
--- a/operator/redisfailover/checker_test.go
+++ b/operator/redisfailover/checker_test.go
@@ -364,8 +364,11 @@ func TestCheckAndHeal(t *testing.T) {
 					mrfc.On("GetMasterIP", rf).Twice().Return(master, nil)
 					if test.slavesOK {
 						mrfc.On("CheckAllSlavesFromMaster", master, rf).Once().Return(nil)
+						mrfc.On("CheckNumberRedisConnectedSlaves", master, rf).Once().Return(nil)
 					} else {
 						mrfc.On("CheckAllSlavesFromMaster", master, rf).Once().Return(errors.New(""))
+						mrfc.On("CheckNumberRedisConnectedSlaves", master, rf).Once().Return(errors.New(""))
+						mrfh.On("ResetReplicaConnections", master, rf).Once().Return(nil)
 						if test.redisSetMasterOnAllOK {
 							mrfh.On("SetMasterOnAll", master, rf).Once().Return(nil)
 						} else {
diff --git a/operator/redisfailover/service/check.go b/operator/redisfailover/service/check.go
index 0048e5e3b..ab3081319 100644
--- a/operator/redisfailover/service/check.go
+++ b/operator/redisfailover/service/check.go
@@ -23,6 +23,7 @@ type RedisFailoverCheck interface {
 	CheckSentinelNumber(rFailover *redisfailoverv1.RedisFailover) error
 	CheckAllSlavesFromMaster(master string, rFailover *redisfailoverv1.RedisFailover) error
 	CheckSentinelNumberInMemory(sentinel string, rFailover *redisfailoverv1.RedisFailover) error
+	CheckNumberRedisConnectedSlaves(masterIP string, rFailover *redisfailoverv1.RedisFailover) error
 	CheckSentinelSlavesNumberInMemory(sentinel string, rFailover *redisfailoverv1.RedisFailover) error
 	CheckSentinelQuorum(rFailover *redisfailoverv1.RedisFailover) (int, error)
 	CheckIfMasterLocalhost(rFailover *redisfailoverv1.RedisFailover) (bool, error)
@@ -248,7 +249,19 @@ func (r *RedisFailoverChecker) CheckSentinelSlavesNumberInMemory(sentinel string
 		}
 	}
 	return nil
+}
 
+func (r *RedisFailoverChecker) CheckNumberRedisConnectedSlaves(masterIP string, rf *redisfailoverv1.RedisFailover) error {
+	portString := rf.Spec.Redis.Port.ToString()
+	nSlaves, err := r.redisClient.GetNumberRedisConnectedSlaves(masterIP, portString)
+	if err != nil {
+		return err
+	} else {
+		if nSlaves != rf.Spec.Redis.Replicas-1 {
+			return errors.New("redis number of slaves mismatch")
+		}
+	}
+	return nil
 }
 
 // CheckSentinelMonitor controls if the sentinels are monitoring the expected master
diff --git a/operator/redisfailover/service/check_test.go b/operator/redisfailover/service/check_test.go
index 23bf4b20b..14de5aa8b 100644
--- a/operator/redisfailover/service/check_test.go
+++ b/operator/redisfailover/service/check_test.go
@@ -436,6 +436,50 @@ func TestCheckSentinelMonitorWithPortIPMismatch(t *testing.T) {
 	assert.Error(err)
 }
 
+func TestCheckNumberRedisConnectedSlavesGeConnectedSlavesNumberError(t *testing.T) {
+	assert := assert.New(t)
+
+	rf := generateRF()
+
+	ms := &mK8SService.Services{}
+	mr := &mRedisService.Client{}
+	mr.On("GetNumberRedisConnectedSlaves", "1.1.1.1", "0").Once().Return(int32(0), errors.New("expected error"))
+
+	checker := rfservice.NewRedisFailoverChecker(ms, mr, log.DummyLogger{}, metrics.Dummy)
+
+	err := checker.CheckNumberRedisConnectedSlaves("1.1.1.1", rf)
+	assert.Error(err)
+}
+
+func TestCheckNumberRedisConnectedSlavesGeConnectedSlavesNumberMismatch(t *testing.T) {
+	assert := assert.New(t)
+
+	rf := generateRF()
+
+	ms := &mK8SService.Services{}
+	mr := &mRedisService.Client{}
+	mr.On("GetNumberRedisConnectedSlaves", "1.1.1.1", "0").Once().Return(int32(rf.Spec.Redis.Replicas+1), nil)
+
+	checker := rfservice.NewRedisFailoverChecker(ms, mr, log.DummyLogger{}, metrics.Dummy)
+
+	err := checker.CheckNumberRedisConnectedSlaves("1.1.1.1", rf)
+	assert.Error(err)
+}
+
+func TestCheckNumberRedisConnectedSlaves(t *testing.T) {
+	assert := assert.New(t)
+	rf := generateRF()
+
+	ms := &mK8SService.Services{}
+	mr := &mRedisService.Client{}
+	mr.On("GetNumberRedisConnectedSlaves", "1.1.1.1", "0").Once().Return(rf.Spec.Redis.Replicas-1, nil)
+
+	checker := rfservice.NewRedisFailoverChecker(ms, mr, log.DummyLogger{}, metrics.Dummy)
+
+	err := checker.CheckNumberRedisConnectedSlaves("1.1.1.1", rf)
+	assert.NoError(err)
+}
+
 func TestGetMasterIPGetStatefulSetPodsError(t *testing.T) {
 	assert := assert.New(t)
 
diff --git a/operator/redisfailover/service/heal.go b/operator/redisfailover/service/heal.go
index 503298074..07e98457f 100644
--- a/operator/redisfailover/service/heal.go
+++ b/operator/redisfailover/service/heal.go
@@ -15,6 +15,7 @@ import (
 // RedisFailoverHeal defines the interface able to fix the problems on the redis failovers
 type RedisFailoverHeal interface {
 	MakeMaster(ip string, rFailover *redisfailoverv1.RedisFailover) error
+	ResetReplicaConnections(ip string, rFailover *redisfailoverv1.RedisFailover) error
 	SetOldestAsMaster(rFailover *redisfailoverv1.RedisFailover) error
 	SetMasterOnAll(masterIP string, rFailover *redisfailoverv1.RedisFailover) error
 	SetExternalMasterOnAll(masterIP string, masterPort string, rFailover *redisfailoverv1.RedisFailover) error
@@ -85,6 +86,21 @@ func (r *RedisFailoverHealer) MakeMaster(ip string, rf *redisfailoverv1.RedisFai
 	return nil
 }
 
+func (r *RedisFailoverHealer) ResetReplicaConnections(ip string, rf *redisfailoverv1.RedisFailover) error {
+	password, err := k8s.GetRedisPassword(r.k8sService, rf)
+	if err != nil {
+		return err
+	}
+
+	port := rf.Spec.Redis.Port.ToString()
+	err = r.redisClient.ResetReplicaConnections(ip, port, password)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
 // SetOldestAsMaster puts all redis to the same master, choosen by order of appearance
 func (r *RedisFailoverHealer) SetOldestAsMaster(rf *redisfailoverv1.RedisFailover) error {
 	ssp, err := r.k8sService.GetStatefulSetPods(rf.Namespace, GetRedisName(rf))
diff --git a/operator/redisfailover/service/heal_test.go b/operator/redisfailover/service/heal_test.go
index 3a14ef987..80fce3ee3 100644
--- a/operator/redisfailover/service/heal_test.go
+++ b/operator/redisfailover/service/heal_test.go
@@ -17,6 +17,34 @@ import (
 	rfservice "github.com/spotahome/redis-operator/operator/redisfailover/service"
 )
 
+func TestResetReplicaConnectionsError(t *testing.T) {
+	assert := assert.New(t)
+	rf := generateRF()
+
+	ms := &mK8SService.Services{}
+	mr := &mRedisService.Client{}
+	mr.On("ResetReplicaConnections", "0.0.0.0", "0", "").Once().Return(errors.New(""))
+
+	healer := rfservice.NewRedisFailoverHealer(ms, mr, log.DummyLogger{})
+
+	err := healer.ResetReplicaConnections("0.0.0.0", rf)
+	assert.Error(err)
+}
+
+func TestResetReplicaConnections(t *testing.T) {
+	assert := assert.New(t)
+	rf := generateRF()
+
+	ms := &mK8SService.Services{}
+	mr := &mRedisService.Client{}
+	mr.On("ResetReplicaConnections", "0.0.0.0", "0", "").Once().Return(nil)
+
+	healer := rfservice.NewRedisFailoverHealer(ms, mr, log.DummyLogger{})
+
+	err := healer.ResetReplicaConnections("0.0.0.0", rf)
+	assert.NoError(err)
+}
+
 func TestSetOldestAsMasterNewMasterError(t *testing.T) {
 	assert := assert.New(t)
 
diff --git a/service/redis/client.go b/service/redis/client.go
index 451eed322..8ac11c61a 100644
--- a/service/redis/client.go
+++ b/service/redis/client.go
@@ -18,12 +18,14 @@ import (
 type Client interface {
 	GetNumberSentinelsInMemory(ip string, port string) (int32, error)
 	GetNumberSentinelSlavesInMemory(ip string, port string) (int32, error)
+	GetNumberRedisConnectedSlaves(ip string, port string) (int32, error)
 	ResetSentinel(ip string, port string) error
 	GetSlaveOf(ip, port, password string) (string, error)
 	IsMaster(ip, port, password string) (bool, error)
 	MonitorRedis(ip, monitor, quorum, password string, port string) error
 	MonitorRedisWithPort(ip, monitor, port, quorum, password string, sentinelPort string) error
 	MakeMaster(ip, port, password string) error
+	ResetReplicaConnections(ip, port, password string) error
 	MakeSlaveOf(ip, masterIP, password string) error
 	MakeSlaveOfWithPort(ip, masterIP, masterPort, password string) error
 	GetSentinelMonitor(ip string, port string) (string, string, error)
@@ -45,23 +47,26 @@ func New(metricsRecorder metrics.Recorder) Client {
 }
 
 const (
-	sentinelsNumberREString = "sentinels=([0-9]+)"
-	slaveNumberREString     = "slaves=([0-9]+)"
-	sentinelStatusREString  = "status=([a-z]+)"
-	redisMasterHostREString = "master_host:([0-9.]+)"
-	redisRoleMaster         = "role:master"
-	redisSyncing            = "master_sync_in_progress:1"
-	redisMasterSillPending  = "master_host:127.0.0.1"
-	redisLinkUp             = "master_link_status:up"
-	redisPort               = "6379"
-	masterName              = "mymaster"
+	sentinelsNumberREString      = "sentinels=([0-9]+)"
+	slaveNumberREString          = "slaves=([0-9]+)"
+	sentinelStatusREString       = "status=([a-z]+)"
+	redisMasterHostREString      = "master_host:([0-9.]+)"
+	redisConnectedSlavesREString = "connected_slaves:([0-9]+)"
+	redisRoleMaster              = "role:master"
+	redisSyncing                 = "master_sync_in_progress:1"
+	redisMasterSillPending       = "master_host:127.0.0.1"
+	redisLinkUp                  = "master_link_status:up"
+
+	redisPort  = "6379"
+	masterName = "mymaster"
 )
 
 var (
-	sentinelNumberRE  = regexp.MustCompile(sentinelsNumberREString)
-	sentinelStatusRE  = regexp.MustCompile(sentinelStatusREString)
-	slaveNumberRE     = regexp.MustCompile(slaveNumberREString)
-	redisMasterHostRE = regexp.MustCompile(redisMasterHostREString)
+	sentinelNumberRE       = regexp.MustCompile(sentinelsNumberREString)
+	sentinelStatusRE       = regexp.MustCompile(sentinelStatusREString)
+	slaveNumberRE          = regexp.MustCompile(slaveNumberREString)
+	redisMasterHostRE      = regexp.MustCompile(redisMasterHostREString)
+	redisConnectedSlavesRE = regexp.MustCompile(redisConnectedSlavesREString)
 )
 
 // GetNumberSentinelsInMemory return the number of sentinels that the requested sentinel has
@@ -128,6 +133,35 @@ func (c *client) GetNumberSentinelSlavesInMemory(ip string, sentinelPort string)
 	return int32(nSlaves), nil
 }
 
+// GetNumberRedisConnectedSlaves return the number of slaves that the requested redis has
+func (c *client) GetNumberRedisConnectedSlaves(ip string, sentinelPort string) (int32, error) {
+	options := &rediscli.Options{
+		Addr:     net.JoinHostPort(ip, sentinelPort),
+		Password: "",
+		DB:       0,
+	}
+	rClient := rediscli.NewClient(options)
+	defer rClient.Close()
+	info, err := rClient.Info(context.TODO(), "replication").Result()
+	if err != nil {
+		c.metricsRecorder.RecordRedisOperation(metrics.KIND_REDIS, ip, metrics.GET_NUM_REDIS_SLAVES_IN_MEM, metrics.FAIL, getRedisError(err))
+		return 0, err
+	}
+
+	match := redisConnectedSlavesRE.FindStringSubmatch(info)
+	if len(match) == 0 {
+		c.metricsRecorder.RecordRedisOperation(metrics.KIND_REDIS, ip, metrics.GET_NUM_REDIS_SLAVES_IN_MEM, metrics.FAIL, metrics.REGEX_NOT_FOUND)
+		return 0, errors.New("slaves regex not found")
+	}
+	nSlaves, err := strconv.Atoi(match[1])
+	if err != nil {
+		c.metricsRecorder.RecordRedisOperation(metrics.KIND_REDIS, ip, metrics.GET_NUM_REDIS_SLAVES_IN_MEM, metrics.FAIL, metrics.MISC)
+		return 0, err
+	}
+	c.metricsRecorder.RecordRedisOperation(metrics.KIND_REDIS, ip, metrics.GET_NUM_REDIS_SLAVES_IN_MEM, metrics.SUCCESS, metrics.NOT_APPLICABLE)
+	return int32(nSlaves), nil
+}
+
 func isSentinelReady(info string) error {
 	matchStatus := sentinelStatusRE.FindStringSubmatch(info)
 	if len(matchStatus) == 0 || matchStatus[1] != "ok" {
@@ -262,6 +296,22 @@ func (c *client) MakeMaster(ip string, port string, password string) error {
 	return nil
 }
 
+func (c client) ResetReplicaConnections(ip string, port string, password string) error {
+	options := &rediscli.Options{
+		Addr:     net.JoinHostPort(ip, port),
+		Password: password,
+		DB:       0,
+	}
+	rClient := rediscli.NewClient(options)
+	defer rClient.Close()
+	if res := rClient.ClientKillByFilter(context.TODO(), "TYPE", "REPLICA"); res.Err() != nil {
+		c.metricsRecorder.RecordRedisOperation(metrics.KIND_REDIS, ip, metrics.RESET_REPLICA_CONNECTIONS, metrics.FAIL, getRedisError(res.Err()))
+		return res.Err()
+	}
+	c.metricsRecorder.RecordRedisOperation(metrics.KIND_REDIS, ip, metrics.RESET_REPLICA_CONNECTIONS, metrics.SUCCESS, metrics.NOT_APPLICABLE)
+	return nil
+}
+
 func (c *client) MakeSlaveOf(ip, masterIP, password string) error {
 	return c.MakeSlaveOfWithPort(ip, masterIP, redisPort, password)
 }