De-duplicate EndpointSlice endpoint addresses in the resolver

coredns/resolver/endpoint_slice.go

@@ -25,6 +25,7 @@ import (
 	"strings"
 
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/apimachinery/pkg/util/sets"
 
 	"github.com/submariner-io/admiral/pkg/resource"
 	"github.com/submariner-io/lighthouse/coredns/constants"
@@ -142,6 +143,8 @@ func (i *Interface) putHeadlessEndpointSlices(key, clusterID string, endpointSli
 
 	serviceInfo.clusters[clusterID] = clusterInfo
 
+	allAddresses := sets.New[string]()
+
 	for _, endpointSlice := range endpointSlices {
 		mcsPorts := mcsServicePortsFrom(endpointSlice.Ports)
 		publishNotReadyAddresses := endpointSlice.Annotations[constants.PublishNotReadyAddresses] == strconv.FormatBool(true)
@@ -169,15 +172,19 @@ func (i *Interface) putHeadlessEndpointSlices(key, clusterID string, endpointSli
 			}
 
 			for _, address := range endpoint.Addresses {
+				if allAddresses.Has(address) {
+					continue
+				}
+
+				allAddresses.Insert(address)
 
 				record := DNSRecord{
 					IP:          address,
 					Ports:       mcsPorts,
 					ClusterName: clusterID,
+					HostName:    hostname,
 				}
 
-				record.HostName = hostname
-
 				records = append(records, record)
 			}
 

coredns/resolver/headless_service_test.go

@@ -246,6 +246,67 @@ func testHeadlessService() {
 			})
 		})
 	})
+
+	When("a service has multiple EndpointSlices with duplicate addresses", func() {
+		BeforeEach(func() {
+			endpointSlice = newEndpointSlice(namespace1, service1, clusterID1, []mcsv1a1.ServicePort{port1},
+				discovery.Endpoint{
+					Addresses:  []string{endpointIP1},
+					Conditions: discovery.EndpointConditions{Ready: &ready},
+				},
+				discovery.Endpoint{
+					Addresses:  []string{endpointIP2},
+					Conditions: discovery.EndpointConditions{Ready: &ready},
+				},
+				discovery.Endpoint{
+					Addresses:  []string{endpointIP3},
+					Conditions: discovery.EndpointConditions{Ready: &ready},
+				},
+			)
+		})
+
+		JustBeforeEach(func() {
+			Expect(t.resolver.PutEndpointSlices(endpointSlice, newEndpointSlice(namespace1, service1, clusterID1, []mcsv1a1.ServicePort{port1},
+				discovery.Endpoint{
+					Addresses:  []string{endpointIP1},
+					Conditions: discovery.EndpointConditions{Ready: &ready},
+				},
+				discovery.Endpoint{
+					Addresses:  []string{endpointIP3},
+					Conditions: discovery.EndpointConditions{Ready: &ready},
+				},
+				discovery.Endpoint{
+					Addresses:  []string{endpointIP4},
+					Conditions: discovery.EndpointConditions{Ready: &ready},
+				},
+			))).To(BeFalse())
+		})
+
+		It("should return DNS records with unique addresses", func() {
+			t.assertDNSRecordsFound(namespace1, service1, "", "", true,
+				resolver.DNSRecord{
+					IP:          endpointIP1,
+					Ports:       []mcsv1a1.ServicePort{port1},
+					ClusterName: clusterID1,
+				},
+				resolver.DNSRecord{
+					IP:          endpointIP2,
+					Ports:       []mcsv1a1.ServicePort{port1},
+					ClusterName: clusterID1,
+				},
+				resolver.DNSRecord{
+					IP:          endpointIP3,
+					Ports:       []mcsv1a1.ServicePort{port1},
+					ClusterName: clusterID1,
+				},
+				resolver.DNSRecord{
+					IP:          endpointIP4,
+					Ports:       []mcsv1a1.ServicePort{port1},
+					ClusterName: clusterID1,
+				},
+			)
+		})
+	})
 }
 
 func testHeadlessServiceInMultipleClusters() {