PCSM-219: Use embedded options for Clone and Repl

Author: chupe

main.go

@@ -712,45 +712,46 @@ func resolveStartOptions(cfg *config.Config, params startRequest) (*pcsm.StartOp
 		PauseOnInitialSync: params.PauseOnInitialSync,
 		IncludeNamespaces:  params.IncludeNamespaces,
 		ExcludeNamespaces:  params.ExcludeNamespaces,
+		Repl: pcsm.ReplOptions{
+			UseCollectionBulkWrite: cfg.UseCollectionBulkWrite,
+		},
+		Clone: pcsm.CloneOptions{},
 	}
 
 	// Clone parallelism: HTTP > CLI > default
 	if params.CloneNumParallelCollections != nil {
-		options.CloneParallelism = *params.CloneNumParallelCollections
+		options.Clone.Parallelism = *params.CloneNumParallelCollections
 	} else {
-		options.CloneParallelism = cfg.Clone.NumParallelCollections
+		options.Clone.Parallelism = cfg.Clone.NumParallelCollections
 	}
 
 	// Clone read workers: HTTP > CLI > default
 	if params.CloneNumReadWorkers != nil {
-		options.CloneReadWorkers = *params.CloneNumReadWorkers
+		options.Clone.ReadWorkers = *params.CloneNumReadWorkers
 	} else {
-		options.CloneReadWorkers = cfg.Clone.NumReadWorkers
+		options.Clone.ReadWorkers = cfg.Clone.NumReadWorkers
 	}
 
 	// Clone insert workers: HTTP > CLI > default
 	if params.CloneNumInsertWorkers != nil {
-		options.CloneInsertWorkers = *params.CloneNumInsertWorkers
+		options.Clone.InsertWorkers = *params.CloneNumInsertWorkers
 	} else {
-		options.CloneInsertWorkers = cfg.Clone.NumInsertWorkers
+		options.Clone.InsertWorkers = cfg.Clone.NumInsertWorkers
 	}
 
 	// Clone segment size: HTTP > CLI > default
 	segmentSize, err := resolveCloneSegmentSize(cfg, params.CloneSegmentSize)
 	if err != nil {
 		return nil, err
 	}
-	options.CloneSegmentSizeBytes = segmentSize
+	options.Clone.SegmentSizeBytes = segmentSize
 
 	// Clone read batch size: HTTP > CLI > default
 	batchSize, err := resolveCloneReadBatchSize(cfg, params.CloneReadBatchSize)
 	if err != nil {
 		return nil, err
 	}
-	options.CloneReadBatchSizeBytes = batchSize
-
-	// UseCollectionBulkWrite: internal only, always from config (CLI + env var via Viper)
-	options.UseCollectionBulkWrite = cfg.UseCollectionBulkWrite
+	options.Clone.ReadBatchSizeBytes = batchSize
 
 	return options, nil
 }

pcsm/clone.go

@@ -22,13 +22,32 @@ import (
 	"github.com/percona/percona-clustersync-mongodb/topo"
 )
 
