Repository: submariner-io/lighthouse
Pull request #351 (2020-10-21 09:47:50)

Remove  Endpoint Controller
*Removes endpoint controller
*Distributes endpointslices for clusterIp servies and uses it to
check if a service is healthy

Fixes: submariner-io/lighthouse#332

Signed-off-by: Aswin Surayanarayanan <asuryana@redhat.com>

submariner-bot: 🤖 Created branch: z_pr351/aswinsuryan/remove-endpoint-controller
submariner-bot: 🤖 Updated branch: z_pr351/aswinsuryan/remove-endpoint-controller
submariner-bot: 🤖 Updated branch: z_pr351/aswinsuryan/remove-endpoint-controller
submariner-bot: 🤖 Updated branch: z_pr351/aswinsuryan/remove-endpoint-controller
submariner-bot: 🤖 Updated branch: z_pr351/aswinsuryan/remove-endpoint-controller
submariner-bot: 🤖 Updated branch: z_pr351/aswinsuryan/remove-endpoint-controller
submariner-bot: 🤖 Updated branch: z_pr351/aswinsuryan/remove-endpoint-controller
submariner-bot: 🤖 Updated branch: z_pr351/aswinsuryan/remove-endpoint-controller
submariner-bot: 🤖 Updated branch: z_pr351/aswinsuryan/remove-endpoint-controller
submariner-bot: 🤖 Updated branch: z_pr351/aswinsuryan/remove-endpoint-controller
submariner-bot: 🤖 Updated branch: z_pr351/aswinsuryan/remove-endpoint-controller
submariner-bot: 🤖 Updated branch: z_pr351/aswinsuryan/remove-endpoint-controller
submariner-bot: 🤖 Updated branch: z_pr351/aswinsuryan/remove-endpoint-controller
submariner-bot: 🤖 Updated branch: z_pr351/aswinsuryan/remove-endpoint-controller
submariner-bot: 🤖 Updated branch: z_pr351/aswinsuryan/remove-endpoint-controller
submariner-bot: 🤖 Updated branch: z_pr351/aswinsuryan/remove-endpoint-controller
submariner-bot: 🤖 Updated branch: z_pr351/aswinsuryan/remove-endpoint-controller

Patch:
diff --git a/go.sum b/go.sum
index 1e68fd9..9b25184 100644
--- a/go.sum
+++ b/go.sum
@@ -394,6 +394,7 @@ github.com/miekg/dns v1.1.15/go.mod h1:W1PPwlIAgtquWBMBEV9nkV9Cazfe8ScdGz/Lj7v3N
 github.com/miekg/dns v1.1.27/go.mod h1:KNUDUusw/aVsxyTYZM1oqvCicbwhgbNgztCETuNZ7xM=
 github.com/miekg/dns v1.1.31 h1:sJFOl9BgwbYAWOGEwr61FU28pqsBNdpRBnhGXtO06Oo=
 github.com/miekg/dns v1.1.31/go.mod h1:KNUDUusw/aVsxyTYZM1oqvCicbwhgbNgztCETuNZ7xM=
+github.com/miekg/dns v1.1.34 h1:SgTzfkN+oLoIHF1bgUP+C71mzuDl3AhLApHzCCIAMWM=
 github.com/miekg/dns v1.1.34/go.mod h1:KNUDUusw/aVsxyTYZM1oqvCicbwhgbNgztCETuNZ7xM=
 github.com/miekg/dns v1.1.35/go.mod h1:KNUDUusw/aVsxyTYZM1oqvCicbwhgbNgztCETuNZ7xM=
 github.com/mitchellh/go-homedir v1.1.0 h1:lukF9ziXFxDFPkA1vsr5zpc1XuPDn/wFntq5mG+4E0Y=
@@ -429,8 +430,11 @@ github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+W
 github.com/onsi/ginkgo v1.7.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
 github.com/onsi/ginkgo v1.10.1/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
 github.com/onsi/ginkgo v1.12.1/go.mod h1:zj2OWP4+oCPe1qIXoGWkgMRwljMUYCdkwsT2108oapk=
+github.com/onsi/ginkgo v1.14.0 h1:2mOpI4JVVPBN+WQRa0WKH2eXR+Ey+uK4n7Zj0aYpIQA=
+github.com/onsi/ginkgo v1.14.0/go.mod h1:iSB4RoI2tjJc9BBv4NKIKWKya62Rps+oPG/Lv9klQyY=
 github.com/onsi/ginkgo v1.14.1 h1:jMU0WaQrP0a/YAEq8eJmJKjBoMs+pClEr1vDMlM/Do4=
 github.com/onsi/ginkgo v1.14.1/go.mod h1:iSB4RoI2tjJc9BBv4NKIKWKya62Rps+oPG/Lv9klQyY=
+github.com/onsi/ginkgo v1.14.2 h1:8mVmC9kjFFmA8H4pKMUhcblgifdkOIXPvbhN1T36q1M=
 github.com/onsi/ginkgo v1.14.2/go.mod h1:iSB4RoI2tjJc9BBv4NKIKWKya62Rps+oPG/Lv9klQyY=
 github.com/onsi/gomega v0.0.0-20170829124025-dcabb60a477c/go.mod h1:C1qb7wdrVGGVU+Z6iS04AVkA3Q65CEZX59MT0QO5uiA=
 github.com/onsi/gomega v1.4.2/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY=
