CNTRLPLANE-3461: refactor defrag to minimize database lock time

server/config/config.go

@@ -201,6 +201,12 @@ type ServerConfig struct {
 	// consider running defrag during bootstrap. Needs to be set to non-zero value to take effect.
 	BootstrapDefragThresholdMegabytes uint `json:"bootstrap-defrag-threshold-megabytes"`
 
+	// DefragJournalMaxOps sets the maximum number of write operations buffered
+	// in the non-blocking defrag journal before backpressure is applied. Only
+	// effective when NonBlockingDefrag feature gate is enabled. 0 means
+	// unlimited (no backpressure).
+	DefragJournalMaxOps int `json:"defrag-journal-max-ops"`
+
 	// MaxLearners sets a limit to the number of learner members that can exist in the cluster membership.
 	MaxLearners int `json:"max-learners"`
 

server/embed/config.go

@@ -52,6 +52,7 @@ import (
 	"go.etcd.io/etcd/server/v3/etcdserver/api/v3compactor"
 	"go.etcd.io/etcd/server/v3/etcdserver/api/v3discovery"
 	"go.etcd.io/etcd/server/v3/features"
+	"go.etcd.io/etcd/server/v3/storage/backend"
 )
 
 const (
@@ -459,6 +460,11 @@ type Config struct {
 	ExperimentalBootstrapDefragThresholdMegabytes uint `json:"experimental-bootstrap-defrag-threshold-megabytes"`
 	// BootstrapDefragThresholdMegabytes is the minimum number of megabytes needed to be freed for etcd server to
 	BootstrapDefragThresholdMegabytes uint `json:"bootstrap-defrag-threshold-megabytes"`
+	// DefragJournalMaxOps sets the maximum number of write operations buffered
+	// in the defrag journal before backpressure is applied to writers. Only
+	// effective when the NonBlockingDefrag feature gate is enabled. 0 means
+	// unlimited (no backpressure).
+	DefragJournalMaxOps int `json:"defrag-journal-max-ops"`
 	// WarningUnaryRequestDuration is the time duration after which a warning is generated if applying
 	// unary request takes more time than this value.
 	WarningUnaryRequestDuration time.Duration `json:"warning-unary-request-duration"`
@@ -712,6 +718,7 @@ func NewConfig() *Config {
 		ExperimentalMemoryMlock:             false,
 		ExperimentalStopGRPCServiceOnDefrag: false,
 		MaxLearners:                         membership.DefaultMaxLearners,
+		DefragJournalMaxOps:                 backend.DefaultDefragJournalMaxOps,
 
 		ExperimentalTxnModeWriteWithSharedBuffer:  DefaultExperimentalTxnModeWriteWithSharedBuffer,
 		ExperimentalDistributedTracingAddress:     DefaultDistributedTracingAddress,
@@ -950,6 +957,7 @@ func (cfg *Config) AddFlags(fs *flag.FlagSet) {
 	// TODO: delete in v3.7
 	fs.UintVar(&cfg.ExperimentalBootstrapDefragThresholdMegabytes, "experimental-bootstrap-defrag-threshold-megabytes", 0, "Enable the defrag during etcd server bootstrap on condition that it will free at least the provided threshold of disk space. Needs to be set to non-zero value to take effect. It's deprecated, and will be decommissioned in v3.7. Use --bootstrap-defrag-threshold-megabytes instead.")
 	fs.UintVar(&cfg.BootstrapDefragThresholdMegabytes, "bootstrap-defrag-threshold-megabytes", 0, "Enable the defrag during etcd server bootstrap on condition that it will free at least the provided threshold of disk space. Needs to be set to non-zero value to take effect.")
+	fs.IntVar(&cfg.DefragJournalMaxOps, "defrag-journal-max-ops", backend.DefaultDefragJournalMaxOps, "Maximum number of write operations buffered in the non-blocking defrag journal before backpressure is applied. Requires --feature-gates=NonBlockingDefrag=true. Set to 0 for unlimited.")
 	// TODO: delete in v3.7
 	fs.IntVar(&cfg.MaxLearners, "max-learners", membership.DefaultMaxLearners, "Sets the maximum number of learners that can be available in the cluster membership.")
 	fs.Uint64Var(&cfg.ExperimentalSnapshotCatchUpEntries, "experimental-snapshot-catchup-entries", cfg.ExperimentalSnapshotCatchUpEntries, "Number of entries for a slow follower to catch up after compacting the raft storage entries. Deprecated in v3.6 and will be decommissioned in v3.7. Use --snapshot-catchup-entries instead.")

server/embed/etcd.go

@@ -233,6 +233,7 @@ func StartEtcd(inCfg *Config) (e *Etcd, err error) {
 		WarningUnaryRequestDuration:       cfg.WarningUnaryRequestDuration,
 		MemoryMlock:                       cfg.MemoryMlock,
 		BootstrapDefragThresholdMegabytes: cfg.BootstrapDefragThresholdMegabytes,
+		DefragJournalMaxOps:               cfg.DefragJournalMaxOps,
 		MaxLearners:                       cfg.MaxLearners,
 		V2Deprecation:                     cfg.V2DeprecationEffective(),
 		ExperimentalLocalAddress:          cfg.InferLocalAddr(),

server/etcdmain/help.go

@@ -319,6 +319,8 @@ Experimental feature:
     Enable the defrag during etcd server bootstrap on condition that it will free at least the provided threshold of disk space. Needs to be set to non-zero value to take effect. Deprecated in v3.6 and will be decommissioned in v3.7. Use '--bootstrap-defrag-threshold-megabytes' instead.
   --bootstrap-defrag-threshold-megabytes
     Enable the defrag during etcd server bootstrap on condition that it will free at least the provided threshold of disk space. Needs to be set to non-zero value to take effect.
+  --defrag-journal-max-ops '50000'
+    Maximum number of write operations buffered in the non-blocking defrag journal before backpressure is applied. Requires --feature-gates=NonBlockingDefrag=true. Set to 0 for unlimited.
   --experimental-warning-unary-request-duration '300ms'
     Set time duration after which a warning is generated if a unary request takes more than this duration. Deprecated in v3.6 and will be decommissioned in v3.7. Use '--warning-unary-request-duration' instead.
   --max-learners '1'

server/etcdserver/server.go

@@ -260,6 +260,7 @@ type EtcdServer struct {
 	kv         mvcc.WatchableKV
 	lessor     lease.Lessor
 	bemu       sync.RWMutex
+	defragMu   sync.Mutex
 	be         backend.Backend
 	beHooks    *serverstorage.BackendHooks
 	authStore  auth.AuthStore
@@ -975,8 +976,10 @@ func (s *EtcdServer) Cleanup() {
 }
 
 func (s *EtcdServer) Defragment() error {
-	s.bemu.Lock()
-	defer s.bemu.Unlock()
+	s.defragMu.Lock()
+	defer s.defragMu.Unlock()
+	s.bemu.RLock()
+	defer s.bemu.RUnlock()
 	return s.be.Defrag()
 }
 

server/features/etcd_features.go

@@ -35,6 +35,14 @@ const (
 	// of code conflicts because changes are more likely to be scattered
 	// across the file.
 
+	// NonBlockingDefrag enables a non-blocking defragmentation algorithm that
+	// allows writes to continue during the bulk data copy phase. A write journal
+	// captures mutations during the copy; only the journal replay and database
+	// swap require the write lock, reducing disruption from O(db_size) to
+	// O(writes_during_copy).
+	// owner: @dusk125
+	// alpha: v3.7
+	NonBlockingDefrag featuregate.Feature = "NonBlockingDefrag"
 	// StopGRPCServiceOnDefrag enables etcd gRPC service to stop serving client requests on defragmentation.
 	// owner: @chaochn47
 	// alpha: v3.6
@@ -78,6 +86,7 @@ const (
 
 var (
 	DefaultEtcdServerFeatureGates = map[featuregate.Feature]featuregate.FeatureSpec{
+		NonBlockingDefrag:            {Default: false, PreRelease: featuregate.Alpha},
 		StopGRPCServiceOnDefrag:      {Default: false, PreRelease: featuregate.Alpha},
 		InitialCorruptCheck:          {Default: false, PreRelease: featuregate.Alpha},
 		CompactHashCheck:             {Default: false, PreRelease: featuregate.Alpha},

server/storage/backend.go

@@ -23,6 +23,7 @@ import (
 
 	"go.etcd.io/etcd/server/v3/config"
 	"go.etcd.io/etcd/server/v3/etcdserver/api/snap"
+	"go.etcd.io/etcd/server/v3/features"
 	"go.etcd.io/etcd/server/v3/storage/backend"
 	"go.etcd.io/etcd/server/v3/storage/schema"
 	"go.etcd.io/raft/v3/raftpb"
@@ -52,6 +53,10 @@ func newBackend(cfg config.ServerConfig, hooks backend.Hooks) backend.Backend {
 	}
 	bcfg.Mlock = cfg.MemoryMlock
 	bcfg.Hooks = hooks
+	if cfg.ServerFeatureGate != nil {
+		bcfg.NonBlockingDefrag = cfg.ServerFeatureGate.Enabled(features.NonBlockingDefrag)
+	}
+	bcfg.DefragJournalMaxOps = cfg.DefragJournalMaxOps
 	return backend.New(bcfg)
 }