登录/ 注册
 找回密码
 立即注册
关闭

CSDN热搜

程序园 精品问答 技术交流 资源下载
返回列表 发新帖
首页 › 业界区 › 业界 ›hashicorp/raft模块实现的raft集群存在节点跨集群身份冲 ...

hashicorp/raft模块实现的raft集群存在节点跨集群身份冲突问题

奄幂牛 7 天前
问题场景描述

我通过模块github.com/hashicorp/raft使用golang实现了一个raft集群功能，发现如下场景中会遇到一个问题：
测试启动如下2个raft集群，集群名称，和集群node与IP地址如下，raft集群均通过BootstrapCluster方法初始化：
Cluster1 BootstrapCluster servers:
  1. - node1: {raft.ServerID: c1-node1, raft.ServerAddress: 192.168.100.1:7000}
  2. - node2: {raft.ServerID: c1-node2, raft.ServerAddress: 192.168.100.2:7000}
  3. - node3: {raft.ServerID: c1-node3, raft.ServerAddress: 192.168.100.3:7000}
复制代码
Cluster2 BootstrapCluster servers：
  1. - node3: {raft.ServerID: c2-node3, raft.ServerAddress: 192.168.100.3:7000}
  2. - node4: {raft.ServerID: c2-node4, raft.ServerAddress: 192.168.100.4:7000}
  3. - node5: {raft.ServerID: c2-node5, raft.ServerAddress: 192.168.100.5:7000}
复制代码
其中，"node3"的地址会存在2个集群中。

  • "node1","node2"按照"Cluster1"启动:
sudo ./raft_svr -cluster 'c1-node1,127.0.0.1,800;c1-node2,127.0.0.2,800;c1-node3,127.0.0.3,800' -id c1-node1
sudo ./raft_svr -cluster 'c1-node1,127.0.0.1,800;c1-node2,127.0.0.2,800;c1-node3,127.0.0.3,800' -id c1-node2

  • "node3","node4","node5"先按照"Cluster2"启动:
sudo ./raft_svr -cluster 'c2-node3,127.0.0.3,800;c2-node4,127.0.0.4,800;c2-node5,127.0.0.5,800' -id c2-node3
sudo ./raft_svr -cluster 'c2-node3,127.0.0.3,800;c2-node4,127.0.0.4,800;c2-node5,127.0.0.5,800' -id c2-node4
sudo ./raft_svr -cluster 'c2-node3,127.0.0.3,800;c2-node4,127.0.0.4,800;c2-node5,127.0.0.5,800' -id c2-node5
然后就会发现"node3"会在"Cluster1"和"Cluster2"之间来回切换，一会属于"Cluster1"，一会属于"Cluster2".
  1. INFO[0170] current state:Follower, leader address:127.0.0.5:800, servers:[{Suffrage:Voter ID:c2-node3 Address:127.0.0.3:800} {Suffrage:Voter ID:c2-node4 Address:127.0.0.4:800} {Suffrage:Voter ID:c2-node5 Address:127.0.0.5:800}], last contact:2025-05-14 15:35:53.330867 +0800 CST m=+169.779019126
  2. INFO[0171] current state:Follower, leader address:127.0.0.1:800, servers:[{Suffrage:Voter ID:c2-node3 Address:127.0.0.3:800} {Suffrage:Voter ID:c2-node4 Address:127.0.0.4:800} {Suffrage:Voter ID:c2-node5 Address:127.0.0.5:800}], last contact:2025-05-14 15:35:54.308388 +0800 CST m=+170.756576126