+// CloneOptions configures the clone behavior.
+type CloneOptions struct {
+	// Parallelism is the number of collections to clone in parallel.
+	// Default: 2 (config.DefaultCloneNumParallelCollection)
+	Parallelism int
+	// ReadWorkers is the number of read workers during clone.
+	// Default: auto (0 = runtime.NumCPU()/4)
+	ReadWorkers int
+	// InsertWorkers is the number of insert workers during clone.
+	// Default: auto (0 = runtime.NumCPU()*2)
+	InsertWorkers int
+	// SegmentSizeBytes is the segment size for clone operations in bytes.
+	// Default: auto (0 = calculated per collection)
+	SegmentSizeBytes int64
+	// ReadBatchSizeBytes is the read batch size during clone in bytes.
+	// Default: ~47.5MB (config.DefaultCloneReadBatchSizeBytes)
+	ReadBatchSizeBytes int32
+}
+
 // Clone handles the cloning of data from a source MongoDB to a target MongoDB.
 type Clone struct {
 	source   *mongo.Client // Source MongoDB client
 	target   *mongo.Client // Target MongoDB client
 	catalog  *Catalog      // Catalog for managing collections and indexes
 	nsFilter sel.NSFilter  // Namespace filter
-	options  *StartOptions // Clone options from StartOptions
+	options  *CloneOptions // Clone options
 
 	lock sync.Mutex
 	err  error // Error encountered during the cloning process
@@ -80,7 +99,7 @@ func NewClone(
 	source, target *mongo.Client,
 	catalog *Catalog,
 	nsFilter sel.NSFilter,
-	opts *StartOptions,
+	opts *CloneOptions,
 ) *Clone {
 	return &Clone{
 		source:   source,
@@ -300,18 +319,18 @@ func (c *Clone) run() error {
 func (c *Clone) doClone(ctx context.Context, namespaces []namespaceInfo) error {
 	cloneLogger := log.Ctx(ctx)
 
-	numParallelCollections := c.options.CloneParallelism
+	numParallelCollections := c.options.Parallelism
 	if numParallelCollections < 1 {
 		numParallelCollections = config.DefaultCloneNumParallelCollection
 	}
 
 	cloneLogger.Debugf("NumParallelCollections: %d", numParallelCollections)
 
 	copyManager := NewCopyManager(c.source, c.target, CopyManagerOptions{
-		NumReadWorkers:     c.options.CloneReadWorkers,
-		NumInsertWorkers:   c.options.CloneInsertWorkers,
-		SegmentSizeBytes:   c.options.CloneSegmentSizeBytes,
-		ReadBatchSizeBytes: c.options.CloneReadBatchSizeBytes,
+		NumReadWorkers:     c.options.ReadWorkers,
+		NumInsertWorkers:   c.options.InsertWorkers,
+		SegmentSizeBytes:   c.options.SegmentSizeBytes,
+		ReadBatchSizeBytes: c.options.ReadBatchSizeBytes,
 	})
 	defer copyManager.Close()
 

pcsm/pcsm.go

@@ -168,9 +168,8 @@ func (ml *PCSM) Recover(ctx context.Context, data []byte) error {
 	nsFilter := sel.MakeFilter(cp.NSInclude, cp.NSExclude)
 	catalog := NewCatalog(ml.target)
 	// Use default options for recovery (clone tuning is less relevant when resuming from checkpoint)
-	defaultOpts := &StartOptions{}
-	clone := NewClone(ml.source, ml.target, catalog, nsFilter, defaultOpts)
-	repl := NewRepl(ml.source, ml.target, catalog, nsFilter, false)
+	clone := NewClone(ml.source, ml.target, catalog, nsFilter, &CloneOptions{})
+	repl := NewRepl(ml.source, ml.target, catalog, nsFilter, &ReplOptions{})
 
 	if cp.Catalog != nil {
 		err = catalog.Recover(cp.Catalog)
@@ -285,27 +284,17 @@ func (ml *PCSM) resetError() {
 
 // StartOptions represents the options for starting the PCSM.
 type StartOptions struct {
-	// PauseOnInitialSync indicates whether to finalize after the initial sync.
+	// PauseOnInitialSync indicates whether to pause after the initial sync completes.
 	PauseOnInitialSync bool
-	// IncludeNamespaces are the namespaces to include.
+	// IncludeNamespaces are the namespaces to include in replication.
 	IncludeNamespaces []string
-	// ExcludeNamespaces are the namespaces to exclude.
+	// ExcludeNamespaces are the namespaces to exclude from replication.
 	ExcludeNamespaces []string
 
-	// Clone tuning options
-	// CloneParallelism is the number of collections to clone in parallel.
-	CloneParallelism int
-	// CloneReadWorkers is the number of read workers during clone.
-	CloneReadWorkers int
-	// CloneInsertWorkers is the number of insert workers during clone.
-	CloneInsertWorkers int
-	// CloneSegmentSizeBytes is the segment size for clone operations in bytes.
-	CloneSegmentSizeBytes int64
-	// CloneReadBatchSizeBytes is the read batch size during clone in bytes.
-	CloneReadBatchSizeBytes int32
-
-	// UseCollectionBulkWrite indicates whether to use collection-level bulk write.
-	UseCollectionBulkWrite bool
+	// Clone contains clone tuning options.
+	Clone CloneOptions
+	// Repl contains replication behavior options.
+	Repl ReplOptions
 }
 
 // Start starts the replication process with the given options.
@@ -336,8 +325,8 @@ func (ml *PCSM) Start(_ context.Context, options *StartOptions) error {
 	ml.nsFilter = sel.MakeFilter(ml.nsInclude, ml.nsExclude)
 	ml.pauseOnInitialSync = options.PauseOnInitialSync
 	ml.catalog = NewCatalog(ml.target)
-	ml.clone = NewClone(ml.source, ml.target, ml.catalog, ml.nsFilter, options)
-	ml.repl = NewRepl(ml.source, ml.target, ml.catalog, ml.nsFilter, options.UseCollectionBulkWrite)
+	ml.clone = NewClone(ml.source, ml.target, ml.catalog, ml.nsFilter, &options.Clone)
+	ml.repl = NewRepl(ml.source, ml.target, ml.catalog, ml.nsFilter, &options.Repl)
 	ml.state = StateRunning
 
 	go ml.run()

pcsm/repl.go

@@ -28,6 +28,13 @@ var (
 
 const advanceTimePseudoEvent = "@tick"
 
+// ReplOptions configures the replication behavior.
+type ReplOptions struct {
+	// UseCollectionBulkWrite indicates whether to use collection-level bulk write
+	// instead of client bulk write. Default: false (use client bulk write).
+	UseCollectionBulkWrite bool
+}
+
 // Repl handles replication from a source MongoDB to a target MongoDB.
 type Repl struct {
 	source *mongo.Client // Source MongoDB client
@@ -36,7 +43,7 @@ type Repl struct {
 	nsFilter sel.NSFilter // Namespace filter
 	catalog  *Catalog     // Catalog for managing collections and indexes
 
-	useCollectionBulkWrite bool // Whether to use collection-level bulk write
+	options *ReplOptions // Replication options
 
 	lastReplicatedOpTime bson.Timestamp
 
@@ -91,16 +98,16 @@ func NewRepl(
 	source, target *mongo.Client,
 	catalog *Catalog,
 	nsFilter sel.NSFilter,
-	useCollectionBulkWrite bool,
+	opts *ReplOptions,
 ) *Repl {
 	return &Repl{
-		source:                 source,
-		target:                 target,
-		nsFilter:               nsFilter,
-		catalog:                catalog,
-		useCollectionBulkWrite: useCollectionBulkWrite,
-		pauseC:                 make(chan struct{}),
-		doneSig:                make(chan struct{}),
+		source:   source,
+		target:   target,
+		nsFilter: nsFilter,
+		catalog:  catalog,
+		options:  opts,
+		pauseC:   make(chan struct{}),
+		doneSig:  make(chan struct{}),
 	}
 }
 
@@ -230,7 +237,7 @@ func (r *Repl) Start(ctx context.Context, startAt bson.Timestamp) error {
 		return errors.Wrap(err, "major version")
 	}
 
-	if topo.Support(targetVer).ClientBulkWrite() && !r.useCollectionBulkWrite {
+	if topo.Support(targetVer).ClientBulkWrite() && !r.options.UseCollectionBulkWrite {
 		r.bulkWrite = newClientBulkWrite(config.BulkOpsSize, targetVer.Major() < 8) //nolint:mnd
 	} else {
 		r.bulkWrite = newCollectionBulkWrite(config.BulkOpsSize, targetVer.Major() < 8) //nolint:mnd