@@ -441,6 +445,7 @@ github.com/onsi/gomega v1.10.1 h1:o0+MgICZLuZ7xjH7Vx6zS/zcu93/BEp1VwkIW1mEXCE=
 github.com/onsi/gomega v1.10.1/go.mod h1:iN09h71vgCQne3DLsj+A5owkum+a2tYe+TOCB1ybHNo=
 github.com/onsi/gomega v1.10.2 h1:aY/nuoWlKJud2J6U0E3NWsjlg+0GtwXxgEqthRdzlcs=
 github.com/onsi/gomega v1.10.2/go.mod h1:iN09h71vgCQne3DLsj+A5owkum+a2tYe+TOCB1ybHNo=
+github.com/onsi/gomega v1.10.3 h1:gph6h/qe9GSUw1NhH1gp+qb+h8rXD8Cy60Z32Qw3ELA=
 github.com/onsi/gomega v1.10.3/go.mod h1:V9xEwhxec5O8UDM77eCW8vLymOMltsqPVYWrpDsH8xc=
 github.com/opentracing-contrib/go-observer v0.0.0-20170622124052-a52f23424492 h1:lM6RxxfUMrYL/f8bWEUqdXrANWtrL7Nndbm9iFN0DlU=
 github.com/opentracing-contrib/go-observer v0.0.0-20170622124052-a52f23424492/go.mod h1:Ngi6UdF0k5OKD5t5wlmGhe/EDKPoUM3BXZSSfIuJbis=
@@ -544,6 +549,7 @@ github.com/submariner-io/admiral v0.7.0 h1:1Ddjx8PvmeUjqO9+qMAPoFgkDSq8asmXl/tls
 github.com/submariner-io/admiral v0.7.0/go.mod h1:S5Aemqu4Ac2sZWW9AEfPIhSC4eMCWmCVQ/rSSEHRt2o=
 github.com/submariner-io/shipyard v0.7.1 h1:2+CdN9c7BZX7AUKvBrRkqvha+qqEfbWRjgutbFNaiKo=
 github.com/submariner-io/shipyard v0.7.1/go.mod h1:m2Gs15WS6bCDMY2QUjjeC3p4Q95H+pG2UW4k2S9rhQc=
+github.com/submariner-io/shipyard v0.7.2 h1:jlg8AHfBkAqWKJXyby1VEBN1aCipDgwfCvBoWr5Qb6M=
 github.com/submariner-io/shipyard v0.7.2/go.mod h1:m4Qj1bHIgQBmp/CoPQOhsQDaQJvaF9YV3oyDLNfecRg=
 github.com/tidwall/pretty v1.0.0/go.mod h1:XNkn88O1ChpSDQmQeStsy+sBenx6DDtFZJxhVysOjyk=
 github.com/timewasted/linode v0.0.0-20160829202747-37e84520dcf7/go.mod h1:imsgLplxEC/etjIhdr3dNzV3JeT27LbVu5pYWm0JCBY=
@@ -605,6 +611,7 @@ golang.org/x/crypto v0.0.0-20191206172530-e9b2fee46413/go.mod h1:LzIPMQfyMNhhGPh
 golang.org/x/crypto v0.0.0-20200311171314-f7b00557c8c4/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20200323165209-0ec3e9974c59 h1:3zb4D3T4G8jdExgVU/95+vQXfpEPiMdCaZgmGVxjNHM=
 golang.org/x/crypto v0.0.0-20200323165209-0ec3e9974c59/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
+golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9 h1:psW17arqaxU48Z5kZ0CQnkZWQJsqcURM6tKiBApRjXI=
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
 golang.org/x/exp v0.0.0-20190125153040-c74c464bbbf2/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
@@ -649,6 +656,7 @@ golang.org/x/net v0.0.0-20200520004742-59133d7f0dd7 h1:AeiKBIuRw3UomYXSbLy0Mc2dD
 golang.org/x/net v0.0.0-20200520004742-59133d7f0dd7/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=
 golang.org/x/net v0.0.0-20200602114024-627f9648deb9 h1:pNX+40auqi2JqRfOP1akLGtYcn15TUbkhwuCO3foqqM=
 golang.org/x/net v0.0.0-20200602114024-627f9648deb9/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=
+golang.org/x/net v0.0.0-20201006153459-a7d1128ccaa0 h1:wBouT66WTYFXdxfVdz9sVWARVd/2vfGcmI45D2gj45M=
 golang.org/x/net v0.0.0-20201006153459-a7d1128ccaa0/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
@@ -697,12 +705,14 @@ golang.org/x/sys v0.0.0-20200519105757-fe76b779f299 h1:DYfZAGf2WMFjMxbgTjaC+2HC7
 golang.org/x/sys v0.0.0-20200519105757-fe76b779f299/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200615200032-f1bc736245b1 h1:ogLJMz+qpzav7lGMh10LMvAkM/fAoGlaiiHYiFYdm80=
 golang.org/x/sys v0.0.0-20200615200032-f1bc736245b1/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f h1:+Nyd8tzPX9R7BWHguqsrbFdRx3WQ/1ib8I44HXV5yTA=
 golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/text v0.0.0-20160726164857-2910a502d2bf/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.2 h1:tW2bmiBqwgJj/UpqtC8EpXEZVYOwU0yG4iWbprSVAcs=
 golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
