首页 > 其他分享 >使用hashicorp Raft开发分布式服务

使用hashicorp Raft开发分布式服务

时间：2023-06-16 22:55:42浏览次数：58  
标签：log hashicorp FSM 192.168 分布式服务 snapshot Raft 节点

使用hashicorp Raft开发高可用服务

开发raft时用到的比较主流的两个库是Etcd Raft 和hashicorp Raft，网上也有一些关于这两个库的讨论。之前分析过etcd Raft，发现该库相对hashicorp Raft比较难以理解，其最大的问题是没有实现网络层，实现难度比较大，因此本文在实现时使用了hashicorp Raft。

下文中会参考consul的一致性协议来讲解如何实现Raft协议。

Raft概述

术语

  • Log entry：Raft的主要单元。Raft将一致性问题分解为日志复制。日志是一个有序的表项，其包含了Raft的集群变更信息(如添加/移除节点)以及对应用数据的操作等。
  • FSM：Finite State Machine。FSM是有限状态的集合。当一条日志被Raft apply后，可以对FSM进行状态转换。相同顺序的日志在apply之后必须产生相同的结果，即行为必须是确定性的。
  • Peer set：指所有参与日志复制的成员。
  • Quorum：仲裁指peer set中的大部分成员：对于包含N个成员的peer set，仲裁要求有(N/2)+1个成员。如果出于某些原因导致仲裁节点不可用，则集群会变为unavailable状态，且新的日志也不会被commit。
  • Committed Entry：当一个Log entry持久化到仲裁数量的节点后，该认为该Log entry是Committed的。只有当Log entry 被Committed之后，它才会被FSM apply。
  • Leader：任何时间，peer set会选举一个节点作为leader。leader负责处理新的Log entry，并将其复制给follower，以及决定何时将Log entry判定为committed状态。

Raft机制简介

Raft节点总是处于三种状态之一： follower, candidate, leader。一开始，所有的节点都是follower，该状态下，节点可以从leader接收log，并参与选举，如果一段时间内没有接收到任何Log entry，则节点会自提升到candidate状态。在candidate状态下，节点会请求其他节点的选举，如果一个candidate接收到大部分节点(仲裁数目)的认同，就会被提升为leader。leader必须接收新的Log entry，并复制到所有其他follower。

如果用户无法接受旧的数据，则所有的请求必须由leader执行。

一旦一个集群有了leader，就可以接收新的Log entry。客户端可以请求leader追加一个新的Log entry。Leader会将Log entry写入持久化存储，并尝试将其复制给仲裁数目的follower。一旦Log entry被commit，就可以将该Log entry apply到FSM。FSM是应用特定的存储，在Consul中，使用 MemDB来维护集群状态。

无限量复制log的方式是不可取的。Raft 提供了一种机制，可以对当前状态进行快照并压缩log。由于 FSM 的抽象，FSM 的状态恢复必须与replay log的状态相同。Raft 可以捕获某个时刻的 FSM 状态，然后移除用于达到该状态的所有log。这些操作可以在没有用户干预的情况下自动执行，防止无限使用磁盘，同时最小化replay log所花费的时间。

Raft Consensus中所有的操作都必须经过Leader，因此需要保证所有的请求都能够发送到Leader节点，然后由Leader将请求发送给所有Follower，并等待大部分(仲裁数目的)节点处理完成该命令，leader通过选举机制产生。之后每个follower会执行如下操作：

  1. 在接收到命令之后，使用WAL方式将数据保存为Log entry
  2. 在成功写入log entry之后，将数据发给FSM进行处理
  3. 在FSM成功处理完数据之后，返回数据。之后Leader会注意到该节点已经成功完成数据处理。

如下表述来自Raft Protocol Overview，它的意思是，如果是查询类的请求，直接从FSM返回结果即可，如果是修改类的请求，则需要通过raft.Apply来保证变更的一致性

所有raft集群中的成员都知道当前的leader，当一个RPC请求到达一个非leader的成员时，它会将该请求转发给当前的leader。如果是查询类型的RPC，意味着它是只读的，leader会基于FSM的当前状态生成结果；如果是事务类型的RPC，意味着它需要修改状态，Leader会生成一条新的log entry，并执行Raft.Apply，当log entry被提交并apply到FSM之后，事务才算执行完成。

接口和原理描述

如下是官方给出的Raft Apply的流程图：

sequenceDiagram autonumber participant client participant leadermain as leader:main participant leaderfsm as leader:fsm participant leaderreplicate as leader:replicate (each peer) participant followermain as follower:main (each peer) participant followerfsm as follower:fsm (each peer) client-)leadermain: applyCh to dispatchLogs leadermain->>leadermain: store logs to disk leadermain-)leaderreplicate: triggerCh leaderreplicate-->>followermain: Transport.AppendEntries RPC followermain->>followermain: store logs to disk opt leader commit index is ahead of peer commit index followermain-)followerfsm: fsmMutateCh <br>apply committed logs followerfsm->>followerfsm: fsm.Apply end followermain-->>leaderreplicate: respond success=true leaderreplicate->>leaderreplicate: update commitment opt quorum commit index has increased leaderreplicate-)leadermain: commitCh leadermain-)leaderfsm: fsmMutateCh leaderfsm->>leaderfsm: fsm.Apply leaderfsm-)client: future.respond end

