Merge pull request etcd-io#6670 from fanminshi/lease_stressor

functional-tester: add lease stresser

tools/functional-tester/etcd-tester/checks.go

@@ -16,7 +16,10 @@ package main
 
 import (
 	"fmt"
+	"strings"
 	"time"
+
+	"golang.org/x/net/context"
 )
 
 type Checker interface {
@@ -34,6 +37,8 @@ type hashChecker struct {
 
 func newHashChecker(hrg hashAndRevGetter) Checker { return &hashChecker{hrg} }
 
+const leaseCheckerTimeout = 10 * time.Second
+
 func (hc *hashChecker) Check() (err error) {
 	plog.Printf("fetching current revisions...")
 	var (
@@ -69,6 +74,100 @@ func (hc *hashChecker) Check() (err error) {
 	return nil
 }
 
+type leaseChecker struct {
+	leaseStressers []Stresser
+}
+
+func newLeaseChecker(leaseStressers []Stresser) Checker { return &leaseChecker{leaseStressers} }
+
+func (lc *leaseChecker) Check() error {
+	plog.Info("lease stresser invariant check...")
+	errc := make(chan error)
+	for _, ls := range lc.leaseStressers {
+		go func(s Stresser) { errc <- lc.checkInvariant(s) }(ls)
+	}
+	var errs []error
+	for i := 0; i < len(lc.leaseStressers); i++ {
+		if err := <-errc; err != nil {
+			errs = append(errs, err)
+		}
+	}
+
+	if len(errs) == 0 {
+		return nil
+	}
+	return fmt.Errorf("lease stresser encounters error: (%v)", fromErrsToString(errs))
+}
+
+func fromErrsToString(errs []error) string {
+	stringArr := make([]string, len(errs))
+	for i, err := range errs {
+		stringArr[i] = err.Error()
+	}
+	return strings.Join(stringArr, ",")
+}
+
+func (lc *leaseChecker) checkInvariant(lStresser Stresser) error {
+	ls := lStresser.(*leaseStresser)
+	if err := checkLeasesExpired(ls); err != nil {
+		return err
+	}
+	ls.revokedLeases = &atomicLeases{leases: make(map[int64]time.Time)}
+	return checkLeasesAlive(ls)
+}
+
+func checkLeasesExpired(ls *leaseStresser) error {
+	plog.Infof("revoked leases %v", ls.revokedLeases.getLeasesMap())
+	return checkLeases(true, ls, ls.revokedLeases.getLeasesMap())
+}
+
+func checkLeasesAlive(ls *leaseStresser) error {
+	plog.Infof("alive leases %v", ls.aliveLeases.getLeasesMap())
+	return checkLeases(false, ls, ls.aliveLeases.getLeasesMap())
+}
+
+func checkLeases(expired bool, ls *leaseStresser, leases map[int64]time.Time) error {
+	ctx, cancel := context.WithTimeout(context.Background(), leaseCheckerTimeout)
+	defer cancel()
+	for leaseID := range leases {
+		keysExpired, err := ls.hasKeysAttachedToLeaseExpired(ctx, leaseID)
+		if err != nil {
+			plog.Errorf("hasKeysAttachedToLeaseExpired error: (%v)", err)
+			return err
+		}
+		leaseExpired, err := ls.hasLeaseExpired(ctx, leaseID)
+		if err != nil {
+			plog.Errorf("hasLeaseExpired error: (%v)", err)
+			return err
+		}
+		if leaseExpired != keysExpired {
+			return fmt.Errorf("lease %v expiration mismatch (lease expired=%v, keys expired=%v)", leaseID, leaseExpired, keysExpired)
+		}
+		if leaseExpired != expired {
+			return fmt.Errorf("lease %v expected expired=%v, got %v", leaseID, expired, leaseExpired)
+		}
+	}
+	return nil
+}
+
+type compositeChecker struct {
+	checkers []Checker
+}
+
+func newCompositeChecker(checkers []Checker) Checker {
+	return &compositeChecker{checkers}
+}
+
+func (cchecker *compositeChecker) Check() error {
+	for _, checker := range cchecker.checkers {
+		if err := checker.Check(); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
 type noChecker struct{}
 
 func newNoChecker() Checker        { return &noChecker{} }

tools/functional-tester/etcd-tester/cluster.go

@@ -45,9 +45,10 @@ type cluster struct {
 	consistencyCheck bool
 	Size             int
 
-	Stressers     []Stresser
-	stressBuilder stressBuilder
-	Checker       Checker
+	Stressers            []Stresser
+	stressBuilder        stressBuilder
+	leaseStresserBuilder leaseStresserBuilder
+	Checker              Checker
 
 	Members []*member
 }
@@ -99,18 +100,27 @@ func (c *cluster) bootstrap() error {
 		}
 	}
 
-	c.Stressers = make([]Stresser, len(members))
+	c.Stressers = make([]Stresser, 0)
+	leaseStressers := make([]Stresser, len(members))
 	for i, m := range members {
-		c.Stressers[i] = c.stressBuilder(m)
+		lStresser := c.leaseStresserBuilder(m)
+		leaseStressers[i] = lStresser
+		c.Stressers = append(c.Stressers, c.stressBuilder(m), lStresser)
+	}
+
+	for i := range c.Stressers {
 		go c.Stressers[i].Stress()
 	}
 
+	var checkers []Checker
 	if c.consistencyCheck && !c.v2Only {
-		c.Checker = newHashChecker(hashAndRevGetter(c))
+		checkers = append(checkers, newHashChecker(hashAndRevGetter(c)), newLeaseChecker(leaseStressers))
 	} else {
-		c.Checker = newNoChecker()
+		checkers = append(checkers, newNoChecker())
 	}
 
+	c.Checker = newCompositeChecker(checkers)
+
 	c.Size = size
 	c.Members = members
 	return nil
@@ -176,6 +186,7 @@ func (c *cluster) Cleanup() error {
 	for _, s := range c.Stressers {
 		s.Cancel()
 	}
+
 	return lasterr
 }