乐趣区

关于etcd:etcd的租约是怎么实现的

作者：

jiezi

在

租约是什么

咱们都晓得Redis能够通过expire命令对key设置过期工夫，来实现缓存的ttl，etcd同样有一种个性能够对key设置过期工夫，也就是租约（Lease）。不过相较来说，两者的实用场景并不相同，etcd的Lease宽泛的用在服务注册与保活上，redis则次要用于淘汰缓存。上面介绍一下etcd的Lease机制，会从应用形式，以及实现原理来逐渐探索。

应用形式

首先通过一个案例简略介绍它的应用形式。

package main

import (
    "context"
    "log"
    "os"
    "os/signal"
    "syscall"
    "time"
​
    clientv3 "go.etcd.io/etcd/client/v3"
)
​
func main() {
​
    key := "linugo-lease"
​
    cli, err := clientv3.New(clientv3.Config{
        Endpoints:   []string{"127.0.0.1:23790"},
        DialTimeout: time.Second,
    })
    if err != nil {
        log.Fatal("new client err: ", err)
    }
​
    //首先创立一个Lease并通过Grant办法申请一个租约，设置ttl为20秒，没有续约的话，该租约会在20s后隐没
    ls := clientv3.NewLease(cli)
    grantResp, err := ls.Grant(context.TODO(), 20)
    if err != nil {
        log.Fatal("grant err: ", err)
    }
    log.Printf("grant id: %x\n", grantResp.ID)
​
    //接下来插入一个键值对绑定该租约，该键值对会随着租约的到期而相应被删除
    putResp, err := cli.Put(context.TODO(), key, "value", clientv3.WithLease(grantResp.ID))
    if err != nil {
        log.Fatal("put err: ", err)
    }
​
    log.Printf("create version: %v\n", putResp.Header.Revision)
    //通过KeepAliveOnce办法对该租约进行续期，每隔5s会将该租约续期到初始的20s
    go func() {
        for {
            time.Sleep(time.Second * 5)
            resp, err := ls.KeepAliveOnce(context.TODO(), grantResp.ID)
            if err != nil {
                log.Println("keep alive once err: ", err)
                break
            }
            log.Println("keep alive: ", resp.TTL)
        }
    }()
​
    sigC := make(chan os.Signal, 1)
    signal.Notify(sigC, os.Interrupt, syscall.SIGTERM)
    s := <-sigC
    log.Println("exit with: ", s.String())
}

咱们能够通过上述形式实现某个服务模块的保活，能够将节点的地址注册到etcd中，并绑定适当时长的租约，定时进行续约操作，若节点宕机，超过了租约时长，etcd中该节点的信息就会被移除掉，实现服务的主动摘除，通常配合etcd的watch个性来做到实时的感知。
v3版的客户端接口除了上述的Grant，KeepAliveOnce办法，还包含了一些其余重要的办法如Revoke删除某个租约，TimeToLive查看某个租约残余时长等。
etcd服务端面向租约对客户端服务的有5个接口，别离对client端的办法给予了实现。本次次要对服务端的实现办法进行剖析。

type LeaseServer interface {
    //对应客户端的Grant办法，创立租约
    LeaseGrant(context.Context, *LeaseGrantRequest) (*LeaseGrantResponse, error)
    //删除某个租约
    LeaseRevoke(context.Context, *LeaseRevokeRequest) (*LeaseRevokeResponse, error)
    //租约某个续期
    LeaseKeepAlive(Lease_LeaseKeepAliveServer) error
    //租约残余时长查问
    LeaseTimeToLive(context.Context, *LeaseTimeToLiveRequest) (*LeaseTimeToLiveResponse, error)
    //查看所有租约
    LeaseLeases(context.Context, *LeaseLeasesRequest) (*LeaseLeasesResponse, error)
}

初始化

在etcd启动时候，会初始化一个lessor，lessor外部存储了所有无关租约的信息，包含租约ID，到期工夫，租约绑定的键值对等；lessor实现了一系列接口，是租约性能的具体实现逻辑，包含Grant(创立)，Revoke(撤销)，Renew(续租)等。

