PCSM-219: Align all env var config with CLI and HTTP request options

.vscode/settings.json

@@ -23,7 +23,6 @@
   "go.formatFlags": [
     "-extra"
   ],
-  "go.formatTool": "gofumpt",
   "go.lintTool": "golangci-lint-v2",
   "go.useLanguageServer": true,
   "gopls": {
@@ -40,5 +39,31 @@
     "tests"
   ],
   "python.testing.pytestEnabled": true,
-  "python.testing.unittestEnabled": false
+  "python.testing.unittestEnabled": false,
+  "cSpell.words": [
+    "bson",
+    "clustersync",
+    "cmdutil",
+    "codegen",
+    "colls",
+    "connstring",
+    "contextcheck",
+    "Debugf",
+    "errgroup",
+    "errorlint",
+    "Infof",
+    "keygen",
+    "mapstructure",
+    "nolint",
+    "opencode",
+    "pcsm",
+    "pipefail",
+    "readconcern",
+    "readpref",
+    "Warnf",
+    "wrapcheck",
+    "Wrapf",
+    "writeconcern",
+    "zerolog"
+  ]
 }

README.md

@@ -150,7 +150,7 @@ When starting the PCSM server, you can use the following options:
 - `--target`: The MongoDB connection string for the target cluster
 - `--log-level`: The log level (default: "info")
 - `--log-json`: Output log in JSON format with disabled color
-- `--no-color`: Disable log ASCI color
+- `--log-no-color`: Disable log ASCI color
 
 Example:
 
@@ -213,7 +213,11 @@ Example:
 
 ```json
 {
-    "includeNamespaces": ["dbName.*", "anotherDB.collName1", "anotherDB.collName2"],
+    "includeNamespaces": [
+        "dbName.*",
+        "anotherDB.collName1",
+        "anotherDB.collName2"
+    ],
     "excludeNamespaces": ["dbName.collName"]
 }
 ```
@@ -303,7 +307,6 @@ The /status endpoint provides the current state of the PCSM replication process,
 - `lastReplicatedOpTime.ts`: op time timestamp
 - `lastReplicatedOpTime.isoDate`: op time ts in human-readable form
 
-
 - `initialSync.completed`: indicates if the initial sync is completed.
 - `initialSync.lagTimeSeconds`: the lag time in logical seconds until the initial sync completed.
 

config/config.go

