Make service mirror controller per target cluster

cli/cmd/install.go

@@ -438,7 +438,7 @@ func (options *installOptions) validateAndBuild(stage string, flags *pflag.FlagS
 
 	if options.enableEndpointSlices {
 		if err = validateEndpointSlicesFeature(); err != nil {
-			return nil, nil, fmt.Errorf("--enableEndpointSlice=true not supported:%s", err)
+			return nil, nil, fmt.Errorf("--enableEndpointSlice=true not supported: %s", err)
 		}
 	}
 

cli/cmd/multicluster.go

@@ -542,6 +542,14 @@ func newLinkCommand() *cobra.Command {
 			for _, ingress := range gateway.Status.LoadBalancer.Ingress {
 				gatewayAddresses = append(gatewayAddresses, ingress.IP)
 			}
+			if len(gatewayAddresses) == 0 {
+				return fmt.Errorf("Gateway %s.%s has no ingress addresses", gateway.Name, gateway.Namespace)
+			}
+
+			gatewayIdentity, ok := gateway.Annotations[k8s.GatewayIdentity]
+			if !ok || gatewayIdentity == "" {
+				return fmt.Errorf("Gatway %s.%s has no %s annotation", gateway.Name, gateway.Namespace, k8s.GatewayIdentity)
+			}
 
 			probeSpec, err := mc.ExtractProbeSpec(gateway)
 			if err != nil {
@@ -562,7 +570,7 @@ func newLinkCommand() *cobra.Command {
 				ClusterCredentialsSecret:      fmt.Sprintf("cluster-credentials-%s", opts.clusterName),
 				GatewayAddress:                strings.Join(gatewayAddresses, ","),
 				GatewayPort:                   gatewayPort,
-				GatewayIdentity:               gateway.Annotations[k8s.GatewayIdentity],
+				GatewayIdentity:               gatewayIdentity,
 				ProbeSpec:                     probeSpec,
 			}
 

controller/api/destination/watcher/endpoints_watcher.go

@@ -129,7 +129,6 @@ var endpointsVecs = newEndpointsMetricsVecs()
 // NewEndpointsWatcher creates an EndpointsWatcher and begins watching the
 // k8sAPI for pod, service, and endpoint changes. An EndpointsWatcher will
 // watch on Endpoints or EndpointSlice resources, depending on cluster configuration.
-//TODO: Allow EndpointSlice resources to be used once opt-in functionality is supported.
 func NewEndpointsWatcher(k8sAPI *k8s.API, log *logging.Entry, enableEndpointSlices bool) *EndpointsWatcher {
 	ew := &EndpointsWatcher{
 		publishers:           make(map[ServiceID]*servicePublisher),
@@ -161,7 +160,6 @@ func NewEndpointsWatcher(k8sAPI *k8s.API, log *logging.Entry, enableEndpointSlic
 			UpdateFunc: ew.updateEndpointSlice,
 		})
 	} else {
-		// ew.log.Debugf("Cluster does not have EndpointSlice access:%v", err)
 		ew.log.Debugf("Watching Endpoints resources")
 		k8sAPI.Endpoint().Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
 			AddFunc:    ew.addEndpoints,

controller/cmd/destination/main.go

@@ -67,7 +67,17 @@ func Main(args []string) {
 		}
 	}
 
+	// we need to create a separate client to check for EndpointSlice access in k8s cluster
+	// when slices are enabled and registered, k8sAPI is initialized with 'ES' resource
+	k8Client, err := pkgK8s.NewAPI(*kubeConfigPath, "", "", []string{}, 0)
+	if err != nil {
+		log.Fatalf("Failed to initialize K8s API Client: %s", err)
+	}
 	enableEndpointSlices := global.GetEndpointSliceEnabled()
+	err = pkgK8s.EndpointSliceAccess(k8Client)
+	if enableEndpointSlices && err != nil {
+		log.Fatalf("Failed to start with EndpointSlices enabled: %s", err)
+	}
 
 	var k8sAPI *k8s.API
 	if enableEndpointSlices {
@@ -85,11 +95,6 @@ func Main(args []string) {
 		log.Fatalf("Failed to initialize K8s API: %s", err)
 	}
 
-	err = consts.EndpointSliceAccess(k8sAPI.Client)
-	if enableEndpointSlices && err != nil {
-		log.Fatalf("Failed to start with EndpointSlices enabled: %s", err)
-	}
-
 	server := destination.NewServer(
 		*addr,
 		*controllerNamespace,

controller/cmd/service-mirror/cluster_watcher.go

@@ -259,10 +259,7 @@ func (rcsw *RemoteClusterServiceWatcher) cleanupOrphanedServices() error {
 // created. This piece of code is responsible for doing just that. It takes care of
 // services, endpoints and namespaces (if needed)
 func (rcsw *RemoteClusterServiceWatcher) cleanupMirroredResources() error {
-	matchLabels := map[string]string{
-		consts.MirroredResourceLabel:  "true",
-		consts.RemoteClusterNameLabel: rcsw.link.TargetClusterName,
-	}
+	matchLabels := rcsw.getMirroredServiceLabels()
 
 	services, err := rcsw.localAPIClient.Svc().Lister().List(labels.Set(matchLabels).AsSelector())
 	if err != nil {
@@ -315,20 +312,14 @@ func (rcsw *RemoteClusterServiceWatcher) cleanupMirroredResources() error {
 // Deletes a locally mirrored service as it is not present on the remote cluster anymore
 func (rcsw *RemoteClusterServiceWatcher) handleRemoteServiceDeleted(ev *RemoteServiceDeleted) error {
 	localServiceName := rcsw.mirroredResourceName(ev.Name)
-	rcsw.log.Infof("Deleting mirrored service %s/%s and its corresponding Endpoints", ev.Namespace, localServiceName)
+	rcsw.log.Infof("Deleting mirrored service %s/%s", ev.Namespace, localServiceName)
 	var errors []error
 	if err := rcsw.localAPIClient.Client.CoreV1().Services(ev.Namespace).Delete(localServiceName, &metav1.DeleteOptions{}); err != nil {
 		if !kerrors.IsNotFound(err) {
 			errors = append(errors, fmt.Errorf("could not delete Service: %s/%s: %s", ev.Namespace, localServiceName, err))
 		}
 	}
 
-	if err := rcsw.localAPIClient.Client.CoreV1().Endpoints(ev.Namespace).Delete(localServiceName, &metav1.DeleteOptions{}); err != nil {
-		if !kerrors.IsNotFound(err) {
-			errors = append(errors, fmt.Errorf("could not delete Endpoints: %s/%s: %s", ev.Namespace, localServiceName, err))
-		}
-	}
-
 	if len(errors) > 0 {
 		return RetryableError{errors}
 	}
@@ -350,11 +341,10 @@ func (rcsw *RemoteClusterServiceWatcher) handleRemoteServiceUpdated(ev *RemoteSe
 		},
 	}
 
-	if rcsw.link.GatewayIdentity != "" {
-		copiedEndpoints.Annotations[consts.RemoteGatewayIdentity] = rcsw.link.GatewayIdentity
-	} else {
-		delete(copiedEndpoints.Annotations, consts.RemoteGatewayIdentity)
+	if copiedEndpoints.Annotations == nil {
+		copiedEndpoints.Annotations = make(map[string]string)
 	}
+	copiedEndpoints.Annotations[consts.RemoteGatewayIdentity] = rcsw.link.GatewayIdentity
 
 	if _, err := rcsw.localAPIClient.Client.CoreV1().Endpoints(copiedEndpoints.Namespace).Update(copiedEndpoints); err != nil {
 		return RetryableError{[]error{err}}
@@ -646,6 +636,11 @@ func (rcsw *RemoteClusterServiceWatcher) Start() error {
 	)
 	go rcsw.processEvents()
 
+	// We need to issue a RepairEndpoints immediately to populate the gateway
+	// mirror endpoints.
+	ev := RepairEndpoints{}
+	rcsw.eventsQueue.Add(&ev)
+
 	go func() {
 		ticker := time.NewTicker(rcsw.repairPeriod)
 		for {
@@ -746,11 +741,10 @@ func (rcsw *RemoteClusterServiceWatcher) repairEndpoints() {
 			},
 		}
 
-		if rcsw.link.GatewayIdentity != "" {
-			updatedEndpoints.Annotations[consts.RemoteGatewayIdentity] = rcsw.link.GatewayIdentity
-		} else {
-			delete(updatedEndpoints.Annotations, consts.RemoteGatewayIdentity)
+		if updatedEndpoints.Annotations == nil {
+			updatedEndpoints.Annotations = make(map[string]string)
 		}
+		updatedEndpoints.Annotations[consts.RemoteGatewayIdentity] = rcsw.link.GatewayIdentity
 
 		_, err = rcsw.localAPIClient.Client.CoreV1().Services(updatedService.Namespace).Update(updatedService)
 		if err != nil {
@@ -777,12 +771,12 @@ func (rcsw *RemoteClusterServiceWatcher) createOrUpdateEndpoints(ep *corev1.Endp
 		} else {
 			return err
 		}
-	} else {
-		// Exists so we should update it.
-		_, err = rcsw.localAPIClient.Client.CoreV1().Endpoints(ep.Namespace).Update(ep)
-		if err != nil {
-			return err
-		}
 	}
+	// Exists so we should update it.
+	_, err = rcsw.localAPIClient.Client.CoreV1().Endpoints(ep.Namespace).Update(ep)
+	if err != nil {
+		return err
+	}
+
 	return nil
 }

controller/cmd/service-mirror/cluster_watcher_mirroring_test.go

@@ -51,15 +51,8 @@ func (tc *mirroringTestCase) run(t *testing.T) {
 		}
 
 		if tc.expectedLocalEndpoints == nil {
-			// ensure the are no local endpoints
-			endpoints, err := localAPI.Client.CoreV1().Endpoints(corev1.NamespaceAll).List(metav1.ListOptions{})
-			if err != nil {
-				t.Fatal(err)
-			}
-			if len(endpoints.Items) > 0 {
-				t.Fatalf("Was expecting no local endpoints but instead found %d", len(endpoints.Items))
-
-			}
+			// In a real Kubernetes cluster, deleting the service is sufficient
+			// to delete the endpoints.
 		} else {
 			for _, expected := range tc.expectedLocalEndpoints {
 				actual, err := localAPI.Client.CoreV1().Endpoints(expected.Namespace).Get(expected.Name, metav1.GetOptions{})

pkg/healthcheck/healthcheck_multicluster.go

@@ -109,7 +109,6 @@ func (hc *HealthChecker) multiClusterCategory() []category {
 				{
 					description: "all gateway mirrors are healthy",
 					hintAnchor:  "l5d-multicluster-gateways-endpoints",
-					warning:     true,
 					check: func(ctx context.Context) error {
 						if hc.Options.MultiCluster {
 							return hc.checkIfGatewayMirrorsHaveEndpoints(ctx)

pkg/k8s/authz.go

@@ -109,14 +109,12 @@ func ServiceProfilesAccess(k8sClient kubernetes.Interface) error {
 
 // EndpointSliceAccess verifies whether the K8s cluster has
 // access to EndpointSlice resources.
-//TODO: Uncomment function and change return type once EndpointSlices
-// are supported and made opt-in through install flag
 func EndpointSliceAccess(k8sClient kubernetes.Interface) error {
-	// gv := discovery.SchemeGroupVersion.String()
-	// res, err := k8sClient.Discovery().ServerResourcesForGroupVersion(gv)
-	// if err != nil {
-	// 	return err
-	// }
+	gv := discovery.SchemeGroupVersion.String()
+	res, err := k8sClient.Discovery().ServerResourcesForGroupVersion(gv)
+	if err != nil {
+		return err
+	}
 
 	if res.GroupVersion == gv {
 		for _, apiRes := range res.APIResources {
@@ -126,8 +124,7 @@ func EndpointSliceAccess(k8sClient kubernetes.Interface) error {
 		}
 	}
 
-	// return errors.New("EndpointSlice resource not found")
-	return errors.New("EndpointSlice not supported")
+	return errors.New("EndpointSlice resource not found")
 }
 
 func checkEndpointSlicesExist(k8sClient kubernetes.Interface) error {