etcd 的Put请求处理
在介绍etcdctl的内容中，我们知道了etcdctl实际上是向etcd服务端执行了grpc请求获取对应的结果，这一篇主要介绍当向etcd服务端执行Get/Put指令的时候究竟做了哪些工作。
Client发出请求
Put指令和之前介绍到的Get指令类似，通过grpc client发起请求并带上键值对的参数。在这里实际进行操作的是client类型中的未命名属性KV接口类型完成的
put command 代码：https://github.com/etcd-io/etcd/blob/v3.5.15/etcdctl/ctlv3/command/put_command.go
kv 代码：https://github.com/etcd-io/etcd/blob/v3.5.15/client/v3/kv.go
type Client struct {ClusterKVLeaseWatcherAuthMaintenance...
}type kv struct {remote pb.KVClientcallOpts []grpc.CallOption
}
kv 类型会将不同的操作包装成Op类型，由Do函数统一操作，这种命令模式有助于实现操作的解耦，对可能的新增命令有更好的扩展性。 在Do函数中通过对Op对象类型的判断由remote对象执行具体的请求。我们也可以从代码中看到目前etcd一共支持四种操作类型，get并不是对应了Get类型而是Range类型。
const (// A default Op has opType 0, which is invalid.tRange opType = iota + 1tPuttDeleteRangetTxn
)
raft请求处理
通过pb.KVClient类型我们可以找到grpc定义的对应的Server类型。顺藤摸瓜在server文件夹下找到了对应的接口服务代码。
https://github.com/etcd-io/etcd/blob/v3.5.15/server/etcdserver/v3_server.go
func (s *EtcdServer) Put(ctx context.Context, r *pb.PutRequest) (*pb.PutResponse, error) {ctx = context.WithValue(ctx, traceutil.StartTimeKey{}, time.Now())resp, err := s.raftRequest(ctx, pb.InternalRaftRequest{Put: r})if err != nil {return nil, err}return resp.(*pb.PutResponse), nil
}func (s *EtcdServer) raftRequest(ctx context.Context, r pb.InternalRaftRequest) (proto.Message, error) {return s.raftRequestOnce(ctx, r)
}func (s *EtcdServer) raftRequestOnce(ctx context.Context, r pb.InternalRaftRequest) (proto.Message, error) {result, err := s.processInternalRaftRequestOnce(ctx, r)if err != nil {return nil, err}if result.Err != nil {return nil, result.Err}if startTime, ok := ctx.Value(traceutil.StartTimeKey{}).(time.Time); ok && result.Trace != nil {applyStart := result.Trace.GetStartTime()// The trace object is created in toApply. Here reset the start time to trace// the raft request time by the difference between the request start time// and toApply start timeresult.Trace.SetStartTime(startTime)result.Trace.InsertStep(0, applyStart, "process raft request")result.Trace.LogIfLong(traceThreshold)}return result.Resp, nil
}
因此对请求的处理是：
- 将操作封装成raftRequest
- 在
processInternalRaftRequestOnce方法里，将raftRequest提交给raft node进行处理。这时会检查一下apply index和commint index的gap，从代码上看就是先apply， 然后再commit。 后面会对权限做一些校验，通过后会将raftRequest序列化由raft node处理。这部分代码在https://github.com/etcd-io/raft 里实现。
func (n *node) Propose(ctx context.Context, data []byte) error {return n.stepWait(ctx, pb.Message{Type: pb.MsgProp, Entries: []pb.Entry{{Data: data}}})
}
stepWithWaitOption方法中将消息扔给node的channel处理并等待结果, 可以看看node的这个channelpropc的接收者是如何处理的。
select {// TODO: maybe buffer the config propose if there exists one (the way// described in raft dissertation)// Currently it is dropped in Step silently.case pm := <-propc:m := pm.mm.From = r.iderr := r.Step(m)if pm.result != nil {pm.result <- errclose(pm.result)}}
在Step方法中对不同类型的message进行了处理， 这里应该会有一些raft共识相关的信息类型， 但是对于前面生成的MsgProp消息，调用了raft.step方法，raft.step 是函数类型type stepFunc func(r *raft, m pb.Message) error, 在node成为leader, follower, 或是candidate的时候确定的。
func (r *raft) Step(m pb.Message) error {// Handle the message term, which may result in our stepping down to a follower....switch m.Type {case pb.MsgHup:...case pb.MsgStorageAppendResp:...case pb.MsgStorageApplyResp:...case pb.MsgVote, pb.MsgPreVote:...default:err := r.step(r, m)if err != nil {return err}}return nil
}
如果node是leader的话，raft.step 对应stepLeader，会根据entries记录raftlog日志，将msg添加到node的msgs切片中，并通知其他节点当前的提交位置。而如果是candidate则会记录没有leader的错误日志并返回error。
// 代码中对msgs的描述// msgs contains the list of messages that should be sent out immediately to
// other nodes.
//
// Messages in this list must target other nodes.
msgs []pb.Message
- 最终会新创建
Ready对象， 而相关的entries会被写入到存储中。 这里存储在内存中和数据库文件中是如何刷入的，后面有机会在看。
if err := r.storage.Save(rd.HardState, rd.Entries); err != nil {r.lg.Fatal("failed to save Raft hard state and entries", zap.Error(err))
}