PCSM-219: Integrate target compressors into Viper config

Author: chupe

config/config.go

@@ -3,7 +3,6 @@ package config
 
 import (
 	"math"
-	"os"
 	"slices"
 	"strings"
 
@@ -36,11 +35,11 @@ func Load(cmd *cobra.Command) (*Config, error) {
 		return nil, errors.Wrap(err, "unmarshal config")
 	}
 
+	cfg.MongoDB.TargetCompressors = parseCompressors(cfg.MongoDB.TargetCompressorsRaw)
+
 	return &cfg, nil
 }
 
-// bindEnvVars binds environment variable names to Viper keys.
-// Note: Clone tuning options are CLI/HTTP only (no env var support).
 func bindEnvVars() {
 	_ = viper.BindEnv("port", "PCSM_PORT")
 
@@ -54,23 +53,23 @@ func bindEnvVars() {
 	_ = 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")
 }
 
-// UseTargetClientCompressors returns a list of enabled compressors (from "zstd", "zlib", "snappy")
-// for the target MongoDB client connection, as specified by the comma-separated environment
-// variable PCSM_DEV_TARGET_CLIENT_COMPRESSORS. If unset or empty, returns nil.
-func UseTargetClientCompressors() []string {
-	s := strings.TrimSpace(os.Getenv("PCSM_DEV_TARGET_CLIENT_COMPRESSORS"))
+//nolint:gochecknoglobals
+var allowedCompressors = []string{"zstd", "zlib", "snappy"}
+
+func parseCompressors(raw string) []string {
+	s := strings.TrimSpace(raw)
 	if s == "" {
 		return nil
 	}
 
-	allowCompressors := []string{"zstd", "zlib", "snappy"}
-
-	rv := make([]string, 0, min(len(s), len(allowCompressors)))
+	rv := make([]string, 0, len(allowedCompressors))
 	for a := range strings.SplitSeq(s, ",") {
 		a = strings.TrimSpace(a)
-		if slices.Contains(allowCompressors, a) && !slices.Contains(rv, a) {
+		if slices.Contains(allowedCompressors, a) && !slices.Contains(rv, a) {
 			rv = append(rv, a)
 		}
 	}
@@ -95,7 +94,7 @@ func ParseAndValidateCloneSegmentSize(value string) (int64, error) {
 }
 
 // ParseAndValidateCloneReadBatchSize parses a byte size string and validates it.
-// It allows 0 (auto) or values within [MinCloneReadBatchSizeBytes, MaxCloneReadBatchSizeBytes].
+// It allows 0 (auto) or values within [[MinCloneReadBatchSizeBytes], [MaxCloneReadBatchSizeBytes]].
 func ParseAndValidateCloneReadBatchSize(value string) (int32, error) {
 	sizeBytes, err := humanize.ParseBytes(value)
 	if err != nil {

config/config_test.go

@@ -146,84 +146,3 @@ func TestParseAndValidateCloneReadBatchSize(t *testing.T) {
 		})
 	}
 }
-
-func TestUseTargetClientCompressors(t *testing.T) {
-	tests := []struct {
-		name    string
-		envVal  string
-		want    []string
-		wantNil bool
-	}{
-		{
-			name:    "empty env - returns nil",
-			envVal:  "",
-			want:    nil,
-			wantNil: true,
-		},
-		{
-			name:   "single valid compressor zstd",
-			envVal: "zstd",
-			want:   []string{"zstd"},
-		},
-		{
-			name:   "single valid compressor zlib",
-			envVal: "zlib",
-			want:   []string{"zlib"},
-		},
-		{
-			name:   "single valid compressor snappy",
-			envVal: "snappy",
-			want:   []string{"snappy"},
-		},
-		{
-			name:   "multiple valid compressors",
-			envVal: "zstd,zlib,snappy",
-			want:   []string{"zstd", "zlib", "snappy"},
-		},
-		{
-			name:   "compressors with spaces",
-			envVal: " zstd , zlib , snappy ",
-			want:   []string{"zstd", "zlib", "snappy"},
-		},
-		{
-			name:   "invalid compressor ignored",
-			envVal: "zstd,invalid,zlib",
-			want:   []string{"zstd", "zlib"},
-		},
-		{
-			name:   "all invalid compressors - returns empty slice",
-			envVal: "invalid,gzip,lz4",
-			want:   []string{},
-		},
-		{
-			name:   "duplicate compressors - deduplicated",
-			envVal: "zstd,zstd,zlib,zstd",
-			want:   []string{"zstd", "zlib"},
-		},
-		{
-			name:    "whitespace only - returns nil",
-			envVal:  "   ",
-			want:    nil,
-			wantNil: true,
-		},
-		{
-			name:   "mixed valid and invalid with spaces",
-			envVal: " zstd , invalid , snappy ",
-			want:   []string{"zstd", "snappy"},
-		},
-	}
-
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			t.Setenv("PCSM_DEV_TARGET_CLIENT_COMPRESSORS", tt.envVal)
-
-			got := config.UseTargetClientCompressors()
-
-			if tt.wantNil {
-				assert.Nil(t, got)
-			} else {
-				assert.Equal(t, tt.want, got)
-			}
-		})
-	}
-}

