Merge branch 'main' into release-0.7.0

Author: inelpandzic

.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"
+  ]
 }

config/config.go

@@ -0,0 +1,205 @@
+// Package config provides configuration management for PCSM using Viper.
+package config
+
+import (
+	"context"
+	"math"
+	"os"
+	"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"
+	"github.com/percona/percona-clustersync-mongodb/log"
+)
+
+// 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 populates the provided Config.
+func Load(cmd *cobra.Command, cfg *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()
+
+	err := viper.Unmarshal(cfg, viper.DecodeHook(
+		mapstructure.ComposeDecodeHookFunc(
+			mapstructure.StringToTimeDurationHookFunc(),
+			mapstructure.StringToSliceHookFunc(","),
+		),
+	))
+	if err != nil {
+		return errors.Wrap(err, "unmarshal config")
+	}
+
+	cfg.MongoDB.TargetCompressors = filterCompressors(cfg.MongoDB.TargetCompressors)
+
+	if viper.GetBool("no-color") {
+		cfg.Log.NoColor = true
+	}
+
+	return nil
+}
+
+// WarnDeprecatedEnvVars logs warnings for any deprecated environment variables that are set.
+// Expects the logger to be initialized.
+func WarnDeprecatedEnvVars(ctx context.Context) {
+	deprecated := map[string]string{
+		"PLM_MONGODB_CLI_OPERATION_TIMEOUT": "PCSM_MONGODB_OPERATION_TIMEOUT",
+		"PCSM_NO_COLOR":                     "PCSM_LOG_NO_COLOR",
+	}
+
+	for old, replacement := range deprecated {
+		if _, ok := os.LookupEnv(old); ok {
+			log.Ctx(ctx).Warnf(
+				"Environment variable %s is deprecated; use %s instead",
+				old, replacement,
+			)
+		}
+	}
+}
+
+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", // deprecated
+	)
+
+	_ = viper.BindEnv("mongodb-operation-timeout",
+		"PCSM_MONGODB_OPERATION_TIMEOUT",
+		"PLM_MONGODB_CLI_OPERATION_TIMEOUT", // deprecated
+	)
+
+	_ = 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
+}

config/config_test.go

@@ -0,0 +1,148 @@
+package config_test
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/dustin/go-humanize"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/percona/percona-clustersync-mongodb/config"
+)
+
+func TestParseAndValidateCloneSegmentSize(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name    string
+		value   string
+		want    int64
+		wantErr string
+	}{
+		{
+			name:  "valid size 500MB (above minimum)",
+			value: "500MB",
+			want:  500 * humanize.MByte,
+		},
+		{
+			name:  "valid size 1GiB",
+			value: "1GiB",
+			want:  humanize.GiByte,
+		},
+		{
+			name:  "zero value (auto)",
+			value: "0",
+			want:  0,
+		},
+		{
+			name:    "below minimum (100MB)",
+			value:   "100MB",
+			wantErr: "cloneSegmentSize must be at least",
+		},
+		{
+			name:    "above maximum",
+			value:   "100GiB",
+			wantErr: "cloneSegmentSize must be at most",
+		},
+		{
+			name:  "at minimum boundary (using exact bytes)",
+			value: fmt.Sprintf("%dB", config.MinCloneSegmentSizeBytes),
+			want:  int64(config.MinCloneSegmentSizeBytes),
+		},
+		{
+			name:  "at maximum boundary",
+			value: "64GiB",
+			want:  int64(config.MaxCloneSegmentSizeBytes),
+		},
+		{
+			name:    "invalid format",
+			value:   "abc",
+			wantErr: "invalid cloneSegmentSize value",
+		},
+		{
+			name:    "empty string",
+			value:   "",
+			wantErr: "invalid cloneSegmentSize value",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			got, err := config.ParseAndValidateCloneSegmentSize(tt.value)
+
+			if tt.wantErr == "" {
+				require.NoError(t, err)
+				assert.Equal(t, tt.want, got)
+			} else {
+				require.Error(t, err)
+				assert.Contains(t, err.Error(), tt.wantErr)
+			}
+		})
+	}
+}
+
+func TestParseAndValidateCloneReadBatchSize(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name    string
+		value   string
+		want    int32
+		wantErr string
+	}{
+		{
+			name:  "valid size 16MiB",
+			value: "16MiB",
+			want:  16 * humanize.MiByte,
+		},
+		{
+			name:  "valid size 48MB",
+			value: "48MB",
+			want:  48 * humanize.MByte,
+		},
+		{
+			name:  "zero value (auto)",
+			value: "0",
+			want:  0,
+		},
+		{
+			name:    "below minimum",
+			value:   "1KB",
+			wantErr: "cloneReadBatchSize must be at least",
+		},
+		{
+			name:  "at minimum boundary (using exact bytes)",
+			value: fmt.Sprintf("%dB", config.MinCloneReadBatchSizeBytes),
+			want:  config.MinCloneReadBatchSizeBytes,
+		},
+		{
+			name:    "invalid format",
+			value:   "xyz",
+			wantErr: "invalid cloneReadBatchSize value",
+		},
+		{
+			name:    "empty string",
+			value:   "",
+			wantErr: "invalid cloneReadBatchSize value",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			got, err := config.ParseAndValidateCloneReadBatchSize(tt.value)
+
+			if tt.wantErr == "" {
+				require.NoError(t, err)
+				assert.Equal(t, tt.want, got)
+			} else {
+				require.Error(t, err)
+				assert.Contains(t, err.Error(), tt.wantErr)
+			}
+		})
+	}
+}