refactor(source/wrappers): move wrappers logic away from execute file

controller/execute.go

@@ -446,24 +446,14 @@ func buildSource(ctx context.Context, cfg *externaldns.Config) (source.Source, e
 	if err != nil {
 		return nil, err
 	}
-	// Combine multiple sources into a single, deduplicated source.
-	combinedSource := wrappers.NewDedupSource(wrappers.NewMultiSource(sources, sourceCfg.DefaultTargets, sourceCfg.ForceDefaultTargets))
-	cfg.AddSourceWrapper("dedup")
-	if len(cfg.NAT64Networks) > 0 {
-		combinedSource, err = wrappers.NewNAT64Source(combinedSource, cfg.NAT64Networks)
-		if err != nil {
-			return nil, fmt.Errorf("failed to create NAT64 source wrapper: %w", err)
-		}
-		cfg.AddSourceWrapper("nat64")
-	}
-	// Filter targets
-	targetFilter := endpoint.NewTargetNetFilterWithExclusions(cfg.TargetNetFilter, cfg.ExcludeTargetNets)
-	if targetFilter.IsEnabled() {
-		combinedSource = wrappers.NewTargetFilterSource(combinedSource, targetFilter)
-		cfg.AddSourceWrapper("target-filter")
-	}
-	combinedSource = wrappers.NewPostProcessor(combinedSource, wrappers.WithTTL(cfg.MinTTL))
-	return combinedSource, nil
+	opts := wrappers.NewConfig(
+		wrappers.WithDefaultTargets(cfg.DefaultTargets),
+		wrappers.WithForceDefaultTargets(cfg.ForceDefaultTargets),
+		wrappers.WithNAT64Networks(cfg.NAT64Networks),
+		wrappers.WithTargetNetFilter(cfg.TargetNetFilter),
+		wrappers.WithExcludeTargetNets(cfg.ExcludeTargetNets),
+		wrappers.WithMinTTL(cfg.MinTTL))
+	return wrappers.WrapSources(sources, opts)
 }
 
 // RegexDomainFilter overrides DomainFilter

controller/execute_test.go

@@ -449,9 +449,6 @@ func TestBuildSourceWithWrappers(t *testing.T) {
 				Sources:         []string{"fake"},
 				TargetNetFilter: []string{"10.0.0.0/8"},
 			},
-			asserts: func(t *testing.T, cfg *externaldns.Config) {
-				assert.True(t, cfg.IsSourceWrapperInstrumented("target-filter"))
-			},
 		},
 		{
 			name: "configuration with nat64 networks",
@@ -460,29 +457,20 @@ func TestBuildSourceWithWrappers(t *testing.T) {
 				Sources:       []string{"fake"},
 				NAT64Networks: []string{"2001:db8::/96"},
 			},
-			asserts: func(t *testing.T, cfg *externaldns.Config) {
-				assert.True(t, cfg.IsSourceWrapperInstrumented("nat64"))
-			},
 		},
 		{
 			name: "default configuration",
 			cfg: &externaldns.Config{
 				APIServerURL: svr.URL,
 				Sources:      []string{"fake"},
 			},
-			asserts: func(t *testing.T, cfg *externaldns.Config) {
-				assert.True(t, cfg.IsSourceWrapperInstrumented("dedup"))
-				assert.False(t, cfg.IsSourceWrapperInstrumented("nat64"))
-				assert.False(t, cfg.IsSourceWrapperInstrumented("target-filter"))
-			},
 		},
 	}
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			_, err := buildSource(t.Context(), tt.cfg)
 			require.NoError(t, err)
-			tt.asserts(t, tt.cfg)
 		})
 	}
 }

pkg/apis/externaldns/types.go