config/schema.go

@@ -6,21 +6,17 @@ import (
 
 // Config holds all PCSM configuration.
 type Config struct {
-	// Connection
 	Port   int    `mapstructure:"port"`
 	Source string `mapstructure:"source"`
 	Target string `mapstructure:"target"`
 
-	// Logging (squash keeps flat keys)
 	Log LogConfig `mapstructure:",squash"`
 
-	// MongoDB client options
 	MongoDB MongoDBConfig `mapstructure:",squash"`
 
-	// Internal options
 	UseCollectionBulkWrite bool `mapstructure:"use-collection-bulk-write"`
 
-	// Hidden startup flags
+	// hidden startup flags
 	Start              bool `mapstructure:"start"`
 	ResetState         bool `mapstructure:"reset-state"`
 	PauseOnInitialSync bool `mapstructure:"pause-on-initial-sync"`
@@ -36,6 +32,14 @@ type LogConfig struct {
 // MongoDBConfig holds MongoDB client configuration.
 type MongoDBConfig struct {
 	OperationTimeout string `mapstructure:"mongodb-operation-timeout"`
+
+	// TargetCompressorsRaw is the raw comma-separated env var value (dev option).
+	// Parsed into TargetCompressors during Load().
+	TargetCompressorsRaw string `mapstructure:"dev-target-client-compressors"`
+
+	// TargetCompressors is the parsed list of compressors for the target connection.
+	// Populated from TargetCompressorsRaw during Load().
+	TargetCompressors []string `mapstructure:"-"`
 }
 
 // OperationTimeoutDuration returns the parsed timeout or default.

main.go

@@ -514,9 +514,7 @@ func createServer(ctx context.Context, cfg *config.Config) (*Server, error) {
 	lg.Infof("Connected to source cluster [%s]: %s://%s",
 		sourceVersion.FullString(), cs.Scheme, strings.Join(cs.Hosts, ","))
 
-	target, err := topo.ConnectWithOptions(ctx, cfg.Target, cfg, &topo.ConnectOptions{
-		Compressors: config.UseTargetClientCompressors(),
-	})
+	target, err := topo.Connect(ctx, cfg.Target, cfg)
 	if err != nil {
 		return nil, errors.Wrap(err, "connect to target cluster")
 	}

topo/connect.go

@@ -19,24 +19,8 @@ import (
 	"github.com/percona/percona-clustersync-mongodb/util"
 )
 
-type ConnectOptions struct {
-	Compressors []string
-}
-
 // Connect establishes a connection to a MongoDB instance using the provided URI.
-// If the URI is empty, it returns an error.
 func Connect(ctx context.Context, uri string, cfg *config.Config) (*mongo.Client, error) {
-	return ConnectWithOptions(ctx, uri, cfg, &ConnectOptions{})
-}
-
-// ConnectWithOptions establishes a connection to a MongoDB instance using the provided URI and options.
-// If the URI is empty, it returns an error.
-func ConnectWithOptions(
-	ctx context.Context,
-	uri string,
-	cfg *config.Config,
-	connOpts *ConnectOptions,
-) (*mongo.Client, error) {
 	if uri == "" {
 		return nil, errors.New("invalid MongoDB URI")
 	}
@@ -61,8 +45,8 @@ func ConnectWithOptions(
 		SetWriteConcern(writeconcern.Majority()).
 		SetTimeout(cfg.MongoDB.OperationTimeoutDuration())
 
-	if connOpts != nil && connOpts.Compressors != nil {
-		opts.SetCompressors(connOpts.Compressors)
+	if uri == cfg.Target && len(cfg.MongoDB.TargetCompressors) > 0 {
+		opts.SetCompressors(cfg.MongoDB.TargetCompressors)
 	}
 
 	if config.MongoLogEnabled {