Apply是数据进入Raft的接口，整个Raft的主要作用是维护数据操作的一致性。在上图中，由两个apply：一个是Raft.Apply(内部会通过applyCh传递log)，其也是外部数据的入口。另一个是FSM.Apply，其数据源头是Raft.Apply。FSM基于Raft实现了一致性读写。在上图中可以看到，leader的FSM.Apply是在数据commit成功(仲裁成功)之后才执行的，这样就能以Raft的方式保证分布式场景下应用数据的一致性，可以将FSM.Apply理解为应用数据的写入操作。

Raft中的一条log表示一个操作。使用hashicorp/raft时应该将实现分为两层：一层是底层的Raft，支持Raft数据的存储、快照等，集群的选举和恢复等，这一部分由Raft模块自实现；另一层是应用层，需要由用户实现FSM接口，FSM的接口并不对外，在Raft的处理过程中会调用FSM的接口来实现应用数据的存储、备份和恢复等操作。这两层都有数据的读写和快照实现，因此在理解上需要进行区分。

Raft节点的初始化

如果是新建的Raft节点，可以使用BootstrapCluster方法初始化该节点。为避免非新的节点被初始化，在调用BootstrapCluster前可以使用raft.HasExistingState来判断实例中是否包含相关状态(logs，当前term或snapshot)：

if (s.config.Bootstrap) && !s.config.ReadReplica {
  hasState, err := raft.HasExistingState(log, stable, snap)
  if err != nil {
    return err
  }
  if !hasState {
    configuration := raft.Configuration{
      Servers: []raft.Server{
        {
          ID:      s.config.RaftConfig.LocalID,
          Address: trans.LocalAddr(),
        },
      },
    }
    if err := raft.BootstrapCluster(s.config.RaftConfig,
      log, stable, snap, trans, configuration); err != nil {
      return err
    }
  }
}

Raft节点的创建

Raft节点的创建方法如下，如果存储非空，则Raft会尝试恢复该节点：