@@ -217,7 +217,6 @@ type Config struct {
 	ExcludeUnschedulable                          bool
 	EmitEvents                                    []string
 	ForceDefaultTargets                           bool
-	sourceWrappers                                map[string]bool // map of source wrappers, e.g. "targetfilter", "nat64"
 }
 
 var defaultConfig = &Config{
@@ -383,7 +382,6 @@ var defaultConfig = &Config{
 	WebhookServer:                false,
 	ZoneIDFilter:                 []string{},
 	ForceDefaultTargets:          false,
-	sourceWrappers:               map[string]bool{},
 }
 
 var providerNames = []string{
@@ -808,22 +806,6 @@ func bindFlags(b FlagBinder, cfg *Config) {
 	b.BoolVar("webhook-server", "When enabled, runs as a webhook server instead of a controller. (default: false).", defaultConfig.WebhookServer, &cfg.WebhookServer)
 }
 
-func (cfg *Config) AddSourceWrapper(name string) {
-	if cfg.sourceWrappers == nil {
-		cfg.sourceWrappers = make(map[string]bool)
-	}
-	cfg.sourceWrappers[name] = true
-}
-
-// IsSourceWrapperInstrumented returns whether a source wrapper is enabled or not.
-func (cfg *Config) IsSourceWrapperInstrumented(name string) bool {
-	if cfg.sourceWrappers == nil {
-		return false
-	}
-	_, ok := cfg.sourceWrappers[name]
-	return ok
-}
-
 func App(cfg *Config) *kingpin.Application {
 	app := kingpin.New("external-dns", "ExternalDNS synchronizes exposed Kubernetes Services and Ingresses with DNS providers.\n\nNote that all flags may be replaced with env vars - `--flag` -> `EXTERNAL_DNS_FLAG=1` or `--flag value` -> `EXTERNAL_DNS_FLAG=value`")
 	app.Version(Version)

pkg/apis/externaldns/types_test.go

@@ -1079,13 +1079,6 @@ func TestNewCobraCommandValidationValid(t *testing.T) {
 	require.NoError(t, err)
 }
 
-func TestSourceWrapperHelpers(t *testing.T) {
-	cfg := NewConfig()
-	assert.False(t, cfg.IsSourceWrapperInstrumented("nat64"))
-	cfg.AddSourceWrapper("nat64")
-	assert.True(t, cfg.IsSourceWrapperInstrumented("nat64"))
-}
-
 // Accepted binder/backend differences:
 // - Enum validation
 // - Boolean negation form

source/wrappers/types.go

@@ -0,0 +1,126 @@
+/*
+Copyright 2025 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package wrappers
+
+import (
+	"fmt"
+	"time"
+
+	"sigs.k8s.io/external-dns/endpoint"
+	"sigs.k8s.io/external-dns/source"
+)
+
+type Config struct {
+	defaultTargets      []string
+	forceDefaultTargets bool
+	nat64Networks       []string
+	targetNetFilter     []string
+	excludeTargetNets   []string
+	minTTL              time.Duration
+	sourceWrappers      map[string]bool // map of source wrappers, e.g. "targetfilter", "nat64"
+}
+
+func NewConfig(opts ...Option) *Config {
+	o := &Config{}
+	for _, opt := range opts {
+		opt(o)
+	}
+	return o
+}
+
+type Option func(config *Config)
+
+func WithDefaultTargets(input []string) Option {
+	return func(o *Config) {
+		o.defaultTargets = input
+	}
+}
+
+func WithForceDefaultTargets(input bool) Option {
+	return func(o *Config) {
+		o.forceDefaultTargets = input
+	}
+}
+
+func WithNAT64Networks(input []string) Option {
+	return func(o *Config) {
+		o.nat64Networks = input
+	}
+}
+
+func WithTargetNetFilter(input []string) Option {
+	return func(o *Config) {
+		o.targetNetFilter = input
+	}
+}
+
+func WithExcludeTargetNets(input []string) Option {
+	return func(o *Config) {
+		o.excludeTargetNets = input
+	}
+}
+
+func WithMinTTL(ttl time.Duration) Option {
+	return func(o *Config) {
+		o.minTTL = ttl
+	}
+}
+
+// addSourceWrapper registers a source wrapper by name in the Config.
+// It initializes the sourceWrappers map if it is nil.
+func (o *Config) addSourceWrapper(name string) {
+	if o.sourceWrappers == nil {
+		o.sourceWrappers = make(map[string]bool)
+	}
+	o.sourceWrappers[name] = true
+}
+
+// isSourceWrapperInstrumented returns whether a source wrapper is enabled or not.
+func (o *Config) isSourceWrapperInstrumented(name string) bool {
+	if o.sourceWrappers == nil {
+		return false
+	}
+	_, ok := o.sourceWrappers[name]
+	return ok
+}
+
+// WrapSources combines multiple sources into a single source,
+// applies optional NAT64 and target network filtering wrappers, and sets a minimum TTL.
+// It registers each applied wrapper in the Config for instrumentation.
+func WrapSources(
+	sources []source.Source,
+	opts *Config,
+) (source.Source, error) {
+	combinedSource := NewDedupSource(NewMultiSource(sources, opts.defaultTargets, opts.forceDefaultTargets))
+	opts.addSourceWrapper("dedup")
+	if len(opts.nat64Networks) > 0 {
+		var err error
+		combinedSource, err = NewNAT64Source(combinedSource, opts.nat64Networks)
+		if err != nil {
+			return nil, fmt.Errorf("failed to create NAT64 source wrapper: %w", err)
+		}
+		opts.addSourceWrapper("nat64")
+	}
+	targetFilter := endpoint.NewTargetNetFilterWithExclusions(opts.targetNetFilter, opts.excludeTargetNets)
+	if targetFilter.IsEnabled() {
+		combinedSource = NewTargetFilterSource(combinedSource, targetFilter)
+		opts.addSourceWrapper("target-filter")
+	}
+	combinedSource = NewPostProcessor(combinedSource, WithTTL(opts.minTTL))
+	opts.addSourceWrapper("post-processor")
+	return combinedSource, nil
+}

source/wrappers/types_test.go

@@ -0,0 +1,142 @@
+/*
+Copyright 2025 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package wrappers
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestBuildSourceWithWrappers(t *testing.T) {
+	tests := []struct {
+		name    string
+		cfg     *Config
+		asserts func(*testing.T, *Config)
+	}{
+		{
+			name: "configuration with target filter wrapper",
+			cfg: NewConfig(
+				WithTargetNetFilter([]string{"10.0.0.0/8"}),
+			),
+			asserts: func(t *testing.T, cfg *Config) {
+				assert.True(t, cfg.isSourceWrapperInstrumented("target-filter"))
+			},
+		},
+		{
+			name: "configuration with nat64 networks",
+			cfg: NewConfig(
+				WithNAT64Networks([]string{"2001:db8::/96"}),
+			),
+			asserts: func(t *testing.T, cfg *Config) {
+				assert.True(t, cfg.isSourceWrapperInstrumented("nat64"))
+			},
+		},
+		{
+			name: "default configuration",
+			cfg:  NewConfig(),
+			asserts: func(t *testing.T, cfg *Config) {
+				assert.True(t, cfg.isSourceWrapperInstrumented("dedup"))
+				assert.False(t, cfg.isSourceWrapperInstrumented("nat64"))
+				assert.False(t, cfg.isSourceWrapperInstrumented("target-filter"))
+			},
+		},
+		{
+			name: "with TTL and NAT64",
+			cfg: NewConfig(
+				WithMinTTL(300),
+				WithNAT64Networks([]string{"2001:db8::/96"}),
+			),
+			asserts: func(t *testing.T, cfg *Config) {
+				assert.True(t, cfg.isSourceWrapperInstrumented("dedup"))
+				assert.True(t, cfg.isSourceWrapperInstrumented("nat64"))
+				assert.True(t, cfg.isSourceWrapperInstrumented("post-processor"))
+				assert.False(t, cfg.isSourceWrapperInstrumented("target-filter"))
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			_, err := WrapSources(nil, tt.cfg)
+			require.NoError(t, err)
+			tt.asserts(t, tt.cfg)
+		})
+	}
+}
+
+func TestWrapSources_NAT64Error(t *testing.T) {
+	cfg := NewConfig(WithNAT64Networks([]string{"badnet"}))
+	src, err := WrapSources(nil, cfg)
+	assert.Nil(t, src)
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "failed to create NAT64 source wrapper")
+}
+
+func TestWithDefaultTargets(t *testing.T) {
+	cfg := &Config{}
+	opt := WithDefaultTargets([]string{"1.2.3.4"})
+	opt(cfg)
+	assert.Equal(t, []string{"1.2.3.4"}, cfg.defaultTargets)
+}
+
+func TestWithForceDefaultTargets(t *testing.T) {
+	cfg := &Config{}
+	opt := WithForceDefaultTargets(true)
+	opt(cfg)
+	assert.True(t, cfg.forceDefaultTargets)
+}
+
+func TestWithNAT64Networks(t *testing.T) {
+	cfg := &Config{}
+	opt := WithNAT64Networks([]string{"2001:db8::/96"})
+	opt(cfg)
+	assert.Equal(t, []string{"2001:db8::/96"}, cfg.nat64Networks)
+}
+
+func TestWithTargetNetFilter(t *testing.T) {
+	cfg := &Config{}
+	opt := WithTargetNetFilter([]string{"10.0.0.0/8"})
+	opt(cfg)
+	assert.Equal(t, []string{"10.0.0.0/8"}, cfg.targetNetFilter)
+}
+
+func TestWithExcludeTargetNets(t *testing.T) {
+	cfg := &Config{}
+	opt := WithExcludeTargetNets([]string{"192.168.0.0/16"})
+	opt(cfg)
+	assert.Equal(t, []string{"192.168.0.0/16"}, cfg.excludeTargetNets)
+}
+
+func TestWithMinTTL(t *testing.T) {
+	cfg := &Config{}
+	opt := WithMinTTL(300 * time.Second)
+	opt(cfg)
+	assert.Equal(t, 300*time.Second, cfg.minTTL)
+}
+
+func TestAddSourceWrapperAndIsSourceWrapperInstrumented(t *testing.T) {
+	cfg := &Config{}
+	assert.False(t, cfg.isSourceWrapperInstrumented("dedup"))
+	cfg.addSourceWrapper("dedup")
+	assert.True(t, cfg.isSourceWrapperInstrumented("dedup"))
+	cfg.addSourceWrapper("nat64")
+	assert.True(t, cfg.isSourceWrapperInstrumented("nat64"))
+	assert.False(t, cfg.isSourceWrapperInstrumented("target-filter"))
+}