@@ -0,0 +1,180 @@
+// Package config provides configuration management for PCSM using Viper.
+package config
+
+import (
+	"math"
+	"slices"
+	"strings"
+	"time"
+
+	"github.com/dustin/go-humanize"
+	"github.com/go-viper/mapstructure/v2"
+	"github.com/spf13/cobra"
+	"github.com/spf13/viper"
+
+	"github.com/percona/percona-clustersync-mongodb/errors"
+)
+
+// Config holds all PCSM configuration.
+type Config struct {
+	Port   int    `mapstructure:"port"`
+	Source string `mapstructure:"source"`
+	Target string `mapstructure:"target"`
+
+	Log LogConfig `mapstructure:",squash"`
+
+	MongoDB MongoDBConfig `mapstructure:",squash"`
+
+	UseCollectionBulkWrite bool `mapstructure:"use-collection-bulk-write"`
+
+	Clone CloneConfig `mapstructure:",squash"`
+
+	// hidden startup flags
+	Start              bool `mapstructure:"start"`
+	ResetState         bool `mapstructure:"reset-state"`
+	PauseOnInitialSync bool `mapstructure:"pause-on-initial-sync"`
+}
+
+// LogConfig holds logging configuration.
+type LogConfig struct {
+	Level   string `mapstructure:"log-level"`
+	JSON    bool   `mapstructure:"log-json"`
+	NoColor bool   `mapstructure:"log-no-color"`
+}
+
+// MongoDBConfig holds MongoDB client configuration.
+type MongoDBConfig struct {
+	OperationTimeout  time.Duration `mapstructure:"mongodb-operation-timeout"`
+	TargetCompressors []string      `mapstructure:"dev-target-client-compressors"`
+}
+
+// CloneConfig holds clone operation configuration.
+type CloneConfig struct {
+	// NumParallelCollections is the number of collections to clone in parallel.
+	// 0 means auto (calculated at runtime).
+	NumParallelCollections int `mapstructure:"clone-num-parallel-collections"`
+	// NumReadWorkers is the number of read workers during clone.
+	// 0 means auto (calculated at runtime).
+	NumReadWorkers int `mapstructure:"clone-num-read-workers"`
+	// NumInsertWorkers is the number of insert workers during clone.
+	// 0 means auto (calculated at runtime).
+	NumInsertWorkers int `mapstructure:"clone-num-insert-workers"`
+	// SegmentSize is the segment size for clone operations (e.g., "500MB", "1GiB").
+	// Empty string means auto (calculated at runtime for each collection).
+	SegmentSize string `mapstructure:"clone-segment-size"`
+	// ReadBatchSize is the read batch size during clone (e.g., "16MiB", "100MB").
+	// Empty string means auto (calculated at runtime for each collection).
+	ReadBatchSize string `mapstructure:"clone-read-batch-size"`
+}
+
+// Load initializes Viper and returns a validated Config.
+func Load(cmd *cobra.Command) (*Config, error) {
+	viper.SetEnvPrefix("PCSM")
+	viper.SetEnvKeyReplacer(strings.NewReplacer("-", "_"))
+	viper.AutomaticEnv()
+
+	if cmd.PersistentFlags() != nil {
+		_ = viper.BindPFlags(cmd.PersistentFlags())
+	}
+
+	if cmd.Flags() != nil {
+		_ = viper.BindPFlags(cmd.Flags())
+	}
+
+	bindEnvVars()
+
+	var cfg Config
+
+	err := viper.Unmarshal(&cfg, viper.DecodeHook(
+		mapstructure.ComposeDecodeHookFunc(
+			mapstructure.StringToTimeDurationHookFunc(),
+			mapstructure.StringToSliceHookFunc(","),
+		),
+	))
+	if err != nil {
+		return nil, errors.Wrap(err, "unmarshal config")
+	}
+
+	cfg.MongoDB.TargetCompressors = filterCompressors(cfg.MongoDB.TargetCompressors)
+
+	if viper.GetBool("no-color") {
+		cfg.Log.NoColor = true
+	}
+
+	return &cfg, nil
+}
+
+func bindEnvVars() {
+	_ = viper.BindEnv("port", "PCSM_PORT")
+
+	_ = viper.BindEnv("source", "PCSM_SOURCE_URI")
+	_ = viper.BindEnv("target", "PCSM_TARGET_URI")
+
+	_ = viper.BindEnv("log-level", "PCSM_LOG_LEVEL")
+	_ = viper.BindEnv("log-json", "PCSM_LOG_JSON")
+	_ = viper.BindEnv("log-no-color", "PCSM_LOG_NO_COLOR", "PCSM_NO_COLOR")
+
+	_ = viper.BindEnv("mongodb-operation-timeout", "PCSM_MONGODB_OPERATION_TIMEOUT")
+
+	_ = viper.BindEnv("use-collection-bulk-write", "PCSM_USE_COLLECTION_BULK_WRITE")
+
+	_ = viper.BindEnv("dev-target-client-compressors", "PCSM_DEV_TARGET_CLIENT_COMPRESSORS")
+
+	_ = viper.BindEnv("clone-num-parallel-collections", "PCSM_CLONE_NUM_PARALLEL_COLLECTIONS")
+	_ = viper.BindEnv("clone-num-read-workers", "PCSM_CLONE_NUM_READ_WORKERS")
+	_ = viper.BindEnv("clone-num-insert-workers", "PCSM_CLONE_NUM_INSERT_WORKERS")
+	_ = viper.BindEnv("clone-segment-size", "PCSM_CLONE_SEGMENT_SIZE")
+	_ = viper.BindEnv("clone-read-batch-size", "PCSM_CLONE_READ_BATCH_SIZE")
+}
+
+//nolint:gochecknoglobals
+var allowedCompressors = []string{"zstd", "zlib", "snappy"}
+
+func filterCompressors(compressors []string) []string {
+	if len(compressors) == 0 {
+		return nil
+	}
+
+	filtered := make([]string, 0, len(allowedCompressors))
+
+	for _, c := range compressors {
+		c = strings.TrimSpace(c)
+		if slices.Contains(allowedCompressors, c) && !slices.Contains(filtered, c) {
+			filtered = append(filtered, c)
+		}
+	}
+
+	return filtered
+}
+
+// ParseAndValidateCloneSegmentSize parses a byte size string and validates it.
+// It allows 0 (auto) or values within [MinCloneSegmentSizeBytes, MaxCloneSegmentSizeBytes].
+func ParseAndValidateCloneSegmentSize(value string) (int64, error) {
+	sizeBytes, err := humanize.ParseBytes(value)
+	if err != nil {
+		return 0, errors.Wrapf(err, "invalid cloneSegmentSize value: %s", value)
+	}
+
+	err = ValidateCloneSegmentSize(sizeBytes)
+	if err != nil {
+		return 0, err
+	}
+
+	return int64(min(sizeBytes, math.MaxInt64)), nil //nolint:gosec
+}
+
+// ParseAndValidateCloneReadBatchSize parses a byte size string and validates it.
+// It allows 0 (auto) or values within [[MinCloneReadBatchSizeBytes], [MaxCloneReadBatchSizeBytes]].
+func ParseAndValidateCloneReadBatchSize(value string) (int32, error) {
+	sizeBytes, err := humanize.ParseBytes(value)
+	if err != nil {
+		return 0, errors.Wrapf(err, "invalid cloneReadBatchSize value: %s", value)
+	}
+
+	err = ValidateCloneReadBatchSize(sizeBytes)
+	if err != nil {
+		return 0, err
+	}
+
+	return int32(min(sizeBytes, math.MaxInt32)), nil //nolint:gosec
+}