lease: Add new configuration flag to toggle persisting expiration of …

…leases

Add a persist expiry flag that allows the expiry field of leases to be
persisted along with the ID and ttl. Defaults to false. Addresses etcd-io#9496

embed/config.go

@@ -225,6 +225,8 @@ type Config struct {
 	ExperimentalCorruptCheckTime    time.Duration `json:"experimental-corrupt-check-time"`
 	ExperimentalEnableV2V3          string        `json:"experimental-enable-v2v3"`
 
+	PersistExpiry                   bool          `json:"persist-expiry"`
+
 	// ForceNewCluster starts a new cluster even if previously started; unsafe.
 	ForceNewCluster bool `json:"force-new-cluster"`
 }
@@ -289,6 +291,7 @@ func NewConfig() *Config {
 		HostWhitelist:         map[string]struct{}{"*": {}},
 		AuthToken:             "simple",
 		PreVote:               false, // TODO: enable by default in v3.5
+		PersistExpiry:         false,
 	}
 	cfg.InitialCluster = cfg.InitialClusterFromName(cfg.Name)
 	return cfg

embed/etcd.go

@@ -175,6 +175,7 @@ func StartEtcd(inCfg *Config) (e *Etcd, err error) {
 		PreVote:                 cfg.PreVote,
 		Debug:                   cfg.Debug,
 		ForceNewCluster:         cfg.ForceNewCluster,
+		PersistExpiry:           cfg.PersistExpiry,
 	}
 
 	if e.Server, err = etcdserver.NewServer(srvcfg); err != nil {

etcdmain/config.go

@@ -237,6 +237,8 @@ func newConfig() *config {
 	fs.DurationVar(&cfg.ec.ExperimentalCorruptCheckTime, "experimental-corrupt-check-time", cfg.ec.ExperimentalCorruptCheckTime, "Duration of time between cluster corruption check passes.")
 	fs.StringVar(&cfg.ec.ExperimentalEnableV2V3, "experimental-enable-v2v3", cfg.ec.ExperimentalEnableV2V3, "v3 prefix for serving emulated v2 state.")
 
+	fs.BoolVar(&cfg.ec.PersistExpiry, "persist-expiry", cfg.ec.PersistExpiry, "Persist expiry values for leases.")
+
 	// unsafe
 	fs.BoolVar(&cfg.ec.ForceNewCluster, "force-new-cluster", false, "Force to create a new one member cluster.")
 

etcdserver/config.go

@@ -83,6 +83,8 @@ type ServerConfig struct {
 	Debug bool
 
 	ForceNewCluster bool
+
+	PersistExpiry bool
 }
 
 // VerifyBootstrap sanity-checks the initial config for bootstrap case

etcdserver/server.go

@@ -453,7 +453,7 @@ func NewServer(cfg ServerConfig) (srv *EtcdServer, err error) {
 
 	// always recover lessor before kv. When we recover the mvcc.KV it will reattach keys to its leases.
 	// If we recover mvcc.KV first, it will attach the keys to the wrong lessor before it recovers.
-	srv.lessor = lease.NewLessor(srv.be, int64(math.Ceil(minTTL.Seconds())))
+	srv.lessor = lease.NewLessor(srv.be, int64(math.Ceil(minTTL.Seconds())), cfg.PersistExpiry)
 	srv.kv = mvcc.New(srv.be, srv.lessor, &srv.consistIndex)
 	if beExist {
 		kvindex := srv.kv.ConsistentIndex()

lease/leasehttp/http_test.go

@@ -32,7 +32,7 @@ func TestRenewHTTP(t *testing.T) {
 	defer os.Remove(tmpPath)
 	defer be.Close()
 
-	le := lease.NewLessor(be, int64(5))
+	le := lease.NewLessor(be, int64(5), false)
 	le.Promote(time.Second)
 	l, err := le.Grant(1, int64(5))
 	if err != nil {
@@ -56,7 +56,7 @@ func TestTimeToLiveHTTP(t *testing.T) {
 	defer os.Remove(tmpPath)
 	defer be.Close()
 
-	le := lease.NewLessor(be, int64(5))
+	le := lease.NewLessor(be, int64(5), false)
 	le.Promote(time.Second)
 	l, err := le.Grant(1, int64(5))
 	if err != nil {
@@ -97,7 +97,7 @@ func testApplyTimeout(t *testing.T, f func(*lease.Lease, string) error) {
 	defer os.Remove(tmpPath)
 	defer be.Close()
 
-	le := lease.NewLessor(be, int64(5))
+	le := lease.NewLessor(be, int64(5), false)
 	le.Promote(time.Second)
 	l, err := le.Grant(1, int64(5))
 	if err != nil {

lease/leasepb/lease.proto

@@ -13,6 +13,7 @@ option (gogoproto.goproto_enum_prefix_all) = false;
 message Lease {
   int64 ID = 1;
   int64 TTL = 2;
+  int64 Expiry = 3;
 }
 
 message LeaseInternalRequest {

lease/lessor.go

@@ -149,13 +149,15 @@ type lessor struct {
 	stopC chan struct{}
 	// doneC is a channel whose closure indicates that the lessor is stopped.
 	doneC chan struct{}
+
+	persistExpiry bool
 }
 
-func NewLessor(b backend.Backend, minLeaseTTL int64) Lessor {
-	return newLessor(b, minLeaseTTL)
+func NewLessor(b backend.Backend, minLeaseTTL int64, persistExpiry bool) Lessor {
+	return newLessor(b, minLeaseTTL, persistExpiry)
 }
 
-func newLessor(b backend.Backend, minLeaseTTL int64) *lessor {
+func newLessor(b backend.Backend, minLeaseTTL int64, persistExpiry bool) *lessor {
 	l := &lessor{
 		leaseMap:    make(map[LeaseID]*Lease),
 		itemMap:     make(map[LeaseItem]LeaseID),
@@ -165,7 +167,9 @@ func newLessor(b backend.Backend, minLeaseTTL int64) *lessor {
 		expiredC: make(chan []*Lease, 16),
 		stopC:    make(chan struct{}),
 		doneC:    make(chan struct{}),
+		persistExpiry: persistExpiry,
 	}
+
 	l.initAndRecover()
 
 	go l.runLoop()
@@ -226,14 +230,14 @@ func (le *lessor) Grant(id LeaseID, ttl int64) (*Lease, error) {
 		l.ttl = le.minLeaseTTL
 	}
 
-	if le.isPrimary() {
+	if le.isPrimary() || le.persistExpiry {
 		l.refresh(0)
 	} else {
 		l.forever()
 	}
 
 	le.leaseMap[id] = l
-	l.persistTo(le.b)
+	l.persistTo(le.b, le.persistExpiry)
 
 	return l, nil
 }
@@ -315,6 +319,9 @@ func (le *lessor) Renew(id LeaseID) (int64, error) {
 	}
 
 	l.refresh(0)
+	if le.persistExpiry {
+		l.persistTo(le.b, le.persistExpiry)
+	}
 	return l.ttl, nil
 }
 
@@ -345,7 +352,9 @@ func (le *lessor) Promote(extend time.Duration) {
 	defer le.mu.Unlock()
 
 	le.demotec = make(chan struct{})
-
+	if le.persistExpiry {
+		return
+	}
 	// refresh the expiries of all leases.
 	for _, l := range le.leaseMap {
 		l.refresh(extend)
@@ -397,9 +406,11 @@ func (le *lessor) Demote() {
 	le.mu.Lock()
 	defer le.mu.Unlock()
 
-	// set the expiries of all leases to forever
-	for _, l := range le.leaseMap {
-		l.forever()
+	if !le.persistExpiry {
+		// set the expiries of all leases to forever
+		for _, l := range le.leaseMap {
+			l.forever()
+		}
 	}
 
 	if le.demotec != nil {
@@ -550,13 +561,19 @@ func (le *lessor) initAndRecover() {
 		if lpb.TTL < le.minLeaseTTL {
 			lpb.TTL = le.minLeaseTTL
 		}
+		var expiry time.Time
+		if le.persistExpiry && lpb.Expiry != 0{
+			expiry = time.Unix(0, lpb.Expiry)
+		}else{
+			expiry = forever
+		}
 		le.leaseMap[ID] = &Lease{
 			ID:  ID,
 			ttl: lpb.TTL,
 			// itemSet will be filled in when recover key-value pairs
 			// set expiry to forever, refresh when promoted
 			itemSet: make(map[LeaseItem]struct{}),
-			expiry:  forever,
+			expiry:  expiry,
 			revokec: make(chan struct{}),
 		}
 	}
@@ -583,10 +600,14 @@ func (l *Lease) expired() bool {
 	return l.Remaining() <= 0
 }
 
-func (l *Lease) persistTo(b backend.Backend) {
+func (l *Lease) persistTo(b backend.Backend, expiry bool) {
 	key := int64ToBytes(int64(l.ID))
-
-	lpb := leasepb.Lease{ID: int64(l.ID), TTL: int64(l.ttl)}
+	var lpb leasepb.Lease
+	if expiry{
+		lpb = leasepb.Lease{ID: int64(l.ID), TTL: int64(l.ttl), Expiry: l.expiry.UnixNano()}
+	}else{
+		lpb = leasepb.Lease{ID: int64(l.ID), TTL: int64(l.ttl)}
+	}
 	val, err := lpb.Marshal()
 	if err != nil {
 		panic("failed to marshal lease proto item")