*: add experimental flag for watch notify interval

Signed-off-by: Gyuho Lee <leegyuho@amazon.com>

embed/config.go

@@ -284,6 +284,7 @@ type Config struct {
 	// ExperimentalEnableLeaseCheckpoint enables primary lessor to persist lease remainingTTL to prevent indefinite auto-renewal of long lived leases.
 	ExperimentalEnableLeaseCheckpoint       bool          `json:"experimental-enable-lease-checkpoint"`
 	ExperimentalCompactionBatchLimit        int           `json:"experimental-compaction-batch-limit"`
+	ExperimentalWatchProgressNotifyInterval time.Duration `json:"experimental-watch-progress-notify-interval"`
 
 	// ForceNewCluster starts a new cluster even if previously started; unsafe.
 	ForceNewCluster bool `json:"force-new-cluster"`

embed/etcd.go

@@ -208,6 +208,7 @@ func StartEtcd(inCfg *Config) (e *Etcd, err error) {
 		UnsafeNoFsync:              cfg.UnsafeNoFsync,
 		EnableLeaseCheckpoint:      cfg.ExperimentalEnableLeaseCheckpoint,
 		CompactionBatchLimit:       cfg.ExperimentalCompactionBatchLimit,
+		WatchProgressNotifyInterval: cfg.ExperimentalWatchProgressNotifyInterval,
 	}
 	print(e.cfg.logger, *cfg, srvcfg, memberInitialized)
 	if e.Server, err = etcdserver.NewServer(srvcfg); err != nil {

etcdmain/config.go

@@ -257,6 +257,7 @@ func newConfig() *config {
 	fs.StringVar(&cfg.ec.ExperimentalBackendFreelistType, "experimental-backend-bbolt-freelist-type", cfg.ec.ExperimentalBackendFreelistType, "ExperimentalBackendFreelistType specifies the type of freelist that boltdb backend uses(array and map are supported types)")
 	fs.BoolVar(&cfg.ec.ExperimentalEnableLeaseCheckpoint, "experimental-enable-lease-checkpoint", false, "Enable to persist lease remaining TTL to prevent indefinite auto-renewal of long lived leases.")
 	fs.IntVar(&cfg.ec.ExperimentalCompactionBatchLimit, "experimental-compaction-batch-limit", cfg.ec.ExperimentalCompactionBatchLimit, "Sets the maximum revisions deleted in each compaction batch.")
+	fs.DurationVar(&cfg.ec.ExperimentalWatchProgressNotifyInterval, "experimental-watch-progress-notify-interval", cfg.ec.ExperimentalWatchProgressNotifyInterval, "Duration of periodic watch progress notifications.")
 
 	// unsafe
 	fs.BoolVar(&cfg.ec.UnsafeNoFsync, "unsafe-no-fsync", false, "Disables fsync, unsafe, will cause data loss.")

etcdmain/help.go

@@ -210,6 +210,8 @@ Experimental feature:
     ExperimentalCompactionBatchLimit sets the maximum revisions deleted in each compaction batch.
   --experimental-peer-skip-client-san-verification 'false'
     Skip verification of SAN field in client certificate for peer connections.
+  --experimental-watch-progress-notify-interval '10m'
+    Duration of periodical watch progress notification.
 
 Unsafe feature:
   --force-new-cluster 'false'

etcdserver/api/v3rpc/watch.go

@@ -31,6 +31,8 @@ import (
 	"go.uber.org/zap"
 )
 
+const minWatchProgressInterval = 100 * time.Millisecond
+
 type watchServer struct {
 	lg *zap.Logger
 
@@ -58,6 +60,20 @@ func NewWatchServer(s *etcdserver.EtcdServer) pb.WatchServer {
 		watchable: s.Watchable(),
 		ag:        s,
 	}
+	if srv.lg == nil {
+		srv.lg = zap.NewNop()
+	}
+	if s.Cfg.WatchProgressNotifyInterval > 0 {
+		if s.Cfg.WatchProgressNotifyInterval < minWatchProgressInterval {
+			srv.lg.Warn(
+				"adjusting watch progress notify interval to minimum period",
+				zap.Duration("min-watch-progress-notify-interval", minWatchProgressInterval),
+			)
+			s.Cfg.WatchProgressNotifyInterval = minWatchProgressInterval
+		}
+		SetProgressReportInterval(s.Cfg.WatchProgressNotifyInterval)
+	}
+	return srv
 }
 
 var (

etcdserver/config.go

@@ -159,6 +159,8 @@ type ServerConfig struct {
 
 	EnableGRPCGateway bool
 
+	WatchProgressNotifyInterval time.Duration
+
 	// UnsafeNoFsync disables all uses of fsync.
 	// Setting this is unsafe and will cause data loss.
 	UnsafeNoFsync bool `json:"unsafe-no-fsync"`