在上一讲,我们一起看了etcd server是怎么匹配到对应的处理函数的,如果忘记了请回顾一下。
今天,我们再进一步,看看PUT
操作接下来是怎么执行的。
HTTP1部分
request_KV_Put_0
整个函数主要分为两步:
- 解析请求到
etcdserverpb.PutRequest
数据结构; client
执行PUT
操作;
关于解析部分,我们暂时不用关心如何反序列化的(反序列化是一种可替换的插件,常见的如json/protobuffer/xml),重点看看它的数据结构:
type PutRequest struct {
Key []byte `protobuf:"bytes,1,opt,name=key,proto3" json:"key,omitempty"`
Value []byte `protobuf:"bytes,2,opt,name=value,proto3" json:"value,omitempty"`
Lease int64 `protobuf:"varint,3,opt,name=lease,proto3" json:"lease,omitempty"`
PrevKv bool `protobuf:"varint,4,opt,name=prev_kv,json=prevKv,proto3" json:"prev_kv,omitempty"`
IgnoreValue bool `protobuf:"varint,5,opt,name=ignore_value,json=ignoreValue,proto3" json:"ignore_value,omitempty"`
IgnoreLease bool `protobuf:"varint,6,opt,name=ignore_lease,json=ignoreLease,proto3" json:"ignore_lease,omitempty"`
}
从我们执行的etcdctl put mykey "this is awesome"
为例,不难猜到:
- Key - mykey
- Value - this is awesome
接下来,我们去看看client是如何执行PUT
的。
etcdserverpb.kVClient
request_KV_Put_0
函数中的client是一个接口KVClient
,包括Range/Put/DeleteRange/Txn/Compact五种操作。
这里提一下,很多开源库将接口与其实现,用大小写来区分,来强制要求外部模块依赖其接口:
比如KVClient作为接口,而kVClient作为其实现是小写的,所以外部模块无法直接使用kVClient这个数据结构。
它的实现可以很容易地翻阅代码找到,是etcdserverpb.kVClient
。我们去看看对应的PUT
方法。
func (c *kVClient) Put(ctx context.Context, in *PutRequest, opts ...grpc.CallOption) (*PutResponse, error) {
out := new(PutResponse)
err := grpc.Invoke(ctx, "/etcdserverpb.KV/Put", in, out, c.cc, opts...)
if err != nil {
return nil, err
}
return out, nil
}
这里,我们就找到了HTTP调用gRPC的影子,也就是这个Invoke
方法。
gRPC部分
proto文件
关于gRPC的调用部分,我比较推荐从最原始的proto
文件开始阅读,主要包括2个文件:
etcd/etcdserver/etcdserverpb/rpc.proto
原始文件etcd/etcdserver/etcdserverpb/rpc.pb.go
生成文件
从下面的定义可以看到HTTP1采用了POST
方法,对应URL为/v3/kv/put
:
rpc Put(PutRequest) returns (PutResponse) {
option (google.api.http) = {
post: "/v3/kv/put"
body: "*"
};
}
etcdserverpb.RegisterKVServer
我们要注意,proto文件及其生成的go代码只是定义了server的接口,具体的实现需要开发者自行编码实现,通过注册函数RegisterKVServer
将两者串联起来。
查找该函数的调用,分为三个,各有用途:
grpc.go
- server的调用处grpc_proxy.go
- proxy代理模式,忽略mockserver.go
- mock服务,忽略
跳转到1对应的代码处,我们看到了注册函数pb.RegisterKVServer(grpcServer, NewQuotaKVServer(s))
。
NewQuotaKVServer
进一步跳转,来到了NewKVServer
函数中。
NewKVServer
这个函数新建了一个kvServer
对象,它实现接口etcdserverpb.KVServer
。我们再看对应的PUT
方法。
(*kvServer) Put
Put
方法代码很少:
func (s *kvServer) Put(ctx context.Context, r *pb.PutRequest) (*pb.PutResponse, error) {
if err := checkPutRequest(r); err != nil {
return nil, err
}
resp, err := s.kv.Put(ctx, r)
if err != nil {
return nil, togRPCError(err)
}
s.hdr.fill(resp.Header)
return resp, nil
}
而这里的s.kv
,其定义为接口etcdserver.RaftKV
,定义了如下五个方法:
type RaftKV interface {
// 范围操作
Range(ctx context.Context, r *pb.RangeRequest) (*pb.RangeResponse, error)
// KV操作
Put(ctx context.Context, r *pb.PutRequest) (*pb.PutResponse, error)
// 删除范围
DeleteRange(ctx context.Context, r *pb.DeleteRangeRequest) (*pb.DeleteRangeResponse, error)
// 事务
Txn(ctx context.Context, r *pb.TxnRequest) (*pb.TxnResponse, error)
// 压缩
Compact(ctx context.Context, r *pb.CompactionRequest) (*pb.CompactionResponse, error)
}
etcd server集群之间采用的是RAFT协议,而RaftKV
则是实现的关键。查找RaftKV的具体实现EtcdServer
,我们就找到了如下代码:
(*EtcdServer) Put
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
}
值得注意的是,这里将多种请求命令(如PUT/RANGE),都封装到了一个结构体InternalRaftRequest
中。
我们继续跳转。
(*EtcdServer) raftRequest
func (s *EtcdServer) raftRequest(ctx context.Context, r pb.InternalRaftRequest) (proto.Message, error) {
return s.raftRequestOnce(ctx, r)
}
一般来说,带Once
关键字的函数,强调只执行一次,简单的可以用sync.Once
函数实现,复杂的会结合sync
和atomic
进行针对性的设计。
我们再进一步跳转。
(*EtcdServer) processInternalRaftRequestOnce
这部分的代码我做了个精简,如下:
// 发起RAFT提案Propose(分布式共识算法的术语,不清楚的同学有个初步印象即可)
err = s.r.Propose(cctx, data)
// 监控的metrics,表示提案处于Pending计数+1,退出则-1
proposalsPending.Inc()
defer proposalsPending.Dec()
// 处理结果异步返回,分为三个情况
select {
// 正常返回结果
case x := <-ch:
return x.(*applyResult), nil
// 超时等异常处理
case <-cctx.Done():
proposalsFailed.Inc()
s.w.Trigger(id, nil) // GC wait
return nil, s.parseProposeCtxErr(cctx.Err(), start)
// 被正常关闭
case <-s.done:
return nil, ErrStopped
}
raftNode部分
(raftNode)Propose
如果我们对Propose
方法感兴趣,就需要深入学习raftNode
这一大块了,它是对RAFT协议的整体封装。
在etcd
里,raftNode
是一个比较独立的模块,我们会在后续模块专门分析。
小结
通过本篇的代码阅读,我们经历了 HTTP1 -> gRPC -> raftNode 三层,对整个PUT
调用链有了一个基本印象。
.png)
我在图中特别标注了一些关键的接口与实现。