+golang.org/x/text v0.3.3 h1:cokOdA+Jmi5PJGXLlLllQSgYigAEfHXJAERHVMaCc2k=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/time v0.0.0-20180412165947-fbb02b2291d2/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
diff --git a/pkg/agent/controller/agent.go b/pkg/agent/controller/agent.go
index 7bce8f5..de8f9ba 100644
--- a/pkg/agent/controller/agent.go
+++ b/pkg/agent/controller/agent.go
@@ -13,7 +13,6 @@ import (
 	lhconstants "github.com/submariner-io/lighthouse/pkg/constants"
 	corev1 "k8s.io/api/core/v1"
 	discovery "k8s.io/api/discovery/v1beta1"
-	"k8s.io/apimachinery/pkg/api/equality"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -141,22 +140,6 @@ func NewWithDetail(spec *AgentSpecification, syncerConf *broker.SyncerConfig, re
 		return nil, err
 	}
 
-	agentController.endpointSyncer, err = syncer.NewResourceSyncer(&syncer.ResourceSyncerConfig{
-		Name:                "Endpoint events",
-		SourceClient:        localClient,
-		SourceNamespace:     metav1.NamespaceAll,
-		Direction:           syncer.LocalToRemote,
-		RestMapper:          restMapper,
-		Federator:           agentController.serviceExportSyncer.GetBrokerFederator(),
-		ResourceType:        &corev1.Endpoints{},
-		Transform:           agentController.endpointToRemoteServiceImport,
-		ResourcesEquivalent: agentController.endpointEquivalent,
-		Scheme:              runtimeScheme,
-	})
-	if err != nil {
-		return nil, err
-	}
-
 	agentController.serviceImportController, err = newServiceImportController(spec, agentController.serviceSyncer, restMapper,
 		localClient, runtimeScheme)
 	if err != nil {
@@ -183,10 +166,6 @@ func (a *Controller) Start(stopCh <-chan struct{}) error {
 		return err
 	}
 
-	if err := a.endpointSyncer.Start(stopCh); err != nil {
-		return err
-	}
-
 	if err := a.endpointSliceSyncer.Start(stopCh); err != nil {
 		return err
 	}
@@ -352,66 +331,6 @@ func (a *Controller) serviceToRemoteServiceImport(obj runtime.Object, op syncer.
 	return serviceImport, false
 }
 
-func (a *Controller) endpointToRemoteServiceImport(obj runtime.Object, op syncer.Operation) (runtime.Object, bool) {
-	ep := obj.(*corev1.Endpoints)
-	obj, found, err := a.serviceExportSyncer.GetLocalResource(ep.Name, ep.Namespace, &lighthousev2a1.ServiceExport{})
-	if err != nil {
-		klog.Errorf("Error retrieving ServiceExport for Endpoints (%s/%s): %v", ep.Namespace, ep.Name, err)
-		return nil, true
-	}
-
-	if !found {
-		// Service Export not created yet
-		return nil, false
-	}
-
-	svcExport := obj.(*lighthousev2a1.ServiceExport)
-
-	serviceImport, found, err := a.serviceImportController.getServiceImport(
-		a.getObjectNameWithClusterId(svcExport.Name, svcExport.Namespace), a.namespace)
-	if err != nil {
-		klog.Errorf("Error retrieving ServiceImport for Endpoints (%s/%s): %v", ep.Namespace, ep.Name, err)
-
-		// Requeue
-		return nil, true
-	}
-
-	if !found {
-		return nil, true
-	}
-
-	if serviceImport.Spec.Type == lighthousev2a1.Headless {
-		return nil, false
-	}
-
-	ipList := getIPsFromEndpoint(ep)
-	var ip string
-
-	if serviceImport.Spec.Type == lighthousev2a1.ClusterSetIP && len(ipList) > 0 {
-		/*
-			When there's no healthy pods, the IPs in the Endpoint will become empty and
-			thus we also clear the ServiceImport IPs to avoid clients sending a request
-			to a cluster which has no active pods to handle the request.
-
-			Once at least one backing Endpoint pod becomes healthy, we need to re-establish
-			the ServiceImport IPs from the previously cached clusterIP annotation.
-		*/
-		ip = serviceImport.Annotations[clusterIP]
-	} else {
-		ip = ""
-	}
-
-	oldClusterIP := serviceImport.Spec.IP
-	if oldClusterIP == ip {
-		klog.V(log.DEBUG).Infof("Old and new cluster status are same")
-		return nil, false
-	}
-
-	serviceImport.Spec.IP = ip
-
-	return serviceImport, false
-}
-
 func (a *Controller) updateExportedServiceStatus(name, namespace string, condType lighthousev2a1.ServiceExportConditionType,
 	status corev1.ConditionStatus, reason, msg string) {
 	retryErr := retry.RetryOnConflict(retry.DefaultRetry, func() error {
@@ -476,11 +395,6 @@ func (a *Controller) getServiceExport(name, namespace string) (*lighthousev2a1.S
 	return se, nil
 }
 
-func (a *Controller) endpointEquivalent(obj1, obj2 *unstructured.Unstructured) bool {
-	return equality.Semantic.DeepEqual(util.GetNestedField(obj1, "subsets"),
-		util.GetNestedField(obj2, "subsets"))
-}
-
 func serviceExportConditionEqual(c1, c2 *lighthousev2a1.ServiceExportCondition) bool {
 	return c1.Type == c2.Type && c1.Status == c2.Status && reflect.DeepEqual(c1.Reason, c2.Reason) &&
 		reflect.DeepEqual(c1.Message, c2.Message)
diff --git a/pkg/agent/controller/serviceimport.go b/pkg/agent/controller/serviceimport.go
index de77970..038da93 100644
--- a/pkg/agent/controller/serviceimport.go
+++ b/pkg/agent/controller/serviceimport.go
@@ -69,8 +69,7 @@ func (c *ServiceImportController) serviceImportCreatedOrUpdated(serviceImport *l
 		return false
 	}
 
-	if serviceImport.Spec.Type != lighthousev2a1.Headless ||
-		serviceImport.GetLabels()[lhconstants.LabelSourceCluster] != c.clusterID {
+	if serviceImport.GetLabels()[lhconstants.LabelSourceCluster] != c.clusterID {
 		return false
 	}
 
@@ -130,12 +129,3 @@ func (c *ServiceImportController) serviceImportToEndpointController(obj runtime.
 
 	return nil, c.serviceImportDeleted(serviceImport, key)
 }
-
-func (c *ServiceImportController) getServiceImport(name, namespace string) (*lighthousev2a1.ServiceImport, bool, error) {
-	obj, found, err := c.serviceImportSyncer.GetResource(name, namespace)
-	if obj != nil {
-		return obj.(*lighthousev2a1.ServiceImport), found, err
-	}
-
-	return nil, found, err
-}
diff --git a/pkg/agent/controller/types.go b/pkg/agent/controller/types.go
index 10a7e60..3ed964c 100644
--- a/pkg/agent/controller/types.go
+++ b/pkg/agent/controller/types.go
@@ -21,7 +21,6 @@ type Controller struct {
 	serviceExportSyncer     *broker.Syncer
 	endpointSliceSyncer     *broker.Syncer
 	serviceSyncer           syncer.Interface
-	endpointSyncer          syncer.Interface
 	serviceImportController *ServiceImportController
 }
 
diff --git a/pkg/endpointslice/controller.go b/pkg/endpointslice/controller.go
index b2351e0..481bf11 100644
--- a/pkg/endpointslice/controller.go
+++ b/pkg/endpointslice/controller.go
@@ -15,21 +15,35 @@ import (
 	"k8s.io/klog"
 )
 
+type NewClientsetFunc func(kubeConfig *rest.Config) (kubernetes.Interface, error)
+
+// Indirection hook for unit tests to supply fake client sets
+var NewClientset NewClientsetFunc
+
 type Controller struct {
 	// Indirection hook for unit tests to supply fake client sets
-	NewClientset func(kubeConfig *rest.Config) (kubernetes.Interface, error)
+	NewClientset NewClientsetFunc
 	epsInformer  cache.Controller
 	stopCh       chan struct{}
 	store        Store
+	clientSet    kubernetes.Interface
 }
 
 func NewController(endpointSliceStore Store) *Controller {
 	return &Controller{
-		NewClientset: func(c *rest.Config) (kubernetes.Interface, error) {
-			return kubernetes.NewForConfig(c)
-		},
-		stopCh: make(chan struct{}),
-		store:  endpointSliceStore,
+		NewClientset: getNewClientsetFunc(),
+		stopCh:       make(chan struct{}),
+		store:        endpointSliceStore,
+	}
+}
+
+func getNewClientsetFunc() NewClientsetFunc {
+	if NewClientset != nil {
+		return NewClientset
+	}
+
+	return func(c *rest.Config) (kubernetes.Interface, error) {
+		return kubernetes.NewForConfig(c)
 	}
 }
 
@@ -41,6 +55,7 @@ func (c *Controller) Start(kubeConfig *rest.Config) error {
 		return fmt.Errorf("Error creating client set: %v", err)
 	}
 
+	c.clientSet = clientSet
 	labelMap := map[string]string{
 		discovery.LabelManagedBy: lhconstants.LabelValueManagedBy,
 	}
@@ -98,3 +113,16 @@ func (c *Controller) Stop() {
 
 	klog.Infof("EndpointSlice Controller stopped")
 }
+
+func (c *Controller) IsHealthy(name, namespace, clusterId string) bool {
+	key := keyFunc(name, namespace)
+	endpointInfo := c.store.Get(key)
+	if endpointInfo != nil && endpointInfo.clusterInfo != nil {
+		info := endpointInfo.clusterInfo[clusterId]
+		if info != nil {
+			return len(info.ipList) > 0
+		}
+	}
+
+	return false
+}
diff --git a/pkg/endpointslice/map.go b/pkg/endpointslice/map.go
index 501889d..121a965 100644
--- a/pkg/endpointslice/map.go
+++ b/pkg/endpointslice/map.go
@@ -132,6 +132,14 @@ func (m *Map) Remove(es *discovery.EndpointSlice) {
 	}
 }
 
+func (m *Map) Get(key string) *endpointInfo {
+	m.RLock()
+	defer m.RUnlock()
+	endpointInfo := m.epMap[key]
+
+	return endpointInfo
+}
+
 func getKey(es *discovery.EndpointSlice) (string, bool) {
 	name, ok := es.Labels[constants.LabelSourceName]
 
diff --git a/pkg/endpointslice/store.go b/pkg/endpointslice/store.go
index 3b23a58..d8c2d4d 100644
--- a/pkg/endpointslice/store.go
+++ b/pkg/endpointslice/store.go
@@ -6,4 +6,6 @@ type Store interface {
 	Put(endpointSlice *discovery.EndpointSlice)
 
 	Remove(endpointSlice *discovery.EndpointSlice)
+
+	Get(key string) *endpointInfo
 }
diff --git a/pkg/serviceimport/map.go b/pkg/serviceimport/map.go
index c7ffcbf..9b781a8 100644
--- a/pkg/serviceimport/map.go
+++ b/pkg/serviceimport/map.go
@@ -35,7 +35,8 @@ type Map struct {
 	sync.RWMutex
 }
 
-func (m *Map) selectIP(queue []clusterInfo, counter *uint64, checkCluster func(string) bool) string {
+func (m *Map) selectIP(queue []clusterInfo, counter *uint64, name, namespace string, checkCluster func(string) bool,
+	checkEndpoint func(string, string, string) bool) string {
 	queueLength := len(queue)
 	for i := 0; i < queueLength; i++ {
 		c := atomic.LoadUint64(counter)
@@ -44,7 +45,7 @@ func (m *Map) selectIP(queue []clusterInfo, counter *uint64, checkCluster func(s
 
 		atomic.AddUint64(counter, 1)
 
-		if checkCluster(info.name) {
+		if checkCluster(info.name) && checkEndpoint(name, namespace, info.name) {
 			return info.ip
 		}
 	}
@@ -52,7 +53,8 @@ func (m *Map) selectIP(queue []clusterInfo, counter *uint64, checkCluster func(s
 	return ""
 }
 
-func (m *Map) GetIP(namespace, name, cluster string, checkCluster func(string) bool) (string, bool) {
+func (m *Map) GetIP(namespace, name, cluster string, checkCluster func(string) bool,
+	checkEndpoint func(string, string, string) bool) (string, bool) {
 	clusterIPs, queue, counter, isHeadless := func() (map[string]string, []clusterInfo, *uint64, bool) {
 		m.RLock()
 		defer m.RUnlock()
@@ -74,7 +76,7 @@ func (m *Map) GetIP(namespace, name, cluster string, checkCluster func(string) b
 		return ip, found
 	}
 
-	ip := m.selectIP(queue, counter, checkCluster)
+	ip := m.selectIP(queue, counter, name, namespace, checkCluster, checkEndpoint)
 	if ip != "" {
 		return ip, true
 	}
diff --git a/plugin/lighthouse/handler.go b/plugin/lighthouse/handler.go
index 0a34507..0d86ace 100644
--- a/plugin/lighthouse/handler.go
+++ b/plugin/lighthouse/handler.go
@@ -46,7 +46,8 @@ func (lh *Lighthouse) ServeDNS(ctx context.Context, w dns.ResponseWriter, r *dns
 
 	var ips []string
 	var found bool
-	ip, found := lh.serviceImports.GetIP(pReq.namespace, pReq.service, pReq.cluster, lh.clusterStatus.IsConnected)
+	ip, found := lh.serviceImports.GetIP(pReq.namespace, pReq.service, pReq.cluster, lh.clusterStatus.IsConnected,
+		lh.endpointsStatus.IsHealthy)
 
 	if !found {
 		ips, found = lh.endpointSlices.GetIPs(pReq.hostname, pReq.cluster, pReq.namespace, pReq.service, lh.clusterStatus.IsConnected)
diff --git a/plugin/lighthouse/lighthouse.go b/plugin/lighthouse/lighthouse.go
index bfc599d..aec6279 100644
--- a/plugin/lighthouse/lighthouse.go
+++ b/plugin/lighthouse/lighthouse.go
@@ -25,17 +25,22 @@ var (
 var log = clog.NewWithPlugin("lighthouse")
 
 type Lighthouse struct {
-	Next           plugin.Handler
-	Fall           fall.F
-	Zones          []string
-	ttl            uint32
-	serviceImports *serviceimport.Map
-	endpointSlices *endpointslice.Map
-	clusterStatus  ClusterStatus
+	Next            plugin.Handler
+	Fall            fall.F
+	Zones           []string
+	ttl             uint32
+	serviceImports  *serviceimport.Map
+	endpointSlices  *endpointslice.Map
+	clusterStatus   ClusterStatus
+	endpointsStatus EndpointsStatus
 }
 
 type ClusterStatus interface {
 	IsConnected(clusterId string) bool
 }
 
+type EndpointsStatus interface {
+	IsHealthy(name, namespace, clusterId string) bool
+}
+
 var _ plugin.Handler = &Lighthouse{}
diff --git a/plugin/lighthouse/setup.go b/plugin/lighthouse/setup.go
index 6e589fd..9b85e6c 100644
--- a/plugin/lighthouse/setup.go
+++ b/plugin/lighthouse/setup.go
@@ -82,7 +82,8 @@ func lighthouseParse(c *caddy.Controller) (*Lighthouse, error) {
 		return nil
 	})
 
-	lh := &Lighthouse{ttl: defaultTtl, serviceImports: siMap, clusterStatus: gwController, endpointSlices: epMap}
+	lh := &Lighthouse{ttl: defaultTtl, serviceImports: siMap, clusterStatus: gwController, endpointSlices: epMap,
+		endpointsStatus: epController}
 
 	// Changed `for` to `if` to satisfy golint:
 	//	 SA4004: the surrounding loop is unconditionally terminated (staticcheck)


Test patch:
diff --git a/pkg/agent/controller/agent_test.go b/pkg/agent/controller/agent_test.go
index c770654..51f0745 100644
--- a/pkg/agent/controller/agent_test.go
+++ b/pkg/agent/controller/agent_test.go
@@ -118,23 +118,6 @@ var _ = Describe("ServiceImport syncing", func() {
 		})
 	})
 
-	When("the Service's pod endpoint IPs are lost and then reestablished", func() {
-		It("should clear and restore the ServiceImport's IPs", func() {
-			t.createService()
-			t.createEndpoints()
-			t.createServiceExport()
-			t.awaitServiceExported(t.service.Spec.ClusterIP, 0)
-
-			t.endpoints.Subsets[0].Addresses = nil
-			t.updateEndpoints()
-			t.awaitUpdatedServiceImport("")
-
-			t.endpoints.Subsets[0].Addresses = append(t.endpoints.Subsets[0].Addresses, corev1.EndpointAddress{IP: "192.168.5.10"})
-			t.updateEndpoints()
-			t.awaitUpdatedServiceImport("")
-		})
-	})
-
 	When("the ServiceExportCondition list count reaches MaxExportStatusConditions", func() {
 		var oldMaxExportStatusConditions int
 
diff --git a/pkg/endpointslice/controller_test.go b/pkg/endpointslice/controller_test.go
new file mode 100644
index 0000000..1efdc91
--- /dev/null
+++ b/pkg/endpointslice/controller_test.go
@@ -0,0 +1,158 @@
+package endpointslice_test
+
+import (
+	"time"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+	lhconstants "github.com/submariner-io/lighthouse/pkg/constants"
+	"github.com/submariner-io/lighthouse/pkg/endpointslice"
+	"k8s.io/api/discovery/v1beta1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+	fakeKubeClient "k8s.io/client-go/kubernetes/fake"
+	"k8s.io/client-go/rest"
+)
+
+const cluster1EndPointIP1 = "192.168.0.1"
+const cluster2EndPointIP1 = "192.169.0.1"
+const cluster1HostNamePod1 = "cl1host1"
+const cluster2HostNamePod1 = "cl2host1"
+const remoteClusterID1 = "west"
+const remoteClusterID2 = "south"
+const testName1 = "testName1"
+const testName2 = "testName2"
+const testService1 = "testService1"
+const testService2 = "testService2"
+const testNS1 = "testNameSpace1"
+const testNS2 = "testNameSpace2"
+
+var _ = Describe("EndpointSlice controller", func() {
+	t := newEndpointSliceTestDiver()
+
+	When("a service has a valid endpoint", func() {
+		When("IsHealthy is called for the service with a valid cluster", func() {
+			It("should return true", func() {
+				esName := testName1 + remoteClusterID1
+				endPoint1 := t.newEndpoint(cluster1HostNamePod1, cluster1EndPointIP1)
+				endpointSlice := t.newEndpointSliceFromEndpoint(testService1, remoteClusterID1, esName, testNS1, []v1beta1.Endpoint{endPoint1})
+				t.createEndpointSlice(testNS1, endpointSlice)
+				t.awaitIsHealthy(testService1, testNS1, remoteClusterID1)
+			})
+		})
+	})
+
+	When("IsHealthy is called for a non-existent service", func() {
+		It("should return false", func() {
+			Expect(t.controller.IsHealthy(testService1, testNS1, remoteClusterID1)).To(BeFalse())
+		})
+	})
+
+	When("IsHealthy is called for a service with no endpoints", func() {
+		It("should return false", func() {
+			esName := testName1 + remoteClusterID1
+			endpointSlice := t.newEndpointSliceFromEndpoint(testService1, remoteClusterID1, esName, testNS1, []v1beta1.Endpoint{})
+			t.createEndpointSlice(testNS1, endpointSlice)
+			t.awaitNotIsHealthy(testService1, testNS1, remoteClusterID1)
+		})
+	})
+
+	When("a service exists in multiple clusters with valid endpoints", func() {
+		When("IsHealthy is called for each cluster", func() {
+			It("should return true", func() {
+				esName1 := testName1 + remoteClusterID1
+				endPoint1 := t.newEndpoint(cluster1HostNamePod1, cluster1EndPointIP1)
+				endpointSlice := t.newEndpointSliceFromEndpoint(testService1, remoteClusterID1, esName1, testNS1, []v1beta1.Endpoint{endPoint1})
+				t.createEndpointSlice(testNS1, endpointSlice)
+
+				esName2 := testName2 + remoteClusterID2
+				endPoint2 := t.newEndpoint(cluster2HostNamePod1, cluster2EndPointIP1)
+				endpointSlice2 := t.newEndpointSliceFromEndpoint(testService2, remoteClusterID2, esName2, testNS2, []v1beta1.Endpoint{endPoint2})
+				t.createEndpointSlice(testNS2, endpointSlice2)
+
+				t.awaitIsHealthy(testService1, testNS1, remoteClusterID1)
+				t.awaitIsHealthy(testService2, testNS2, remoteClusterID2)
+			})
+		})
+	})
+
+	When("IsHealthy is called for a non-existent cluster", func() {
+		It("should return false", func() {
+			esName1 := testName1 + remoteClusterID1
+			endPoint1 := t.newEndpoint(cluster1HostNamePod1, cluster1EndPointIP1)
+			endpointSlice := t.newEndpointSliceFromEndpoint(testService1, remoteClusterID1, esName1, testNS1, []v1beta1.Endpoint{endPoint1})
+			t.createEndpointSlice(testNS1, endpointSlice)
+
+			t.awaitNotIsHealthy(testService1, testNS1, "randomcluster")
+		})
+	})
+})
+
+type endpointSliceTestDriver struct {
+	controller *endpointslice.Controller
+	kubeClient kubernetes.Interface
+	epMap      *endpointslice.Map
+}
+
+func newEndpointSliceTestDiver() *endpointSliceTestDriver {
+	t := &endpointSliceTestDriver{}
+
+	BeforeEach(func() {
+		t.kubeClient = fakeKubeClient.NewSimpleClientset()
+		t.epMap = endpointslice.NewMap()
+		t.controller = endpointslice.NewController(t.epMap)
+		t.controller.NewClientset = func(c *rest.Config) (kubernetes.Interface, error) {
+			return t.kubeClient, nil
+		}
+
+		Expect(t.controller.Start(&rest.Config{})).To(Succeed())
+	})
+
+	AfterEach(func() {
+		t.controller.Stop()
+	})
+
+	return t
+}
+
+func (t *endpointSliceTestDriver) createEndpointSlice(namespace string, endpointSlice *v1beta1.EndpointSlice) {
+	_, err := t.kubeClient.DiscoveryV1beta1().EndpointSlices(namespace).Create(endpointSlice)
+	Expect(err).To(Succeed())
+}
+
+func (t *endpointSliceTestDriver) newEndpointSliceFromEndpoint(serviceName, remoteCluster, esName,
+	esNs string, endPoints []v1beta1.Endpoint) *v1beta1.EndpointSlice {
+	return &v1beta1.EndpointSlice{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      esName,
+			Namespace: esNs,
+			Labels: map[string]string{
+				v1beta1.LabelManagedBy:           lhconstants.LabelValueManagedBy,
+				lhconstants.LabelSourceName:      serviceName,
+				lhconstants.LabelSourceCluster:   remoteCluster,
+				lhconstants.LabelSourceNamespace: esNs,
+			},
+		},
+		Endpoints: endPoints,
+	}
+}
+
+func (t *endpointSliceTestDriver) newEndpoint(hostname, ipaddress string) v1beta1.Endpoint {
+	return v1beta1.Endpoint{
+		Addresses:  []string{ipaddress},
+		Conditions: v1beta1.EndpointConditions{},
+		Hostname:   &hostname,
+	}
+}
+
+func (t *endpointSliceTestDriver) awaitIsHealthy(name, nameSpace, clusterId string) {
+	Eventually(func() bool {
+		return t.controller.IsHealthy(name, nameSpace, clusterId)
+	}, 5).Should(BeTrue())
+}
+
+func (t *endpointSliceTestDriver) awaitNotIsHealthy(name, nameSpace, clusterId string) {
+	Consistently(func() bool {
+		return t.controller.IsHealthy(name, nameSpace, clusterId)
+	}, 500*time.Millisecond).Should(BeFalse())
+}
diff --git a/pkg/serviceimport/map_test.go b/pkg/serviceimport/map_test.go
index 578e0be..a0013bd 100644
--- a/pkg/serviceimport/map_test.go
+++ b/pkg/serviceimport/map_test.go
@@ -20,21 +20,27 @@ var _ = Describe("ServiceImport Map", func() {
 	)
 
 	var (
-		clusterStatusMap map[string]bool
-		serviceImportMap *serviceimport.Map
+		clusterStatusMap  map[string]bool
+		serviceImportMap  *serviceimport.Map
+		endpointStatusMap map[string]bool
 	)
 
 	BeforeEach(func() {
 		clusterStatusMap = map[string]bool{clusterID1: true, clusterID2: true, clusterID3: true}
 		serviceImportMap = serviceimport.NewMap()
+		endpointStatusMap = map[string]bool{clusterID1: true, clusterID2: true, clusterID3: true}
 	})
 
 	checkCluster := func(id string) bool {
 		return clusterStatusMap[id]
 	}
 
+	checkEndpoint := func(name, namespace, id string) bool {
+		return endpointStatusMap[id]
+	}
+
 	getIPs := func(ns, name, cluster string) string {
-		ip, found := serviceImportMap.GetIP(ns, name, cluster, checkCluster)
+		ip, found := serviceImportMap.GetIP(ns, name, cluster, checkCluster, checkEndpoint)
 		Expect(found).To(BeTrue())
 		return ip
 	}
@@ -177,7 +183,7 @@ var _ = Describe("ServiceImport Map", func() {
 
 	When("a service does not exist", func() {
 		It("should return not found", func() {
-			_, found := serviceImportMap.GetIP(namespace1, service1, "", checkCluster)
+			_, found := serviceImportMap.GetIP(namespace1, service1, "", checkCluster, checkEndpoint)
 			Expect(found).To(BeFalse())
 		})
 	})
@@ -199,7 +205,7 @@ var _ = Describe("ServiceImport Map", func() {
 			Expect(getIP(namespace1, service1)).To(Equal(serviceIP1))
 
 			serviceImportMap.Remove(si)
-			_, found := serviceImportMap.GetIP(namespace1, service1, "", checkCluster)
+			_, found := serviceImportMap.GetIP(namespace1, service1, "", checkCluster, checkEndpoint)
 			Expect(found).To(BeFalse())
 
 			// Should be a no-op
diff --git a/plugin/lighthouse/handler_test.go b/plugin/lighthouse/handler_test.go
index e3e7f12..0d6beb9 100644
--- a/plugin/lighthouse/handler_test.go
+++ b/plugin/lighthouse/handler_test.go
@@ -54,6 +54,18 @@ func (m *MockClusterStatus) IsConnected(clusterId string) bool {
 	return m.clusterStatusMap[clusterId]
 }
 
+type MockEndpointStatus struct {
+	endpointStatusMap map[string]bool
+}
+
+func NewMockEndpointStatus() *MockEndpointStatus {
+	return &MockEndpointStatus{endpointStatusMap: make(map[string]bool)}
+}
+
+func (m *MockEndpointStatus) IsHealthy(name, namespace, clusterId string) bool {
+	return m.endpointStatusMap[clusterId]
+}
+
 func (w *FailingResponseWriter) WriteMsg(m *dns.Msg) error {
 	return errors.New(w.errorMsg)
 }
@@ -67,12 +79,16 @@ func testWithoutFallback() {
 	BeforeEach(func() {
 		mockCs := NewMockClusterStatus()
 		mockCs.clusterStatusMap[clusterID] = true
+		mockEs := NewMockEndpointStatus()
+		mockEs.endpointStatusMap[clusterID] = true
+
 		lh = &Lighthouse{
-			Zones:          []string{"clusterset.local."},
-			serviceImports: setupServiceImportMap(),
-			endpointSlices: setupEndpointSliceMap(),
-			clusterStatus:  mockCs,
-			ttl:            defaultTtl,
+			Zones:           []string{"clusterset.local."},
+			serviceImports:  setupServiceImportMap(),
+			endpointSlices:  setupEndpointSliceMap(),
+			clusterStatus:   mockCs,
+			endpointsStatus: mockEs,
+			ttl:             defaultTtl,
 		}
 
 		rec = dnstest.NewRecorder(&test.ResponseWriter{})
@@ -193,14 +209,17 @@ func testWithFallback() {
 	BeforeEach(func() {
 		mockCs := NewMockClusterStatus()
 		mockCs.clusterStatusMap[clusterID] = true
+		mockEs := NewMockEndpointStatus()
+		mockEs.endpointStatusMap[clusterID] = true
 		lh = &Lighthouse{
-			Zones:          []string{"clusterset.local."},
-			Fall:           fall.F{Zones: []string{"clusterset.local."}},
-			Next:           test.NextHandler(dns.RcodeBadCookie, errors.New("dummy plugin")),
-			serviceImports: setupServiceImportMap(),
-			endpointSlices: setupEndpointSliceMap(),
-			clusterStatus:  mockCs,
-			ttl:            defaultTtl,
+			Zones:           []string{"clusterset.local."},
+			Fall:            fall.F{Zones: []string{"clusterset.local."}},
+			Next:            test.NextHandler(dns.RcodeBadCookie, errors.New("dummy plugin")),
+			serviceImports:  setupServiceImportMap(),
+			endpointSlices:  setupEndpointSliceMap(),
+			clusterStatus:   mockCs,
+			endpointsStatus: mockEs,
+			ttl:             defaultTtl,
 		}
 
 		rec = dnstest.NewRecorder(&test.ResponseWriter{})
@@ -270,12 +289,16 @@ func testClusterStatus() {
 		mockCs = NewMockClusterStatus()
 		mockCs.clusterStatusMap[clusterID] = true
 		mockCs.clusterStatusMap[clusterID2] = true
+		mockEs := NewMockEndpointStatus()
+		mockEs.endpointStatusMap[clusterID] = true
+		mockEs.endpointStatusMap[clusterID2] = true
 		lh = &Lighthouse{
-			Zones:          []string{"clusterset.local."},
-			serviceImports: setupServiceImportMap(),
-			endpointSlices: setupEndpointSliceMap(),
-			clusterStatus:  mockCs,
-			ttl:            defaultTtl,
+			Zones:           []string{"clusterset.local."},
+			serviceImports:  setupServiceImportMap(),
+			endpointSlices:  setupEndpointSliceMap(),
+			clusterStatus:   mockCs,
+			endpointsStatus: mockEs,
+			ttl:             defaultTtl,
 		}
 		lh.serviceImports.Put(newServiceImport(namespace1, service1, clusterID2, serviceIP2, lighthousev2a1.ClusterSetIP))
 
@@ -364,17 +387,22 @@ func testHeadlessService() {
 		rec    *dnstest.Recorder
 		lh     *Lighthouse
 		mockCs *MockClusterStatus
+		mockEs *MockEndpointStatus
 	)
 
 	BeforeEach(func() {
 		mockCs = NewMockClusterStatus()
 		mockCs.clusterStatusMap[clusterID] = true
+		mockEs = NewMockEndpointStatus()
+		mockEs.endpointStatusMap[clusterID] = true
+		mockEs.endpointStatusMap[clusterID2] = true
 		lh = &Lighthouse{
-			Zones:          []string{"clusterset.local."},
-			serviceImports: serviceimport.NewMap(),
-			endpointSlices: setupEndpointSliceMap(),
-			clusterStatus:  mockCs,
-			ttl:            defaultTtl,
+			Zones:           []string{"clusterset.local."},
+			serviceImports:  serviceimport.NewMap(),
+			endpointSlices:  setupEndpointSliceMap(),
+			clusterStatus:   mockCs,
+			endpointsStatus: mockEs,
+			ttl:             defaultTtl,
 		}
 
 		rec = dnstest.NewRecorder(&test.ResponseWriter{})
diff --git a/plugin/lighthouse/setup_test.go b/plugin/lighthouse/setup_test.go
index dae5065..cf3b81b 100644
--- a/plugin/lighthouse/setup_test.go
+++ b/plugin/lighthouse/setup_test.go
@@ -4,6 +4,8 @@ import (
 	"context"
 	"errors"
 
+	"k8s.io/client-go/kubernetes"
+
 	"github.com/caddyserver/caddy"
 	"github.com/coredns/coredns/core/dnsserver"
 	"github.com/coredns/coredns/plugin/pkg/fall"
@@ -12,11 +14,13 @@ import (
 	. "github.com/onsi/gomega"
 	lighthouseClientset "github.com/submariner-io/lighthouse/pkg/client/clientset/versioned"
 	fakeLighthouseClientset "github.com/submariner-io/lighthouse/pkg/client/clientset/versioned/fake"
+	"github.com/submariner-io/lighthouse/pkg/endpointslice"
 	"github.com/submariner-io/lighthouse/pkg/gateway"
 	"github.com/submariner-io/lighthouse/pkg/serviceimport"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/dynamic"
 	fakeClient "k8s.io/client-go/dynamic/fake"
+	fakeKubeClient "k8s.io/client-go/kubernetes/fake"
 	"k8s.io/client-go/rest"
 )
 
@@ -40,6 +44,10 @@ var _ = Describe("Plugin setup", func() {
 		serviceimport.NewClientset = func(kubeConfig *rest.Config) (lighthouseClientset.Interface, error) {
 			return fakeLighthouseClientset.NewSimpleClientset(), nil
 		}
+
+		endpointslice.NewClientset = func(kubeConfig *rest.Config) (kubernetes.Interface, error) {
+			return fakeKubeClient.NewSimpleClientset(), nil
+		}
 	})
 
 	AfterEach(func() {