Merge pull request #2 from jingyih/expensive_read_instrumentation

*: add flag for backend expensive read limit

embed/config.go

@@ -75,6 +75,8 @@ const (
 	// maxElectionMs specifies the maximum value of election timeout.
 	// More details are listed in ../Documentation/tuning.md#time-parameters.
 	maxElectionMs = 50000
+
+	DefaultBackendExpensiveReadLimit = 1000
 )
 
 var (
@@ -270,9 +272,10 @@ type Config struct {
 	AuthToken  string `json:"auth-token"`
 	BcryptCost uint   `json:"bcrypt-cost"`
 
-	ExperimentalInitialCorruptCheck bool          `json:"experimental-initial-corrupt-check"`
-	ExperimentalCorruptCheckTime    time.Duration `json:"experimental-corrupt-check-time"`
-	ExperimentalEnableV2V3          string        `json:"experimental-enable-v2v3"`
+	ExperimentalInitialCorruptCheck       bool          `json:"experimental-initial-corrupt-check"`
+	ExperimentalCorruptCheckTime          time.Duration `json:"experimental-corrupt-check-time"`
+	ExperimentalEnableV2V3                string        `json:"experimental-enable-v2v3"`
+	ExperimentalBackendExpensiveReadLimit int           `json:"experimental-backend-expensive-read-limit"`
 
 	// ForceNewCluster starts a new cluster even if previously started; unsafe.
 	ForceNewCluster bool `json:"force-new-cluster"`

embed/etcd.go

@@ -201,6 +201,7 @@ func StartEtcd(inCfg *Config) (e *Etcd, err error) {
 		Debug:                      cfg.Debug,
 		ForceNewCluster:            cfg.ForceNewCluster,
 		EnableGRPCGateway:          cfg.EnableGRPCGateway,
+		BackendExpensiveReadLimit:  cfg.ExperimentalBackendExpensiveReadLimit,
 	}
 	print(e.cfg.logger, *cfg, srvcfg, memberInitialized)
 	if e.Server, err = etcdserver.NewServer(srvcfg); err != nil {

etcdmain/config.go

@@ -248,6 +248,7 @@ func newConfig() *config {
 	fs.BoolVar(&cfg.ec.ExperimentalInitialCorruptCheck, "experimental-initial-corrupt-check", cfg.ec.ExperimentalInitialCorruptCheck, "Enable to check data corruption before serving any client/peer traffic.")
 	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.IntVar(&cfg.ec.ExperimentalBackendExpensiveReadLimit, "experimental-backend-expensive-read-limit", embed.DefaultBackendExpensiveReadLimit, "The number of keys in expensive read request in etcd backend")
 
 	// unsafe
 	fs.BoolVar(&cfg.ec.ForceNewCluster, "force-new-cluster", false, "Force to create a new one member cluster.")

etcdmain/help.go

@@ -200,6 +200,8 @@ Experimental feature:
     Duration of time between cluster corruption check passes.
   --experimental-enable-v2v3 ''
     Serve v2 requests through the v3 backend under a given prefix.
+  --experimental-backend-expensive-read-limit '1000'
+    The number of keys in expensive read request in etcd backend
 
 Unsafe feature:
   --force-new-cluster 'false'

etcdserver/backend.go

@@ -48,6 +48,7 @@ func newBackend(cfg ServerConfig) backend.Backend {
 		// permit 10% excess over quota for disarm
 		bcfg.MmapSize = uint64(cfg.QuotaBackendBytes + cfg.QuotaBackendBytes/10)
 	}
+	bcfg.ExpensiveReadLimit = cfg.BackendExpensiveReadLimit
 	return backend.New(bcfg)
 }
 

etcdserver/config.go

@@ -150,6 +150,8 @@ type ServerConfig struct {
 	LeaseCheckpointInterval time.Duration
 
 	EnableGRPCGateway bool
+
+	BackendExpensiveReadLimit int
 }
 
 // VerifyBootstrap sanity-checks the initial config for bootstrap case

mvcc/backend/backend.go

@@ -71,6 +71,7 @@ type Backend interface {
 	Defrag() error
 	ForceCommit()
 	Close() error
+	ExpensiveReadLimit() int
 }
 
 type Snapshot interface {
@@ -108,6 +109,8 @@ type backend struct {
 	donec chan struct{}
 
 	lg *zap.Logger
+
+	expensiveReadLimit int
 }
 
 type BackendConfig struct {
@@ -121,6 +124,8 @@ type BackendConfig struct {
 	MmapSize uint64
 	// Logger logs backend-side operations.
 	Logger *zap.Logger
+	// ExpensiveReadLimit is the number of keys in expensive read request
+	ExpensiveReadLimit int
 }
 
 func DefaultBackendConfig() BackendConfig {
@@ -178,6 +183,8 @@ func newBackend(bcfg BackendConfig) *backend {
 		donec: make(chan struct{}),
 
 		lg: bcfg.Logger,
+
+		expensiveReadLimit: bcfg.ExpensiveReadLimit,
 	}
 	b.batchTx = newBatchTxBuffered(b)
 	go b.run()
@@ -570,3 +577,7 @@ func (s *snapshot) Close() error {
 	<-s.donec
 	return s.Tx.Rollback()
 }
+
+func (b *backend) ExpensiveReadLimit() int {
+	return b.expensiveReadLimit
+}

mvcc/kvstore_txn.go

@@ -22,9 +22,8 @@ import (
 )
 
 const (
-	expensiveReadLimit = 1000
-	readonly           = true
-	readwrite          = false
+	readonly  = true
+	readwrite = false
 )
 
 type storeTxnRead struct {
@@ -40,6 +39,8 @@ type storeTxnRead struct {
 
 	firstRev int64
 	rev      int64
+
+	expensiveReadLimit int
 }
 
 func (s *store) Read() TxnRead {
@@ -49,13 +50,14 @@ func (s *store) Read() TxnRead {
 	firstRev, rev := s.compactMainRev, s.currentRev
 	s.revMu.RUnlock()
 	return newMetricsTxnRead(&storeTxnRead{
-		s:               s,
-		tx:              tx,
-		b:               s.b,
-		ro:              readonly,
-		isCommittedRead: false,
-		firstRev:        firstRev,
-		rev:             rev})
+		s:                  s,
+		tx:                 tx,
+		b:                  s.b,
+		ro:                 readonly,
+		isCommittedRead:    false,
+		firstRev:           firstRev,
+		rev:                rev,
+		expensiveReadLimit: s.b.ExpensiveReadLimit()})
 }
 
 func (tr *storeTxnRead) FirstRev() int64       { return tr.firstRev }
@@ -162,8 +164,8 @@ func (tr *storeTxnRead) rangeKeys(key, end []byte, curRev int64, ro RangeOptions
 		limit = len(revpairs)
 	}
 
-	if limit > expensiveReadLimit && !tr.txlocked && tr.ro { // first expensive read in a read only transcation
-		// too many keys to range. upgrade the read transcation to concurrent read tx.
+	if limit > tr.expensiveReadLimit && !tr.txlocked && tr.ro { // first expensive read in a read only transaction
+		// too many keys to range. upgrade the read transaction to concurrent read tx.
 		tr.tx = tr.b.CommittedReadTx()
 		tr.isCommittedRead = true
 	}