func NewRaft(conf *Config,
    fsm FSM,
    logs LogStore,
    stable StableStore,
    snaps SnapshotStore,
    trans Transport) (*Raft, error) {

包括：

  • fsm：由应用实现，用于处理应用数据。FSM中的数据来自底层的Raft log。

  • logs，stable和snaps：logs(存储Raft log) ，stable(保存Raft选举信息，如角色、term等信息) 可以使用raftboltdb.New进行初始化， snaps用于Leader和follower之间的批量数据同步以及(手动或自动)集群恢复，可以使用raft.NewFileSnapshotStore或raft.NewFileSnapshotStoreWithLogger进行初始化。

  • trans：Transport是raft集群内部节点之间的信息通道，节点之间需要通过该通道来同步log、选举leader等。下面接口中的AppendEntriesPipeline和AppendEntries方法用于log同步，RequestVote用于leader选举，InstallSnapshot用于在follower 的log落后过多的情况下，给follower发送snapshot(批量log)。

    可以使用raft.NewTCPTransport、raft.NewTCPTransportWithLogger或raft.NewNetworkTransportWithConfig方法来初始化trans。

    type Transport interface {
  ...
  AppendEntriesPipeline(id ServerID, target ServerAddress) (AppendPipeline, error)
  AppendEntries(id ServerID, target ServerAddress, args *AppendEntriesRequest, resp *AppendEntriesResponse) error

  // RequestVote sends the appropriate RPC to the target node.
  RequestVote(id ServerID, target ServerAddress, args *RequestVoteRequest, resp *RequestVoteResponse) error
  InstallSnapshot(id ServerID, target ServerAddress, args *InstallSnapshotRequest, resp *InstallSnapshotResponse, data io.Reader) error
  ...
}

NewRaft方法中会运行如下后台任务:

r.goFunc(r.run)           //处理角色变更和RPC请求
r.goFunc(r.runFSM)        //负责将logs apply到FSM
r.goFunc(r.runSnapshots)  //管理FSM的snapshot

监控Leader变化

为保证数据的一致性，只能通过leader写入数据，因此需要及时了解leader的变更信息，在Raft的配置中有一个变量NotifyCh chan<- bool，当Raft变为leader时会将true写入该chan，通过读取该chan来判断本节点是否是leader。在初始化Raft配置的时候传入即可：

  leaderNotifyCh := make(chan bool, 10)
  raftConfig.NotifyCh = leaderNotifyCh

还有其他方式可以获取leader变更状态：

如下方法可以生成一个chan，当本节点变为Leader时会发送true，当本节点丢失Leader角色时发送false，该方法的用途与上述方式相同，但由于该方法没有缓存，可能导致丢失变更信号，因此推荐使用上面的方式。

func (r *Raft) LeaderCh() <-chan bool

实现FSM

至此已经完成了Raft的初始化。下面就是要实现初始化函数中要求实现的内容，主要就是实现FSM接口。其中logs、stable、snaps和trans已经提到，使用现成的方法初始化即可。对于存储来说，也可以根据需要采用其他方式，如S3。

下面是LogStore、SnapshotStore和StableStore的接口定义。

type LogStore interface {//用于存储Raft log
  // FirstIndex returns the first index written. 0 for no entries.
  FirstIndex() (uint64, error)

  // LastIndex returns the last index written. 0 for no entries.
  LastIndex() (uint64, error)

  // GetLog gets a log entry at a given index.
  GetLog(index uint64, log *Log) error

  // StoreLog stores a log entry.
  StoreLog(log *Log) error

  // StoreLogs stores multiple log entries.
  StoreLogs(logs []*Log) error

  // DeleteRange deletes a range of log entries. The range is inclusive.
  DeleteRange(min, max uint64) error
}

type SnapshotStore interface {//用于快照的生成和恢复
  // Create is used to begin a snapshot at a given index and term, and with
  // the given committed configuration. The version parameter controls
  // which snapshot version to create.
  Create(version SnapshotVersion, index, term uint64, configuration Configuration,
    configurationIndex uint64, trans Transport) (SnapshotSink, error)

  // List is used to list the available snapshots in the store.
  // It should return then in descending order, with the highest index first.
  List() ([]*SnapshotMeta, error)

  // Open takes a snapshot ID and provides a ReadCloser. Once close is
  // called it is assumed the snapshot is no longer needed.
  Open(id string) (*SnapshotMeta, io.ReadCloser, error)
}

type StableStore interface { //用于存储集群元数据
  Set(key []byte, val []byte) error

  // Get returns the value for key, or an empty byte slice if key was not found.
  Get(key []byte) ([]byte, error)

  SetUint64(key []byte, val uint64) error

  // GetUint64 returns the uint64 value for key, or 0 if key was not found.
  GetUint64(key []byte) (uint64, error)
}

FSM基于Raft来实现，包含三个方法：

  • Apply：在Raft完成commit索引之后，保存应用数据。
  • Snapshot：用于支持log压缩，可以保存某个时间点的FSM快照。需要注意的是，由于Apply和Snapshot运行在同一个线程中(如run和runFSM线程)，因此要求函数能够快速返回，否则会阻塞Apply的执行。在实现中，该函数只需捕获指向当前状态的指针，而对于IO开销较大的操作，则放到FSMSnapshot.Persist中执行。
  • Restore：用于从snapshot恢复FSM
type FSM interface {
  // Apply is called once a log entry is committed by a majority of the cluster.
  //
  // Apply should apply the log to the FSM. Apply must be deterministic and
  // produce the same result on all peers in the cluster.
  //
  // The returned value is returned to the client as the ApplyFuture.Response.
  Apply(*Log) interface{}

  // Snapshot returns an FSMSnapshot used to: support log compaction, to
  // restore the FSM to a previous state, or to bring out-of-date followers up
  // to a recent log index.
  //
  // The Snapshot implementation should return quickly, because Apply can not
  // be called while Snapshot is running. Generally this means Snapshot should
  // only capture a pointer to the state, and any expensive IO should happen
  // as part of FSMSnapshot.Persist.
  //
  // Apply and Snapshot are always called from the same thread, but Apply will
  // be called concurrently with FSMSnapshot.Persist. This means the FSM should
  // be implemented to allow for concurrent updates while a snapshot is happening.
  Snapshot() (FSMSnapshot, error)

  // Restore is used to restore an FSM from a snapshot. It is not called
  // concurrently with any other command. The FSM must discard all previous
  // state before restoring the snapshot.
  Restore(snapshot io.ReadCloser) error
}

FSMSnapshot是实现快照需要实现的另一个接口，用于保存持久化FSM状态，后续可以通过FSM.Restore方法恢复FSM。该接口不会阻塞Raft.Apply，但在持久化FSM的数据时需要保证不影响Raft.Apply的并发访问。

FSMSnapshot.Persist的入参sink是调用SnapshotStore.Creates时的返回值。如果是通过raft.NewFileSnapshotStore初始化了SnapshotStore，则入参sink的类型就是FileSnapshotStore。

FSMSnapshot.Persist执行结束之后需要执行SnapshotSink.Close() ，如果出现错误，则执行SnapshotSink.Cancel() 。

// FSMSnapshot is returned by an FSM in response to a Snapshot
// It must be safe to invoke FSMSnapshot methods with concurrent
// calls to Apply.
type FSMSnapshot interface {
  // Persist should dump all necessary state to the WriteCloser 'sink',
  // and call sink.Close() when finished or call sink.Cancel() on error.
  Persist(sink SnapshotSink) error

  // Release is invoked when we are finished with the snapshot.
  Release()
}

FSM的备份和恢复

FSM的备份和恢复的逻辑比较难理解，一方面备份的数据存储在Raft中，FSM接口是由Raft主动调用的，另一方面又需要由用户实现FSM的备份和恢复逻辑，因此需要了解Raft是如何与FSM交互的。

FSM依赖snapshot来实现备份和恢复，snapshot中保存的也都是FSM信息。

何时备份和恢复
备份的时机

  • 当用户执行RecoverCluster接口时会调用FSM.Snapshot触发创建一个新的FSM snapshot

  • 手动调用如下接口也会触发创建FSM snapshot：

    func (r *Raft) Snapshot() SnapshotFuture

  • Raft自动备份也会触发创建FSM snapshot，默认时间为[120s, 240s]之间的随机时间。

恢复的时机

  • 当用户执行RecoverCluster接口时会调用FSM.Restore，用于手动恢复集群

  • 当用户执行Raft.Restore接口时会调用FSM.Restore，用于手动恢复集群

  • 通过NewRaft创建Raft节点时会尝试恢复snapshot(Raft.restoreSnapshot-->Raft.tryRestoreSingleSnapshot-->fsmRestoreAndMeasure-->fsm.Restore)

因此在正常情况下，Raft会不定期创建snapshot，且在创建Raft节点(新建或重启)的时候也会尝试通过snapshot来恢复FSM。

备份和恢复的内部逻辑

FSM的备份和恢复与SnapshotStore接口息息相关。

在备份FSM时的逻辑如下，首先通过SnapshotStore.Create创建一个snapshot，然后初始化一个FSMSnapshot实例，并通过FSMSnapshot.Persist将FSM保存到创建出的snapshot中：

sink, err := snaps.Create(version, lastIndex, lastTerm, configuration, 1, trans) //创建一个snapshot
snapshot, err := fsm.Snapshot() //初始化一个FSMSnapshot实例
snapshot.Persist(sink)          //调用FSMSnapshot.Persist将FSM保存到上面的snapshot中

恢复FSM的逻辑如下，首先通过SnapshotStore.List获取snapshots，然后通过SnapshotStore.Open逐个打开获取到的snapshot，最后调用FSM.Restore恢复FSM，其入参可以看做是snapshot的文件描述符：

snapshots, err = snaps.List()
for _, snapshot := range snapshots {
  _, source, err = snaps.Open(snapshot.ID)
  crc := newCountingReadCloser(source)
  err = fsm.Restore(crc)
  // Close the source after the restore has completed
  source.Close()
}

下面以consul的实现为例看下它是如何进行FSM的备份和恢复的。

备份

FSM.Snapshot()的作用就是返回一个SnapshotSink接口对象，进而调用SnapshotSink.Persist来持久化FSM。

下面是consul的SnapshotSink实现，逻辑比较简单，它将FSM持久化到了一个snapshot中，注意它在写入snapshot前做了编码(编码类型为ChunkingStateType)：

// Persist saves the FSM snapshot out to the given sink.
func (s *snapshot) Persist(sink raft.SnapshotSink) error {
  ...
  // Write the header
  header := SnapshotHeader{
    LastIndex: s.state.LastIndex(),
  }
  encoder := codec.NewEncoder(sink, structs.MsgpackHandle)
  if err := encoder.Encode(&header); err != nil {
    sink.Cancel()
    return err
  }
  ...

  if _, err := sink.Write([]byte{byte(structs.ChunkingStateType)}); err != nil {
    return err
  }
  if err := encoder.Encode(s.chunkState); err != nil {
    return err
  }

  return nil
}

func (s *snapshot) Release() {
  s.state.Close()
}
恢复

备份时将FSM保存在了snapshot中，恢复时读取并解码对应类型的snapshot即可：

// Restore streams in the snapshot and replaces the current state store with a
// new one based on the snapshot if all goes OK during the restore.
func (c *FSM) Restore(old io.ReadCloser) error {
  defer old.Close()
  ...

  handler := func(header *SnapshotHeader, msg structs.MessageType, dec *codec.Decoder) error {
    switch {
    case msg == structs.ChunkingStateType: //解码数据
      chunkState := &raftchunking.State{
        ChunkMap: make(raftchunking.ChunkMap),
      }
      if err := dec.Decode(chunkState); err != nil {
        return err
      }
      if err := c.chunker.State(chunkState); err != nil {
        return err
      }
      ...
    default:
      if msg >= 64 {
        return fmt.Errorf("msg type <%d> is a Consul Enterprise log entry. Consul OSS cannot restore it", msg)
      } else {
        return fmt.Errorf("Unrecognized msg type %d", msg)
      }
    }
    return nil
  }
  if err := ReadSnapshot(old, handler); err != nil {
    return err
  }
  ...

  return nil
}

// ReadSnapshot decodes each message type and utilizes the handler function to
// process each message type individually
func ReadSnapshot(r io.Reader, handler func(header *SnapshotHeader, msg structs.MessageType, dec *codec.Decoder) error) error {
  // Create a decoder
  dec := codec.NewDecoder(r, structs.MsgpackHandle)

  // Read in the header
  var header SnapshotHeader
  if err := dec.Decode(&header); err != nil {
    return err
  }

  // Populate the new state
  msgType := make([]byte, 1)
  for {
    // Read the message type
    _, err := r.Read(msgType)
    if err == io.EOF {
      return nil
    } else if err != nil {
      return err
    }

    // Decode
    msg := structs.MessageType(msgType[0])

    if err := handler(&header, msg, dec); err != nil {
      return err
    }
  }
}

至此已经完成了Raft的开发介绍。需要注意的是，FSM接口都是Raft内部调用的，用户并不会直接与之交互。

更多参见：Raft Developer Documentation

Raft关键对外接口

Raft节点管理

将节点添加到集群中，节点刚添加到集群中时状态是staging，当其ready之后就会被提升为voter，参与选举。如果节点已经是voter，则该操作会更新服务地址。该方法必须在leader上调用：

func (r *Raft) AddVoter(id ServerID, address ServerAddress, prevIndex uint64, timeout time.Duration) IndexFuture

如下方法用于添加一个只接收log entry、但不参与投票或commit log的节点：该方法必须在leader上调用

func (r *Raft) AddNonvoter(id ServerID, address ServerAddress, prevIndex uint64, timeout time.Duration) IndexFuture

将节点从集群中移除，如果移除的节点是leader，则会触发leader选举。该方法必须在leader上调用：

func (r *Raft) RemoveServer(id ServerID, prevIndex uint64, timeout time.Duration) IndexFuture

取消节点的投票权，节点不再参与投票或commit log。该方法必须在leader上调用：

func (r *Raft) DemoteVoter(id ServerID, prevIndex uint64, timeout time.Duration) IndexFuture

重新加载节点配置：

func (r *Raft) ReloadConfig(rc ReloadableConfig) error
Raft数据的存储和读取

用于阻塞等待FSM apply所有操作。该方法必须在leader上调用：

func (r *Raft) Barrier(timeout time.Duration) Future

apply一个命令到FSM，该方法必须在leader上调用：

func (r *Raft) Apply(cmd []byte, timeout time.Duration) ApplyFuture

从上面接口可以看到，在Raft协议中，必须通过leader才能写入(apply)数据，在非leader的节点上执行Apply()会返回ErrNotLeader的错误。

Apply方法会调用LogStore接口的StoreLogs方法存储log(cmd)。Raft.applyCh负责将log发送给FSM进行处理，最后通过dispatchLogs将log分发给其他节点(dispatchLogs会调用Transport.AppendEntries来将log分发给对端)。

在分布式环境中，外部请求可能通过LB转发到非leader节点上，此时非leader节点需要将请求转发到leader节点上进行处理，在consul中会通过ForwardRPC将请求转发给leader，再由leader执行Apply操作。

集群恢复

当集群中的节点少于仲裁数目时，集群将无法正常运作，此时可以手动调用如下接口尝试恢复集群，但这样会可能会导致原本正在复制的日志被commit。

最佳方式是停止所有节点，并在所有节点上运行RecoverCluster，当集群重启之后，会发生选举，Raft也会恢复运作。

func RecoverCluster(conf *Config, fsm FSM, logs LogStore, stable StableStore,
  snaps SnapshotStore, trans Transport, configuration Configuration) error

通过如下方式可以让集群使用外部snapshot(如备份的snapshot)。注意该操作只适用于DR，且只能在Leader上运行。

func (r *Raft) Restore(meta *SnapshotMeta, reader io.Reader, timeout time.Duration) error
状态获取

获取节点的状态信息：

func (r *Raft) Stats() map[string]string

返回当前leader的地址和集群ID。如果当前没有leader则返回空：

func (r *Raft) LeaderWithID() (ServerAddress, ServerID)
节点数据交互

各个节点之间主要通过RPC来交互log和选举信息，可以分为RPC客户端和RPC服务端。

RPC客户端通过调用Transport接口方法来传递数据(如Leader执行Raft.Apply log之后会调用Transport.AppendEntries来分发log)。

RPC服务端的实现如下，其处理了不同类型的RPC请求，如AppendEntriesRequest就是Leader执行Transport.AppendEntries传递的请求内容：

func (r *Raft) processRPC(rpc RPC) {
  if err := r.checkRPCHeader(rpc); err != nil {
    rpc.Respond(nil, err)
    return
  }

  switch cmd := rpc.Command.(type) {
  case *AppendEntriesRequest:
    r.appendEntries(rpc, cmd)
  case *RequestVoteRequest:
    r.requestVote(rpc, cmd)
  case *InstallSnapshotRequest:
    r.installSnapshot(rpc, cmd)
  case *TimeoutNowRequest:
    r.timeoutNow(rpc, cmd)
  default:
    r.logger.Error("got unexpected command",
      "command", hclog.Fmt("%#v", rpc.Command))
    rpc.Respond(nil, fmt.Errorf("unexpected command"))
  }
}

实现描述

实现Raft时需要考虑如下几点：

  • 实现FSM接口，包含FSM和FSMSnapshot这两个接口
  • 如何实现Raft节点的自动发现，包含节点的加入和退出
  • 客户端和应用的交互接口，主要用于应用数据的增删改等查等操作，对FSM的修改必须通过Raft.Apply接口实现，以保证FSM的数据一致性，而在读取应用数据时，如果要求数据强一致，则需要从leader的FSM读取，否则也可以从follower的FSM读取
  • 在非Leader节点接收到客户端的修改类请求后，如何将请求转发给Leader节点

在此次实现Raft的过程中，主要参考了stcache和consul的源代码，其中FSM的实现参考了前者，而Raft的初始化和节点发现参考了后者。

源代码结构如下：

- src
    discovery #节点发现代码
    raft      #raft管理代码
    rpc       #请求转发代码
    service   #主服务管理代码

  • discovery：采用serf来实现节点发现，它底层采用的还是memberlist，通过gossip来管理节点。

  • rpc：实现了非Leader节点向Leader节点转发请求的功能，本demo仅实现了/api/v1/set接口转发，对于/api/v1/get接口，则直接从本节点的FSM中获取数据，因此get接口不是强一致性的。

    使用如下命令可以生成rpc模块的pb.go文件：

    $ protoc --go_out=. --go_opt=paths=source_relative  --go-grpc_out=. --go-grpc_opt=paths=source_relative ./forward.proto

启动demo

下面启动3个节点来组成Raft集群：

入参描述如下：

  • httpAddress：与用户交互的服务
  • raftTCPAddress：Raft服务
  • rpcAddress：请求转发的gRpc服务
  • serfAddress：serf节点发现服务
  • dataDir：Raft存储路径，创建Raft节点时会用到
  • bootstrap：该节点是否需要使用bootstrap方式启动
  • joinAddress：加入Raft集群的地址，为serfAddress，可以添加多个，如add1,add2

  1. 第一个节点启动时并没有需要加入的集群，因此第一个节点以bootstrap方式启动，启动后成为leader。

    $ raft-example --httpAddress 0.0.0.0:5000 --raftTCPAddress 192.168.1.42:6000 --rpcAddress=0.0.0.0:7000 --serfAddress 192.168.1.42:8000 --dataDir /Users/charlie.liu/home/raftDatadir/node0 --bootstrap true

    注意：raftTCPAddress不能为0.0.0.0，否则raft会报错误："local bind address is not advertisable"，serfAddress的地址最好也不要使用0.0.0.0。

  2. 启动第2、3个节点，后续的节点启动的时候需要加入集群，启动的时候指定第一个节点的地址：

    $ raft-example --httpAddress 0.0.0.0:5001 --raftTCPAddress 192.168.1.42:6001 --rpcAddress=0.0.0.0:7001 --serfAddress 192.168.1.42:8001 --dataDir /Users/charlie.liu/home/raftDatadir/node1 --joinAddress 192.168.1.42:8000

    $ raft-example --httpAddress 0.0.0.0:5002 --raftTCPAddress 192.168.1.42:6002 --rpcAddress=0.0.0.0:7002 --serfAddress 192.168.1.42:8002 --dataDir /Users/charlie.liu/home/raftDatadir/node2 --joinAddress 192.168.1.42:8000

在节点启动之后，就可以在Leader的标准输出中可以看到Raft集群中的成员信息：

[INFO]  raft: updating configuration: command=AddVoter server-id=192.168.1.42:6002 server-addr=192.168.1.42:6002 servers="[{Suffrage:Voter ID:192.168.1.42:6000 Address:192.168.1.42:6000} {Suffrage:Voter ID:192.168.1.42:6001 Address:192.168.1.42:6001} {Suffrage:Voter ID:192.168.1.42:6002 Address:192.168.1.42:6002}]"

使用/api/maintain/stats接口可以查看各个节点的状态，num_peers展示了对端节点数目，state展示了当前节点的角色。

$ curl 0.0.0.0:5000/api/maintain/stats|jq   //node0为Leader
{
  "applied_index": "6",
  "commit_index": "6",
  "fsm_pending": "0",
  "last_contact": "0",
  "last_log_index": "6",
  "last_log_term": "2",
  "last_snapshot_index": "0",
  "last_snapshot_term": "0",
  "latest_configuration": "[{Suffrage:Voter ID:192.168.1.42:6000 Address:192.168.1.42:6000} {Suffrage:Voter ID:192.168.1.42:6001 Address:192.168.1.42:6001} {Suffrage:Voter ID:192.168.1.42:6002 Address:192.168.1.42:6002}]",
  "latest_configuration_index": "0",
  "num_peers": "2",
  "protocol_version": "3",
  "protocol_version_max": "3",
  "protocol_version_min": "0",
  "snapshot_version_max": "1",
  "snapshot_version_min": "0",
  "state": "Leader",
  "term": "2"
}

$ curl 0.0.0.0:5001/api/maintain/stats|jq   //node2为Follower
{
  "applied_index": "6",
  "commit_index": "6",
  "fsm_pending": "0",
  "last_contact": "15.996792ms",
  "last_log_index": "6",
  "last_log_term": "2",
  "last_snapshot_index": "0",
  "last_snapshot_term": "0",
  "latest_configuration": "[{Suffrage:Voter ID:192.168.1.42:6000 Address:192.168.1.42:6000} {Suffrage:Voter ID:192.168.1.42:6001 Address:192.168.1.42:6001} {Suffrage:Voter ID:192.168.1.42:6002 Address:192.168.1.42:6002}]",
  "latest_configuration_index": "0",
  "num_peers": "2",
  "protocol_version": "3",
  "protocol_version_max": "3",
  "protocol_version_min": "0",
  "snapshot_version_max": "1",
  "snapshot_version_min": "0",
  "state": "Follower",
  "term": "2"
}

$ curl 0.0.0.0:5002/api/maintain/stats|jq   //node2为Follower
{
  "applied_index": "6",
  "commit_index": "6",
  "fsm_pending": "0",
  "last_contact": "76.764584ms",
  "last_log_index": "6",
  "last_log_term": "2",
  "last_snapshot_index": "0",
  "last_snapshot_term": "0",
  "latest_configuration": "[{Suffrage:Voter ID:192.168.1.42:6000 Address:192.168.1.42:6000} {Suffrage:Voter ID:192.168.1.42:6001 Address:192.168.1.42:6001} {Suffrage:Voter ID:192.168.1.42:6002 Address:192.168.1.42:6002}]",
  "latest_configuration_index": "0",
  "num_peers": "2",
  "protocol_version": "3",
  "protocol_version_max": "3",
  "protocol_version_min": "0",
  "snapshot_version_max": "1",
  "snapshot_version_min": "0",
  "state": "Follower",
  "term": "2"
}

Leader切换

停掉上述Demo中的Leader节点(node0)，可以看到node1称为新的leader，且term变为4：

$ curl 0.0.0.0:5001/api/maintain/stats|jq  //新的Leader
{
  "applied_index": "15",
  "commit_index": "15",
  "fsm_pending": "0",
  "last_contact": "0",
  "last_log_index": "15",
  "last_log_term": "4",
  "last_snapshot_index": "0",
  "last_snapshot_term": "0",
  "latest_configuration": "[{Suffrage:Voter ID:192.168.1.42:6000 Address:192.168.1.42:6000} {Suffrage:Voter ID:192.168.1.42:6001 Address:192.168.1.42:6001} {Suffrage:Voter ID:192.168.1.42:6002 Address:192.168.1.42:6002}]",
  "latest_configuration_index": "0",
  "num_peers": "2",
  "protocol_version": "3",
  "protocol_version_max": "3",
  "protocol_version_min": "0",
  "snapshot_version_max": "1",
  "snapshot_version_min": "0",
  "state": "Leader",
  "term": "4"
}

$ curl 0.0.0.0:5002/api/maintain/stats|jq
{
  "applied_index": "15",
  "commit_index": "15",
  "fsm_pending": "0",
  "last_contact": "42.735ms",
  "last_log_index": "15",
  "last_log_term": "4",
  "last_snapshot_index": "0",
  "last_snapshot_term": "0",
  "latest_configuration": "[{Suffrage:Voter ID:192.168.1.42:6000 Address:192.168.1.42:6000} {Suffrage:Voter ID:192.168.1.42:6001 Address:192.168.1.42:6001} {Suffrage:Voter ID:192.168.1.42:6002 Address:192.168.1.42:6002}]",
  "latest_configuration_index": "0",
  "num_peers": "2",
  "protocol_version": "3",
  "protocol_version_max": "3",
  "protocol_version_min": "0",
  "snapshot_version_max": "1",
  "snapshot_version_min": "0",
  "state": "Follower",
  "term": "4"
}

在本实现中，如果停止一个Raft节点，则Leader节点会一直打印连接该节点失败的日志，原因是在Ctrl+c停止Raft节点的时候没有调用Raft.RemoveServer来移除该节点。这种处理方式是合理的，因为当一个节点重启或故障的时候，不应该从Raft中移除，此时应该查明原因，恢复集群。

本实现中没有主动移除Raft节点的接口，也可以添加一个接口来调用Raft.RemoveServer，进而移除预期的节点，注意只能在Leader节点上执行Raft.RemoveServer。

应用数据的读写

下面我们验证应用数据的写入和读取。

  1. 向非Leader节点写入数据，其会将写入请求转发给leader，由leader执行数据写入。下面展示向非Leader节写入数据的场景：

    $ curl 0.0.0.0:5001/api/maintain/stats|jq
{
  "applied_index": "64",
  "commit_index": "64",
  "fsm_pending": "0",
  "last_contact": "4.312667ms",
  "last_log_index": "64",
  "last_log_term": "137",
  "last_snapshot_index": "0",
  "last_snapshot_term": "0",
  "latest_configuration": "[{Suffrage:Voter ID:0.0.0.0:7000 Address:192.168.1.42:6000} {Suffrage:Voter ID:0.0.0.0:7001 Address:192.168.1.42:6001} {Suffrage:Voter ID:0.0.0.0:7002 Address:192.168.1.42:6002}]",
  "latest_configuration_index": "0",
  "num_peers": "2",
  "protocol_version": "3",
  "protocol_version_max": "3",
  "protocol_version_min": "0",
  "snapshot_version_max": "1",
  "snapshot_version_min": "0",
  "state": "Follower",  #非Leader节点
  "term": "137"
}

$ curl -XPOST localhost:5001/api/v1/set --header 'Content-Type: application/json' --header 'Content-Type: application/json' -d '
{
    "key" : "testKey",
    "value" : "testValue"
}'

  2. 向所有节点查询写入的数据，可以看到所有节点都可以查询到该数据：

    $ curl -XGET localhost:5000/api/v1/get --header 'Content-Type: application/json' --header 'Content-Type: application/json' -d '
{
    "key" : "testKey"
}'
testValue

$ curl -XGET localhost:5001/api/v1/get --header 'Content-Type: application/json' --header 'Content-Type: application/json' -d '
{
    "key" : "testKey"
}'
testValue

$curl -XGET localhost:5002/api/v1/get --header 'Content-Type: application/json' --header 'Content-Type: application/json' -d '
{
    "key" : "testKey"
}'
testValue

TIPS

  • 验证场景下，如果节点IP发生变动，可以通过删除--dataDir目录来清除集群元数据
  • 如果集群中的节点不足仲裁数目，则节点可能处理candidate状态，无法变为Leader，因此要保证集群中有足够的节点，避免一次停掉过多节点。

参考

标签：log,hashicorp,FSM,192.168,分布式服务,snapshot,Raft,节点
From： https://www.cnblogs.com/charlieroro/p/17486646.html

相关文章

  • 【翻译】6.824 lab2 Raft （自用不负责）
    介绍这是系列实验中的第一个，您将在这些实验中构建容错键/值存储系统。在本实验中，您将实现Raft，一个复制状态机协议。在下一个实验中，您将在Raft之上构建键/值服务。然后，您将在多个复制的状态机上“共享”您的服务，以获得更高的性能。复制的服务通过将其状态(即数据)的完整副本存储......
  • Pasos和RAFT算法
    Paxos提出时间1990年，RAFT提出时间2013年。RAFT是Paxos的简化版，或者说是提高投票效率，但是降低了投票公平性的妥协方案。RAFT分布式raft(ReplicatedAndFaultTolerant)选举算法原理分成三个角色，领导者，跟随者，和候选者。在没有领导者的时候和领导者联系不上的时候。跟随者......
  • Raft一致性算法
    分布式的高可用方案都会考虑容灾，那么redis高可用是如何做到故障自动切换的？1增加主客观下线判定。对于主客观下线判定，当某个哨兵节点与主节点连接超时，则将其标志位主观下线，然后开始将主节点下线这个信息与其他哨兵节点同步，其他哨兵节点根据自身与主节点的通信情况，做出赞成或者......
  • Nacos集群Raft反序列化漏洞-修复
    近日，奇安信CERT监测到 Nacos集群Raft反序列化漏洞(QVD-2023-13065)，在Nacos集群处理部分Jraft请求时，攻击者可以无限制使用hessian进行反序列化利用，最终实现代码执行。鉴于该漏洞仅影响集群间通信端口7848（默认配置下），若部署时已进行限制或未暴露则风险可控，建议客户做好自查及防护......
  • 怎样打造一个分布式数据库——rocksDB, raft, mvcc，本质上是为了解决跨数据中心的复制
    为什么我们要创建另外一个数据库？在前面三十年基本上是关系数据库的时代，那个时代创建了很多伟大的公司，比如说IBM、Oracle、微软也有自己的数据库，早期还有一个公司叫Sybase，有一部分特别老的程序员同学在当年的教程里面还可以找到这些东西，但是现在基本上看不到了。另外是NoSQL。NoS......
  • nacos服务下线操作时报错：The Raft Group [naming_instance_metadata] did not find th
    【问题描述】caused:errCode:500,errMsg:dometadataoperationfailed;caused:com.alibaba.nacos.consistency.exception.ConsistencyException:TheRaftGroup[naming_instance_metadata]didnotfindtheLeadernode;caused:TheRaftGroup[naming_instance_metad......
  • raft笔记
    目的：一致性算法，允许一组机器作为一个一致的组来工作，这些组可以承受某些成员的故障，提高可用性领导选举，日志同步，快照，集群变动复制状态机用于解决分布式系统中的各种容错问题，会出现共识算法共识和复制状态机通过保持复制日志的一致性raft是一种日志复制算法Raft通过首先选举一个......
  • java-raft框架之atomix进行分布式管理
    共识算法在一个分布式的系统中，管理各个节点的一致性（共识）一直是个很有难度的问题。在近几十年的发展中，于1990年诞生的Paxos算法是其中最为经典的代表，并一统江湖数几十载。如著名的zookeeper、chubby都是基于Paxos算法的经典应用。不过Paxos算法的复杂度和难以理解性，也吸引了一些大佬......
  • J. Joy of Handcraft
    J.JoyofHandcraft题意：给定n个灯泡的时间周期以及对应的亮度值，求1~m的时刻，每一时刻的灯泡最大亮度分析：按时间轴建树，维护时间区间的亮度最大值按亮度值递减排序，遍历灯泡时只modify为相同周期中亮度值最大的一个灯泡作为区间亮度最大值区间修改，单点查询实现：#incl......
  • 关于Minecraft bukkit开发的随笔 （玩家/实体带颜色发光机制简谈）
    Bukkit（原版同理）的玩家/实体带颜色发光机制简谈众所周知，在Bukkit的代表实体的org.bukkit.entity.Entity对象中有setGlowing(booleanflag)这么一个方法，根据Javadoc中的描述：点击查看代码Entityentity=...Scoreboardsb=Bukkit.getScoreboardManager().getMainSco......

赞助商

阅读排行

网站主页关于我们联系我们网站地图

本网站内容转载自其他媒体，侵权联系[admin##ips99.com]。

Copyright © 2020-2023 IPS99 版权所有 IPS99