From 3abbe9856a2aa9d3ff78569a964245e0f40eab1d Mon Sep 17 00:00:00 2001 From: bocharov Date: Fri, 17 Jul 2026 11:21:50 -0700 Subject: [PATCH] Force-apply topology on reconcile and add a per-cluster sync endpoint Kvrocks cluster nodes do not gossip: the controller is the sole source of topology and must reliably push one complete, identical view to every node. Three gaps let a node's view drift and never recover (see #395): 1. The probe loop re-pushed only a node whose epoch was behind the store, so a node that had drifted at an equal epoch (a dropped push, a partial view, a stale/phantom node entry) was never repaired. 2. CLUSTERX SETNODES was sent without force, so even a re-push at an equal version is a no-op on the server (the version gate rejects a lower version and no-ops an equal one). The only prior recovery was CLUSTER RESET, which needs an empty DB and is unusable once a node holds data. 3. A push dropped by a transient error was not retried before the next tick. Changes: - SyncClusterInfo takes a force argument and, when set, sends the SETNODES force flag so the topology applies unconditionally. A forced push replaces the node's whole topology, so it also clears stale/phantom entries without a CLUSTER RESET and without data loss. Wrapped in a bounded retry-with-backoff. - The reconcile loop parses cluster_known_nodes and cluster_slots_ok from CLUSTER INFO and force-pushes a node whose peer count or slot coverage disagrees with the desired topology, not only one whose epoch lags. Coverage is compared against the desired covered-slot count; a node that omits these fields falls back to epoch-only reconcile. The node-ahead adopt branch is preserved. - New endpoint POST /namespaces/{namespace}/clusters/{cluster}/sync force-pushes the stored topology to every node and reports per-node results. It performs no CLUSTER RESET, so it is safe to run on a populated cluster. - CheckNewNodes and Shard.ToSlotsString reject a blank or port-less node address, so a half-resolved address fails loudly instead of being pushed as a phantom. Unit tests cover the CLUSTER INFO parsing, address validation, the blank-address guard, the divergence check, and the sync handler. --- controller/cluster.go | 59 +++++++++++++++---- controller/cluster_test.go | 110 ++++++++++++++++++++++++++++++++++++ server/api/cluster.go | 43 ++++++++++++++ server/api/cluster_test.go | 62 +++++++++++++++++++- server/api/shard.go | 4 +- server/api/shard_test.go | 4 +- server/route.go | 1 + store/cluster.go | 4 +- store/cluster_mock_node.go | 15 ++++- store/cluster_node.go | 78 ++++++++++++++++++++----- store/cluster_node_test.go | 21 ++++++- store/cluster_shard.go | 8 ++- store/cluster_shard_test.go | 28 +++++++++ store/store.go | 10 +++- store/store_test.go | 3 + 15 files changed, 414 insertions(+), 36 deletions(-) diff --git a/controller/cluster.go b/controller/cluster.go index b6f78bcb..59778d34 100755 --- a/controller/cluster.go +++ b/controller/cluster.go @@ -117,7 +117,7 @@ func (c *ClusterChecker) WithFailoverOptions(opts store.FailoverOptions) *Cluste return c } -func (c *ClusterChecker) probeNode(ctx context.Context, node store.Node) (int64, error) { +func (c *ClusterChecker) probeNode(ctx context.Context, node store.Node) (*store.ClusterInfo, error) { clusterInfo, err := node.GetClusterInfo(ctx) if err != nil { // We need to use the string contains to check the error message @@ -125,14 +125,14 @@ func (c *ClusterChecker) probeNode(ctx context.Context, node store.Node) (int64, // And it's fixed in PR: https://github.com/apache/kvrocks/pull/2362, // but we need to be compatible with the old version here. if strings.Contains(err.Error(), ErrRestoringBackUp.Error()) { - return -1, ErrRestoringBackUp + return nil, ErrRestoringBackUp } else if strings.Contains(err.Error(), ErrClusterNotInitialized.Error()) { - return -1, ErrClusterNotInitialized + return nil, ErrClusterNotInitialized } else { - return -1, err + return nil, err } } - return clusterInfo.CurrentEpoch, nil + return clusterInfo, nil } func (c *ClusterChecker) increaseFailureCount(shardIndex int, node store.Node) int64 { @@ -230,8 +230,9 @@ func (c *ClusterChecker) syncClusterToNodes(ctx context.Context) error { zap.Int64("version", version), zap.String("node_id", n.ID()), zap.String("addr", n.Addr())) - // sync the clusterName to the latest version - if err := n.SyncClusterInfo(ctx, clusterInfo); err != nil { + // sync the clusterName to the latest version; force so a drifted-but-equal-version + // node is actually repaired rather than no-op'd by the server's version gate. + if err := n.SyncClusterInfo(ctx, clusterInfo, true); err != nil { log.Error("Failed to sync the cluster topology to the node", zap.Error(err)) } else { log.Info("Succeed to sync the cluster topology to the node") @@ -242,12 +243,38 @@ func (c *ClusterChecker) syncClusterToNodes(ctx context.Context) error { return nil } +// topologyDiverged reports whether a probed node's applied topology disagrees with the desired one: +// a wrong peer count or wrong slot coverage. It returns false when the node did not report these +// fields (KnownNodes/SlotsOk < 0), so an older kvrocks that omits them falls back to epoch-only +// reconcile instead of being force-pushed on every tick. +func topologyDiverged(info *store.ClusterInfo, wantNodes, wantSlots int64) bool { + if info == nil || info.KnownNodes < 0 || info.SlotsOk < 0 { + return false + } + return info.KnownNodes != wantNodes || info.SlotsOk != wantSlots +} + func (c *ClusterChecker) parallelProbeNodes(ctx context.Context, cluster *store.Cluster) { var mu sync.Mutex var latestNodeVersion int64 = 0 var latestClusterNodesStr string var wg sync.WaitGroup + // Desired peer count and covered-slot count from the stored (authoritative) topology. Comparing a + // node's applied view against these detects drift at an equal epoch; using the DESIRED coverage + // (not a hardcoded 16384) avoids false drift while a cluster is intentionally mid-scale with some + // slots not yet assigned. + wantNodes := int64(0) + wantSlots := int64(0) + for _, shard := range cluster.Shards { + wantNodes += int64(len(shard.Nodes)) + for _, sr := range shard.SlotRanges { + if sr.Start >= 0 && sr.Stop >= sr.Start { + wantSlots += int64(sr.Stop - sr.Start + 1) + } + } + } + for i, shard := range cluster.Shards { for _, node := range shard.Nodes { wg.Add(1) @@ -259,7 +286,7 @@ func (c *ClusterChecker) parallelProbeNodes(ctx context.Context, cluster *store. zap.Bool("is_master", n.IsMaster()), zap.String("addr", n.Addr()), ) - version, err := c.probeNode(ctx, n) + info, err := c.probeNode(ctx, n) // Don't sync the cluster info to the node if it is restoring the db from backup if errors.Is(err, ErrRestoringBackUp) { log.Error("The node is restoring the db from backup") @@ -274,10 +301,20 @@ func (c *ClusterChecker) parallelProbeNodes(ctx context.Context, cluster *store. } log.Debug("Probe the clusterName node") + // An uninitialized node (nil info) reports version -1, so the "behind" test below pushes it. + var version int64 = -1 + if info != nil { + version = info.CurrentEpoch + } + clusterVersion := cluster.Version.Load() - if version < clusterVersion { - // sync the clusterName to the latest version - if err := n.SyncClusterInfo(ctx, cluster); err != nil { + // Push when the node is behind the store, OR when it matches the epoch but its applied + // topology has drifted (a dropped/partial push the epoch comparison alone cannot see — + // issue #395). Force so an equal-version repair is actually applied: the server no-ops an + // equal-version SETNODES otherwise. A forced push replaces the node's whole topology, so + // it also clears phantom/empty-address entries without a data-destroying CLUSTER RESET. + if version < clusterVersion || (version == clusterVersion && topologyDiverged(info, wantNodes, wantSlots)) { + if err := n.SyncClusterInfo(ctx, cluster, true); err != nil { log.With(zap.Error(err)).Error("Failed to sync the clusterName info") } } else if version > clusterVersion { diff --git a/controller/cluster_test.go b/controller/cluster_test.go index a3774446..38dba39b 100644 --- a/controller/cluster_test.go +++ b/controller/cluster_test.go @@ -307,3 +307,113 @@ func TestCluster_MigrateSlot(t *testing.T) { defer ticker.Stop() <-ticker.C } + +// TestCluster_ReconcileForcePush drives the real reconcile path and asserts when it pushes. The +// equal-epoch-but-diverged case is the regression guard: the previous epoch-only logic re-pushed only +// when a node's version was strictly behind, so a node drifted at an equal epoch was never repaired. +func TestCluster_ReconcileForcePush(t *testing.T) { + ctx := context.Background() + ns, clusterName := "test-ns", "reconcile" + + newChecker := func() (*ClusterChecker, *store.Cluster, []*store.ClusterMockNode) { + s := NewMockClusterStore() + nodes := make([]*store.ClusterMockNode, 3) + for i := range nodes { + nodes[i] = store.NewClusterMockNode() + nodes[i].SetRole(store.RoleSlave) + } + nodes[0].SetRole(store.RoleMaster) + cluster := &store.Cluster{ + Name: clusterName, + Shards: []*store.Shard{{ + Nodes: []store.Node{nodes[0], nodes[1], nodes[2]}, + SlotRanges: []store.SlotRange{{Start: 0, Stop: 16383}}, + MigratingSlot: &store.MigratingSlot{IsMigrating: false}, + TargetShardIndex: -1, + }}, + } + cluster.Version.Store(1) + require.NoError(t, s.CreateCluster(ctx, ns, cluster)) + c := &ClusterChecker{ + clusterStore: s, + namespace: ns, + clusterName: clusterName, + options: ClusterCheckOptions{pingInterval: time.Second, maxFailureCount: 3}, + failureCounts: make(map[string]int64), + syncCh: make(chan struct{}, 1), + } + return c, cluster, nodes + } + + forcedPushes := func(nodes []*store.ClusterMockNode) int { + n := 0 + for _, node := range nodes { + for _, force := range node.SyncForceCalls { + require.True(t, force, "reconcile must push with force=true") + n++ + } + } + return n + } + + t.Run("converged: no push", func(t *testing.T) { + c, cluster, nodes := newChecker() + for _, node := range nodes { + node.MockClusterInfo = &store.ClusterInfo{CurrentEpoch: 1, KnownNodes: 3, SlotsOk: 16384} + } + c.parallelProbeNodes(ctx, cluster) + require.Equal(t, 0, forcedPushes(nodes)) + }) + + t.Run("equal epoch but incomplete coverage: force push (regression)", func(t *testing.T) { + c, cluster, nodes := newChecker() + for _, node := range nodes { + node.MockClusterInfo = &store.ClusterInfo{CurrentEpoch: 1, KnownNodes: 3, SlotsOk: 16000} + } + c.parallelProbeNodes(ctx, cluster) + require.Equal(t, 3, forcedPushes(nodes)) + }) + + t.Run("equal epoch but wrong peer count: force push (regression)", func(t *testing.T) { + c, cluster, nodes := newChecker() + for _, node := range nodes { + node.MockClusterInfo = &store.ClusterInfo{CurrentEpoch: 1, KnownNodes: 2, SlotsOk: 16384} + } + c.parallelProbeNodes(ctx, cluster) + require.Equal(t, 3, forcedPushes(nodes)) + }) + + t.Run("behind epoch: force push", func(t *testing.T) { + c, cluster, nodes := newChecker() + for _, node := range nodes { + node.MockClusterInfo = &store.ClusterInfo{CurrentEpoch: 0, KnownNodes: 3, SlotsOk: 16384} + } + c.parallelProbeNodes(ctx, cluster) + require.Equal(t, 3, forcedPushes(nodes)) + }) + + t.Run("fields unreported at equal epoch: no push", func(t *testing.T) { + c, cluster, nodes := newChecker() + for _, node := range nodes { + node.MockClusterInfo = &store.ClusterInfo{CurrentEpoch: 1, KnownNodes: -1, SlotsOk: -1} + } + c.parallelProbeNodes(ctx, cluster) + require.Equal(t, 0, forcedPushes(nodes)) + }) +} + +func TestTopologyDiverged(t *testing.T) { + const wantNodes, wantSlots = int64(6), int64(16384) + + // Converged: peer count and coverage both match desired. + require.False(t, topologyDiverged(&store.ClusterInfo{KnownNodes: 6, SlotsOk: 16384}, wantNodes, wantSlots)) + // Wrong peer count (e.g. a phantom or missing node). + require.True(t, topologyDiverged(&store.ClusterInfo{KnownNodes: 7, SlotsOk: 16384}, wantNodes, wantSlots)) + // Incomplete coverage. + require.True(t, topologyDiverged(&store.ClusterInfo{KnownNodes: 6, SlotsOk: 16000}, wantNodes, wantSlots)) + // Fields not reported (older kvrocks): fall back to epoch-only, never treated as diverged. + require.False(t, topologyDiverged(&store.ClusterInfo{KnownNodes: -1, SlotsOk: -1}, wantNodes, wantSlots)) + require.False(t, topologyDiverged(nil, wantNodes, wantSlots)) + // A cluster intentionally mid-scale (partial desired coverage) is converged when the node matches. + require.False(t, topologyDiverged(&store.ClusterInfo{KnownNodes: 6, SlotsOk: 8192}, wantNodes, 8192)) +} diff --git a/server/api/cluster.go b/server/api/cluster.go index 52178891..f7330aad 100644 --- a/server/api/cluster.go +++ b/server/api/cluster.go @@ -161,6 +161,49 @@ func (handler *ClusterHandler) MigrateSlot(c *gin.Context) { helper.ResponseOK(c, gin.H{"cluster": cluster}) } +// Sync force-pushes the stored (authoritative) topology to every node in the cluster and reports +// which nodes accepted it. Unlike the probe loop — which only re-pushes a node whose epoch is behind +// — this re-asserts the full topology unconditionally, so it repairs nodes that have drifted at an +// equal epoch or hold phantom/empty-address entries (apache/kvrocks-controller#395). It performs no +// CLUSTER RESET and never wipes data, so it is safe to run on a populated cluster. +func (handler *ClusterHandler) Sync(c *gin.Context) { + namespace := c.Param("namespace") + clusterName := c.Param("cluster") + + lock := handler.getLock(namespace, clusterName) + lock.Lock() + defer lock.Unlock() + + cluster, err := handler.s.GetCluster(c, namespace, clusterName) + if err != nil { + helper.ResponseError(c, err) + return + } + + synced := make([]string, 0) + failures := make(map[string]string) + for _, shard := range cluster.Shards { + for _, node := range shard.Nodes { + // Skip nodes the controller already knows are down (e.g. a drained replica); pushing to + // them would only add expected errors to the report. + if node.Failed() { + continue + } + if err := node.SyncClusterInfo(c, cluster, true); err != nil { + failures[node.Addr()] = err.Error() + } else { + synced = append(synced, node.Addr()) + } + } + } + + if len(failures) > 0 { + helper.ResponseError(c, fmt.Errorf("synced %d node(s); %d failed: %v", len(synced), len(failures), failures)) + return + } + helper.ResponseOK(c, gin.H{"synced": synced, "version": cluster.Version.Load()}) +} + func (handler *ClusterHandler) Import(c *gin.Context) { namespace := c.Param("namespace") clusterName := c.Param("cluster") diff --git a/server/api/cluster_test.go b/server/api/cluster_test.go index d9e14705..96b60c31 100644 --- a/server/api/cluster_test.go +++ b/server/api/cluster_test.go @@ -23,6 +23,7 @@ import ( "bytes" "context" "encoding/json" + "errors" "io" "net/http" "net/http/httptest" @@ -180,7 +181,7 @@ func TestClusterImport(t *testing.T) { require.NoError(t, err) ctx := context.Background() require.NoError(t, cluster.Reset(ctx)) - require.NoError(t, clusterNode.SyncClusterInfo(ctx, cluster)) + require.NoError(t, clusterNode.SyncClusterInfo(ctx, cluster, true)) defer func() { // clean up the cluster information to avoid affecting other tests require.NoError(t, cluster.Reset(ctx)) @@ -303,3 +304,62 @@ func TestClusterMigrateData(t *testing.T) { } }) } + +// syncFakeStore returns a fixed cluster (with mock nodes) so the Sync handler can be tested without +// serializing through the storage engine, which would turn mock nodes back into real ClusterNodes. +type syncFakeStore struct { + store.Store + cluster *store.Cluster +} + +func (s *syncFakeStore) GetCluster(ctx context.Context, ns, name string) (*store.Cluster, error) { + if s.cluster == nil { + return nil, consts.ErrNotFound + } + return s.cluster, nil +} + +func TestClusterHandler_Sync(t *testing.T) { + ns, name := "test-ns", "sync-cluster" + full := []store.SlotRange{{Start: 0, Stop: 16383}} + + newHandler := func(nodes []store.Node) *ClusterHandler { + cluster := &store.Cluster{Name: name, Shards: store.Shards{{Nodes: nodes, SlotRanges: full}}} + cluster.Version.Store(1) + return &ClusterHandler{s: &syncFakeStore{Store: store.NewClusterStore(engine.NewMock()), cluster: cluster}} + } + call := func(h *ClusterHandler) *httptest.ResponseRecorder { + recorder := httptest.NewRecorder() + ctx := GetTestContext(recorder) + ctx.Params = []gin.Param{{Key: "namespace", Value: ns}, {Key: "cluster", Value: name}} + h.Sync(ctx) + return recorder + } + + t.Run("force-pushes healthy nodes and skips failed", func(t *testing.T) { + n0 := store.NewClusterMockNode() + n0.SetRole(store.RoleMaster) + n1 := store.NewClusterMockNode() + n1.SetRole(store.RoleSlave) + n2 := store.NewClusterMockNode() + n2.SetRole(store.RoleSlave) + n2.SetStatus(store.NodeStatusFailed) + + rec := call(newHandler([]store.Node{n0, n1, n2})) + require.Equal(t, http.StatusOK, rec.Code) + require.Equal(t, []bool{true}, n0.SyncForceCalls, "healthy node force-pushed") + require.Equal(t, []bool{true}, n1.SyncForceCalls, "healthy node force-pushed") + require.Empty(t, n2.SyncForceCalls, "failed node must be skipped") + }) + + t.Run("reports push failures with a non-OK status", func(t *testing.T) { + n0 := store.NewClusterMockNode() + n0.SetRole(store.RoleMaster) + n1 := store.NewClusterMockNode() + n1.SetRole(store.RoleSlave) + n1.SyncErr = errors.New("push failed") + + rec := call(newHandler([]store.Node{n0, n1})) + require.NotEqual(t, http.StatusOK, rec.Code) + }) +} diff --git a/server/api/shard.go b/server/api/shard.go index 502b23ab..d9b70df9 100644 --- a/server/api/shard.go +++ b/server/api/shard.go @@ -192,13 +192,13 @@ func (handler *ShardHandler) Failover(c *gin.Context) { wg.Add(2) go func() { defer wg.Done() - if e := oldMaster.SyncClusterInfo(c, cluster); e != nil { + if e := oldMaster.SyncClusterInfo(c, cluster, true); e != nil { logger.Get().With(zap.Error(e), zap.String("node", oldMaster.Addr())).Warn("Failed to sync cluster info to old master") } }() go func() { defer wg.Done() - if e := newMaster.SyncClusterInfo(c, cluster); e != nil { + if e := newMaster.SyncClusterInfo(c, cluster, true); e != nil { logger.Get().With(zap.Error(e), zap.String("node", newMaster.Addr())).Warn("Failed to sync cluster info to new master") } }() diff --git a/server/api/shard_test.go b/server/api/shard_test.go index c1315588..6f8a73ec 100644 --- a/server/api/shard_test.go +++ b/server/api/shard_test.go @@ -258,8 +258,8 @@ func TestClusterFailover(t *testing.T) { } // sync cluster info to each node - require.NoError(t, node0.SyncClusterInfo(ctx, gotCluster)) - require.NoError(t, node1.SyncClusterInfo(ctx, gotCluster)) + require.NoError(t, node0.SyncClusterInfo(ctx, gotCluster, true)) + require.NoError(t, node1.SyncClusterInfo(ctx, gotCluster, true)) clusterNodeInfo0, err := node0.GetClusterNodeInfo(ctx) require.NoError(t, err) diff --git a/server/route.go b/server/route.go index 113a1a09..04a22abf 100644 --- a/server/route.go +++ b/server/route.go @@ -71,6 +71,7 @@ func (srv *Server) initHandlers() { clusters.GET("/:cluster", middleware.RequiredCluster, handler.Cluster.Get) clusters.DELETE("/:cluster", middleware.RequiredCluster, handler.Cluster.Remove) clusters.POST("/:cluster/migrate", handler.Cluster.MigrateSlot) + clusters.POST("/:cluster/sync", middleware.RequiredCluster, handler.Cluster.Sync) clusters.POST("/:cluster/nodes/status", middleware.RequiredCluster, handler.Node.SetStatus) } diff --git a/store/cluster.go b/store/cluster.go index b402594b..30474faa 100644 --- a/store/cluster.go +++ b/store/cluster.go @@ -148,7 +148,9 @@ func (cluster *Cluster) PromoteNewMaster(ctx context.Context, func (cluster *Cluster) SyncToNodes(ctx context.Context) error { for i := 0; i < len(cluster.Shards); i++ { for _, node := range cluster.Shards[i].Nodes { - if err := node.SyncClusterInfo(ctx, cluster); err != nil { + // force: this runs right after a topology change, so assert the new state unconditionally + // rather than let the server's version gate no-op an equal-version push. + if err := node.SyncClusterInfo(ctx, cluster, true); err != nil { return err } } diff --git a/store/cluster_mock_node.go b/store/cluster_mock_node.go index 200ae93a..338e3c24 100644 --- a/store/cluster_mock_node.go +++ b/store/cluster_mock_node.go @@ -34,6 +34,13 @@ type ClusterMockNode struct { MasterReplOffset uint64 // used when simulating master in GetReplicationInfo SlaveOffset uint64 // used when simulating slave offset in GetReplicationInfo SlaveAddr string // when master, slave Addr for matching; empty means use mock.Addr() + + // MockClusterInfo, when set, is what GetClusterInfo returns (else a zero-value ClusterInfo). + MockClusterInfo *ClusterInfo + // SyncForceCalls records the `force` argument of every SyncClusterInfo call, for assertions. + SyncForceCalls []bool + // SyncErr, when set, is returned by SyncClusterInfo (to simulate a push failure). + SyncErr error } var _ Node = (*ClusterMockNode)(nil) @@ -49,11 +56,15 @@ func (mock *ClusterMockNode) GetClusterNodeInfo(ctx context.Context) (*ClusterNo } func (mock *ClusterMockNode) GetClusterInfo(ctx context.Context) (*ClusterInfo, error) { + if mock.MockClusterInfo != nil { + return mock.MockClusterInfo, nil + } return &ClusterInfo{}, nil } -func (mock *ClusterMockNode) SyncClusterInfo(ctx context.Context, cluster *Cluster) error { - return nil +func (mock *ClusterMockNode) SyncClusterInfo(ctx context.Context, cluster *Cluster, force bool) error { + mock.SyncForceCalls = append(mock.SyncForceCalls, force) + return mock.SyncErr } func (mock *ClusterMockNode) Reset(ctx context.Context) error { diff --git a/store/cluster_node.go b/store/cluster_node.go index bce333fc..4a80d81c 100755 --- a/store/cluster_node.go +++ b/store/cluster_node.go @@ -84,7 +84,7 @@ type Node interface { Reset(ctx context.Context) error GetClusterNodeInfo(ctx context.Context) (*ClusterNodeInfo, error) GetClusterInfo(ctx context.Context) (*ClusterInfo, error) - SyncClusterInfo(ctx context.Context, cluster *Cluster) error + SyncClusterInfo(ctx context.Context, cluster *Cluster, force bool) error CheckClusterMode(ctx context.Context) (int64, error) MigrateSlot(ctx context.Context, slot SlotRange, NodeID string) error @@ -111,6 +111,12 @@ type ClusterInfo struct { CurrentEpoch int64 `json:"cluster_current_epoch"` MigratingSlot *MigratingSlot `json:"migrating_slot"` MigratingState string `json:"migrating_state"` + // KnownNodes and SlotsOk are parsed from CLUSTER INFO so the reconcile loop can detect a node + // whose applied topology has drifted from the desired one even at an equal epoch — the case a + // pure epoch comparison misses (see apache/kvrocks-controller#395). SlotsOk is the count of + // assigned slots (16384 when fully covered); KnownNodes is the node's peer count. + KnownNodes int64 `json:"cluster_known_nodes"` + SlotsOk int64 `json:"cluster_slots_ok"` } type ClusterNodeInfo struct { @@ -252,8 +258,13 @@ func (n *ClusterNode) GetClusterInfo(ctx context.Context) (*ClusterInfo, error) if err != nil { return nil, err } + return parseClusterInfo(infoStr) +} - clusterInfo := &ClusterInfo{CurrentEpoch: -1} +// parseClusterInfo parses the CLUSTER INFO reply. Fields absent from the reply keep their sentinel -1 +// so a partial/older kvrocks reply never looks "converged" (all-slots-covered / expected-peer-count). +func parseClusterInfo(infoStr string) (*ClusterInfo, error) { + clusterInfo := &ClusterInfo{CurrentEpoch: -1, KnownNodes: -1, SlotsOk: -1} lines := strings.Split(infoStr, "\r\n") for _, line := range lines { fields := strings.Split(line, ":") @@ -261,22 +272,26 @@ func (n *ClusterNode) GetClusterInfo(ctx context.Context) (*ClusterInfo, error) continue } fields[1] = strings.TrimSpace(fields[1]) + var err error switch strings.ToLower(strings.TrimSpace(fields[0])) { case "cluster_current_epoch": clusterInfo.CurrentEpoch, err = strconv.ParseInt(fields[1], 10, 64) - if err != nil { - return nil, err - } + case "cluster_known_nodes": + clusterInfo.KnownNodes, err = strconv.ParseInt(fields[1], 10, 64) + case "cluster_slots_ok": + clusterInfo.SlotsOk, err = strconv.ParseInt(fields[1], 10, 64) case "migrating_slot", "migrating_slot(s)": // TODO(@git-hulk): handle multiple migrating slots - slotRange, err := ParseSlotRange(fields[1]) - if err != nil { - return nil, err + var slotRange *SlotRange + if slotRange, err = ParseSlotRange(fields[1]); err == nil { + clusterInfo.MigratingSlot = FromSlotRange(*slotRange) } - clusterInfo.MigratingSlot = FromSlotRange(*slotRange) case "migrating_state": clusterInfo.MigratingState = fields[1] } + if err != nil { + return nil, err + } } return clusterInfo, nil } @@ -315,17 +330,52 @@ func (n *ClusterNode) GetClusterNodesString(ctx context.Context) (string, error) return strings.TrimRight(clusterNodesStr, "\n"), nil } -func (n *ClusterNode) SyncClusterInfo(ctx context.Context, cluster *Cluster) error { +// syncMaxRetries and syncRetryBaseDelay bound the per-node push retry. kvrocks nodes do not gossip, +// so the controller is the sole topology source; a push dropped by a transient blip (node restarting, +// network hiccup) is never re-tried by any other path until the next probe tick, leaving that node on a +// stale/divergent view (apache/kvrocks-controller#395). A short bounded backoff closes that window. +const ( + syncMaxRetries = 3 + syncRetryBaseDelay = 150 * time.Millisecond +) + +// SyncClusterInfo pushes the authoritative topology to this node via CLUSTERX SETNODEID + SETNODES. +// +// When force is true the SETNODES carries the `force` flag, which makes kvrocks apply the topology +// unconditionally — bypassing its version gate (reject-if-lower, no-op-if-equal). This is what lets the +// controller REPAIR a node that has drifted at an equal epoch (the divergence #395 describes): a normal +// equal-version push is silently no-op'd by the server, so it can never fix a bad-but-same-version view. +// Because SETNODES replaces the node's entire topology, a forced push also removes stale/phantom node +// entries (e.g. empty-address ghosts) without the data-destroying CLUSTER RESET the alternative requires. +// Safe because the controller is the single writer of topology; callers must not force a version older +// than a node legitimately ahead of the store (the probe loop keeps the `node ahead` branch for that). +func (n *ClusterNode) SyncClusterInfo(ctx context.Context, cluster *Cluster, force bool) error { clusterStr, err := cluster.ToSlotString() if err != nil { return err } + args := []interface{}{"CLUSTERX", "SETNODES", clusterStr, cluster.Version.Load()} + if force { + args = append(args, "force") + } redisCli := n.GetClient() - err = redisCli.Do(ctx, "CLUSTERX", "SETNODEID", n.id).Err() - if err != nil { - return err + var lastErr error + for attempt := 0; attempt < syncMaxRetries; attempt++ { + if attempt > 0 { + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(syncRetryBaseDelay << (attempt - 1)): + } + } + if lastErr = redisCli.Do(ctx, "CLUSTERX", "SETNODEID", n.id).Err(); lastErr != nil { + continue + } + if lastErr = redisCli.Do(ctx, args...).Err(); lastErr == nil { + return nil + } } - return redisCli.Do(ctx, "CLUSTERX", "SETNODES", clusterStr, cluster.Version.Load()).Err() + return lastErr } func (n *ClusterNode) Reset(ctx context.Context) error { diff --git a/store/cluster_node_test.go b/store/cluster_node_test.go index 41bf55f8..df05cd3c 100644 --- a/store/cluster_node_test.go +++ b/store/cluster_node_test.go @@ -57,7 +57,7 @@ func TestClusterNode(t *testing.T) { }} cluster.Version.Store(1) - require.NoError(t, node0.SyncClusterInfo(ctx, cluster)) + require.NoError(t, node0.SyncClusterInfo(ctx, cluster, true)) clusterInfo, err := node0.GetClusterInfo(ctx) require.NoError(t, err) require.EqualValues(t, 1, clusterInfo.CurrentEpoch) @@ -104,3 +104,22 @@ func TestNodeInfo_Validate(t *testing.T) { node.addr = "1.2.3.4" require.NoError(t, node.Validate()) } + +func TestParseClusterInfo(t *testing.T) { + // Absent fields keep the -1 sentinel so a partial reply never looks converged. + info, err := parseClusterInfo("cluster_state:ok\r\ncluster_current_epoch:7\r\n") + require.NoError(t, err) + require.EqualValues(t, 7, info.CurrentEpoch) + require.EqualValues(t, -1, info.KnownNodes) + require.EqualValues(t, -1, info.SlotsOk) + + info, err = parseClusterInfo("cluster_current_epoch:3\r\ncluster_known_nodes:6\r\ncluster_slots_ok:16384\r\n") + require.NoError(t, err) + require.EqualValues(t, 3, info.CurrentEpoch) + require.EqualValues(t, 6, info.KnownNodes) + require.EqualValues(t, 16384, info.SlotsOk) + + // A malformed integer surfaces as an error rather than a silent zero. + _, err = parseClusterInfo("cluster_known_nodes:not-a-number\r\n") + require.Error(t, err) +} diff --git a/store/cluster_shard.go b/store/cluster_shard.go index ba1c10d0..29c45363 100644 --- a/store/cluster_shard.go +++ b/store/cluster_shard.go @@ -24,6 +24,7 @@ import ( "encoding/json" "errors" "fmt" + "net" "strings" "sync" "time" @@ -41,7 +42,6 @@ const ( NotMigratingInt = -1 ) - // FailoverOptions configures manual failover behavior. type FailoverOptions struct { WaitForSync bool // whether to wait for replication gap to reach 0 @@ -420,6 +420,12 @@ func (shard *Shard) ToSlotsString() (string, error) { } for i, node := range shard.Nodes { + // A blank host (":port") would emit a malformed line and register a phantom, unreachable node. + // Fail loudly here rather than push corruption to every kvrocks node. + host, port, err := net.SplitHostPort(node.Addr()) + if err != nil || host == "" || port == "" { + return "", fmt.Errorf("node %s has invalid address %q", node.ID(), node.Addr()) + } builder.WriteString(node.ID()) builder.WriteByte(' ') builder.WriteString(strings.Replace(node.Addr(), ":", " ", 1)) diff --git a/store/cluster_shard_test.go b/store/cluster_shard_test.go index 97216fcf..7789e9e0 100644 --- a/store/cluster_shard_test.go +++ b/store/cluster_shard_test.go @@ -99,6 +99,34 @@ func TestToSlotsString_WithFailedSlave(t *testing.T) { require.Contains(t, result, "slave,fail "+master.ID()) } +func TestToSlotsString_RejectsBlankAddr(t *testing.T) { + // An unresolved pod hostname yields a blank host (":6380"); this must not serialize into a + // phantom node line — ToSlotsString should fail loudly instead, for either a master or a slave. + t.Run("slave", func(t *testing.T) { + shard := NewShard() + shard.SlotRanges = []SlotRange{{Start: 0, Stop: 100}} + master := NewClusterNode("127.0.0.1:6379", "") + master.SetRole(RoleMaster) + slave := NewClusterNode(":6380", "") + slave.SetRole(RoleSlave) + shard.Nodes = []Node{master, slave} + + _, err := shard.ToSlotsString() + require.Error(t, err) + }) + + t.Run("master", func(t *testing.T) { + shard := NewShard() + shard.SlotRanges = []SlotRange{{Start: 0, Stop: 100}} + master := NewClusterNode(":6379", "") + master.SetRole(RoleMaster) + shard.Nodes = []Node{master} + + _, err := shard.ToSlotsString() + require.Error(t, err) + }) +} + func TestReplicaAppliedReplOffset(t *testing.T) { require.Equal(t, uint64(0), ReplicaAppliedReplOffset(nil)) require.Equal(t, uint64(10), ReplicaAppliedReplOffset(&ReplicationInfo{Role: RoleMaster, MasterReplOffset: 10})) diff --git a/store/store.go b/store/store.go index fc319fcc..078ec02c 100644 --- a/store/store.go +++ b/store/store.go @@ -24,9 +24,11 @@ import ( "context" "encoding/json" "fmt" + "net" + "sync" + "github.com/apache/kvrocks-controller/logger" "go.uber.org/zap" - "sync" "github.com/apache/kvrocks-controller/consts" "github.com/apache/kvrocks-controller/store/engine" @@ -274,6 +276,12 @@ func (s *ClusterStore) RemoveCluster(ctx context.Context, ns, cluster string) er func (s *ClusterStore) CheckNewNodes(ctx context.Context, nodes []string) error { newNodes := make(map[string]bool, 0) for _, node := range nodes { + // Reject a blank/half-formed address (e.g. ":6666" from an unresolved hostname) at the API + // boundary: it would otherwise serialize into a malformed CLUSTERX SETNODES line and register + // a phantom, unreachable node. + if host, port, err := net.SplitHostPort(node); err != nil || host == "" || port == "" { + return fmt.Errorf("%w: node address must be host:port, got %q", consts.ErrInvalidArgument, node) + } newNodes[node] = true } diff --git a/store/store_test.go b/store/store_test.go index b0517f70..b8609430 100644 --- a/store/store_test.go +++ b/store/store_test.go @@ -103,5 +103,8 @@ func TestClusterStore(t *testing.T) { require.NoError(t, store.CheckNewNodes(ctx, []string{"127.0.0.1:4444", "127.0.0.1:5555"})) require.NotNil(t, store.CheckNewNodes(ctx, []string{"127.0.0.1:3333", "127.0.0.1:4444"})) require.NotNil(t, store.CheckNewNodes(ctx, []string{"127.0.0.1:2222", "127.0.0.1:3333"})) + // Blank / port-less addresses are rejected at the boundary (no phantom nodes). + require.Error(t, store.CheckNewNodes(ctx, []string{":6666"})) + require.Error(t, store.CheckNewNodes(ctx, []string{"127.0.0.1"})) }) }