Derive exported EndpointSlices from K8s EndpointSlices instead of Endpoints

coredns/go.mod

@@ -11,7 +11,7 @@ require (
 	github.com/onsi/gomega v1.27.8
 	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.16.0
-	github.com/submariner-io/admiral v0.16.0-m2
+	github.com/submariner-io/admiral v0.16.0-m2.0.20230720122518-897928217fb3
 	k8s.io/api v0.27.3
 	k8s.io/apimachinery v0.27.3
 	k8s.io/client-go v0.27.3

coredns/go.sum

@@ -689,8 +689,8 @@ github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/
 github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
 github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk=
 github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
-github.com/submariner-io/admiral v0.16.0-m2 h1:U8pM8LPxlltNNvWz6wZdNh6P/wNHwMyy37rEg0baSRU=
-github.com/submariner-io/admiral v0.16.0-m2/go.mod h1:06na0fD+cJHuc4MFTWxrhum9q8tMQikj2D75jwX4IhU=
+github.com/submariner-io/admiral v0.16.0-m2.0.20230720122518-897928217fb3 h1:QNLKinCcAQREeC0c+328iohD8EI47V7GO62oGZFKIV0=
+github.com/submariner-io/admiral v0.16.0-m2.0.20230720122518-897928217fb3/go.mod h1:06na0fD+cJHuc4MFTWxrhum9q8tMQikj2D75jwX4IhU=
 github.com/subosito/gotenv v1.2.0/go.mod h1:N0PQaV/YGNqwC0u51sEeR/aUtSLEXKX9iv69rRypqCw=
 github.com/syndtr/goleveldb v1.0.1-0.20210819022825-2ae1ddf74ef7/go.mod h1:q4W45IWZaF22tdD+VEXcAWRA037jwmWEB5VWYORlTpc=
 github.com/tidwall/pretty v1.0.0/go.mod h1:XNkn88O1ChpSDQmQeStsy+sBenx6DDtFZJxhVysOjyk=

coredns/plugin/handler_test.go

@@ -100,7 +100,7 @@ func testWithoutFallback() {
 
 		t.lh.Resolver.PutServiceImport(newServiceImport(namespace1, service1, mcsv1a1.ClusterSetIP))
 
-		t.lh.Resolver.PutEndpointSlice(newEndpointSlice(namespace1, service1, clusterID, []mcsv1a1.ServicePort{port1},
+		t.lh.Resolver.PutEndpointSlices(newEndpointSlice(namespace1, service1, clusterID, []mcsv1a1.ServicePort{port1},
 			newEndpoint(serviceIP, "", true)))
 
 		rec = dnstest.NewRecorder(&test.ResponseWriter{})
@@ -164,7 +164,7 @@ func testWithoutFallback() {
 		BeforeEach(func() {
 			t.lh.Resolver.PutServiceImport(newServiceImport(namespace2, service1, mcsv1a1.ClusterSetIP))
 
-			t.lh.Resolver.PutEndpointSlice(newEndpointSlice(namespace2, service1, clusterID, []mcsv1a1.ServicePort{port1},
+			t.lh.Resolver.PutEndpointSlices(newEndpointSlice(namespace2, service1, clusterID, []mcsv1a1.ServicePort{port1},
 				newEndpoint(serviceIP, "", true)))
 		})
 
@@ -322,7 +322,7 @@ func testWithFallback() {
 
 		t.lh.Resolver.PutServiceImport(newServiceImport(namespace1, service1, mcsv1a1.ClusterSetIP))
 
-		t.lh.Resolver.PutEndpointSlice(newEndpointSlice(namespace1, service1, clusterID, []mcsv1a1.ServicePort{port1},
+		t.lh.Resolver.PutEndpointSlices(newEndpointSlice(namespace1, service1, clusterID, []mcsv1a1.ServicePort{port1},
 			newEndpoint(serviceIP, "", true)))
 
 		rec = dnstest.NewRecorder(&test.ResponseWriter{})
@@ -443,10 +443,10 @@ func testClusterStatus() {
 
 		t.lh.Resolver.PutServiceImport(newServiceImport(namespace1, service1, mcsv1a1.ClusterSetIP))
 
-		t.lh.Resolver.PutEndpointSlice(newEndpointSlice(namespace1, service1, clusterID, []mcsv1a1.ServicePort{port1},
+		t.lh.Resolver.PutEndpointSlices(newEndpointSlice(namespace1, service1, clusterID, []mcsv1a1.ServicePort{port1},
 			newEndpoint(serviceIP, "", true)))
 
-		t.lh.Resolver.PutEndpointSlice(newEndpointSlice(namespace1, service1, clusterID2, []mcsv1a1.ServicePort{port1, port2},
+		t.lh.Resolver.PutEndpointSlices(newEndpointSlice(namespace1, service1, clusterID2, []mcsv1a1.ServicePort{port1, port2},
 			newEndpoint(serviceIP2, "", true)))
 
 		rec = dnstest.NewRecorder(&test.ResponseWriter{})
@@ -584,7 +584,7 @@ func testHeadlessService() {
 	})
 
 	JustBeforeEach(func() {
-		t.lh.Resolver.PutEndpointSlice(newEndpointSlice(namespace1, service1, clusterID, []mcsv1a1.ServicePort{port1}, endpoints...))
+		t.lh.Resolver.PutEndpointSlices(newEndpointSlice(namespace1, service1, clusterID, []mcsv1a1.ServicePort{port1}, endpoints...))
 	})
 
 	When("a headless service has no endpoints", func() {
@@ -720,7 +720,7 @@ func testHeadlessService() {
 		BeforeEach(func() {
 			t.lh.Resolver.PutServiceImport(newServiceImport(namespace1, service1, mcsv1a1.Headless))
 
-			t.lh.Resolver.PutEndpointSlice(newEndpointSlice(namespace1, service1, clusterID2, []mcsv1a1.ServicePort{port1},
+			t.lh.Resolver.PutEndpointSlices(newEndpointSlice(namespace1, service1, clusterID2, []mcsv1a1.ServicePort{port1},
 				newEndpoint(endpointIP2, hostName2, true)))
 
 			endpoints = append(endpoints, newEndpoint(endpointIP, hostName1, true))
@@ -774,12 +774,12 @@ func testLocalService() {
 
 		t.lh.Resolver.PutServiceImport(newServiceImport(namespace1, service1, mcsv1a1.ClusterSetIP))
 
-		t.lh.Resolver.PutEndpointSlice(newEndpointSlice(namespace1, service1, clusterID, []mcsv1a1.ServicePort{port1},
+		t.lh.Resolver.PutEndpointSlices(newEndpointSlice(namespace1, service1, clusterID, []mcsv1a1.ServicePort{port1},
 			newEndpoint(serviceIP, "", true)))
 
 		t.lh.Resolver.PutServiceImport(newServiceImport(namespace1, service1, mcsv1a1.ClusterSetIP))
 
-		t.lh.Resolver.PutEndpointSlice(newEndpointSlice(namespace1, service1, clusterID2, []mcsv1a1.ServicePort{port1, port2},
+		t.lh.Resolver.PutEndpointSlices(newEndpointSlice(namespace1, service1, clusterID2, []mcsv1a1.ServicePort{port1, port2},
 			newEndpoint(serviceIP2, "", true)))
 
 		rec = dnstest.NewRecorder(&test.ResponseWriter{})
@@ -853,7 +853,7 @@ func testLocalService() {
 
 	When("service is in local and remote clusters and local has no active endpoints", func() {
 		BeforeEach(func() {
-			t.lh.Resolver.PutEndpointSlice(newEndpointSlice(namespace1, service1, clusterID, []mcsv1a1.ServicePort{port1},
+			t.lh.Resolver.PutEndpointSlices(newEndpointSlice(namespace1, service1, clusterID, []mcsv1a1.ServicePort{port1},
 				newEndpoint(serviceIP, "", false)))
 		})
 
@@ -884,7 +884,7 @@ func testSRVMultiplePorts() {
 
 		t.lh.Resolver.PutServiceImport(newServiceImport(namespace1, service1, mcsv1a1.ClusterSetIP))
 
-		t.lh.Resolver.PutEndpointSlice(newEndpointSlice(namespace1, service1, clusterID, []mcsv1a1.ServicePort{port1, port2},
+		t.lh.Resolver.PutEndpointSlices(newEndpointSlice(namespace1, service1, clusterID, []mcsv1a1.ServicePort{port1, port2},
 			newEndpoint(endpointIP, "", true)))
 
 		rec = dnstest.NewRecorder(&test.ResponseWriter{})

coredns/resolver/clusterip_service_test.go

@@ -372,7 +372,7 @@ func testClusterIPServiceMisc() {
 	When("a cluster's EndpointSlice is initially created before the ServiceImport", func() {
 		It("should eventually process them and return its DNS record", func() {
 			es := newClusterIPEndpointSlice(namespace1, service1, clusterID1, serviceIP1, true)
-			Expect(t.resolver.PutEndpointSlice(es)).To(BeTrue())
+			Expect(t.resolver.PutEndpointSlices(es)).To(BeTrue())
 
 			t.awaitDNSRecords(namespace1, service1, clusterID1, "", false)
 
@@ -412,7 +412,7 @@ func testClusterIPServiceMisc() {
 
 			t.resolver.PutServiceImport(serviceImport)
 
-			Expect(t.resolver.PutEndpointSlice(newClusterIPEndpointSlice(namespace1, service1, clusterID1, serviceIP1, true))).To(BeTrue())
+			Expect(t.resolver.PutEndpointSlices(newClusterIPEndpointSlice(namespace1, service1, clusterID1, serviceIP1, true))).To(BeTrue())
 
 			t.resolver.RemoveServiceImport(serviceImport)
 		})
@@ -435,7 +435,7 @@ func testClusterIPServiceMisc() {
 
 			t.resolver.PutServiceImport(serviceImport)
 
-			Expect(t.resolver.PutEndpointSlice(newClusterIPEndpointSlice(namespace1, service1, clusterID1, serviceIP1, true))).To(BeTrue())
+			Expect(t.resolver.PutEndpointSlices(newClusterIPEndpointSlice(namespace1, service1, clusterID1, serviceIP1, true))).To(BeTrue())
 
 			t.resolver.RemoveServiceImport(serviceImport)
 		})

coredns/resolver/controller.go

@@ -26,6 +26,7 @@ import (
 	discovery "k8s.io/api/discovery/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
+	k8slabels "k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	logf "sigs.k8s.io/controller-runtime/pkg/log"
 	mcsv1a1 "sigs.k8s.io/mcs-api/pkg/apis/v1alpha1"
@@ -34,8 +35,9 @@ import (
 var logger = log.Logger{Logger: logf.Log.WithName("Resolver")}
 
 type controller struct {
-	resolver *Interface
-	stopCh   chan struct{}
+	resolver        *Interface
+	resourceWatcher watcher.Interface
+	stopCh          chan struct{}
 }
 
 func NewController(r *Interface) *controller {
@@ -72,12 +74,14 @@ func (c *controller) Start(config watcher.Config) error {
 		},
 	}
 
-	resourceWatcher, err := watcher.New(&config)
+	var err error
+
+	c.resourceWatcher, err = watcher.New(&config)
 	if err != nil {
 		return errors.Wrap(err, "error creating the resource watcher")
 	}
 
-	err = resourceWatcher.Start(c.stopCh)
+	err = c.resourceWatcher.Start(c.stopCh)
 	if err != nil {
 		return errors.Wrap(err, "error starting the resource watcher")
 	}
@@ -92,12 +96,49 @@ func (c *controller) Stop() {
 }
 
 func (c *controller) onEndpointSliceCreateOrUpdate(obj runtime.Object, _ int) bool {
-	return c.resolver.PutEndpointSlice(obj.(*discovery.EndpointSlice))
+	endpointSlice := obj.(*discovery.EndpointSlice)
+	if ignoreEndpointSlice(endpointSlice) {
+		return false
+	}
+
+	if !isHeadless(endpointSlice) {
+		return c.resolver.PutEndpointSlices(endpointSlice)
+	}
+
+	return c.resolver.PutEndpointSlices(c.getAllEndpointSlices(endpointSlice)...)
+}
+
+func (c *controller) getAllEndpointSlices(forEPS *discovery.EndpointSlice) []*discovery.EndpointSlice {
+	list := c.resourceWatcher.ListResources(&discovery.EndpointSlice{}, k8slabels.SelectorFromSet(map[string]string{
+		constants.LabelSourceNamespace:  forEPS.Labels[constants.LabelSourceNamespace],
+		mcsv1a1.LabelServiceName:        forEPS.Labels[mcsv1a1.LabelServiceName],
+		constants.MCSLabelSourceCluster: forEPS.Labels[constants.MCSLabelSourceCluster],
+	}))
+
+	epSlices := make([]*discovery.EndpointSlice, len(list))
+	for i := range list {
+		epSlices[i] = list[i].(*discovery.EndpointSlice)
+	}
+
+	return epSlices
 }
 
 func (c *controller) onEndpointSliceDelete(obj runtime.Object, _ int) bool {
-	c.resolver.RemoveEndpointSlice(obj.(*discovery.EndpointSlice))
-	return false
+	endpointSlice := obj.(*discovery.EndpointSlice)
+	if ignoreEndpointSlice(endpointSlice) {
+		return false
+	}
+
+	if !isHeadless(endpointSlice) {
+		c.resolver.RemoveEndpointSlice(endpointSlice)
+	}
+
+	epSlices := c.getAllEndpointSlices(endpointSlice)
+	if len(epSlices) == 0 {
+		c.resolver.RemoveEndpointSlice(endpointSlice)
+	}
+
+	return c.resolver.PutEndpointSlices(epSlices...)
 }
 
 func (c *controller) onServiceImportCreateOrUpdate(obj runtime.Object, _ int) bool {
@@ -109,3 +150,8 @@ func (c *controller) onServiceImportDelete(obj runtime.Object, _ int) bool {
 	c.resolver.RemoveServiceImport(obj.(*mcsv1a1.ServiceImport))
 	return false
 }
+
+func ignoreEndpointSlice(eps *discovery.EndpointSlice) bool {
+	isOnBroker := eps.Namespace != eps.Labels[constants.LabelSourceNamespace]
+	return isOnBroker
+}

coredns/resolver/controller_test.go

@@ -21,6 +21,8 @@ package resolver_test
 import (
 	"context"
 
+	"github.com/submariner-io/lighthouse/coredns/constants"
+
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 	"github.com/submariner-io/admiral/pkg/syncer/test"
@@ -34,13 +36,13 @@ import (
 var _ = Describe("Controller", func() {
 	t := newTestDriver()
 
-	expDNSRecord := resolver.DNSRecord{
-		IP:          serviceIP1,
-		Ports:       []mcsv1a1.ServicePort{port1},
-		ClusterName: clusterID1,
-	}
+	When("a ClusterIP service EndpointSlice is created", func() {
+		expDNSRecord := resolver.DNSRecord{
+			IP:          serviceIP1,
+			Ports:       []mcsv1a1.ServicePort{port1},
+			ClusterName: clusterID1,
+		}
 
-	When("an EndpointSlice is created", func() {
 		var endpointSlice *discovery.EndpointSlice
 
 		JustBeforeEach(func() {
@@ -100,4 +102,117 @@ var _ = Describe("Controller", func() {
 			})
 		})
 	})
+
+	When("there's multiple EndpointSlices for a headless service", func() {
+		var epsName1, epsName2 string
+
+		JustBeforeEach(func() {
+			t.createServiceImport(newHeadlessAggregatedServiceImport(namespace1, service1))
+
+			eps := 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},
+				},
+			)
+			epsName1 = eps.Name
+			t.createEndpointSlice(eps)
+
+			eps = newEndpointSlice(namespace1, service1, clusterID1, []mcsv1a1.ServicePort{port2},
+				discovery.Endpoint{
+					Addresses:  []string{endpointIP3},
+					Conditions: discovery.EndpointConditions{Ready: &ready},
+				},
+				discovery.Endpoint{
+					Addresses:  []string{endpointIP4},
+					Conditions: discovery.EndpointConditions{Ready: &ready},
+				},
+			)
+			epsName2 = eps.Name
+			t.createEndpointSlice(eps)
+		})
+
+		Specify("GetDNSRecords should return their DNS record", func() {
+			t.awaitDNSRecordsFound(namespace1, service1, clusterID1, "", 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{port2},
+					ClusterName: clusterID1,
+				},
+				resolver.DNSRecord{
+					IP:          endpointIP4,
+					Ports:       []mcsv1a1.ServicePort{port2},
+					ClusterName: clusterID1,
+				})
+		})
+
+		Context("and one is deleted", func() {
+			Specify("GetDNSRecords should return the remaining DNS records", func() {
+				t.awaitDNSRecords(namespace1, service1, clusterID1, "", true)
+
+				Expect(t.endpointSlices.Namespace(namespace1).Delete(context.TODO(), epsName1, metav1.DeleteOptions{})).To(Succeed())
+
+				t.awaitDNSRecordsFound(namespace1, service1, clusterID1, "", true,
+					resolver.DNSRecord{
+						IP:          endpointIP3,
+						Ports:       []mcsv1a1.ServicePort{port2},
+						ClusterName: clusterID1,
+					},
+					resolver.DNSRecord{
+						IP:          endpointIP4,
+						Ports:       []mcsv1a1.ServicePort{port2},
+						ClusterName: clusterID1,
+					})
+			})
+		})
+
+		Context("and both are deleted", func() {
+			Specify("GetDNSRecords should return no DNS records", func() {
+				t.awaitDNSRecords(namespace1, service1, clusterID1, "", true)
+
+				Expect(t.endpointSlices.Namespace(namespace1).Delete(context.TODO(), epsName1, metav1.DeleteOptions{})).To(Succeed())
+				Expect(t.endpointSlices.Namespace(namespace1).Delete(context.TODO(), epsName2, metav1.DeleteOptions{})).To(Succeed())
+
+				t.awaitDNSRecords(namespace1, service1, clusterID1, "", false)
+			})
+		})
+	})
+
+	When("an EndpointSlice is on the broker", func() {
+		JustBeforeEach(func() {
+			t.createServiceImport(newAggregatedServiceImport(namespace1, service1))
+			t.createEndpointSlice(&discovery.EndpointSlice{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test",
+					Namespace: test.RemoteNamespace,
+					Labels: map[string]string{
+						constants.MCSLabelSourceCluster: "test",
+						mcsv1a1.LabelServiceName:        "test",
+						constants.LabelSourceNamespace:  namespace1,
+					},
+				},
+			})
+		})
+
+		It("should not process it", func() {
+			Consistently(func() bool {
+				t.awaitDNSRecords(namespace1, service1, clusterID1, "", false)
+				return true
+			}).Should(BeTrue())
+		})
+	})
 })