fix: fix replication on pika node recovering (#3038)

* fix: tryFixReplicationRelationships error (#2991)

* fix invalid arguments error when dashboard send slveof force

Co-authored-by: wangshaoyi <[email protected]>

* fix replication relationship error while recovering from offline

* update proxy slot mappings after slave pika recovered

---------

Co-authored-by: wangshaoyi <[email protected]>

Author: wangshao1

codis/pkg/topom/topom_group.go

@@ -330,7 +330,7 @@ func (s *Topom) GroupPromoteServer(gid int, addr string) error {
 	}
 }
 
-func (s *Topom) tryFixReplicationRelationships(ctx *context, recoveredGroupServers []*redis.ReplicationState, masterOffGroupLen int) {
+func (s *Topom) tryFixReplicationRelationships(ctx *context, recoveredGroupServers []*redis.ReplicationState) {
 	for _, state := range recoveredGroupServers {
 		log.Infof("group-[%d] try to fix server[%v-%v] replication relationship", state.GroupID, state.Index, state.Addr)
 		group, err := ctx.getGroup(state.GroupID)
@@ -346,7 +346,7 @@ func (s *Topom) tryFixReplicationRelationships(ctx *context, recoveredGroupServe
 			continue
 		}
 
-		err = s.tryFixReplicationRelationship(group, state.Server, state, masterOffGroupLen)
+		err = s.tryFixReplicationRelationship(group, state.Server, state)
 		if err != nil {
 			log.Warnf("group-[%d] fix server[%v] replication relationship failed, err: %v", group.Id, state.Addr, err)
 			continue
@@ -371,29 +371,21 @@ func (s *Topom) tryFixReplicationRelationships(ctx *context, recoveredGroupServe
 // only fix which the old state of GroupServer is GroupServerStateOffline.
 // It will only update the state of GroupServer to GroupServerStateNormal, If the GroupServer have right
 // master-slave replication relationship.
-func (s *Topom) tryFixReplicationRelationship(group *models.Group, groupServer *models.GroupServer, state *redis.ReplicationState, masterOffGroupLen int) (err error) {
+func (s *Topom) tryFixReplicationRelationship(group *models.Group, groupServer *models.GroupServer, state *redis.ReplicationState) (err error) {
 	curMasterAddr := group.Servers[0].Addr
 	if isGroupMaster(state, group) {
-		// current server is master,
-		if models.GroupServerRole(state.Replication.Role) == models.RoleMaster {
-			if masterOffGroupLen > 0 {
-				return nil
-			}
-		}
-
 		// execute the command `slaveof no one`
-		if err = promoteServerToNewMaster(state.Addr, s.config.ProductAuth); err != nil {
-			return err
+		if models.GroupServerRole(state.Replication.Role) != models.RoleMaster {
+			if err = promoteServerToNewMaster(state.Addr, s.config.ProductAuth); err != nil {
+				return err
+			}
 		}
 	} else {
-		// skip if it has right replication relationship
-		if state.Replication.GetMasterAddr() == curMasterAddr {
-			return nil
-		}
-
-		// current server is slave, execute the command `slaveof [new master ip] [new master port]`
-		if err = updateMasterToNewOne(groupServer.Addr, curMasterAddr, s.config.ProductAuth); err != nil {
-			return err
+		if state.Replication.GetMasterAddr() != curMasterAddr {
+			// current server is slave, execute the command `slaveof [new master ip] [new master port]`
+			if err = updateMasterToNewOne(groupServer.Addr, curMasterAddr, s.config.ProductAuth); err != nil {
+				return err
+			}
 		}
 	}
 

codis/pkg/topom/topom_sentinel.go

@@ -48,7 +48,7 @@ func (s *Topom) CheckStateAndSwitchSlavesAndMasters(filter func(index int, g *mo
 
 	if len(recoveredGroupServersState) > 0 {
 		// offline GroupServer's service has recovered, check and fix it's master-slave replication relationship
-		s.tryFixReplicationRelationships(ctx, recoveredGroupServersState, len(masterOfflineGroups))
+		s.tryFixReplicationRelationships(ctx, recoveredGroupServersState)
 	}
 
 	return nil
@@ -85,6 +85,18 @@ func (s *Topom) checkAndUpdateGroupServerState(conf *Config, group *models.Group
 			*recoveredGroupServers = append(*recoveredGroupServers, state)
 			// update GroupServer to GroupServerStateNormal state later
 		} else {
+			// This may contains any of following condition:
+			// 1. groupServer.State is Normal
+			// 2. groupServer.State is GroupServerStateSubjectiveOffline and is Master
+			// 3. groupServer.State is GroupServerStateSubjectiveOffline and is Slave
+			// for condition 3, if current server's previous state is SubjectiveOffline
+			// and has been added to slaveofflinegroups before,
+			// should also resync mappings to proxy to enable replicationgroup
+			if groupServer.State == models.GroupServerStateSubjectiveOffline &&
+				!isGroupMaster(state, group) &&
+				group.OutOfSync {
+				*recoveredGroupServers = append(*recoveredGroupServers, state)
+			}
 			// Update the offset information of the state and role nodes
 			groupServer.State = models.GroupServerStateNormal
 			groupServer.ReCallTimes = 0

codis/pkg/utils/redis/client.go

@@ -341,7 +341,7 @@ func (c *Client) SetMaster(master string, force bool) error {
 		}
 
 		if force {
-			if _, err := c.Do("SLAVEOF", host, port, "-f"); err != nil {
+			if _, err := c.Do("SLAVEOF", host, port, "force"); err != nil {
 				return err
 			}
 		} else {

codis/pkg/utils/redis/sentinel.go

@@ -106,8 +106,8 @@ func (i *InfoReplication) UnmarshalJSON(b []byte) error {
 	}
 
 	i.Role = kvmap["role"]
-	i.MasterPort = kvmap["master_host"]
-	i.MasterHost = kvmap["master_port"]
+	i.MasterPort = kvmap["master_port"]
+	i.MasterHost = kvmap["master_host"]
 	i.MasterLinkStatus = kvmap["master_link_status"]
 	i.IsEligibleForMasterElection = kvmap["is_eligible_for_master_election"] == "true"