Repository: kubernetes-sigs/external-dns

Issue:
Cloudflare Regional Services / Hostname are not created when cloudflare region is configured
<!-- Please use this template while reporting a bug and provide as much info as possible. Not doing so may result in your bug not being addressed in a timely manner. Thanks!

-->

**What happened**:
Setting `--cloudflare-region-key=xxx` do nothing.
Cloudflare regional hostname ([Regional Services](https://developers.cloudflare.com/data-localization/regional-services/)) are not created with new records.

**What you expected to happen**:
A regional hostname should be created for each A, AAAA & CNAME records (at least ?). Then removed when the record is deleted.

**How to reproduce it (as minimally and precisely as possible)**:
- Set `--cloudflare-region-key=xxx` (=`us` for example).
- Create and ingress that will create a DNS record in the cloudflare zone.
- Go into your DNS zone and see that the Region is set to `Earth` and not what was configured.

**Anything else we need to know?**:
In the code only update of regional hostname is implemented (https://github.com/kubernetes-sigs/external-dns/pull/4646).

**Environment**:

- External-DNS version (use `external-dns --version`): 1.15.2
- DNS provider: Cloudflare
- Others:


Discussion:

Solution patch:
diff --git a/provider/cloudflare/cloudflare.go b/provider/cloudflare/cloudflare.go
index 5a85c188..0bac3f89 100644
--- a/provider/cloudflare/cloudflare.go
+++ b/provider/cloudflare/cloudflare.go
@@ -27,7 +27,6 @@ import (
 	"sort"
 	"strconv"
 	"strings"
-	"time"
 
 	cloudflare "github.com/cloudflare/cloudflare-go"
 	log "github.com/sirupsen/logrus"
@@ -74,6 +73,11 @@ type CustomHostnameIndex struct {
 
 type CustomHostnamesMap map[CustomHostnameIndex]cloudflare.CustomHostname
 
+type DataLocalizationRegionalHostnameChange struct {
+	Action string
+	cloudflare.RegionalHostname
+}
+
 var recordTypeProxyNotSupported = map[string]bool{
 	"LOC": true,
 	"MX":  true,
@@ -94,6 +98,12 @@ var recordTypeCustomHostnameSupported = map[string]bool{
 	"CNAME": true,
 }
 
+var recordTypeRegionalHostnameSupported = map[string]bool{
+	"A":     true,
+	"AAAA":  true,
+	"CNAME": true,
+}
+
 // cloudFlareDNS is the subset of the CloudFlare API that we actually use.  Add methods as required. Signatures must match exactly.
 type cloudFlareDNS interface {
 	UserDetails(ctx context.Context) (cloudflare.User, error)
@@ -105,7 +115,9 @@ type cloudFlareDNS interface {
 	CreateDNSRecord(ctx context.Context, rc *cloudflare.ResourceContainer, rp cloudflare.CreateDNSRecordParams) (cloudflare.DNSRecord, error)
 	DeleteDNSRecord(ctx context.Context, rc *cloudflare.ResourceContainer, recordID string) error
 	UpdateDNSRecord(ctx context.Context, rc *cloudflare.ResourceContainer, rp cloudflare.UpdateDNSRecordParams) error
+	CreateDataLocalizationRegionalHostname(ctx context.Context, rc *cloudflare.ResourceContainer, rp cloudflare.CreateDataLocalizationRegionalHostnameParams) error
 	UpdateDataLocalizationRegionalHostname(ctx context.Context, rc *cloudflare.ResourceContainer, rp cloudflare.UpdateDataLocalizationRegionalHostnameParams) error
+	DeleteDataLocalizationRegionalHostname(ctx context.Context, rc *cloudflare.ResourceContainer, hostname string) error
 	CustomHostnames(ctx context.Context, zoneID string, page int, filter cloudflare.CustomHostname) ([]cloudflare.CustomHostname, cloudflare.ResultInfo, error)
 	DeleteCustomHostname(ctx context.Context, zoneID string, customHostnameID string) error
 	CreateCustomHostname(ctx context.Context, zoneID string, ch cloudflare.CustomHostname) (*cloudflare.CustomHostnameResponse, error)
@@ -140,11 +152,20 @@ func (z zoneService) UpdateDNSRecord(ctx context.Context, rc *cloudflare.Resourc
 	return err
 }
 
+func (z zoneService) CreateDataLocalizationRegionalHostname(ctx context.Context, rc *cloudflare.ResourceContainer, rp cloudflare.CreateDataLocalizationRegionalHostnameParams) error {
+	_, err := z.service.CreateDataLocalizationRegionalHostname(ctx, rc, rp)
+	return err
+}
+
 func (z zoneService) UpdateDataLocalizationRegionalHostname(ctx context.Context, rc *cloudflare.ResourceContainer, rp cloudflare.UpdateDataLocalizationRegionalHostnameParams) error {
 	_, err := z.service.UpdateDataLocalizationRegionalHostname(ctx, rc, rp)
 	return err
 }
 
+func (z zoneService) DeleteDataLocalizationRegionalHostname(ctx context.Context, rc *cloudflare.ResourceContainer, hostname string) error {
+	return z.service.DeleteDataLocalizationRegionalHostname(ctx, rc, hostname)
+}
+
 func (z zoneService) DeleteDNSRecord(ctx context.Context, rc *cloudflare.ResourceContainer, recordID string) error {
 	return z.service.DeleteDNSRecord(ctx, rc, recordID)
 }
@@ -208,11 +229,19 @@ func updateDNSRecordParam(cfc cloudFlareChange) cloudflare.UpdateDNSRecordParams
 	}
 }
 
+// createDataLocalizationRegionalHostnameParams is a function that returns the appropriate RegionalHostname Param based on the cloudFlareChange passed in
+func createDataLocalizationRegionalHostnameParams(rhc DataLocalizationRegionalHostnameChange) cloudflare.CreateDataLocalizationRegionalHostnameParams {
+	return cloudflare.CreateDataLocalizationRegionalHostnameParams{
+		Hostname:  rhc.Hostname,
+		RegionKey: rhc.RegionKey,
+	}
+}
+
 // updateDataLocalizationRegionalHostnameParams is a function that returns the appropriate RegionalHostname Param based on the cloudFlareChange passed in
-func updateDataLocalizationRegionalHostnameParams(cfc cloudFlareChange) cloudflare.UpdateDataLocalizationRegionalHostnameParams {
+func updateDataLocalizationRegionalHostnameParams(rhc DataLocalizationRegionalHostnameChange) cloudflare.UpdateDataLocalizationRegionalHostnameParams {
 	return cloudflare.UpdateDataLocalizationRegionalHostnameParams{
-		Hostname:  cfc.RegionalHostname.Hostname,
-		RegionKey: cfc.RegionalHostname.RegionKey,
+		Hostname:  rhc.Hostname,
+		RegionKey: rhc.RegionKey,
 	}
 }
 
@@ -466,6 +495,129 @@ func (p *CloudFlareProvider) submitCustomHostnameChanges(ctx context.Context, zo
 	return !failedChange
 }
 
+// submitDataLocalizationRegionalHostnameChanges applies a set of data localization regional hostname changes, returns false if it fails
+func (p *CloudFlareProvider) submitDataLocalizationRegionalHostnameChanges(ctx context.Context, changes []DataLocalizationRegionalHostnameChange, resourceContainer *cloudflare.ResourceContainer) bool {
+	failedChange := false
+
+	for _, change := range changes {
+		logFields := log.Fields{
+			"hostname":   change.Hostname,
+			"region_key": change.RegionKey,
+			"action":     change.Action,
+			"zone":       resourceContainer.Identifier,
+		}
+		log.WithFields(logFields).Info("Changing regional hostname")
+		switch change.Action {
+		case cloudFlareCreate:
+			log.WithFields(logFields).Debug("Creating regional hostname")
+			if p.DryRun {
+				continue
+			}
+			regionalHostnameParam := createDataLocalizationRegionalHostnameParams(change)
+			err := p.Client.CreateDataLocalizationRegionalHostname(ctx, resourceContainer, regionalHostnameParam)
+			if err != nil {
+				var apiErr *cloudflare.Error
+				if errors.As(err, &apiErr) && apiErr.StatusCode == http.StatusConflict {
+					log.WithFields(logFields).Debug("Regional hostname already exists, updating instead")
+					params := updateDataLocalizationRegionalHostnameParams(change)
+					err := p.Client.UpdateDataLocalizationRegionalHostname(ctx, resourceContainer, params)
+					if err != nil {
+						failedChange = true
+						log.WithFields(logFields).Errorf("failed to update regional hostname: %v", err)
+					}
+					continue
+				}
+				failedChange = true
+				log.WithFields(logFields).Errorf("failed to create regional hostname: %v", err)
+			}
+		case cloudFlareUpdate:
+			log.WithFields(logFields).Debug("Updating regional hostname")
+			if p.DryRun {
+				continue
+			}
+			regionalHostnameParam := updateDataLocalizationRegionalHostnameParams(change)
+			err := p.Client.UpdateDataLocalizationRegionalHostname(ctx, resourceContainer, regionalHostnameParam)
+			if err != nil {
+				var apiErr *cloudflare.Error
+				if errors.As(err, &apiErr) && apiErr.StatusCode == http.StatusNotFound {
+					log.WithFields(logFields).Debug("Regional hostname not does not exists, creating instead")
+					params := createDataLocalizationRegionalHostnameParams(change)
+					err := p.Client.CreateDataLocalizationRegionalHostname(ctx, resourceContainer, params)
+					if err != nil {
+						failedChange = true
+						log.WithFields(logFields).Errorf("failed to create regional hostname: %v", err)
+					}
+					continue
+				}
+				failedChange = true
+				log.WithFields(logFields).Errorf("failed to update regional hostname: %v", err)
+			}
+		case cloudFlareDelete:
+			log.WithFields(logFields).Debug("Deleting regional hostname")
+			if p.DryRun {
+				continue
+			}
+			err := p.Client.DeleteDataLocalizationRegionalHostname(ctx, resourceContainer, change.Hostname)
+			if err != nil {
+				var apiErr *cloudflare.Error
+				if errors.As(err, &apiErr) && apiErr.StatusCode == http.StatusNotFound {
+					log.WithFields(logFields).Debug("Regional hostname does not exists, nothing to do")
+					continue
+				}
+				failedChange = true
+				log.WithFields(logFields).Errorf("failed to delete regional hostname: %v", err)
+			}
+		}
+	}
+
+	return !failedChange
+}
+
+// dataLocalizationRegionalHostnamesChanges processes a slice of cloudFlare changes and consolidates them
+// into a list of data localization regional hostname changes.
+// returns nil if no changes are needed
+func dataLocalizationRegionalHostnamesChanges(changes []*cloudFlareChange) ([]DataLocalizationRegionalHostnameChange, error) {
+	regionalHostnameChanges := make(map[string]DataLocalizationRegionalHostnameChange)
+	for _, change := range changes {
+		if change.RegionalHostname.Hostname == "" {
+			continue
+		}
+		if change.RegionalHostname.RegionKey == "" {
+			return nil, fmt.Errorf("region key is empty for regional hostname %q", change.RegionalHostname.Hostname)
+		}
+		regionalHostname, ok := regionalHostnameChanges[change.RegionalHostname.Hostname]
+		switch change.Action {
+		case cloudFlareCreate, cloudFlareUpdate:
+			if !ok {
+				regionalHostnameChanges[change.RegionalHostname.Hostname] = DataLocalizationRegionalHostnameChange{
+					Action:           change.Action,
+					RegionalHostname: change.RegionalHostname,
+				}
+				continue
+			}
+			if regionalHostname.RegionKey != change.RegionalHostname.RegionKey {
+				return nil, fmt.Errorf("conflicting region keys for regional hostname %q: %q and %q", change.RegionalHostname.Hostname, regionalHostname.RegionKey, change.RegionalHostname.RegionKey)
+			}
+			if (change.Action == cloudFlareUpdate && regionalHostname.Action != cloudFlareUpdate) ||
+				regionalHostname.Action == cloudFlareDelete {
+				regionalHostnameChanges[change.RegionalHostname.Hostname] = DataLocalizationRegionalHostnameChange{
+					Action:           cloudFlareUpdate,
+					RegionalHostname: change.RegionalHostname,
+				}
+			}
+		case cloudFlareDelete:
+			if !ok {
+				regionalHostnameChanges[change.RegionalHostname.Hostname] = DataLocalizationRegionalHostnameChange{
+					Action:           cloudFlareDelete,
+					RegionalHostname: change.RegionalHostname,
+				}
+				continue
+			}
+		}
+	}
+	return slices.Collect(maps.Values(regionalHostnameChanges)), nil
+}
+
 // submitChanges takes a zone and a collection of Changes and sends them as a single transaction.
 func (p *CloudFlareProvider) submitChanges(ctx context.Context, changes []*cloudFlareChange) error {
 	// return early if there is nothing to change
@@ -484,6 +636,8 @@ func (p *CloudFlareProvider) submitChanges(ctx context.Context, changes []*cloud
 	var failedZones []string
 	for zoneID, zoneChanges := range changesByZone {
 		var failedChange bool
+		resourceContainer := cloudflare.ZoneIdentifier(zoneID)
+
 		for _, change := range zoneChanges {
 			logFields := log.Fields{
 				"record": change.ResourceRecord.Name,
@@ -499,7 +653,6 @@ func (p *CloudFlareProvider) submitChanges(ctx context.Context, changes []*cloud
 				continue
 			}
 
-			resourceContainer := cloudflare.ZoneIdentifier(zoneID)
 			records, err := p.listDNSRecordsWithAutoPagination(ctx, zoneID)
 			if err != nil {
 				return fmt.Errorf("could not fetch records from zone, %w", err)
@@ -524,13 +677,6 @@ func (p *CloudFlareProvider) submitChanges(ctx context.Context, changes []*cloud
 					failedChange = true
 					log.WithFields(logFields).Errorf("failed to update record: %v", err)
 				}
-				if regionalHostnameParam := updateDataLocalizationRegionalHostnameParams(*change); regionalHostnameParam.RegionKey != "" {
-					regionalHostnameErr := p.Client.UpdateDataLocalizationRegionalHostname(ctx, resourceContainer, regionalHostnameParam)
-					if regionalHostnameErr != nil {
-						failedChange = true
-						log.WithFields(logFields).Errorf("failed to update record when editing region: %v", regionalHostnameErr)
-					}
-				}
 			} else if change.Action == cloudFlareDelete {
 				recordID := p.getRecordID(records, change.ResourceRecord)
 				if recordID == "" {
@@ -557,6 +703,19 @@ func (p *CloudFlareProvider) submitChanges(ctx context.Context, changes []*cloud
 				}
 			}
 		}
+
+		if regionalHostnamesChanges, err := dataLocalizationRegionalHostnamesChanges(zoneChanges); err == nil {
+			if !p.submitDataLocalizationRegionalHostnameChanges(ctx, regionalHostnamesChanges, resourceContainer) {
+				failedChange = true
+			}
+		} else {
+			logFields := log.Fields{
+				"zone": zoneID,
+			}
+			log.WithFields(logFields).Errorf("failed to build data localization regional hostname changes: %v", err)
+			failedChange = true
+		}
+
 		if failedChange {
 			failedZones = append(failedZones, zoneID)
 		}
@@ -649,7 +808,6 @@ func (p *CloudFlareProvider) newCloudFlareChange(action string, ep *endpoint.End
 	if ep.RecordTTL.IsConfigured() {
 		ttl = int(ep.RecordTTL)
 	}
-	dt := time.Now()
 
 	prevCustomHostnames := []string{}
 	newCustomHostnames := map[string]cloudflare.CustomHostname{}
@@ -661,6 +819,13 @@ func (p *CloudFlareProvider) newCloudFlareChange(action string, ep *endpoint.End
 			newCustomHostnames[v] = p.newCustomHostname(v, ep.DNSName)
 		}
 	}
+	regionalHostname := cloudflare.RegionalHostname{}
+	if regionKey := getRegionKey(ep, p.RegionKey); regionKey != "" {
+		regionalHostname = cloudflare.RegionalHostname{
+			Hostname:  ep.DNSName,
+			RegionKey: regionKey,
+		}
+	}
 	return &cloudFlareChange{
 		Action: action,
 		ResourceRecord: cloudflare.DNSRecord{
@@ -671,15 +836,8 @@ func (p *CloudFlareProvider) newCloudFlareChange(action string, ep *endpoint.End
 			Proxied: &proxied,
 			Type:    ep.RecordType,
 			Content: target,
-			Meta: map[string]interface{}{
-				"region": p.RegionKey,
-			},
-		},
-		RegionalHostname: cloudflare.RegionalHostname{
-			Hostname:  ep.DNSName,
-			RegionKey: p.RegionKey,
-			CreatedOn: &dt,
 		},
+		RegionalHostname:    regionalHostname,
 		CustomHostnamesPrev: prevCustomHostnames,
 		CustomHostnames:     newCustomHostnames,
 	}
@@ -787,6 +945,19 @@ func shouldBeProxied(ep *endpoint.Endpoint, proxiedByDefault bool) bool {
 	return proxied
 }
 
+func getRegionKey(endpoint *endpoint.Endpoint, defaultRegionKey string) string {
+	if !recordTypeRegionalHostnameSupported[endpoint.RecordType] {
+		return ""
+	}
+
+	for _, v := range endpoint.ProviderSpecific {
+		if v.Name == source.CloudflareRegionKey {
+			return v.Value
+		}
+	}
+	return defaultRegionKey
+}
+
 func getEndpointCustomHostnames(ep *endpoint.Endpoint) []string {
 	for _, v := range ep.ProviderSpecific {
 		if v.Name == source.CloudflareCustomHostnameKey {
diff --git a/source/source.go b/source/source.go
index 4641c056..4f0b69d4 100644
--- a/source/source.go
+++ b/source/source.go
@@ -72,6 +72,7 @@ const (
 	// The annotation used for determining if traffic will go through Cloudflare
 	CloudflareProxiedKey        = "external-dns.alpha.kubernetes.io/cloudflare-proxied"
 	CloudflareCustomHostnameKey = "external-dns.alpha.kubernetes.io/cloudflare-custom-hostname"
+	CloudflareRegionKey         = "external-dns.alpha.kubernetes.io/cloudflare-region-key"
 
 	SetIdentifierKey = "external-dns.alpha.kubernetes.io/set-identifier"
 )
@@ -202,6 +203,12 @@ func getProviderSpecificAnnotations(annotations map[string]string) (endpoint.Pro
 			Value: v,
 		})
 	}
+	if v, exists := annotations[CloudflareRegionKey]; exists {
+		providerSpecificAnnotations = append(providerSpecificAnnotations, endpoint.ProviderSpecificProperty{
+			Name:  CloudflareRegionKey,
+			Value: v,
+		})
+	}
 	if getAliasFromAnnotations(annotations) {
 		providerSpecificAnnotations = append(providerSpecificAnnotations, endpoint.ProviderSpecificProperty{
 			Name:  "alias",



Test patch:
diff --git a/provider/cloudflare/cloudflare_test.go b/provider/cloudflare/cloudflare_test.go
index 4f3024bb..fd3a97d0 100644
--- a/provider/cloudflare/cloudflare_test.go
+++ b/provider/cloudflare/cloudflare_test.go
@@ -21,16 +21,16 @@ import (
 	"errors"
 	"fmt"
 	"os"
+	"reflect"
 	"slices"
 	"sort"
 	"strings"
 	"testing"
 
 	cloudflare "github.com/cloudflare/cloudflare-go"
+	"github.com/maxatome/go-testdeep/td"
 	log "github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/assert"
-
-	"github.com/maxatome/go-testdeep/td"
 	"sigs.k8s.io/external-dns/endpoint"
 	"sigs.k8s.io/external-dns/internal/testutils"
 	"sigs.k8s.io/external-dns/plan"
@@ -38,10 +38,11 @@ import (
 )
 
 type MockAction struct {
-	Name       string
-	ZoneId     string
-	RecordId   string
-	RecordData cloudflare.DNSRecord
+	Name             string
+	ZoneId           string
+	RecordId         string
+	RecordData       cloudflare.DNSRecord
+	RegionalHostname cloudflare.RegionalHostname
 }
 
 type mockCloudFlareClient struct {
@@ -225,6 +226,19 @@ func (m *mockCloudFlareClient) UpdateDNSRecord(ctx context.Context, rc *cloudfla
 	return nil
 }
 
+func (m *mockCloudFlareClient) CreateDataLocalizationRegionalHostname(ctx context.Context, rc *cloudflare.ResourceContainer, rp cloudflare.CreateDataLocalizationRegionalHostnameParams) error {
+	m.Actions = append(m.Actions, MockAction{
+		Name:     "CreateDataLocalizationRegionalHostname",
+		ZoneId:   rc.Identifier,
+		RecordId: "",
+		RegionalHostname: cloudflare.RegionalHostname{
+			Hostname:  rp.Hostname,
+			RegionKey: rp.RegionKey,
+		},
+	})
+	return nil
+}
+
 func (m *mockCloudFlareClient) UpdateDataLocalizationRegionalHostname(ctx context.Context, rc *cloudflare.ResourceContainer, rp cloudflare.UpdateDataLocalizationRegionalHostnameParams) error {
 	m.Actions = append(m.Actions, MockAction{
 		Name:     "UpdateDataLocalizationRegionalHostname",
@@ -237,6 +251,18 @@ func (m *mockCloudFlareClient) UpdateDataLocalizationRegionalHostname(ctx contex
 	return nil
 }
 
+func (m *mockCloudFlareClient) DeleteDataLocalizationRegionalHostname(ctx context.Context, rc *cloudflare.ResourceContainer, hostname string) error {
+	m.Actions = append(m.Actions, MockAction{
+		Name:     "DeleteDataLocalizationRegionalHostname",
+		ZoneId:   rc.Identifier,
+		RecordId: "",
+		RecordData: cloudflare.DNSRecord{
+			Name: hostname,
+		},
+	})
+	return nil
+}
+
 func (m *mockCloudFlareClient) DeleteDNSRecord(ctx context.Context, rc *cloudflare.ResourceContainer, recordID string) error {
 	m.Actions = append(m.Actions, MockAction{
 		Name:     "Delete",
@@ -745,6 +771,110 @@ func TestCloudflareSetProxied(t *testing.T) {
 	}
 }
 
+func TestCloudflareRegionalHostname(t *testing.T) {
+	endpoints := []*endpoint.Endpoint{
+		{
+			RecordType: "A",
+			DNSName:    "bar.com",
+			Targets:    endpoint.Targets{"127.0.0.1", "127.0.0.2"},
+			ProviderSpecific: endpoint.ProviderSpecific{
+				{
+					Name:  "external-dns.alpha.kubernetes.io/cloudflare-region-key",
+					Value: "eu",
+				},
+			},
+		},
+	}
+
+	AssertActions(t, &CloudFlareProvider{RegionKey: "us"}, endpoints, []MockAction{
+		{
+			Name:     "Create",
+			ZoneId:   "001",
+			RecordId: generateDNSRecordID("A", "bar.com", "127.0.0.1"),
+			RecordData: cloudflare.DNSRecord{
+				ID:      generateDNSRecordID("A", "bar.com", "127.0.0.1"),
+				Type:    "A",
+				Name:    "bar.com",
+				Content: "127.0.0.1",
+				TTL:     1,
+				Proxied: proxyDisabled,
+			},
+		},
+		{
+			Name:     "Create",
+			ZoneId:   "001",
+			RecordId: generateDNSRecordID("A", "bar.com", "127.0.0.2"),
+			RecordData: cloudflare.DNSRecord{
+				ID:      generateDNSRecordID("A", "bar.com", "127.0.0.2"),
+				Type:    "A",
+				Name:    "bar.com",
+				Content: "127.0.0.2",
+				TTL:     1,
+				Proxied: proxyDisabled,
+			},
+		},
+		{
+			Name:   "CreateDataLocalizationRegionalHostname",
+			ZoneId: "001",
+			RegionalHostname: cloudflare.RegionalHostname{
+				Hostname:  "bar.com",
+				RegionKey: "eu",
+			},
+		},
+	},
+		[]string{endpoint.RecordTypeA, endpoint.RecordTypeCNAME},
+	)
+}
+
+func TestCloudflareRegionalHostnameDefaults(t *testing.T) {
+	endpoints := []*endpoint.Endpoint{
+		{
+			RecordType: "A",
+			DNSName:    "bar.com",
+			Targets:    endpoint.Targets{"127.0.0.1", "127.0.0.2"},
+		},
+	}
+
+	AssertActions(t, &CloudFlareProvider{RegionKey: "us"}, endpoints, []MockAction{
+		{
+			Name:     "Create",
+			ZoneId:   "001",
+			RecordId: generateDNSRecordID("A", "bar.com", "127.0.0.1"),
+			RecordData: cloudflare.DNSRecord{
+				ID:      generateDNSRecordID("A", "bar.com", "127.0.0.1"),
+				Type:    "A",
+				Name:    "bar.com",
+				Content: "127.0.0.1",
+				TTL:     1,
+				Proxied: proxyDisabled,
+			},
+		},
+		{
+			Name:     "Create",
+			ZoneId:   "001",
+			RecordId: generateDNSRecordID("A", "bar.com", "127.0.0.2"),
+			RecordData: cloudflare.DNSRecord{
+				ID:      generateDNSRecordID("A", "bar.com", "127.0.0.2"),
+				Type:    "A",
+				Name:    "bar.com",
+				Content: "127.0.0.2",
+				TTL:     1,
+				Proxied: proxyDisabled,
+			},
+		},
+		{
+			Name:   "CreateDataLocalizationRegionalHostname",
+			ZoneId: "001",
+			RegionalHostname: cloudflare.RegionalHostname{
+				Hostname:  "bar.com",
+				RegionKey: "us",
+			},
+		},
+	},
+		[]string{endpoint.RecordTypeA, endpoint.RecordTypeCNAME},
+	)
+}
+
 func TestCloudflareZones(t *testing.T) {
 	provider := &CloudFlareProvider{
 		Client:       NewMockCloudFlareClient(),
@@ -1575,24 +1705,6 @@ func TestCloudFlareProvider_Region(t *testing.T) {
 	}
 }
 
-func TestCloudFlareProvider_updateDataLocalizationRegionalHostnameParams(t *testing.T) {
-	change := &cloudFlareChange{
-		RegionalHostname: cloudflare.RegionalHostname{
-			Hostname:  "example.com",
-			RegionKey: "us",
-		},
-	}
-
-	params := updateDataLocalizationRegionalHostnameParams(*change)
-	if params.Hostname != "example.com" {
-		t.Errorf("expected hostname to be 'example.com', but got '%s'", params.Hostname)
-	}
-
-	if params.RegionKey != "us" {
-		t.Errorf("expected region key to be 'us', but got '%s'", params.RegionKey)
-	}
-}
-
 func TestCloudFlareProvider_newCloudFlareChange(t *testing.T) {
 	_ = os.Setenv("CF_API_KEY", "xxxxxxxxxxxxxxxxx")
 	_ = os.Setenv("CF_API_EMAIL", "test@test.com")
@@ -1609,8 +1721,9 @@ func TestCloudFlareProvider_newCloudFlareChange(t *testing.T) {
 	}
 
 	endpoint := &endpoint.Endpoint{
-		DNSName: "example.com",
-		Targets: []string{"192.0.2.1"},
+		DNSName:    "example.com",
+		RecordType: "A",
+		Targets:    []string{"192.0.2.1"},
 	}
 
 	change := provider.newCloudFlareChange(cloudFlareCreate, endpoint, endpoint.Targets[0], nil)
@@ -2679,3 +2792,334 @@ func TestCloudflareListCustomHostnamesWithPagionation(t *testing.T) {
 	}
 	assert.Equal(t, len(chs), CustomHostnamesNumber)
 }
+
+func Test_getRegionKey(t *testing.T) {
+	type args struct {
+		endpoint         *endpoint.Endpoint
+		defaultRegionKey string
+	}
+	tests := []struct {
+		name string
+		args args
+		want string
+	}{
+		{
+			name: "no region key",
+			args: args{
+				endpoint: &endpoint.Endpoint{
+					RecordType: "A",
+				},
+				defaultRegionKey: "",
+			},
+			want: "",
+		},
+		{
+			name: "default region key",
+			args: args{
+				endpoint: &endpoint.Endpoint{
+					RecordType: "A",
+				},
+				defaultRegionKey: "us",
+			},
+			want: "us",
+		},
+		{
+			name: "endpoint with region key",
+			args: args{
+				endpoint: &endpoint.Endpoint{
+					RecordType: "A",
+					ProviderSpecific: endpoint.ProviderSpecific{
+						{
+							Name:  "external-dns.alpha.kubernetes.io/cloudflare-region-key",
+							Value: "eu",
+						},
+					},
+				},
+				defaultRegionKey: "us",
+			},
+			want: "eu",
+		},
+		{
+			name: "endpoint with empty region key",
+			args: args{
+				endpoint: &endpoint.Endpoint{
+					RecordType: "A",
+					ProviderSpecific: endpoint.ProviderSpecific{
+						{
+							Name:  "external-dns.alpha.kubernetes.io/cloudflare-region-key",
+							Value: "",
+						},
+					},
+				},
+				defaultRegionKey: "us",
+			},
+			want: "",
+		},
+		{
+			name: "unsupported record type",
+			args: args{
+				endpoint: &endpoint.Endpoint{
+					RecordType: "TXT",
+					ProviderSpecific: endpoint.ProviderSpecific{
+						{
+							Name:  "external-dns.alpha.kubernetes.io/cloudflare-region-key",
+							Value: "eu",
+						},
+					},
+				},
+				defaultRegionKey: "us",
+			},
+			want: "",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := getRegionKey(tt.args.endpoint, tt.args.defaultRegionKey); got != tt.want {
+				t.Errorf("getRegionKey() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func Test_dataLocalizationRegionalHostnamesChanges(t *testing.T) {
+	cmpDataLocalizationRegionalHostnameChange := func(i, j DataLocalizationRegionalHostnameChange) int {
+		if i.Action == j.Action {
+			return 0
+		}
+		if i.Hostname < j.Hostname {
+			return -1
+		}
+		return 1
+	}
+	type args struct {
+		changes []*cloudFlareChange
+	}
+	tests := []struct {
+		name    string
+		args    args
+		want    []DataLocalizationRegionalHostnameChange
+		wantErr bool
+	}{
+		{
+			name: "empty input",
+			args: args{
+				changes: []*cloudFlareChange{},
+			},
+			want:    nil,
+			wantErr: false,
+		},
+		{
+			name: "changes without RegionalHostname",
+			args: args{
+				changes: []*cloudFlareChange{
+					{
+						Action: cloudFlareCreate,
+						ResourceRecord: cloudflare.DNSRecord{
+							Name: "example.com",
+						},
+						RegionalHostname: cloudflare.RegionalHostname{}, // Empty
+					},
+				},
+			},
+			want:    nil,
+			wantErr: false,
+		},
+		{
+			name: "change with empty RegionKey",
+			args: args{
+				changes: []*cloudFlareChange{
+					{
+						Action: cloudFlareCreate,
+						ResourceRecord: cloudflare.DNSRecord{
+							Name: "example.com",
+						},
+						RegionalHostname: cloudflare.RegionalHostname{
+							Hostname:  "example.com",
+							RegionKey: "", // Empty region key
+						},
+					},
+				},
+			},
+			wantErr: true,
+		},
+		{
+			name: "conflicting region keys",
+			args: args{
+				changes: []*cloudFlareChange{
+					{
+						Action: cloudFlareCreate,
+						RegionalHostname: cloudflare.RegionalHostname{
+							Hostname:  "example.com",
+							RegionKey: "eu",
+						},
+					},
+					{
+						Action: cloudFlareCreate,
+						RegionalHostname: cloudflare.RegionalHostname{
+							Hostname:  "example.com",
+							RegionKey: "us", // Different region key for same hostname
+						},
+					},
+				},
+			},
+			wantErr: true,
+		},
+		{
+			name: "update takes precedence over create & delete",
+			args: args{
+				changes: []*cloudFlareChange{
+					{
+						Action: cloudFlareCreate,
+						RegionalHostname: cloudflare.RegionalHostname{
+							Hostname:  "example.com",
+							RegionKey: "eu",
+						},
+					},
+					{
+						Action: cloudFlareUpdate,
+						RegionalHostname: cloudflare.RegionalHostname{
+							Hostname:  "example.com",
+							RegionKey: "eu",
+						},
+					},
+					{
+						Action: cloudFlareDelete,
+						RegionalHostname: cloudflare.RegionalHostname{
+							Hostname:  "example.com",
+							RegionKey: "eu",
+						},
+					},
+				},
+			},
+			want: []DataLocalizationRegionalHostnameChange{
+				{
+					Action: cloudFlareUpdate,
+					RegionalHostname: cloudflare.RegionalHostname{
+						Hostname:  "example.com",
+						RegionKey: "eu",
+					},
+				},
+			},
+			wantErr: false,
+		},
+		{
+			name: "create after delete becomes update",
+			args: args{
+				changes: []*cloudFlareChange{
+					{
+						Action: cloudFlareDelete,
+						RegionalHostname: cloudflare.RegionalHostname{
+							Hostname:  "example.com",
+							RegionKey: "eu",
+						},
+					},
+					{
+						Action: cloudFlareCreate,
+						RegionalHostname: cloudflare.RegionalHostname{
+							Hostname:  "example.com",
+							RegionKey: "eu",
+						},
+					},
+				},
+			},
+			want: []DataLocalizationRegionalHostnameChange{
+				{
+					Action: cloudFlareUpdate,
+					RegionalHostname: cloudflare.RegionalHostname{
+						Hostname:  "example.com",
+						RegionKey: "eu",
+					},
+				},
+			},
+			wantErr: false,
+		},
+		{
+			name: "consolidate mixed actions for different hostnames",
+			args: args{
+				changes: []*cloudFlareChange{
+					{
+						Action: cloudFlareCreate,
+						RegionalHostname: cloudflare.RegionalHostname{
+							Hostname:  "example1.com",
+							RegionKey: "eu",
+						},
+					},
+					{
+						Action: cloudFlareUpdate,
+						RegionalHostname: cloudflare.RegionalHostname{
+							Hostname:  "example2.com",
+							RegionKey: "us",
+						},
+					},
+					{
+						Action: cloudFlareDelete,
+						RegionalHostname: cloudflare.RegionalHostname{
+							Hostname:  "example3.com",
+							RegionKey: "ap",
+						},
+					},
+					// duplicated actions
+					{
+						Action: cloudFlareCreate,
+						RegionalHostname: cloudflare.RegionalHostname{
+							Hostname:  "example1.com",
+							RegionKey: "eu",
+						},
+					},
+					{
+						Action: cloudFlareUpdate,
+						RegionalHostname: cloudflare.RegionalHostname{
+							Hostname:  "example2.com",
+							RegionKey: "us",
+						},
+					},
+					{
+						Action: cloudFlareDelete,
+						RegionalHostname: cloudflare.RegionalHostname{
+							Hostname:  "example3.com",
+							RegionKey: "ap",
+						},
+					},
+				},
+			},
+			want: []DataLocalizationRegionalHostnameChange{
+				{
+					Action: cloudFlareCreate,
+					RegionalHostname: cloudflare.RegionalHostname{
+						Hostname:  "example1.com",
+						RegionKey: "eu",
+					},
+				},
+				{
+					Action: cloudFlareUpdate,
+					RegionalHostname: cloudflare.RegionalHostname{
+						Hostname:  "example2.com",
+						RegionKey: "us",
+					},
+				},
+				{
+					Action: cloudFlareDelete,
+					RegionalHostname: cloudflare.RegionalHostname{
+						Hostname:  "example3.com",
+						RegionKey: "ap",
+					},
+				},
+			},
+			wantErr: false,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := dataLocalizationRegionalHostnamesChanges(tt.args.changes)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("dataLocalizationRegionalHostnamesChanges() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			slices.SortFunc(got, cmpDataLocalizationRegionalHostnameChange)
+			slices.SortFunc(tt.want, cmpDataLocalizationRegionalHostnameChange)
+			if !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("dataLocalizationRegionalHostnamesChanges() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
diff --git a/source/source_test.go b/source/source_test.go
index 154650d4..413995a6 100644
--- a/source/source_test.go
+++ b/source/source_test.go
@@ -174,6 +174,41 @@ func TestGetProviderSpecificCloudflareAnnotations(t *testing.T) {
 			t.Errorf("Cloudflare provider specific annotation %s is not set correctly to %s", tc.expectedKey, tc.expectedValue)
 		})
 	}
+
+	for _, tc := range []struct {
+		title         string
+		annotations   map[string]string
+		expectedKey   string
+		expectedValue string
+	}{
+		{
+			title:         "Cloudflare region key annotation is set correctly",
+			annotations:   map[string]string{CloudflareRegionKey: "us"},
+			expectedKey:   CloudflareRegionKey,
+			expectedValue: "us",
+		},
+		{
+			title: "Cloudflare region key annotation among another annotations is set correctly",
+			annotations: map[string]string{
+				"random annotation 1": "random value 1",
+				CloudflareRegionKey:   "us",
+				"random annotation 2": "random value 2",
+			},
+			expectedKey:   CloudflareRegionKey,
+			expectedValue: "us",
+		},
+	} {
+		t.Run(tc.title, func(t *testing.T) {
+			providerSpecificAnnotations, _ := getProviderSpecificAnnotations(tc.annotations)
+			for _, providerSpecificAnnotation := range providerSpecificAnnotations {
+				if providerSpecificAnnotation.Name == tc.expectedKey {
+					assert.Equal(t, tc.expectedValue, providerSpecificAnnotation.Value)
+					return
+				}
+			}
+			t.Errorf("Cloudflare provider specific annotation %s is not set correctly to %v", tc.expectedKey, tc.expectedValue)
+		})
+	}
 }
 
 func TestGetProviderSpecificAliasAnnotations(t *testing.T) {