update

Author: dxyinme

driver/etcddriver.go

@@ -21,13 +21,11 @@ type EtcdDriver struct {
 	nodeID      string
 	serviceName string
 
-	cli    *clientv3.Client
-	nodes  *sync.Map
-	logger dlog.Logger
-
-	leaseTimeout int64
-	leaseID      clientv3.LeaseID
-	lease        clientv3.Lease
+	cli     *clientv3.Client
+	lease   int64
+	nodes   *sync.Map
+	leaseID clientv3.LeaseID
+	logger  dlog.Logger
 
 	ctx    context.Context
 	cancel context.CancelFunc
@@ -49,12 +47,13 @@ func newEtcdDriver(cli *clientv3.Client) *EtcdDriver {
 // 设置key value，绑定租约
 func (e *EtcdDriver) putKeyWithLease(ctx context.Context, key, val string) (clientv3.LeaseID, error) {
 	//设置租约时间，最少5s
-	if e.leaseTimeout < etcdDefaultLease {
-		e.leaseTimeout = etcdDefaultLease
+	if e.lease < etcdDefaultLease {
+		e.lease = etcdDefaultLease
 	}
+
 	subCtx, cancel := context.WithTimeout(ctx, etcdBusinessTimeout)
 	defer cancel()
-	resp, err := e.lease.Grant(ctx, e.leaseTimeout)
+	resp, err := e.cli.Grant(subCtx, e.lease)
 	if err != nil {
 		return 0, err
 	}
@@ -121,26 +120,27 @@ func (e *EtcdDriver) getServices() []string {
 	return addrs
 }
 
-func (e *EtcdDriver) createLease(ctx context.Context, nodeID string) (<-chan *clientv3.LeaseKeepAliveResponse, error) {
+func (e *EtcdDriver) keepAlive(ctx context.Context, nodeID string) (<-chan *clientv3.LeaseKeepAliveResponse, error) {
 	var err error
-	e.lease = clientv3.NewLease(e.cli)
 	e.leaseID, err = e.putKeyWithLease(ctx, nodeID, nodeID)
 	if err != nil {
 		e.logger.Errorf("putKeyWithLease error: %v", err)
 		return nil, err
 	}
-	return e.lease.KeepAlive(ctx, e.leaseID)
+
+	return e.cli.KeepAlive(ctx, e.leaseID)
 }
 
 func (e *EtcdDriver) revoke(ctx context.Context) {
-	_, err := e.lease.Revoke(ctx, e.leaseID)
+	_, err := e.cli.Lease.Revoke(ctx, e.leaseID)
 	if err != nil {
 		e.logger.Infof("lease revoke error: %v", err)
 	}
 }
 
 func (e *EtcdDriver) heartBeat(ctx context.Context) {
-	leaseCh, err := e.createLease(ctx, e.nodeID)
+label:
+	leaseCh, err := e.keepAlive(ctx, e.nodeID)
 	if err != nil {
 		e.logger.Errorf("keep alive error, %v", err)
 		return
@@ -152,11 +152,24 @@ func (e *EtcdDriver) heartBeat(ctx context.Context) {
 				e.logger.Infof("driver stopped")
 				return
 			}
-		case resp := <-leaseCh:
+		case _, ok := <-leaseCh:
 			{
 				// if lease timeout, goto top of
 				// this function to keepalive
-				e.logger.Infof("leaseID=%0x", resp.ID)
+				if !ok {
+					goto label
+				}
+			}
+		case <-time.After(etcdBusinessTimeout):
+			{
+				e.logger.Errorf("ectd cli keepalive timeout")
+				return
+			}
+		case <-time.After(time.Duration(e.lease) * (time.Second) / 2):
+			{
+				// if near to nodes time,
+				// renew the lease
+				goto label
 			}
 		}
 	}
@@ -196,7 +209,7 @@ func (e *EtcdDriver) withOption(opt Option) (err error) {
 	switch opt.Type() {
 	case OptionTypeTimeout:
 		{
-			e.leaseTimeout = int64(opt.(TimeoutOption).timeout.Seconds())
+			e.lease = int64(opt.(TimeoutOption).timeout.Seconds())
 		}
 	case OptionTypeLogger:
 		{