type lessor struct {
    mu sync.RWMutex
    demotec chan struct{}
    //寄存所有无效的lease信息，key为leaseID，value包含该租约的ID，ttl，lease绑定的key等信息
    leaseMap             map[LeaseID]*Lease
    //便于查找lease的一个数据结构，基于最小堆实现，能够将快到期的租约放到队头，查看是否过期时候，只须要查看队头即可
    leaseExpiredNotifier *LeaseExpiredNotifier
    //用于实时更新lease的剩余时间
    leaseCheckpointHeap  LeaseQueue
    //用户寄存的key与lease的绑定关系，通过key能够找到租约
    itemMap              map[LeaseItem]LeaseID
    ......
    //过期的lease会被放到该chan中，被消费者清理
    expiredC chan []*Lease
    ......
}

在lessor被初始化后，同时会启动一个goroutine，用于频繁的查看是否有过期的lease以及更新lease剩余时间。lease的这些查看是集群的leader节点做的，包含更新残余的工夫，保护lease的最小堆，到期时候撤销lease。而follower节点只用于响应leader节点的存储、更新或撤销lease申请。

func (le *lessor) runLoop() {
    defer close(le.doneC)
    for {
        //查看是否有过期的lease
        le.revokeExpiredLeases()
        //checkpoint机制查看并更新lease的剩余时间
        le.checkpointScheduledLeases()
        //每500毫秒查看一次
        select {
        case <-time.After(500 * time.Millisecond):
        case <-le.stopC:
            return
        }
    }
}

为了涵盖大部分场景，咱们假如一个三节点的etcd集群的场景，通过下面的案例代码对其中的一个follower节点发动申请。

创立

当v3客户端调用Grant办法时候，会对应到server端LeaseServer的LeaseGrant办法，该办法会通过一系列的两头步骤（鉴权等）达到etcdServer包装实现的LeaseGrant办法，该办法会调用raft模块并封装一个Lease的提案并进行数据同步流程。因为此时节点是follower，会将申请转交给leader进行解决，leader接到申请后会将该提案封装成一个日志，并播送到follower节点，follower节点执行提案音讯，并回复给leader节点。

在follower节点执行提案内容时候，会解析出该申请是一个创立lease的申请，该流程是在apply模块执行的。apply模块会调用本人包装好的LeaseGrant办法。

func (a *applierV3backend) Apply(r *pb.InternalRaftRequest, shouldApplyV3 membership.ShouldApplyV3) *applyResult {
    op := "unknown"
    ar := &applyResult{}
    ......
​
    // call into a.s.applyV3.F instead of a.F so upper appliers can check individual calls
    switch {
    ......
    case r.LeaseGrant != nil:
        op = "LeaseGrant"
        ar.resp, ar.err = a.s.applyV3.LeaseGrant(r.LeaseGrant)
    ......
    default:
        a.s.lg.Panic("not implemented apply", zap.Stringer("raft-request", r))
    }
    return ar
}

LeaseGrant办法是对lessor实现的Grant办法的进一步封装。

func (a *applierV3backend) LeaseGrant(lc *pb.LeaseGrantRequest) (*pb.LeaseGrantResponse, error) {
    l, err := a.s.lessor.Grant(lease.LeaseID(lc.ID), lc.TTL)
    resp := &pb.LeaseGrantResponse{}
    if err == nil {
        resp.ID = int64(l.ID)
        resp.TTL = l.TTL()
        resp.Header = newHeader(a.s)
    }
    return resp, err
}
lessor通过Grant办法将lease封装并存入到本人的leaseMap，并经lease长久化到boltdb。
func (le *lessor) Grant(id LeaseID, ttl int64) (*Lease, error) {
    ......
    //封装lease
    l := &Lease{
        ID:      id,
        ttl:     ttl,
         //用于寄存该lease绑定的key，用于在lease过期时删除key
        itemSet: make(map[LeaseItem]struct{}),
        revokec: make(chan struct{}),
    }
    ......
    //如果是leader节点，则刷新lease的到期工夫
    if le.isPrimary() {
        l.refresh(0)
    } else {
        //follower节点中没有存储lease的到期工夫
        l.forever()
    }
​
    le.leaseMap[id] = l
    //lease信息长久化
    l.persistTo(le.b)
    //如果是leader节点，就将lease信息放到最小堆中
    if le.isPrimary() {
        item := &LeaseWithTime{id: l.ID, time: l.expiry}
        le.leaseExpiredNotifier.RegisterOrUpdate(item)
        le.scheduleCheckpointIfNeeded(l)
    }
​
    return l, nil
}

