pilot/pkg/serviceregistry/kube/controller/serviceimportcache_test.go - dubbo-go-pixiu - Git at Google

 // Copyright Istio Authors
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
 //     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 // See the License for the specific language governing permissions and
 // limitations under the License.

 package controller

 import (
 	"context"
 	"errors"
 	"fmt"
 	"reflect"
 	"testing"
 	"time"
 )

 import (
 	envoyCore "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
 	. "github.com/onsi/gomega"
 	"istio.io/api/label"
 	kubeMeta "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/tools/cache"
 	mcsapi "sigs.k8s.io/mcs-api/pkg/apis/v1alpha1"
 )

 import (
 	"github.com/apache/dubbo-go-pixiu/pilot/pkg/features"
 	"github.com/apache/dubbo-go-pixiu/pilot/pkg/model"
 	"github.com/apache/dubbo-go-pixiu/pilot/pkg/serviceregistry/kube"
 	"github.com/apache/dubbo-go-pixiu/pkg/config/host"
 	"github.com/apache/dubbo-go-pixiu/pkg/kube/mcs"
 	"github.com/apache/dubbo-go-pixiu/pkg/test"
 	"github.com/apache/dubbo-go-pixiu/pkg/test/util/retry"
 )

 const (
 	serviceImportName      = "test-svc"
 	serviceImportNamespace = "test-ns"
 	serviceImportPodIP     = "128.0.0.2"
 	serviceImportCluster   = "test-cluster"
 )

 var (
 	serviceImportNamespacedName = types.NamespacedName{
 		Namespace: serviceImportNamespace,
 		Name:      serviceImportName,
 	}
 	serviceImportClusterSetHost = serviceClusterSetLocalHostname(serviceImportNamespacedName)
 	serviceImportVIPs           = []string{"1.1.1.1"}
 	serviceImportTimeout        = retry.Timeout(2 * time.Second)
 )

 func TestServiceNotImported(t *testing.T) {
 	for _, mode := range []EndpointMode{EndpointsOnly, EndpointSliceOnly} {
 		t.Run(mode.String(), func(t *testing.T) {
 			c, ic := newTestServiceImportCache(t, mode)

 			ic.createKubeService(t, c)

 			// Check that the service does not have ClusterSet IPs.
 			ic.checkServiceInstances(t)
 		})
 	}
 }

 func TestServiceImportedAfterCreated(t *testing.T) {
 	for _, mode := range []EndpointMode{EndpointsOnly, EndpointSliceOnly} {
 		t.Run(mode.String(), func(t *testing.T) {
 			c, ic := newTestServiceImportCache(t, mode)

 			ic.createKubeService(t, c)
 			ic.createServiceImport(t, mcsapi.ClusterSetIP, serviceImportVIPs)

 			// Check that the service has been assigned ClusterSet IPs.
 			ic.checkServiceInstances(t)
 		})
 	}
 }

 func TestServiceCreatedAfterImported(t *testing.T) {
 	for _, mode := range []EndpointMode{EndpointsOnly, EndpointSliceOnly} {
 		t.Run(mode.String(), func(t *testing.T) {
 			c, ic := newTestServiceImportCache(t, mode)

 			ic.createServiceImport(t, mcsapi.ClusterSetIP, serviceImportVIPs)
 			ic.createKubeService(t, c)

 			// Check that the service has been assigned ClusterSet IPs.
 			ic.checkServiceInstances(t)
 		})
 	}
 }

 func TestUpdateImportedService(t *testing.T) {
 	for _, mode := range []EndpointMode{EndpointsOnly, EndpointSliceOnly} {
 		t.Run(mode.String(), func(t *testing.T) {
 			c, ic := newTestServiceImportCache(t, mode)

 			ic.createKubeService(t, c)
 			ic.createServiceImport(t, mcsapi.ClusterSetIP, serviceImportVIPs)
 			ic.checkServiceInstances(t)

 			// Update the k8s service and verify that both services are updated.
 			ic.updateKubeService(t)
 		})
 	}
 }

 func TestHeadlessServiceImported(t *testing.T) {
 	for _, mode := range []EndpointMode{EndpointsOnly, EndpointSliceOnly} {
 		t.Run(mode.String(), func(t *testing.T) {
 			// Create and run the controller.
 			c, ic := newTestServiceImportCache(t, mode)

 			ic.createKubeService(t, c)
 			ic.createServiceImport(t, mcsapi.Headless, nil)

 			// Verify that we did not generate the synthetic service for the headless service.
 			ic.checkServiceInstances(t)
 		})
 	}
 }

 func TestDeleteImportedService(t *testing.T) {
 	for _, mode := range []EndpointMode{EndpointsOnly, EndpointSliceOnly} {
 		t.Run(mode.String(), func(t *testing.T) {
 			// Create and run the controller.
 			c1, ic := newTestServiceImportCache(t, mode)

 			// Create and run another controller.
 			c2, _ := NewFakeControllerWithOptions(FakeControllerOptions{
 				Stop:      c1.stop,
 				ClusterID: "test-cluster2",
 				Mode:      mode,
 			})
 			go c2.Run(c2.stop)
 			cache.WaitForCacheSync(c2.stop, c2.HasSynced)

 			c1.opts.MeshServiceController.AddRegistryAndRun(c2, c2.stop)

 			ic.createKubeService(t, c1)
 			ic.createServiceImport(t, mcsapi.ClusterSetIP, serviceImportVIPs)
 			ic.checkServiceInstances(t)

 			// create the same service in cluster2
 			createService(c2, serviceImportName, serviceImportNamespace, map[string]string{},
 				[]int32{8080}, map[string]string{"app": "prod-app"}, t)

 			// Delete the k8s service and verify that all internal services are removed.
 			ic.deleteKubeService(t, c2)
 		})
 	}
 }

 func TestUnimportService(t *testing.T) {
 	for _, mode := range []EndpointMode{EndpointsOnly, EndpointSliceOnly} {
 		t.Run(mode.String(), func(t *testing.T) {
 			// Create and run the controller.
 			c, ic := newTestServiceImportCache(t, mode)

 			ic.createKubeService(t, c)
 			ic.createServiceImport(t, mcsapi.ClusterSetIP, serviceImportVIPs)
 			ic.checkServiceInstances(t)

 			ic.unimportService(t)
 		})
 	}
 }

 func TestAddServiceImportVIPs(t *testing.T) {
 	for _, mode := range []EndpointMode{EndpointsOnly, EndpointSliceOnly} {
 		t.Run(mode.String(), func(t *testing.T) {
 			// Create and run the controller.
 			c, ic := newTestServiceImportCache(t, mode)

 			ic.createKubeService(t, c)
 			ic.createServiceImport(t, mcsapi.ClusterSetIP, nil)
 			ic.checkServiceInstances(t)

 			ic.setServiceImportVIPs(t, serviceImportVIPs)
 		})
 	}
 }

 func TestUpdateServiceImportVIPs(t *testing.T) {
 	for _, mode := range []EndpointMode{EndpointsOnly, EndpointSliceOnly} {
 		t.Run(mode.String(), func(t *testing.T) {
 			// Create and run the controller.
 			c, ic := newTestServiceImportCache(t, mode)

 			ic.createKubeService(t, c)
 			ic.createServiceImport(t, mcsapi.ClusterSetIP, serviceImportVIPs)
 			ic.checkServiceInstances(t)

 			updatedVIPs := []string{"1.1.1.1", "1.1.1.2"}
 			ic.setServiceImportVIPs(t, updatedVIPs)
 		})
 	}
 }

 func newTestServiceImportCache(t test.Failer, mode EndpointMode) (c *FakeController, ic *serviceImportCacheImpl) {
 	stopCh := make(chan struct{})
 	test.SetBoolForTest(t, &features.EnableMCSHost, true)
 	t.Cleanup(func() {
 		close(stopCh)
 	})

 	c, _ = NewFakeControllerWithOptions(FakeControllerOptions{
 		Stop:      stopCh,
 		ClusterID: serviceImportCluster,
 		Mode:      mode,
 	})
 	go c.Run(c.stop)
 	cache.WaitForCacheSync(c.stop, c.HasSynced)

 	ic = c.imports.(*serviceImportCacheImpl)
 	return
 }

 func (ic *serviceImportCacheImpl) createKubeService(t *testing.T, c *FakeController) {
 	t.Helper()

 	// Create the test service and endpoints.
 	createService(c, serviceImportName, serviceImportNamespace, map[string]string{},
 		[]int32{8080}, map[string]string{"app": "prod-app"}, t)
 	createEndpoints(t, c, serviceImportName, serviceImportNamespace, []string{"tcp-port"}, []string{serviceImportPodIP}, nil, nil)

 	isImported := ic.isImported(serviceImportNamespacedName)

 	// Wait for the resources to be processed by the controller.
 	retry.UntilSuccessOrFail(t, func() error {
 		clusterLocalHost := ic.clusterLocalHost()
 		if svc := c.GetService(clusterLocalHost); svc == nil {
 			return fmt.Errorf("failed looking up service for host %s", clusterLocalHost)
 		}

 		var expectedHosts map[host.Name]struct{}
 		if isImported {
 			expectedHosts = map[host.Name]struct{}{
 				clusterLocalHost:            {},
 				serviceImportClusterSetHost: {},
 			}
 		} else {
 			expectedHosts = map[host.Name]struct{}{
 				clusterLocalHost: {},
 			}
 		}

 		instances := ic.getProxyServiceInstances()
 		if len(instances) != len(expectedHosts) {
 			return fmt.Errorf("expected 1 service instance, found %d", len(instances))
 		}
 		for _, si := range instances {
 			if si.Service == nil {
 				return fmt.Errorf("proxy ServiceInstance has nil service")
 			}
 			if si.Endpoint == nil {
 				return fmt.Errorf("proxy ServiceInstance has nil endpoint")
 			}
 			if _, found := expectedHosts[si.Service.Hostname]; !found {
 				return fmt.Errorf("found proxy ServiceInstance for unexpected host: %s", si.Service.Hostname)
 			}
 			delete(expectedHosts, si.Service.Hostname)
 		}

 		if len(expectedHosts) > 0 {
 			return fmt.Errorf("failed to find proxy ServiceInstances for hosts: %v", expectedHosts)
 		}

 		return nil
 	}, serviceImportTimeout)
 }

 func (ic *serviceImportCacheImpl) updateKubeService(t *testing.T) {
 	t.Helper()
 	svc, _ := ic.client.CoreV1().Services(serviceImportNamespace).Get(context.TODO(), serviceImportName, kubeMeta.GetOptions{})
 	if svc == nil {
 		t.Fatalf("failed to find k8s service: %s/%s", serviceImportNamespace, serviceImportName)
 	}

 	// Just add a new label.
 	svc.Labels = map[string]string{
 		"foo": "bar",
 	}
 	if _, err := ic.client.CoreV1().Services(serviceImportNamespace).Update(context.TODO(), svc, kubeMeta.UpdateOptions{}); err != nil {
 		t.Fatal(err)
 	}

 	hostNames := []host.Name{
 		ic.clusterLocalHost(),
 		serviceImportClusterSetHost,
 	}

 	// Wait for the services to pick up the label.
 	retry.UntilSuccessOrFail(t, func() error {
 		for _, hostName := range hostNames {
 			svc := ic.GetService(hostName)
 			if svc == nil {
 				return fmt.Errorf("failed to find service for host %s", hostName)
 			}
 			if svc.Attributes.Labels["foo"] != "bar" {
 				return fmt.Errorf("service not updated for %s", hostName)
 			}
 		}

 		return nil
 	}, serviceImportTimeout)
 }

 func (ic *serviceImportCacheImpl) deleteKubeService(t *testing.T, anotherCluster *FakeController) {
 	t.Helper()

 	if err := anotherCluster.client.CoreV1().Services(serviceImportNamespace).Delete(context.TODO(), serviceImportName, kubeMeta.DeleteOptions{}); err != nil {
 		t.Fatal(err)
 	}
 	// Wait for the resources to be processed by the controller.
 	if err := ic.client.CoreV1().Services(serviceImportNamespace).Delete(context.TODO(), serviceImportName, kubeMeta.DeleteOptions{}); err != nil {
 		t.Fatal(err)
 	}

 	// Wait for the resources to be processed by the controller.
 	retry.UntilSuccessOrFail(t, func() error {
 		if svc := ic.GetService(ic.clusterLocalHost()); svc != nil {
 			return fmt.Errorf("found deleted service for host %s", ic.clusterLocalHost())
 		}
 		if svc := ic.GetService(serviceImportClusterSetHost); svc != nil {
 			return fmt.Errorf("found deleted service for host %s", serviceImportClusterSetHost)
 		}

 		instances := ic.getProxyServiceInstances()
 		if len(instances) != 0 {
 			return fmt.Errorf("expected 0 service instance, found %d", len(instances))
 		}

 		return nil
 	}, serviceImportTimeout)
 }

 func (ic *serviceImportCacheImpl) getProxyServiceInstances() []*model.ServiceInstance {
 	return ic.GetProxyServiceInstances(&model.Proxy{
 		Type:            model.SidecarProxy,
 		IPAddresses:     []string{serviceImportPodIP},
 		Locality:        &envoyCore.Locality{Region: "r", Zone: "z"},
 		ConfigNamespace: serviceImportNamespace,
 		Metadata: &model.NodeMetadata{
 			ServiceAccount: "account",
 			ClusterID:      ic.Cluster(),
 			Labels: map[string]string{
 				"app":                      "prod-app",
 				label.SecurityTlsMode.Name: "mutual",
 			},
 		},
 	})
 }

 func (ic *serviceImportCacheImpl) getServiceImport(t *testing.T) *mcsapi.ServiceImport {
 	t.Helper()

 	// Get the ServiceImport as unstructured
 	u, err := ic.client.Dynamic().Resource(mcs.ServiceImportGVR).Namespace(serviceImportNamespace).Get(
 		context.TODO(), serviceImportName, kubeMeta.GetOptions{})
 	if err != nil {
 		return nil
 	}

 	// Convert to ServiceImport
 	si := &mcsapi.ServiceImport{}
 	if err := runtime.DefaultUnstructuredConverter.FromUnstructured(u.Object, si); err != nil {
 		t.Fatal(err)
 	}
 	return si
 }

 func (ic *serviceImportCacheImpl) checkServiceInstances(t *testing.T) {
 	t.Helper()
 	g := NewWithT(t)

 	si := ic.getServiceImport(t)

 	var expectedIPs []string
 	expectedServiceCount := 1
 	expectMCSService := false
 	if si != nil && si.Spec.Type == mcsapi.ClusterSetIP && len(si.Spec.IPs) > 0 {
 		expectedIPs = si.Spec.IPs
 		expectedServiceCount = 2
 		expectMCSService = true
 	}

 	instances := ic.getProxyServiceInstances()
 	g.Expect(instances).To(HaveLen(expectedServiceCount))

 	for _, inst := range instances {
 		svc := inst.Service
 		if svc.Hostname == serviceImportClusterSetHost {
 			if !expectMCSService {
 				t.Fatalf("found ServiceInstance for unimported service %s", serviceImportClusterSetHost)
 			}
 			// Check the ClusterSet IPs.
 			g.Expect(svc.ClusterVIPs.GetAddressesFor(ic.Cluster())).To(Equal(expectedIPs))
 			return
 		}
 	}

 	if expectMCSService {
 		t.Fatalf("failed finding ServiceInstance for %s", serviceImportClusterSetHost)
 	}
 }

 func (ic *serviceImportCacheImpl) createServiceImport(t *testing.T, importType mcsapi.ServiceImportType, vips []string) {
 	t.Helper()

 	// Create the ServiceImport resource in the cluster.
 	_, err := ic.client.Dynamic().Resource(mcs.ServiceImportGVR).Namespace(serviceImportNamespace).Create(context.TODO(),
 		newServiceImport(importType, vips),
 		kubeMeta.CreateOptions{})
 	if err != nil {
 		t.Fatal(err)
 	}

 	shouldCreateMCSService := importType == mcsapi.ClusterSetIP && len(vips) > 0 &&
 		ic.GetService(ic.clusterLocalHost()) != nil

 	// Wait for the import to be processed by the controller.
 	retry.UntilSuccessOrFail(t, func() error {
 		if !ic.isImported(serviceImportNamespacedName) {
 			return fmt.Errorf("serviceImport not found for %s", serviceImportClusterSetHost)
 		}
 		if shouldCreateMCSService && ic.GetService(serviceImportClusterSetHost) == nil {
 			return fmt.Errorf("failed to find service for %s", serviceImportClusterSetHost)
 		}
 		return nil
 	}, serviceImportTimeout)

 	if shouldCreateMCSService {
 		// Wait for the XDS event.
 		ic.waitForXDS(t)
 	}
 }

 func (ic *serviceImportCacheImpl) setServiceImportVIPs(t *testing.T, vips []string) {
 	t.Helper()

 	// Get the ServiceImport
 	si := ic.getServiceImport(t)

 	// Apply the ClusterSet IPs.
 	si.Spec.IPs = vips
 	if _, err := ic.client.Dynamic().Resource(mcs.ServiceImportGVR).Namespace(serviceImportNamespace).Update(
 		context.TODO(), toUnstructured(si), kubeMeta.UpdateOptions{}); err != nil {
 		t.Fatal(err)
 	}

 	if len(vips) > 0 {
 		// Wait for the import to be processed by the controller.
 		retry.UntilSuccessOrFail(t, func() error {
 			svc := ic.GetService(serviceImportClusterSetHost)
 			if svc == nil {
 				return fmt.Errorf("failed to find service for %s", serviceImportClusterSetHost)
 			}

 			actualVIPs := svc.ClusterVIPs.GetAddressesFor(ic.Cluster())
 			if !reflect.DeepEqual(vips, actualVIPs) {
 				return fmt.Errorf("expected ClusterSet VIPs %v, but found %v", vips, actualVIPs)
 			}
 			return nil
 		}, serviceImportTimeout)

 		// Wait for the XDS event.
 		ic.waitForXDS(t)
 	} else {
 		// Wait for the import to be processed by the controller.
 		retry.UntilSuccessOrFail(t, func() error {
 			if svc := ic.GetService(serviceImportClusterSetHost); svc != nil {
 				return fmt.Errorf("found unexpected service for %s", serviceImportClusterSetHost)
 			}
 			return nil
 		}, serviceImportTimeout)
 	}
 }

 func (ic *serviceImportCacheImpl) unimportService(t *testing.T) {
 	t.Helper()

 	if err := ic.client.Dynamic().Resource(mcs.ServiceImportGVR).Namespace(serviceImportNamespace).Delete(
 		context.TODO(), serviceImportName, kubeMeta.DeleteOptions{}); err != nil {
 		t.Fatal(err)
 	}

 	// Wait for the import to be processed by the controller.
 	retry.UntilSuccessOrFail(t, func() error {
 		if ic.isImported(serviceImportNamespacedName) {
 			return fmt.Errorf("serviceImport found for %s", serviceImportClusterSetHost)
 		}
 		if ic.GetService(serviceImportClusterSetHost) != nil {
 			return fmt.Errorf("found MCS service for unimported service %s", serviceImportClusterSetHost)
 		}
 		return nil
 	}, serviceImportTimeout)
 }

 func (ic *serviceImportCacheImpl) isImported(name types.NamespacedName) bool {
 	_, err := ic.lister.ByNamespace(name.Namespace).Get(name.Name)
 	return err == nil
 }

 func (ic *serviceImportCacheImpl) waitForXDS(t *testing.T) {
 	t.Helper()

 	retry.UntilSuccessOrFail(t, func() error {
 		return ic.checkXDS()
 	}, serviceImportTimeout)
 }

 func (ic *serviceImportCacheImpl) checkXDS() error {
 	event := ic.opts.XDSUpdater.(*FakeXdsUpdater).Wait("service")
 	if event == nil {
 		return errors.New("failed waiting for XDS event")
 	}

 	// The name of the event will be the cluster-local hostname.
 	eventID := serviceImportClusterSetHost.String()
 	if event.ID != eventID {
 		return fmt.Errorf("waitForXDS failed: expected event id=%s, but found %s", eventID, event.ID)
 	}
 	return nil
 }

 func (ic *serviceImportCacheImpl) clusterLocalHost() host.Name {
 	return kube.ServiceHostname(serviceImportName, serviceImportNamespace, ic.opts.DomainSuffix)
 }

 func newServiceImport(importType mcsapi.ServiceImportType, vips []string) *unstructured.Unstructured {
 	si := &mcsapi.ServiceImport{
 		TypeMeta: kubeMeta.TypeMeta{
 			Kind:       "ServiceImport",
 			APIVersion: "multicluster.x-k8s.io/v1alpha1",
 		},
 		ObjectMeta: kubeMeta.ObjectMeta{
 			Name:      serviceImportName,
 			Namespace: serviceImportNamespace,
 		},
 		Spec: mcsapi.ServiceImportSpec{
 			Type: importType,
 			IPs:  vips,
 		},
 	}
 	return toUnstructured(si)
 }

 func toUnstructured(o interface{}) *unstructured.Unstructured {
 	u, err := runtime.DefaultUnstructuredConverter.ToUnstructured(o)
 	if err != nil {
 		panic(err)
 	}
 	return &unstructured.Unstructured{Object: u}
 }
	// Copyright Istio Authors
	//
	// Licensed under the Apache License, Version 2.0 (the "License");
	// you may not use this file except in compliance with the License.
	// You may obtain a copy of the License at
	//
	// http://www.apache.org/licenses/LICENSE-2.0
	//
	// Unless required by applicable law or agreed to in writing, software
	// distributed under the License is distributed on an "AS IS" BASIS,
	// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	// See the License for the specific language governing permissions and
	// limitations under the License.

	package controller

	import (
	"context"
	"errors"
	"fmt"
	"reflect"
	"testing"
	"time"
	)

	import (
	envoyCore "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
	. "github.com/onsi/gomega"
	"istio.io/api/label"
	kubeMeta "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/client-go/tools/cache"
	mcsapi "sigs.k8s.io/mcs-api/pkg/apis/v1alpha1"
	)

	import (
	"github.com/apache/dubbo-go-pixiu/pilot/pkg/features"
	"github.com/apache/dubbo-go-pixiu/pilot/pkg/model"
	"github.com/apache/dubbo-go-pixiu/pilot/pkg/serviceregistry/kube"
	"github.com/apache/dubbo-go-pixiu/pkg/config/host"
	"github.com/apache/dubbo-go-pixiu/pkg/kube/mcs"
	"github.com/apache/dubbo-go-pixiu/pkg/test"
	"github.com/apache/dubbo-go-pixiu/pkg/test/util/retry"
	)

	const (
	serviceImportName = "test-svc"
	serviceImportNamespace = "test-ns"
	serviceImportPodIP = "128.0.0.2"
	serviceImportCluster = "test-cluster"
	)

	var (
	serviceImportNamespacedName = types.NamespacedName{
	Namespace: serviceImportNamespace,
	Name: serviceImportName,
	}
	serviceImportClusterSetHost = serviceClusterSetLocalHostname(serviceImportNamespacedName)
	serviceImportVIPs = []string{"1.1.1.1"}
	serviceImportTimeout = retry.Timeout(2 * time.Second)
	)

	func TestServiceNotImported(t *testing.T) {
	for _, mode := range []EndpointMode{EndpointsOnly, EndpointSliceOnly} {
	t.Run(mode.String(), func(t *testing.T) {
	c, ic := newTestServiceImportCache(t, mode)

	ic.createKubeService(t, c)

	// Check that the service does not have ClusterSet IPs.
	ic.checkServiceInstances(t)
	})
	}
	}

	func TestServiceImportedAfterCreated(t *testing.T) {
	for _, mode := range []EndpointMode{EndpointsOnly, EndpointSliceOnly} {
	t.Run(mode.String(), func(t *testing.T) {
	c, ic := newTestServiceImportCache(t, mode)

	ic.createKubeService(t, c)
	ic.createServiceImport(t, mcsapi.ClusterSetIP, serviceImportVIPs)

	// Check that the service has been assigned ClusterSet IPs.
	ic.checkServiceInstances(t)
	})
	}
	}

	func TestServiceCreatedAfterImported(t *testing.T) {
	for _, mode := range []EndpointMode{EndpointsOnly, EndpointSliceOnly} {
	t.Run(mode.String(), func(t *testing.T) {
	c, ic := newTestServiceImportCache(t, mode)

	ic.createServiceImport(t, mcsapi.ClusterSetIP, serviceImportVIPs)
	ic.createKubeService(t, c)

	// Check that the service has been assigned ClusterSet IPs.
	ic.checkServiceInstances(t)
	})
	}
	}

	func TestUpdateImportedService(t *testing.T) {
	for _, mode := range []EndpointMode{EndpointsOnly, EndpointSliceOnly} {
	t.Run(mode.String(), func(t *testing.T) {
	c, ic := newTestServiceImportCache(t, mode)

	ic.createKubeService(t, c)
	ic.createServiceImport(t, mcsapi.ClusterSetIP, serviceImportVIPs)
	ic.checkServiceInstances(t)

	// Update the k8s service and verify that both services are updated.
	ic.updateKubeService(t)
	})
	}
	}

	func TestHeadlessServiceImported(t *testing.T) {
	for _, mode := range []EndpointMode{EndpointsOnly, EndpointSliceOnly} {
	t.Run(mode.String(), func(t *testing.T) {
	// Create and run the controller.
	c, ic := newTestServiceImportCache(t, mode)

	ic.createKubeService(t, c)
	ic.createServiceImport(t, mcsapi.Headless, nil)

	// Verify that we did not generate the synthetic service for the headless service.
	ic.checkServiceInstances(t)
	})
	}
	}

	func TestDeleteImportedService(t *testing.T) {
	for _, mode := range []EndpointMode{EndpointsOnly, EndpointSliceOnly} {
	t.Run(mode.String(), func(t *testing.T) {
	// Create and run the controller.
	c1, ic := newTestServiceImportCache(t, mode)

	// Create and run another controller.
	c2, _ := NewFakeControllerWithOptions(FakeControllerOptions{
	Stop: c1.stop,
	ClusterID: "test-cluster2",
	Mode: mode,
	})
	go c2.Run(c2.stop)
	cache.WaitForCacheSync(c2.stop, c2.HasSynced)

	c1.opts.MeshServiceController.AddRegistryAndRun(c2, c2.stop)

	ic.createKubeService(t, c1)
	ic.createServiceImport(t, mcsapi.ClusterSetIP, serviceImportVIPs)
	ic.checkServiceInstances(t)

	// create the same service in cluster2
	createService(c2, serviceImportName, serviceImportNamespace, map[string]string{},
	[]int32{8080}, map[string]string{"app": "prod-app"}, t)

	// Delete the k8s service and verify that all internal services are removed.
	ic.deleteKubeService(t, c2)
	})
	}
	}

	func TestUnimportService(t *testing.T) {
	for _, mode := range []EndpointMode{EndpointsOnly, EndpointSliceOnly} {
	t.Run(mode.String(), func(t *testing.T) {
	// Create and run the controller.
	c, ic := newTestServiceImportCache(t, mode)

	ic.createKubeService(t, c)
	ic.createServiceImport(t, mcsapi.ClusterSetIP, serviceImportVIPs)
	ic.checkServiceInstances(t)

	ic.unimportService(t)
	})
	}
	}

	func TestAddServiceImportVIPs(t *testing.T) {
	for _, mode := range []EndpointMode{EndpointsOnly, EndpointSliceOnly} {
	t.Run(mode.String(), func(t *testing.T) {
	// Create and run the controller.
	c, ic := newTestServiceImportCache(t, mode)

	ic.createKubeService(t, c)
	ic.createServiceImport(t, mcsapi.ClusterSetIP, nil)
	ic.checkServiceInstances(t)

	ic.setServiceImportVIPs(t, serviceImportVIPs)
	})
	}
	}

	func TestUpdateServiceImportVIPs(t *testing.T) {
	for _, mode := range []EndpointMode{EndpointsOnly, EndpointSliceOnly} {
	t.Run(mode.String(), func(t *testing.T) {
	// Create and run the controller.
	c, ic := newTestServiceImportCache(t, mode)

	ic.createKubeService(t, c)
	ic.createServiceImport(t, mcsapi.ClusterSetIP, serviceImportVIPs)
	ic.checkServiceInstances(t)

	updatedVIPs := []string{"1.1.1.1", "1.1.1.2"}
	ic.setServiceImportVIPs(t, updatedVIPs)
	})
	}
	}

	func newTestServiceImportCache(t test.Failer, mode EndpointMode) (c FakeController, ic serviceImportCacheImpl) {
	stopCh := make(chan struct{})
	test.SetBoolForTest(t, &features.EnableMCSHost, true)
	t.Cleanup(func() {
	close(stopCh)
	})

	c, _ = NewFakeControllerWithOptions(FakeControllerOptions{
	Stop: stopCh,
	ClusterID: serviceImportCluster,
	Mode: mode,
	})
	go c.Run(c.stop)
	cache.WaitForCacheSync(c.stop, c.HasSynced)

	ic = c.imports.(*serviceImportCacheImpl)
	return
	}

	func (ic serviceImportCacheImpl) createKubeService(t testing.T, c *FakeController) {
	t.Helper()

	// Create the test service and endpoints.
	createService(c, serviceImportName, serviceImportNamespace, map[string]string{},
	[]int32{8080}, map[string]string{"app": "prod-app"}, t)
	createEndpoints(t, c, serviceImportName, serviceImportNamespace, []string{"tcp-port"}, []string{serviceImportPodIP}, nil, nil)

	isImported := ic.isImported(serviceImportNamespacedName)

	// Wait for the resources to be processed by the controller.
	retry.UntilSuccessOrFail(t, func() error {
	clusterLocalHost := ic.clusterLocalHost()
	if svc := c.GetService(clusterLocalHost); svc == nil {
	return fmt.Errorf("failed looking up service for host %s", clusterLocalHost)
	}

	var expectedHosts map[host.Name]struct{}
	if isImported {
	expectedHosts = map[host.Name]struct{}{
	clusterLocalHost: {},
	serviceImportClusterSetHost: {},
	}
	} else {
	expectedHosts = map[host.Name]struct{}{
	clusterLocalHost: {},
	}
	}

	instances := ic.getProxyServiceInstances()
	if len(instances) != len(expectedHosts) {
	return fmt.Errorf("expected 1 service instance, found %d", len(instances))
	}
	for _, si := range instances {
	if si.Service == nil {
	return fmt.Errorf("proxy ServiceInstance has nil service")
	}
	if si.Endpoint == nil {
	return fmt.Errorf("proxy ServiceInstance has nil endpoint")
	}
	if _, found := expectedHosts[si.Service.Hostname]; !found {
	return fmt.Errorf("found proxy ServiceInstance for unexpected host: %s", si.Service.Hostname)
	}
	delete(expectedHosts, si.Service.Hostname)
	}

	if len(expectedHosts) > 0 {
	return fmt.Errorf("failed to find proxy ServiceInstances for hosts: %v", expectedHosts)
	}

	return nil
	}, serviceImportTimeout)
	}

	func (ic serviceImportCacheImpl) updateKubeService(t testing.T) {
	t.Helper()
	svc, _ := ic.client.CoreV1().Services(serviceImportNamespace).Get(context.TODO(), serviceImportName, kubeMeta.GetOptions{})
	if svc == nil {
	t.Fatalf("failed to find k8s service: %s/%s", serviceImportNamespace, serviceImportName)
	}

	// Just add a new label.
	svc.Labels = map[string]string{
	"foo": "bar",
	}
	if _, err := ic.client.CoreV1().Services(serviceImportNamespace).Update(context.TODO(), svc, kubeMeta.UpdateOptions{}); err != nil {
	t.Fatal(err)
	}

	hostNames := []host.Name{
	ic.clusterLocalHost(),
	serviceImportClusterSetHost,
	}

	// Wait for the services to pick up the label.
	retry.UntilSuccessOrFail(t, func() error {
	for _, hostName := range hostNames {
	svc := ic.GetService(hostName)
	if svc == nil {
	return fmt.Errorf("failed to find service for host %s", hostName)
	}
	if svc.Attributes.Labels["foo"] != "bar" {
	return fmt.Errorf("service not updated for %s", hostName)
	}
	}

	return nil
	}, serviceImportTimeout)
	}

	func (ic serviceImportCacheImpl) deleteKubeService(t testing.T, anotherCluster *FakeController) {
	t.Helper()

	if err := anotherCluster.client.CoreV1().Services(serviceImportNamespace).Delete(context.TODO(), serviceImportName, kubeMeta.DeleteOptions{}); err != nil {
	t.Fatal(err)
	}
	// Wait for the resources to be processed by the controller.
	if err := ic.client.CoreV1().Services(serviceImportNamespace).Delete(context.TODO(), serviceImportName, kubeMeta.DeleteOptions{}); err != nil {
	t.Fatal(err)
	}

	// Wait for the resources to be processed by the controller.
	retry.UntilSuccessOrFail(t, func() error {
	if svc := ic.GetService(ic.clusterLocalHost()); svc != nil {
	return fmt.Errorf("found deleted service for host %s", ic.clusterLocalHost())
	}
	if svc := ic.GetService(serviceImportClusterSetHost); svc != nil {
	return fmt.Errorf("found deleted service for host %s", serviceImportClusterSetHost)
	}

	instances := ic.getProxyServiceInstances()
	if len(instances) != 0 {
	return fmt.Errorf("expected 0 service instance, found %d", len(instances))
	}

	return nil
	}, serviceImportTimeout)
	}

	func (ic serviceImportCacheImpl) getProxyServiceInstances() []model.ServiceInstance {
	return ic.GetProxyServiceInstances(&model.Proxy{
	Type: model.SidecarProxy,
	IPAddresses: []string{serviceImportPodIP},
	Locality: &envoyCore.Locality{Region: "r", Zone: "z"},
	ConfigNamespace: serviceImportNamespace,
	Metadata: &model.NodeMetadata{
	ServiceAccount: "account",
	ClusterID: ic.Cluster(),
	Labels: map[string]string{
	"app": "prod-app",
	label.SecurityTlsMode.Name: "mutual",
	},
	},
	})
	}

	func (ic serviceImportCacheImpl) getServiceImport(t testing.T) *mcsapi.ServiceImport {
	t.Helper()

	// Get the ServiceImport as unstructured
	u, err := ic.client.Dynamic().Resource(mcs.ServiceImportGVR).Namespace(serviceImportNamespace).Get(
	context.TODO(), serviceImportName, kubeMeta.GetOptions{})
	if err != nil {
	return nil
	}

	// Convert to ServiceImport
	si := &mcsapi.ServiceImport{}
	if err := runtime.DefaultUnstructuredConverter.FromUnstructured(u.Object, si); err != nil {
	t.Fatal(err)
	}
	return si
	}

	func (ic serviceImportCacheImpl) checkServiceInstances(t testing.T) {
	t.Helper()
	g := NewWithT(t)

	si := ic.getServiceImport(t)

	var expectedIPs []string
	expectedServiceCount := 1
	expectMCSService := false
	if si != nil && si.Spec.Type == mcsapi.ClusterSetIP && len(si.Spec.IPs) > 0 {
	expectedIPs = si.Spec.IPs
	expectedServiceCount = 2
	expectMCSService = true
	}

	instances := ic.getProxyServiceInstances()
	g.Expect(instances).To(HaveLen(expectedServiceCount))

	for _, inst := range instances {
	svc := inst.Service
	if svc.Hostname == serviceImportClusterSetHost {
	if !expectMCSService {
	t.Fatalf("found ServiceInstance for unimported service %s", serviceImportClusterSetHost)
	}
	// Check the ClusterSet IPs.
	g.Expect(svc.ClusterVIPs.GetAddressesFor(ic.Cluster())).To(Equal(expectedIPs))
	return
	}
	}

	if expectMCSService {
	t.Fatalf("failed finding ServiceInstance for %s", serviceImportClusterSetHost)
	}
	}

	func (ic serviceImportCacheImpl) createServiceImport(t testing.T, importType mcsapi.ServiceImportType, vips []string) {
	t.Helper()

	// Create the ServiceImport resource in the cluster.
	_, err := ic.client.Dynamic().Resource(mcs.ServiceImportGVR).Namespace(serviceImportNamespace).Create(context.TODO(),
	newServiceImport(importType, vips),
	kubeMeta.CreateOptions{})
	if err != nil {
	t.Fatal(err)
	}

	shouldCreateMCSService := importType == mcsapi.ClusterSetIP && len(vips) > 0 &&
	ic.GetService(ic.clusterLocalHost()) != nil

	// Wait for the import to be processed by the controller.
	retry.UntilSuccessOrFail(t, func() error {
	if !ic.isImported(serviceImportNamespacedName) {
	return fmt.Errorf("serviceImport not found for %s", serviceImportClusterSetHost)
	}
	if shouldCreateMCSService && ic.GetService(serviceImportClusterSetHost) == nil {
	return fmt.Errorf("failed to find service for %s", serviceImportClusterSetHost)
	}
	return nil
	}, serviceImportTimeout)

	if shouldCreateMCSService {
	// Wait for the XDS event.
	ic.waitForXDS(t)
	}
	}

	func (ic serviceImportCacheImpl) setServiceImportVIPs(t testing.T, vips []string) {
	t.Helper()

	// Get the ServiceImport
	si := ic.getServiceImport(t)

	// Apply the ClusterSet IPs.
	si.Spec.IPs = vips
	if _, err := ic.client.Dynamic().Resource(mcs.ServiceImportGVR).Namespace(serviceImportNamespace).Update(
	context.TODO(), toUnstructured(si), kubeMeta.UpdateOptions{}); err != nil {
	t.Fatal(err)
	}

	if len(vips) > 0 {
	// Wait for the import to be processed by the controller.
	retry.UntilSuccessOrFail(t, func() error {
	svc := ic.GetService(serviceImportClusterSetHost)
	if svc == nil {
	return fmt.Errorf("failed to find service for %s", serviceImportClusterSetHost)
	}

	actualVIPs := svc.ClusterVIPs.GetAddressesFor(ic.Cluster())
	if !reflect.DeepEqual(vips, actualVIPs) {
	return fmt.Errorf("expected ClusterSet VIPs %v, but found %v", vips, actualVIPs)
	}
	return nil
	}, serviceImportTimeout)

	// Wait for the XDS event.
	ic.waitForXDS(t)
	} else {
	// Wait for the import to be processed by the controller.
	retry.UntilSuccessOrFail(t, func() error {
	if svc := ic.GetService(serviceImportClusterSetHost); svc != nil {
	return fmt.Errorf("found unexpected service for %s", serviceImportClusterSetHost)
	}
	return nil
	}, serviceImportTimeout)
	}
	}

	func (ic serviceImportCacheImpl) unimportService(t testing.T) {
	t.Helper()

	if err := ic.client.Dynamic().Resource(mcs.ServiceImportGVR).Namespace(serviceImportNamespace).Delete(
	context.TODO(), serviceImportName, kubeMeta.DeleteOptions{}); err != nil {
	t.Fatal(err)
	}

	// Wait for the import to be processed by the controller.
	retry.UntilSuccessOrFail(t, func() error {
	if ic.isImported(serviceImportNamespacedName) {
	return fmt.Errorf("serviceImport found for %s", serviceImportClusterSetHost)
	}
	if ic.GetService(serviceImportClusterSetHost) != nil {
	return fmt.Errorf("found MCS service for unimported service %s", serviceImportClusterSetHost)
	}
	return nil
	}, serviceImportTimeout)
	}

	func (ic *serviceImportCacheImpl) isImported(name types.NamespacedName) bool {
	_, err := ic.lister.ByNamespace(name.Namespace).Get(name.Name)
	return err == nil
	}

	func (ic serviceImportCacheImpl) waitForXDS(t testing.T) {
	t.Helper()

	retry.UntilSuccessOrFail(t, func() error {
	return ic.checkXDS()
	}, serviceImportTimeout)
	}

	func (ic *serviceImportCacheImpl) checkXDS() error {
	event := ic.opts.XDSUpdater.(*FakeXdsUpdater).Wait("service")
	if event == nil {
	return errors.New("failed waiting for XDS event")
	}

	// The name of the event will be the cluster-local hostname.
	eventID := serviceImportClusterSetHost.String()
	if event.ID != eventID {
	return fmt.Errorf("waitForXDS failed: expected event id=%s, but found %s", eventID, event.ID)
	}
	return nil
	}

	func (ic *serviceImportCacheImpl) clusterLocalHost() host.Name {
	return kube.ServiceHostname(serviceImportName, serviceImportNamespace, ic.opts.DomainSuffix)
	}

	func newServiceImport(importType mcsapi.ServiceImportType, vips []string) *unstructured.Unstructured {
	si := &mcsapi.ServiceImport{
	TypeMeta: kubeMeta.TypeMeta{
	Kind: "ServiceImport",
	APIVersion: "multicluster.x-k8s.io/v1alpha1",
	},
	ObjectMeta: kubeMeta.ObjectMeta{
	Name: serviceImportName,
	Namespace: serviceImportNamespace,
	},
	Spec: mcsapi.ServiceImportSpec{
	Type: importType,
	IPs: vips,
	},
	}
	return toUnstructured(si)
	}

	func toUnstructured(o interface{}) *unstructured.Unstructured {
	u, err := runtime.DefaultUnstructuredConverter.ToUnstructured(o)
	if err != nil {
	panic(err)
	}
	return &unstructured.Unstructured{Object: u}
	}