fix(source/wrappers/events): events not triggered (#5687)

Signed-off-by: ivan katliarchuk <[email protected]>

Author: ivankatliarchuk

controller/execute.go

@@ -426,10 +426,15 @@ func buildSource(ctx context.Context, cfg *externaldns.Config) (source.Source, e
 	}
 	// Combine multiple sources into a single, deduplicated source.
 	combinedSource := wrappers.NewDedupSource(wrappers.NewMultiSource(sources, sourceCfg.DefaultTargets, sourceCfg.ForceDefaultTargets))
+	cfg.AddSourceWrapper("dedup")
+	combinedSource = wrappers.NewNAT64Source(combinedSource, cfg.NAT64Networks)
+	cfg.AddSourceWrapper("nat64")
 	// Filter targets
 	targetFilter := endpoint.NewTargetNetFilterWithExclusions(cfg.TargetNetFilter, cfg.ExcludeTargetNets)
-	combinedSource = wrappers.NewNAT64Source(combinedSource, cfg.NAT64Networks)
-	combinedSource = wrappers.NewTargetFilterSource(combinedSource, targetFilter)
+	if targetFilter.IsEnabled() {
+		combinedSource = wrappers.NewTargetFilterSource(combinedSource, targetFilter)
+		cfg.AddSourceWrapper("target-filter")
+	}
 	return combinedSource, nil
 }
 

controller/execute_test.go

@@ -18,7 +18,6 @@ package controller
 
 import (
 	"bytes"
-	"context"
 	"fmt"
 	"net"
 	"net/http"
@@ -36,8 +35,8 @@ import (
 	"github.com/stretchr/testify/require"
 	"sigs.k8s.io/external-dns/endpoint"
 	"sigs.k8s.io/external-dns/pkg/apis/externaldns"
-	"sigs.k8s.io/external-dns/plan"
 	"sigs.k8s.io/external-dns/provider"
+	fakeprovider "sigs.k8s.io/external-dns/provider/fakes"
 )
 
 func TestSelectRegistry(t *testing.T) {
@@ -60,7 +59,7 @@ func TestSelectRegistry(t *testing.T) {
 				ExcludeDNSRecordTypes:  []string{"TXT"},
 				TXTCacheInterval:       60,
 			},
-			provider: &MockProvider{},
+			provider: &fakeprovider.MockProvider{},
 			wantErr:  false,
 			wantType: "DynamoDBRegistry",
 		},
@@ -69,7 +68,7 @@ func TestSelectRegistry(t *testing.T) {
 			cfg: &externaldns.Config{
 				Registry: "noop",
 			},
-			provider: &MockProvider{},
+			provider: &fakeprovider.MockProvider{},
 			wantErr:  false,
 			wantType: "NoopRegistry",
 		},
@@ -84,7 +83,7 @@ func TestSelectRegistry(t *testing.T) {
 				ManagedDNSRecordTypes:  []string{"A", "CNAME"},
 				ExcludeDNSRecordTypes:  []string{"TXT"},
 			},
-			provider: &MockProvider{},
+			provider: &fakeprovider.MockProvider{},
 			wantErr:  false,
 			wantType: "TXTRegistry",
 		},
@@ -94,7 +93,7 @@ func TestSelectRegistry(t *testing.T) {
 				Registry:   "aws-sd",
 				TXTOwnerID: "owner-id",
 			},
-			provider: &MockProvider{},
+			provider: &fakeprovider.MockProvider{},
 			wantErr:  false,
 			wantType: "AWSSDRegistry",
 		},
@@ -103,7 +102,7 @@ func TestSelectRegistry(t *testing.T) {
 			cfg: &externaldns.Config{
 				Registry: "unknown",
 			},
-			provider: &MockProvider{},
+			provider: &fakeprovider.MockProvider{},
 			wantErr:  true,
 			wantType: "",
 		},
@@ -477,21 +476,47 @@ func TestBuildSource(t *testing.T) {
 	}
 }
 
-// mocks
-type MockProvider struct{}
-
-func (m *MockProvider) Records(ctx context.Context) ([]*endpoint.Endpoint, error) {
-	return nil, nil
-}
-
-func (p *MockProvider) ApplyChanges(ctx context.Context, changes *plan.Changes) error {
-	return nil
-}
+func TestBuildSourceWithWrappers(t *testing.T) {
+	svr := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusNotImplemented)
+	}))
+	defer svr.Close()
 