绑定

lease创立好之后，就能够通过Put指令创立一个数据并与lease进行绑定。在Put时候，put的value字段中会有一个leaseID，并存到了boltDB。这样能够在etcd挂掉之后，能够依据长久化存储来复原lease与数据的对应关系。

func (tw *storeTxnWrite) put(key, value []byte, leaseID lease.LeaseID) {
    ......
    kv := mvccpb.KeyValue{
        Key:            key,
        Value:          value,
        CreateRevision: c,
        ModRevision:    rev,
        Version:        ver,
        //lease字段
        Lease:          int64(leaseID),
    }
    //....长久化等操作
    //attach操作
    if leaseID != lease.NoLease {
        if tw.s.le == nil {
            panic("no lessor to attach lease")
        }
        err = tw.s.le.Attach(leaseID, []lease.LeaseItem{{Key: string(key)}})
        if err != nil {
            panic("unexpected error from lease Attach")
        }
    }
    tw.trace.Step("attach lease to kv pair")
}

lessor的Attach操作会将lease与key两者进行绑定并存到本身的itemMap以及lease的itemSet中。

func (le *lessor) Attach(id LeaseID, items []LeaseItem) error {
    ......
    l := le.leaseMap[id]
    l.mu.Lock()
    for _, it := range items {
        //存到lease的itemSet
        l.itemSet[it] = struct{}{}
        //存到lessor的itemMap中
        le.itemMap[it] = id
    }
    l.mu.Unlock()
    return nil
}

保活

客户端提供的keepAlive办法用于lease进行续租，每次调用都会使得lease的剩余时间回到初始化时候设定的剩余时间。因为lease的一些查看以及保护都是由leader节点维持，所以当咱们发送申请到follower时，会间接将申请重定向到leader节点。

func (s *EtcdServer) LeaseRenew(ctx context.Context, id lease.LeaseID) (int64, error) {
    //发送到follower会返回ErrNotPrimary的谬误
    ttl, err := s.lessor.Renew(id)
    if err == nil { // already requested to primary lessor(leader)
        return ttl, nil
    }
    ......
    for cctx.Err() == nil && err != nil {
        //获取leader节点
        leader, lerr := s.waitLeader(cctx)
        if lerr != nil {
            return -1, lerr
        }
        for _, url := range leader.PeerURLs {
            lurl := url + leasehttp.LeasePrefix
             //通过http接口申请到leader的keeplaive接口
            ttl, err = leasehttp.RenewHTTP(cctx, id, lurl, s.peerRt)
            if err == nil || err == lease.ErrLeaseNotFound {
                return ttl, err
            }
        }
        time.Sleep(50 * time.Millisecond)
    }
    ......
    return -1, ErrCanceled
}

达到Leader节点之后会通过Renew更新该lease的剩余时间，过期工夫以及最小堆中的lease。

func (le *lessor) Renew(id LeaseID) (int64, error) {
    le.mu.RLock()
    if !le.isPrimary() {
        le.mu.RUnlock()
        return -1, ErrNotPrimary
    }
​
    demotec := le.demotec
​
    l := le.leaseMap[id]
    if l == nil {
        le.mu.RUnlock()
        return -1, ErrLeaseNotFound
    }
    //当cp（checkpoint办法，须要通过raft做数据同步的办法）不为空而且剩余时间大于0时为true
    clearRemainingTTL := le.cp != nil && l.remainingTTL > 0
​
    le.mu.RUnlock()
    //如果lease过期
    if l.expired() {
        select {
        case <-l.revokec: //revoke时候会间接返回
            return -1, ErrLeaseNotFound
        // The expired lease might fail to be revoked if the primary changes.
        // The caller will retry on ErrNotPrimary.
        case <-demotec:
            return -1, ErrNotPrimary
        case <-le.stopC:
            return -1, ErrNotPrimary
        }
    }
​
    if clearRemainingTTL {
        //通过checkpoint办法同步到各个节点lease的剩余时间
        le.cp(context.Background(), &pb.LeaseCheckpointRequest{Checkpoints: []*pb.LeaseCheckpoint{{ID: int64(l.ID), Remaining_TTL: 0}}})
    }
​
    le.mu.Lock()
    l.refresh(0)
    item := &LeaseWithTime{id: l.ID, time: l.expiry}
    //更新最小堆中的lease
    le.leaseExpiredNotifier.RegisterOrUpdate(item)
    le.mu.Unlock()
    return l.ttl, nil
}

