| 123 | } |
| 124 | |
| 125 | func (c *ClusterChecker) increaseFailureCount(shardIndex int, node store.Node) int64 { |
| 126 | id := node.ID() |
| 127 | c.failureMu.Lock() |
| 128 | if _, ok := c.failureCounts[id]; !ok { |
| 129 | c.failureCounts[id] = 0 |
| 130 | } |
| 131 | c.failureCounts[id] += 1 |
| 132 | count := c.failureCounts[id] |
| 133 | c.failureMu.Unlock() |
| 134 | |
| 135 | // don't add the node into the failover candidates if it's not a master node |
| 136 | if !node.IsMaster() { |
| 137 | return count |
| 138 | } |
| 139 | |
| 140 | log := logger.Get().With( |
| 141 | zap.String("cluster_name", c.clusterName), |
| 142 | zap.String("id", node.ID()), |
| 143 | zap.Bool("is_master", node.IsMaster()), |
| 144 | zap.String("addr", node.Addr())) |
| 145 | if count%c.options.maxFailureCount == 0 || count > c.options.maxFailureCount { |
| 146 | cluster, err := c.clusterStore.GetCluster(c.ctx, c.namespace, c.clusterName) |
| 147 | if err != nil { |
| 148 | log.Error("Failed to get the cluster info", zap.Error(err)) |
| 149 | return count |
| 150 | } |
| 151 | newMasterID, err := cluster.PromoteNewMaster(c.ctx, shardIndex, node.ID(), "") |
| 152 | if err != nil { |
| 153 | log.Error("Failed to promote the new master", zap.Error(err)) |
| 154 | return count |
| 155 | } |
| 156 | err = c.clusterStore.UpdateCluster(c.ctx, c.namespace, cluster) |
| 157 | if err != nil { |
| 158 | log.Error("Failed to update the cluster", zap.Error(err)) |
| 159 | return count |
| 160 | } |
| 161 | // the node is normal if it can be elected as the new master, |
| 162 | // because it requires the node is healthy. |
| 163 | c.resetFailureCount(newMasterID) |
| 164 | log.With(zap.String("new_master_id", newMasterID)).Info("Promote the new master") |
| 165 | } |
| 166 | return count |
| 167 | } |
| 168 | |
| 169 | func (c *ClusterChecker) resetFailureCount(nodeID string) { |
| 170 | c.failureMu.Lock() |