-func (m *MockProvider) AdjustEndpoints(endpoints []*endpoint.Endpoint) ([]*endpoint.Endpoint, error) {
-	return nil, nil
-}
+	tests := []struct {
+		name    string
+		cfg     *externaldns.Config
+		asserts func(*externaldns.Config)
+	}{
+		{
+			name: "configuration with target filter wrapper",
+			cfg: &externaldns.Config{
+				APIServerURL:    svr.URL,
+				Sources:         []string{"fake"},
+				TargetNetFilter: []string{"10.0.0.0/8"},
+			},
+			asserts: func(cfg *externaldns.Config) {
+				assert.True(t, cfg.IsSourceWrapperInstrumented("target-filter"))
+			},
+		},
+		{
+			name: "configuration without target filter wrapper",
+			cfg: &externaldns.Config{
+				APIServerURL: svr.URL,
+				Sources:      []string{"fake"},
+			},
+			asserts: func(cfg *externaldns.Config) {
+				assert.True(t, cfg.IsSourceWrapperInstrumented("dedup"))
+				assert.True(t, cfg.IsSourceWrapperInstrumented("nat64"))
+				assert.False(t, cfg.IsSourceWrapperInstrumented("target-filter"))
+			},
+		},
+	}
 
-func (m *MockProvider) GetDomainFilter() endpoint.DomainFilterInterface {
-	return nil
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			_, err := buildSource(t.Context(), tt.cfg)
+			require.NoError(t, err)
+			tt.asserts(tt.cfg)
+		})
+	}
 }

docs/contributing/source-wrappers.md