撤销

撤销操作能够由两种形式触发，一种是通过客户端间接调用Revoke办法被动触发，一种是leader节点检测到lease过期时候的被动触发。被动触发绝对简略，follower节点收到申请后间接调用raft模块同步该申请，各个节点收到申请后通过lessor被动删除该lease（删除并没有间接删除leaseMap中的lease，而是敞开对应revokec），以及删除绑定在下面的key。

func (le *lessor) Revoke(id LeaseID) error {
    le.mu.Lock()
    l := le.leaseMap[id]
    //敞开告诉的管道
    defer close(l.revokec)
    le.mu.Unlock()
​
    if le.rd == nil {
        return nil
    }
​
    txn := le.rd()
    //Keys办法会将lease中itemSet的key取出
    keys := l.Keys()
    sort.StringSlice(keys).Sort()
    //删除lease绑定的key
    for _, key := range keys {
        txn.DeleteRange([]byte(key), nil)
    }
​
    le.mu.Lock()
    defer le.mu.Unlock()
    delete(le.leaseMap, l.ID)
    //删除boltdb长久化的lease
    le.b.BatchTx().UnsafeDelete(buckets.Lease, int64ToBytes(int64(l.ID)))
​
    txn.End()
    return nil
}

被动触发则通过创立lessor时候启动的异步协程runLoop()，每500ms轮询调用revokeExpiredLeases来查看是否过期。

func (le *lessor) revokeExpiredLeases() {
    var ls []*Lease
    // rate limit
    revokeLimit := leaseRevokeRate / 2
    le.mu.RLock()
    //如果是leader节点
    if le.isPrimary() {
        //在leaseExpiredNotifier最小堆中找到过期的lease
        ls = le.findExpiredLeases(revokeLimit)
    }
    le.mu.RUnlock()
    if len(ls) != 0 {
        select {
        case <-le.stopC:
            return
        case le.expiredC <- ls://将过期的lease发送到expireC中
        default:
        }
    }
}

在etcd启动时候，会另外启动一个异步run协程，会订阅该expireC，收到音讯后发动一个Revoke提案并进行同步操作。

//leassor通过ExpiredLeasesC办法把expiredC裸露进去
func (le *lessor) ExpiredLeasesC() <-chan []*Lease {
    return le.expiredC
}
​
//etcd启动的异步run协程
func (s *EtcdServer) run() {
    ......
    var expiredLeaseC <-chan []*lease.Lease
    if s.lessor != nil {
        expiredLeaseC = s.lessor.ExpiredLeasesC()
    }
    for{
        select{
           case leases := <-expiredLeaseC://接到过期音讯
            s.GoAttach(func() {
                for _, lease := range leases {
                    ......
                    lid := lease.ID
                    s.GoAttach(func() {
                        ctx := s.authStore.WithRoot(s.ctx)
                         //调用revoke办法
                        _, lerr := s.LeaseRevoke(ctx, &pb.LeaseRevokeRequest{ID: int64(lid)})
                        ......
                        <-c
                    })
                }
            })
           ......
            //其余case操作
        }
    }
}
​

小结

为了保持数据的一致性，lease的创立，删除，checkpoint等都须要通过raft模块进行同步，而在续约阶段则间接通过http申请发送到leader节点，所有的保护与查看工作都在leader节点，大体能够用下图来示意。因为作者对raft模块了解不够深刻，所以一笔带过。

​

Reference

  • etcd-v3.5.0源码 – https://github.com/etcd-io/et…
  • etcd 如何实现租约 – 拉钩教育，etcd原理与实际

评论

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注

这个站点使用 Akismet 来减少垃圾评论。了解你的评论数据如何被处理。

更多文章