Add remote-discovery service mirroring (#11201)

We add the ability to mirror services in "remote discovery" mode where no Endpoints are created for the service in the source cluster, but instead the `multicluster.linkerd.io/remote-discovery` and `multicluster.linkerd.io/remote-service` labels are set on the mirror service to indicate that the control plane should perform remote discovery for this service. To accomplish this, we add a new field to the Link resource: `remoteDiscoverySelector` which is a parallel to `selector` but instead it selects Services to export in remote discovery mode. Since this field is purely additive, we do not change the Link CRD version. By treating an empty selector as "Nothing", we remain backwards compatible (an unset `remoteDiscoverySelector` will not export any services in remote discovery mode). Signed-off-by: Alex Leong <[email protected]>
linkerd · Aug 8, 2023 · 084e955 · 084e955
1 parent bc8b4f2
commit 084e955
Show file tree

Hide file tree

Showing 10 changed files with 440 additions and 120 deletions.
diff --git a/multicluster/charts/linkerd-multicluster/templates/link-crd.yaml b/multicluster/charts/linkerd-multicluster/templates/link-crd.yaml
@@ -77,6 +77,34 @@ spec:
                           type: array
                           items:
                             type: string
+              remoteDiscoverySelector:
+                description: Selector for Services to mirror in remote discovery mode
+                type: object
+                properties:
+                  matchLabels:
+                    type: object
+                    x-kubernetes-preserve-unknown-fields: true
+                  matchExpressions:
+                    description: List of selector requirements
+                    type: array
+                    items:
+                      description: A selector item requires a key and an operator
+                      type: object
+                      required:
+                      - key
+                      - operator
+                      properties:
+                        key:
+                          description: Label key that selector should apply to
+                          type: string
+                        operator:
+                          description: Evaluation of a label in relation to set
+                          type: string
+                          enum: [In, NotIn, Exists, DoesNotExist]
+                        values:
+                          type: array
+                          items:
+                            type: string
               targetClusterName:
                 description: Name of target cluster to link to
                 type: string

diff --git a/multicluster/cmd/link.go b/multicluster/cmd/link.go
@@ -47,6 +47,7 @@ type (
 		controlPlaneVersion     string
 		dockerRegistry          string
 		selector                string
+		remoteDiscoverySelector string
 		gatewayAddresses        string
 		gatewayPort             uint32
 		ha                      bool
@@ -250,6 +251,11 @@ A full list of configurable values can be found at https://github.com/linkerd/li
 				return err
 			}
 
+			remoteDiscoverySelector, err := metav1.ParseToLabelSelector(opts.remoteDiscoverySelector)
+			if err != nil {
+				return err
+			}
+
 			link := mc.Link{
 				Name:                          opts.clusterName,
 				Namespace:                     opts.namespace,
@@ -262,6 +268,7 @@ A full list of configurable values can be found at https://github.com/linkerd/li
 				GatewayIdentity:               gatewayIdentity,
 				ProbeSpec:                     probeSpec,
 				Selector:                      *selector,
+				RemoteDiscoverySelector:       *remoteDiscoverySelector,
 			}
 
 			obj, err := link.ToUnstructured()
@@ -326,6 +333,7 @@ A full list of configurable values can be found at https://github.com/linkerd/li
 	cmd.Flags().StringVar(&opts.dockerRegistry, "registry", opts.dockerRegistry,
 		fmt.Sprintf("Docker registry to pull service mirror controller image from ($%s)", flags.EnvOverrideDockerRegistry))
 	cmd.Flags().StringVarP(&opts.selector, "selector", "l", opts.selector, "Selector (label query) to filter which services in the target cluster to mirror")
+	cmd.Flags().StringVar(&opts.remoteDiscoverySelector, "remote-discovery-selector", opts.remoteDiscoverySelector, "Selector (label query) to filter which services in the target cluster to mirror in remote discovery mode")
 	cmd.Flags().StringVar(&opts.gatewayAddresses, "gateway-addresses", opts.gatewayAddresses, "If specified, overwrites gateway addresses when gateway service is not type LoadBalancer (comma separated list)")
 	cmd.Flags().Uint32Var(&opts.gatewayPort, "gateway-port", opts.gatewayPort, "If specified, overwrites gateway port when gateway service is not type LoadBalancer")
 	cmd.Flags().BoolVar(&opts.ha, "ha", opts.ha, "Enable HA configuration for the service-mirror deployment (default false)")
@@ -424,6 +432,7 @@ func newLinkOptionsWithDefault() (*linkOptions, error) {
 		logLevel:                defaults.LogLevel,
 		logFormat:               defaults.LogFormat,
 		selector:                fmt.Sprintf("%s=%s", k8s.DefaultExportedServiceSelector, "true"),
+		remoteDiscoverySelector: fmt.Sprintf("%s=%s", k8s.DefaultExportedServiceSelector, "remote-discovery"),
 		gatewayAddresses:        "",
 		gatewayPort:             0,
 		ha:                      false,

diff --git a/multicluster/cmd/testdata/install_default.golden b/multicluster/cmd/testdata/install_default.golden
diff --git a/multicluster/cmd/testdata/install_ha.golden b/multicluster/cmd/testdata/install_ha.golden
diff --git a/multicluster/cmd/testdata/install_psp.golden b/multicluster/cmd/testdata/install_psp.golden
diff --git a/multicluster/service-mirror/cluster_watcher.go b/multicluster/service-mirror/cluster_watcher.go
@@ -235,6 +235,11 @@ func (rcsw *RemoteClusterServiceWatcher) getMirroredServiceLabels(remoteService
 		return labels
 	}
 
+	if rcsw.isRemoteDiscovery(remoteService) {
+		labels[consts.RemoteDiscoveryLabel] = rcsw.link.TargetClusterName
+		labels[consts.RemoteServiceLabel] = remoteService.GetName()
+	}
+
 	for key, value := range remoteService.ObjectMeta.Labels {
 		if strings.HasPrefix(key, consts.SvcMirrorPrefix) {
 			continue
@@ -430,27 +435,50 @@ func (rcsw *RemoteClusterServiceWatcher) handleRemoteServiceDeleted(ctx context.
 // new gateway being assigned or additional ports exposed. This method takes care of that.
 func (rcsw *RemoteClusterServiceWatcher) handleRemoteServiceUpdated(ctx context.Context, ev *RemoteServiceUpdated) error {
 	rcsw.log.Infof("Updating mirror service %s/%s", ev.localService.Namespace, ev.localService.Name)
-	gatewayAddresses, err := rcsw.resolveGatewayAddress()
-	if err != nil {
-		return err
-	}
 
-	copiedEndpoints := ev.localEndpoints.DeepCopy()
-	copiedEndpoints.Subsets = []corev1.EndpointSubset{
-		{
-			Addresses: gatewayAddresses,
-			Ports:     rcsw.getEndpointsPorts(ev.remoteUpdate),
-		},
-	}
+	if rcsw.isRemoteDiscovery(ev.remoteUpdate) {
+		// The service is mirrored in remote discovery mode and any local
+		// endpoints for it should be deleted if they exist.
+		if ev.localEndpoints != nil {
+			err := rcsw.localAPIClient.Client.CoreV1().Endpoints(ev.localService.Namespace).Delete(ctx, ev.localService.Name, metav1.DeleteOptions{})
+			if err != nil {
+				return RetryableError{[]error{
+					fmt.Errorf("failed to delete mirror endpoints for %s/%s: %w", ev.localService.Namespace, ev.localService.Name, err),
+				}}
+			}
+		}
+	} else if ev.localEndpoints == nil {
+		// The service is mirrored in gateway mode and gateway endpoints should
+		// be created for it.
+		err := rcsw.createGatewayEndpoints(ctx, ev.remoteUpdate)
+		if err != nil {
+			return err
+		}
+	} else {
+		// The service is mirrored in gateway mode and gateway endpoints already
+		// exist for it but may need to be updated.
+		gatewayAddresses, err := rcsw.resolveGatewayAddress()
+		if err != nil {
+			return err
+		}
 
-	if copiedEndpoints.Annotations == nil {
-		copiedEndpoints.Annotations = make(map[string]string)
-	}
-	copiedEndpoints.Annotations[consts.RemoteGatewayIdentity] = rcsw.link.GatewayIdentity
+		copiedEndpoints := ev.localEndpoints.DeepCopy()
+		copiedEndpoints.Subsets = []corev1.EndpointSubset{
+			{
+				Addresses: gatewayAddresses,
+				Ports:     rcsw.getEndpointsPorts(ev.remoteUpdate),
+			},
+		}
 
-	err = rcsw.updateMirrorEndpoints(ctx, copiedEndpoints)
-	if err != nil {
-		return RetryableError{[]error{err}}
+		if copiedEndpoints.Annotations == nil {
+			copiedEndpoints.Annotations = make(map[string]string)
+		}
+		copiedEndpoints.Annotations[consts.RemoteGatewayIdentity] = rcsw.link.GatewayIdentity
+
+		err = rcsw.updateMirrorEndpoints(ctx, copiedEndpoints)
+		if err != nil {
+			return RetryableError{[]error{err}}
+		}
 	}
 
 	ev.localService.Labels = rcsw.getMirroredServiceLabels(ev.remoteUpdate)
@@ -518,6 +546,10 @@ func (rcsw *RemoteClusterServiceWatcher) handleRemoteServiceCreated(ctx context.
 		}
 	}
 
+	if rcsw.isRemoteDiscovery(remoteService) {
+		// For remote discovery services, skip creating gateway endpoints.
+		return nil
+	}
 	return rcsw.createGatewayEndpoints(ctx, remoteService)
 }
 
@@ -657,15 +689,19 @@ func (rcsw *RemoteClusterServiceWatcher) createOrUpdateService(service *corev1.S
 		lastMirroredRemoteVersion, ok := localService.Annotations[consts.RemoteResourceVersionAnnotation]
 		if ok && lastMirroredRemoteVersion != service.ResourceVersion {
 			endpoints, err := rcsw.localAPIClient.Endpoint().Lister().Endpoints(service.Namespace).Get(localName)
-			if err == nil {
-				rcsw.eventsQueue.Add(&RemoteServiceUpdated{
-					localService:   localService,
-					localEndpoints: endpoints,
-					remoteUpdate:   service,
-				})
-				return nil
+			if err != nil {
+				if kerrors.IsNotFound(err) {
+					endpoints = nil
+				} else {
+					return RetryableError{[]error{err}}
+				}
 			}
-			return RetryableError{[]error{err}}
+			rcsw.eventsQueue.Add(&RemoteServiceUpdated{
+				localService:   localService,
+				localEndpoints: endpoints,
+				remoteUpdate:   service,
+			})
+			return nil
 		}
 
 		return nil
@@ -1174,10 +1210,35 @@ func (rcsw *RemoteClusterServiceWatcher) updateReadiness(endpoints *corev1.Endpo
 }
 
 func (rcsw *RemoteClusterServiceWatcher) isExported(l map[string]string) bool {
+	// Treat an empty selector as "Nothing" instead of "Everything" so that
+	// when the selector field is unset, we don't export all Services.
+	if len(rcsw.link.Selector.MatchExpressions)+len(rcsw.link.Selector.MatchLabels) == 0 {
+		return false
+	}
 	selector, err := metav1.LabelSelectorAsSelector(&rcsw.link.Selector)
 	if err != nil {
 		rcsw.log.Errorf("Invalid selector: %s", err)
 		return false
 	}
-	return selector.Matches(labels.Set(l))
+	remoteDiscoverySelector, err := metav1.LabelSelectorAsSelector(&rcsw.link.RemoteDiscoverySelector)
+	if err != nil {
+		rcsw.log.Errorf("Invalid selector: %s", err)
+		return false
+	}
+	return selector.Matches(labels.Set(l)) || remoteDiscoverySelector.Matches(labels.Set(l))
+}
+
+func (rcsw *RemoteClusterServiceWatcher) isRemoteDiscovery(svc *corev1.Service) bool {
+	// Treat an empty remoteDisocverySelector as "Nothing" instead of
+	// "Everything" so that when the remoteDiscoverySelector field is unset, we
+	// don't export all Services.
+	if len(rcsw.link.RemoteDiscoverySelector.MatchExpressions)+len(rcsw.link.RemoteDiscoverySelector.MatchLabels) == 0 {
+		return false
+	}
+	remoteDiscoverySelector, err := metav1.LabelSelectorAsSelector(&rcsw.link.RemoteDiscoverySelector)
+	if err != nil {
+		rcsw.log.Errorf("Invalid selector: %s", err)
+		return false
+	}
+	return remoteDiscoverySelector.Matches(labels.Set(svc.Labels))
 }