@@ -0,0 +1,149 @@
+# 🧩 Source Wrappers/Middleware
+
+## Overview
+
+In ExternalDNS, a **Source** is a component responsible for discovering DNS records from Kubernetes resources (e.g., `Ingress`, `Service`, `Gateway`, etc.).
+
+**Source Wrappers** are middleware-like components that sit between the source and the plan generation. They extend or modify the behavior of the original sources by transforming, filtering, or enriching the DNS records before they're processed by the planner and provider.
+
+---
+
+## Why Wrappers?
+
+Wrappers solve these key challenges:
+
+- ✂️ **Filtering**: Remove unwanted targets or records from sources based on labels, annotations, targets and etc.
+- 🔗 **Aggregation**: Combine Endpoints from multiple underlying sources. For example, from both Kubernetes Services and Ingresses.
+- 🧹 **Deduplication**: Prevent duplicate DNS records across sources.
+- 🌐 **Target transformation**: Rewrite targets for IPv6 networks or alter endpoint attributes like FQDNS or targets.
+- 🧪 **Testing and simulation**: Use the `FakeSource` or wrappers for dry-runs or simulations.
+- 🔁 **Composability**: Chain multiple behaviors without modifying core sources.
+- 🔐 **Access Control**: Limits endpoint exposure based on policies or user access.
+- 📊 **Observability**: Adds logging, debugging, or metrics around source behavior.
+
+---
+
+## Built In Wrappers
+
+|       Wrapper        | Purpose                                 | Use Case                              |
+|:--------------------:|:----------------------------------------|:--------------------------------------|
+|    `MultiSource`     | Combine multiple sources.               | Aggregate `Ingress`, `Service`, etc.  |
+|    `DedupSource`     | Remove duplicate DNS records.           | Avoid duplicate records from sources. |
+| `TargetFilterSource` | Include/exclude targets based on CIDRs. | Exclude internal IPs.                 |
+|    `NAT64Source`     | Add NAT64-prefixed AAAA records.        | Support IPv6 with NAT64.              |
+
+### Use Cases
+
+### 1.1 `TargetFilterSource`
+
+Filters targets (e.g. IPs or hostnames) based on inclusion or exclusion rules.
+
+📌 **Use case**: Only publish public IPs, exclude test environments.
+
+```yaml
+--target-net-filter=192.168.0.0/16
+--exclude-target-nets=10.0.0.0/8
+```
+
+### 2.1 `NAT64Source`
+
+Converts IPv4 targets to IPv6 using NAT64 prefixes.
+
+📌 **Use case**: Publish AAAA records for IPv6-only clients in NAT64 environments.
+
+```yaml
+--nat64-prefix=64:ff9b::/96
+```
+
+---
+
+## How Wrappers Work
+
+Wrappers wrap a `Source` and implement the same `Source` interface (e.g., `Endpoints(ctx)`).
+
+They typically follow this pattern:
+
+```go
+package wrappers
+
+type myWrapper struct {
+	next source.Source
+}
+
+func (m *myWrapper) Endpoints(ctx context.Context) ([]*endpoint.Endpoint, error) {
+	eps, err := m.next.Endpoints(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	// Modify, filter, or enrich endpoints as needed
+	return eps, nil
+}
+
+// AddEventHandler must be implemented to satisfy the source.Source interface.
+func (m *myWrapper) AddEventHandler(ctx context.Context, handler func()) {
+	log.Debugf("myWrapper: adding event handler")
+	m.next.AddEventHandler(ctx, handler)
+}
+```
+
+This allows wrappers to be stacked or composed together.
+
+---
+
+### Composition of Wrappers
+
+Wrappers are often composed like this:
+
+```go
+source := NewMultiSource(actualSources, defaultTargets)
+source = NewDedupSource(source)
+source = NewNAT64Source(source, cfg.NAT64Networks)
+source = NewTargetFilterSource(source, targetFilter)
+```
+
+Each wrapper processes the output of the previous one.
+
+---
+
+## High Level Design
+
+- Source: Implements the base logic for extracting DNS endpoints (e.g. IngressSource, ServiceSource, etc.)
+- Wrappers: Decorate the source (e.g. DedupSource, TargetFilterSource) to enhance or filter endpoint data
+- Plan: Compares the endpoints from Source with DNS state from Provider and produces create/update/delete changes
+- Provider: Applies changes to actual DNS services (e.g. Route53, Cloudflare, Azure DNS)
+
+```mermaid
+sequenceDiagram
+    participant ExternalDNS
+    participant Source
+    participant Wrapper
+    participant DedupWrapper as DedupSource
+    participant Provider
+    participant Plan
+
+    ExternalDNS->>Source: Initialize source (e.g. Ingress, Service)
+    Source-->>ExternalDNS: Implements Source interface
+
+    ExternalDNS->>Wrapper: Wrap with decorators (e.g. dedup, filters)
+    Wrapper->>DedupWrapper: Compose with DedupSource
+    DedupWrapper-->>Wrapper: Return enriched Source
+
+    Wrapper-->>ExternalDNS: Return final wrapped Source
+
+    ExternalDNS->>Plan: Generate plan from Source
+    Plan->>Wrapper: Call Endpoints(ctx)
+    Wrapper->>DedupWrapper: Call Endpoints(ctx)
+    DedupWrapper->>Source: Call Endpoints(ctx)
+    Source-->>DedupWrapper: Return []*Endpoint
+    DedupWrapper-->>Wrapper: Return de-duplicated []*Endpoint
+    Wrapper-->>Plan: Return transformed []*Endpoint
+
+    ExternalDNS->>Provider: ApplyChanges(plan)
+    Provider-->>ExternalDNS: Sync DNS records
+```
+
+## Learn More
+
+- [Source Interface](https://github.com/kubernetes-sigs/external-dns/blob/master/source/source.go)
+- [Wrappers Source Code](https://github.com/kubernetes-sigs/external-dns/tree/master/source/wrappers)

internal/testutils/mock_source.go

@@ -28,10 +28,20 @@ import (
 // MockSource returns mock endpoints.
 type MockSource struct {
 	mock.Mock
+	endpoints []*endpoint.Endpoint
+}
+
+func NewMockSource(endpoints ...*endpoint.Endpoint) *MockSource {
+	m := &MockSource{
+		endpoints: endpoints,
+	}
+	m.On("Endpoints").Return(endpoints, nil)
+	m.On("AddEventHandler", mock.AnythingOfType("*context.cancelCtx")).Return()
+	return m
 }
 
 // Endpoints returns the desired mock endpoints.
-func (m *MockSource) Endpoints(ctx context.Context) ([]*endpoint.Endpoint, error) {
+func (m *MockSource) Endpoints(_ context.Context) ([]*endpoint.Endpoint, error) {
 	args := m.Called()
 
 	endpoints := args.Get(0)
@@ -44,6 +54,10 @@ func (m *MockSource) Endpoints(ctx context.Context) ([]*endpoint.Endpoint, error
 
 // AddEventHandler adds an event handler that should be triggered if something in source changes
 func (m *MockSource) AddEventHandler(ctx context.Context, handler func()) {
+	m.Called(ctx)
+	if handler == nil {
+		return
+	}
 	go func() {
 		ticker := time.NewTicker(5 * time.Second)
 		defer ticker.Stop()

pkg/apis/externaldns/types.go

@@ -213,6 +213,7 @@ type Config struct {
 	NAT64Networks                                 []string
 	ExcludeUnschedulable                          bool
 	ForceDefaultTargets                           bool
+	sourceWrappers                                map[string]bool // map of source wrappers, e.g. "targetfilter", "nat64"
 }
 
 var defaultConfig = &Config{
@@ -376,6 +377,7 @@ var defaultConfig = &Config{
 	WebhookServer:                false,
 	ZoneIDFilter:                 []string{},
 	ForceDefaultTargets:          false,
+	sourceWrappers:               map[string]bool{},
 }
 
 // NewConfig returns new Config object
@@ -427,6 +429,22 @@ func (cfg *Config) ParseFlags(args []string) error {
 	return nil
 }
 
+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)