复制代码
我的代码如下
  1. package main
  3. import (
  4.         "flag"
  5.         "fmt"
  6.         "io"
  7.         "net"
  8.         "os"
  9.         "strconv"
  10.         "strings"
  11.         "time"
  13.         "github.com/hashicorp/raft"
  14.         log "github.com/sirupsen/logrus"
  15. )
  17. type raftCluster struct {
  18.         localRaftID     raft.ServerID
  19.         servers         map[raft.ServerID]raft.ServerAddress // raftID : raftAddressPort
  20.         raft            *raft.Raft
  21.         electionTimeout time.Duration
  22. }
  24. func (r *raftCluster) Start() error {
  25.         config := raft.DefaultConfig()
  26.         config.HeartbeatTimeout = 2000 * time.Millisecond
  27.         config.ElectionTimeout = 5000 * time.Millisecond
  28.         config.CommitTimeout = 2000 * time.Millisecond
  29.         config.LeaderLeaseTimeout = 2000 * time.Millisecond
  30.         config.LocalID = r.localRaftID
  31.         config.LogOutput = log.StandardLogger().Out
  33.         r.electionTimeout = config.ElectionTimeout * time.Duration(len(r.servers)*2)
  35.         localAddressPort := string(r.servers[r.localRaftID])
  36.         tcpAddr, err := net.ResolveTCPAddr("tcp", localAddressPort)
  37.         if err != nil {
  38.                 return fmt.Errorf("resolve tcp address %s, %v", localAddressPort, err)
  39.         }
  40.         transport, err := raft.NewTCPTransport(localAddressPort, tcpAddr, 2, 10*time.Second, log.StandardLogger().Out)
  41.         if err != nil {
  42.                 return fmt.Errorf("fail to create tcp transport, localAddressPort:%s, tcpAddr:%v, %v",
  43.                         localAddressPort, tcpAddr, err)
  44.         }
  45.         snapshots := raft.NewInmemSnapshotStore()
  46.         logStore := raft.NewInmemStore()
  47.         stableStore := raft.NewInmemStore()
  48.         fm := NewFsm()
  49.         r.raft, err = raft.NewRaft(config, fm, logStore, stableStore, snapshots, transport)
  50.         if err != nil {
  51.                 return fmt.Errorf("create raft error, %v", err)
  52.         }
  54.         var configuration raft.Configuration
  55.         for sID, addr := range r.servers {
  56.                 server := raft.Server{
  57.                         ID:      sID,
  58.                         Address: addr,
  59.                 }
  60.                 configuration.Servers = append(configuration.Servers, server)
  61.         }
  62.         err = r.raft.BootstrapCluster(configuration).Error()
  63.         if err != nil {
  64.                 return fmt.Errorf("raft bootstrap faild, conf:%v, %v", configuration, err)
  65.         }
  66.         log.Infof("bootstrap cluster as config: %v", configuration)
  68.         return nil
  69. }
  71. func (r *raftCluster) checkLeaderState() {
  72.         ticker := time.NewTicker(time.Second)
  73.         for {
  74.                 select {
  75.                 case leader := <-r.raft.LeaderCh():
  76.                         log.Infof("im leader:%v, state:%s, leader address:%s", leader, r.raft.State(), r.raft.Leader())
  78.                 case <-ticker.C:
  79.                         verifyErr := r.raft.VerifyLeader().Error()
  80.                         servers := r.raft.GetConfiguration().Configuration().Servers
  81.                         switch verifyErr {
  82.                         case nil:
  83.                                 log.Infof("im leader, servers:%v", servers)
  84.                         case raft.ErrNotLeader:
  85.                                 // check cluster leader
  86.                                 log.Infof("current state:%v, servers:%+v, leader address:%v, last contact:%v",
  87.                                         r.raft.State(), servers, r.raft.Leader(), r.raft.LastContact())
  88.                         }
  89.                 }
  90.         }
  91. }
  93. func main() {
  94.         var (
  95.                 clusters = flag.String("cluster", "",
  96.                         "cluster node address, fmt: ID,IP,Port;ID,IP,Port")
  97.                 clusterId = flag.String("id", "", "cluster id")
  98.         )
  99.         flag.Parse()
  101.         if *clusterId == "" {
  102.                 log.Infof("cluster id messing")
  103.                 os.Exit(1)
  104.         }
  106.         servers := make(map[raft.ServerID]raft.ServerAddress)
  107.         for _, cluster := range strings.Split(*clusters, ";") {
  108.                 info := strings.Split(cluster, ",")
  109.                 var (
  110.                         nid   string
  111.                         nip   net.IP
  112.                         nport int
  113.                         err   error
  114.                 )
  115.                 switch {
  116.                 case len(info) == 3:
  117.                         nid = info[0]
  118.                         nip = net.ParseIP(info[1])
  119.                         if nip == nil {
  120.                                 log.Infof("cluster %s ip %s parse failed", cluster, info[1])
  121.                                 os.Exit(1)
  122.                         }
  123.                         nport, err = strconv.Atoi(info[2])
  124.                         if err != nil {
  125.                                 log.Infof("cluster %s port %s parse failed, %v", cluster, info[2], err)
  126.                         }
  127.                 default:
  128.                         log.Infof("cluster args value is bad format")
  129.                         os.Exit(1)
  130.                 }
  131.                 log.Infof("cluster node id:%s, ip:%v, port:%d", nid, nip, nport)
  132.                 addr := net.TCPAddr{IP: nip, Port: nport}
  133.                 servers[raft.ServerID(nid)] = raft.ServerAddress(addr.String())
  134.         }
  136.         r := raftCluster{
  137.                 localRaftID: raft.ServerID(*clusterId),
  138.                 servers:     servers,
  139.         }
  140.         err := r.Start()
  141.         if err != nil {
  142.                 log.Infof("rafter cluster start failed, %v", err)
  143.                 os.Exit(1)
  144.         }
  145.         r.checkLeaderState()
  146. }
  148. // SimpleFsm: 实现一个简单的Fsm
  150. type SimpleFsm struct {
  151.         db database
  152. }
  154. func NewFsm() *SimpleFsm {
  155.         fsm := &SimpleFsm{
  156.                 db: NewDatabase(),
  157.         }
  158.         return fsm
  159. }
  161. func (f *SimpleFsm) Apply(l *raft.Log) interface{} {
  162.         return nil
  163. }
  165. func (f *SimpleFsm) Snapshot() (raft.FSMSnapshot, error) {
  166.         return &f.db, nil
  167. }
  169. func (f *SimpleFsm) Restore(io.ReadCloser) error {
  170.         return nil
  171. }
  173. type database struct{}
  175. func NewDatabase() database {
  176.         return database{}
  177. }
  179. func (d *database) Get(key string) string {
  180.         return "not implemented"
  181. }
  183. func (d *database) Set(key, value string) {}
  185. func (d *database) Persist(sink raft.SnapshotSink) error {
  186.         _, _ = sink.Write([]byte{})
  187.         _ = sink.Close()
  188.         return nil
  189. }
  191. func (d *database) Release() {}
复制代码
问题排除

重新编译运行后，我们看到node3始终保持在Cluster2中，并且可以看到如下日志
  1. sudo ifconfig lo0 alias 127.0.0.2 up
  2. sudo ifconfig lo0 alias 127.0.0.3 up
  3. sudo ifconfig lo0 alias 127.0.0.4 up
  4. sudo ifconfig lo0 alias 127.0.0.5 up
复制代码
在Cluster1的leader日志中，我们可以看到该leader向node3发送心跳失败的日志：
  1. [WARN]  raft: rejecting appendEntries request since node is not in configuration: from=c1-node1
复制代码
提醒

注意，这个修改方法还没有得到官方的认可，可能会有其他潜在的影响，使用之前应该自我评估。

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

使用道具 举报
相关推荐
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册
发帖
奄幂牛
7 天前

0

粉丝关注

13

主题发布

板块介绍填写区域，请于后台编辑
微信扫一扫