etcd源码分析 - 3.【打通核心流程】PUT键值对的执行链路

简介: 在上一讲,我们一起看了etcd server是怎么匹配到对应的处理函数的,如果忘记了请回顾一下。今天,我们再进一步,看看`PUT`操作接下来是怎么执行的。

在上一讲,我们一起看了etcd server是怎么匹配到对应的处理函数的,如果忘记了请回顾一下。

今天,我们再进一步,看看PUT操作接下来是怎么执行的。

HTTP1部分

request_KV_Put_0

整个函数主要分为两步:

  1. 解析请求到etcdserverpb.PutRequest数据结构;
  2. 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将两者串联起来。

查找该函数的调用,分为三个,各有用途:

  1. grpc.go - server的调用处
  2. grpc_proxy.go - proxy代理模式,忽略
  3. 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函数实现,复杂的会结合syncatomic进行针对性的设计。

我们再进一步跳转。

(*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)

我在图中特别标注了一些关键的接口与实现。

目录
相关文章
|
6月前
|
监控 前端开发 Java
分布式链路监控系统问题之执行原方法时不能调用method.invoke的问题如何解决
分布式链路监控系统问题之执行原方法时不能调用method.invoke的问题如何解决
etcd源码分析 - 2.【打通核心流程】PUT键值对匹配处理函数
在阅读了etcd server的启动流程后,我们对很多关键性函数的入口都有了初步印象。 那么,接下来我们一起看看对键值对的修改,在etcd server内部是怎么流转的。
93 0
etcd源码分析 - 2.【打通核心流程】PUT键值对匹配处理函数
|
IDE Go 开发工具
etcd源码分析 - 5.【打通核心流程】EtcdServer消息的处理函数
在上一讲,我们梳理了`EtcdServer`的关键函数`processInternalRaftRequestOnce`里的四个细节。 其中,`wait.Wait`组件使用里,我们还遗留了一个细节实现,也就是请求的处理结果是怎么通过channel返回的。
127 0
etcd源码分析 - 5.【打通核心流程】EtcdServer消息的处理函数
|
索引
etcd源码分析 - 4.【打通核心流程】processInternalRaftRequestOnce四个细节
在上一讲,我们继续梳理了`PUT`请求到`EtcdServer`这一层的逻辑,并大概阅读了其中的关键函数`processInternalRaftRequestOnce`。
98 0
|
6月前
|
设计模式 NoSQL Redis
Lettuce的特性和内部实现问题之CommandEncoder在通道流水线中的问题如何解决
Lettuce的特性和内部实现问题之CommandEncoder在通道流水线中的问题如何解决
|
4月前
|
存储 安全 Java
ArrayList 核心源码+扩容机制分析
ArrayList的底层是数组队列,相当于动态数组。与 Java 中的数组相比,它的容量能动态增长。在添加大量元素前,应用程序可以使用操作来增加ArrayList实例的容量。这可以减少递增式再分配的数量。ArrayList继承于,实现了ListCloneable这些接口。List: 表明它是一个列表,支持添加、删除、查找等操作,并且可以通过下标进行访问。:这是一个标志接口,表明实现这个接口的List集合是支持快速随机访问的。在ArrayList。
|
6月前
|
NoSQL Redis
Lettuce的特性和内部实现问题之在同步调用模式下,业务线程是如何拿到结果数据的
Lettuce的特性和内部实现问题之在同步调用模式下,业务线程是如何拿到结果数据的
|
7月前
|
数据格式
交易链路设计原则&模式问题之在进行抽象和替换时可能遇到的问题,如何解决
交易链路设计原则&模式问题之在进行抽象和替换时可能遇到的问题,如何解决
|
9月前
|
SQL 设计模式 NoSQL
JedisCluster 通过 Pipeline 实现两套数据轮换更新
本文实现了通过定时任务来调用接口,使两套数据轮换更新。因为要区分两套数据,所以 key 要设置前缀。例如:一天数据一换,今天查的 A 开头的 key ,明天查 B 开头的 key ,后天又查 A 开头的 key。今天查完后,明天更新 B 开头的 key ,但是 A 开头的 key 暂时不动,后天再查的时候,A开头的 key 要进行更新,先删再更新。
186 5
|
Kubernetes 调度 容器
二进制 k8s 集群下线 worker 组件流程分析和实践
二进制 k8s 集群下线 worker 组件流程分析和实践
115 0