Forward porting Gloo service lookup by labels to 1.19.x (#10669)

Author: ryanrolds

changelog/v1.19.0-beta13/discover-gloo-service-with-labels.yaml

@@ -0,0 +1,9 @@
+changelog:
+- type: FIX
+  description: >-
+    Discover the Gloo service using labels.
+    In some environments, services must be renamed.
+    This change allows the service to still be discovered when it's been renamed.
+    In the event that multiple services in the namespace have the gloo=gloo label, an error will occur.
+  issueLink: https://github.com/solo-io/solo-projects/issues/7646
+  resolvesIssue: false

pkg/utils/kubeutils/names.go

@@ -1,12 +1,20 @@
 package kubeutils
 
 const (
-	GlooDeploymentName = "gloo"
-	GlooServiceName    = "gloo"
+	GlooDeploymentName   = "gloo"
+	GlooServiceName      = "gloo"
+	GlooServiceAppLabel  = "gloo"
+	GlooServiceGlooLabel = "gloo"
 
 	// GlooXdsPortName is the name of the port in the Gloo Gateway control plane Kubernetes Service that serves xDS config.
 	// See: install/helm/gloo/templates/2-gloo-service.yaml
 	GlooXdsPortName = "grpc-xds"
 
 	DiscoveryDeploymentName = "discovery"
 )
+
+var (
+	GlooServiceLabels = map[string]string{
+		"gloo": GlooServiceGlooLabel,
+	}
+)

projects/gloo/pkg/syncer/setup/controlplane.go

@@ -7,10 +7,8 @@ import (
 
 	"github.com/rotisserie/eris"
 	"github.com/solo-io/gloo/pkg/utils/kubeutils"
-	"github.com/solo-io/gloo/pkg/utils/namespaces"
 	"github.com/solo-io/solo-kit/pkg/api/v1/clients"
 	skkube "github.com/solo-io/solo-kit/pkg/api/v1/resources/common/kubernetes"
-	apierrors "k8s.io/apimachinery/pkg/api/errors"
 )
 
 var (
@@ -19,42 +17,51 @@ var (
 		return eris.Wrapf(NoXdsPortFoundError, "no port with the name %s found in service %s.%s", portName, svcNamespace, svcName)
 	}
 	NoGlooSvcFoundError = eris.New("failed to find Gloo service")
-	noGlooSvcFoundError = func(err error, svcNamespace string, svcName string) error {
-		wrapped := eris.Wrap(err, NoGlooSvcFoundError.Error())
-		return eris.Wrapf(wrapped, "service %s.%s", svcNamespace, svcName)
+	noGlooSvcFoundError = func(svcNamespace string) error {
+		return eris.Wrapf(NoGlooSvcFoundError, "service in %s with gloo=gloo label", svcNamespace)
+	}
+	MultipleGlooSvcFoundError = eris.New("found multiple Gloo services")
+	multipleGlooSvcFoundError = func(svcNamespace string) error {
+		return eris.Wrapf(MultipleGlooSvcFoundError, "found multiple services in %s with gloo=glo label", svcNamespace)
 	}
 )
 
-// GetControlPlaneXdsPort gets the xDS port from the gloo Service.
-func GetControlPlaneXdsPort(ctx context.Context, svcClient skkube.ServiceClient) (int32, error) {
-	// When this code is invoked from within the running Pod, this will contain the namespace where Gloo is running
-	svcNamespace := namespaces.GetPodNamespace()
-	return GetNamespacedControlPlaneXdsPort(ctx, svcNamespace, svcClient)
-}
-
-// GetNamespacedControlPlaneXdsPort gets the xDS port from the Gloo Service, provided the namespace the Service is running in
-func GetNamespacedControlPlaneXdsPort(ctx context.Context, svcNamespace string, svcClient skkube.ServiceClient) (int32, error) {
-	glooSvc, err := svcClient.Read(svcNamespace, kubeutils.GlooServiceName, clients.ReadOpts{Ctx: ctx})
+func GetControlPlaneService(ctx context.Context, svcNamespace string, svcClient skkube.ServiceClient) (*skkube.Service, error) {
+	opts := clients.ListOpts{
+		Ctx:      ctx,
+		Selector: kubeutils.GlooServiceLabels,
+	}
+	services, err := svcClient.List(svcNamespace, opts)
 	if err != nil {
-		if apierrors.IsNotFound(err) {
-			return 0, noGlooSvcFoundError(err, svcNamespace, kubeutils.GlooServiceName)
-		}
-		return 0, err
+		return nil, err
 	}
 
+	if len(services) == 0 {
+		return nil, noGlooSvcFoundError(svcNamespace)
+	}
+
+	if len(services) > 1 {
+		return nil, multipleGlooSvcFoundError(svcNamespace)
+	}
+
+	return services[0], nil
+}
+
+// GetControlPlaneXdsPort gets the xDS port from the Gloo Service
+func GetControlPlaneXdsPort(service *skkube.Service) (int32, error) {
 	// find the xds port on the Gloo Service
-	for _, port := range glooSvc.Spec.Ports {
+	for _, port := range service.Spec.Ports {
 		if port.Name == kubeutils.GlooXdsPortName {
 			return port.Port, nil
 		}
 	}
-	return 0, noXdsPortFoundError(kubeutils.GlooXdsPortName, svcNamespace, kubeutils.GlooServiceName)
+	return 0, noXdsPortFoundError(kubeutils.GlooXdsPortName, service.Namespace, service.Name)
 }
 
 // GetControlPlaneXdsHost gets the xDS address from the gloo Service.
-func GetControlPlaneXdsHost() string {
+func GetControlPlaneXdsHost(service *skkube.Service) string {
 	return kubeutils.ServiceFQDN(metav1.ObjectMeta{
-		Name:      kubeutils.GlooServiceName,
-		Namespace: namespaces.GetPodNamespace(),
+		Name:      service.Name,
+		Namespace: service.Namespace,
 	})
 }

projects/gloo/pkg/syncer/setup/controlplane_test.go

@@ -7,40 +7,23 @@ import (
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 	"github.com/solo-io/gloo/pkg/utils/kubeutils"
+	"github.com/solo-io/gloo/projects/gloo/pkg/defaults"
 	"github.com/solo-io/gloo/projects/gloo/pkg/syncer/setup"
 	"github.com/solo-io/solo-kit/pkg/api/v1/clients"
 	"github.com/solo-io/solo-kit/pkg/api/v1/clients/factory"
 	"github.com/solo-io/solo-kit/pkg/api/v1/clients/memory"
 	skkube "github.com/solo-io/solo-kit/pkg/api/v1/resources/common/kubernetes"
-	skerrors "github.com/solo-io/solo-kit/pkg/errors"
 	"github.com/solo-io/solo-kit/pkg/utils/statusutils"
 	corev1 "k8s.io/api/core/v1"
 )
 
-var _ = Describe("ControlPlane", func() {
-
-	Context("xds host", func() {
-
-		AfterEach(func() {
-			err := os.Unsetenv(statusutils.PodNamespaceEnvName)
-			Expect(err).NotTo(HaveOccurred())
-		})
-
-		It("respects POD_NAMESPACE", func() {
-			err := os.Setenv(statusutils.PodNamespaceEnvName, "other-ns")
-			Expect(err).NotTo(HaveOccurred())
-			xdsHost := setup.GetControlPlaneXdsHost()
-			Expect(xdsHost).To(Equal("gloo.other-ns.svc.cluster.local"))
-		})
-
-		It("uses default value when POD_NAMESPACE not set", func() {
-			xdsHost := setup.GetControlPlaneXdsHost()
-			Expect(xdsHost).To(Equal("gloo.gloo-system.svc.cluster.local"))
-		})
-	})
+const (
+	otherNS = "other-ns"
+)
 
-	Context("xds port", func() {
+var _ = Describe("ControlPlane", func() {
 
+	Context("xds service", func() {
 		var (
 			ctx       context.Context
 			svcClient skkube.ServiceClient
@@ -58,7 +41,8 @@ var _ = Describe("ControlPlane", func() {
 			svcClient, err = skkube.NewServiceClient(ctx, inMemoryFactory)
 			Expect(err).NotTo(HaveOccurred())
 
-			svc1 = skkube.NewService("gloo-system", kubeutils.GlooServiceName)
+			svc1 = skkube.NewService(defaults.GlooSystem, kubeutils.GlooServiceName)
+			svc1.Labels = kubeutils.GlooServiceLabels
 			svc1.Spec = corev1.ServiceSpec{
 				Ports: []corev1.ServicePort{
 					{
@@ -68,7 +52,8 @@ var _ = Describe("ControlPlane", func() {
 				},
 			}
 
-			svc2 = skkube.NewService("other-ns", kubeutils.GlooServiceName)
+			svc2 = skkube.NewService(otherNS, kubeutils.GlooServiceName)
+			svc2.Labels = kubeutils.GlooServiceLabels
 			svc2.Spec = corev1.ServiceSpec{
 				Ports: []corev1.ServicePort{
 					{
@@ -84,58 +69,92 @@ var _ = Describe("ControlPlane", func() {
 			Expect(err).NotTo(HaveOccurred())
 		})
 
-		It("returns xds port from gloo service in default namespace", func() {
+		It("returns xds service from gloo service in default namespace", func() {
 			// write both services
 			_, err = svcClient.Write(svc1, clients.WriteOpts{Ctx: ctx})
 			Expect(err).NotTo(HaveOccurred())
 			_, err = svcClient.Write(svc2, clients.WriteOpts{Ctx: ctx})
 			Expect(err).NotTo(HaveOccurred())
 
 			// should return the port from gloo service in gloo-system
-			port, err := setup.GetControlPlaneXdsPort(ctx, svcClient)
+			service, err := setup.GetControlPlaneService(ctx, defaults.GlooSystem, svcClient)
 			Expect(err).NotTo(HaveOccurred())
-			Expect(port).To(Equal(int32(1111)))
+			Expect(service).NotTo(BeNil())
+			Expect(service.Name).To(Equal(svc1.Name))
+			Expect(service.Namespace).To(Equal(svc1.Namespace))
 		})
 
-		It("returns xds port from gloo service in POD_NAMESPACE namespace", func() {
-			err := os.Setenv(statusutils.PodNamespaceEnvName, "other-ns")
-			Expect(err).NotTo(HaveOccurred())
-
-			// write both services
+		It("returns error when no gloo service exists in the namespace", func() {
+			// only write svc1
 			_, err = svcClient.Write(svc1, clients.WriteOpts{Ctx: ctx})
 			Expect(err).NotTo(HaveOccurred())
-			_, err = svcClient.Write(svc2, clients.WriteOpts{Ctx: ctx})
-			Expect(err).NotTo(HaveOccurred())
 
-			// should return the port from gloo service in other-ns
-			port, err := setup.GetControlPlaneXdsPort(ctx, svcClient)
-			Expect(err).NotTo(HaveOccurred())
-			Expect(port).To(Equal(int32(2222)))
+			_, err = setup.GetControlPlaneService(ctx, "another-ns", svcClient)
+			Expect(err).To(HaveOccurred())
+			Expect(err).To(MatchError(setup.NoGlooSvcFoundError))
 		})
 
-		It("returns error when no gloo service exists in the POD_NAMESPACE namespace", func() {
-			err := os.Setenv(statusutils.PodNamespaceEnvName, "other-ns")
-			Expect(err).NotTo(HaveOccurred())
+		It("returns error when multiple gloo services exist in the namespace", func() {
+			dupeSvc := skkube.NewService("other-ns", "duplicate-gloo-service")
+			dupeSvc.Labels = kubeutils.GlooServiceLabels
+			dupeSvc.Spec = corev1.ServiceSpec{
+				Ports: []corev1.ServicePort{
+					{
+						Name: kubeutils.GlooXdsPortName,
+						Port: 3333,
+					},
+				},
+			}
 
-			// only write svc1
-			_, err = svcClient.Write(svc1, clients.WriteOpts{Ctx: ctx})
+			// write both services
+			_, err = svcClient.Write(svc2, clients.WriteOpts{Ctx: ctx})
+			Expect(err).NotTo(HaveOccurred())
+			_, err = svcClient.Write(dupeSvc, clients.WriteOpts{Ctx: ctx})
 			Expect(err).NotTo(HaveOccurred())
 
-			_, err = setup.GetControlPlaneXdsPort(ctx, svcClient)
+			_, err = setup.GetControlPlaneService(ctx, otherNS, svcClient)
 			Expect(err).To(HaveOccurred())
-			Expect(skerrors.IsNotExist(err)).To(BeTrue())
+			Expect(err).To(MatchError(setup.MultipleGlooSvcFoundError))
 		})
+	})
 
-		It("returns error when the expected port name is not found", func() {
-			err := os.Setenv(statusutils.PodNamespaceEnvName, "other-ns")
-			Expect(err).NotTo(HaveOccurred())
+	Context("xds host", func() {
+		It("get FQDN for service", func() {
+			svc := skkube.NewService(defaults.GlooSystem, kubeutils.GlooServiceName)
+			xdsHost := setup.GetControlPlaneXdsHost(svc)
+			Expect(xdsHost).To(Equal("gloo.gloo-system.svc.cluster.local"))
+		})
+	})
 
-			// modify the port name
-			svc2.Spec.Ports[0].Name = "test-name"
-			_, err = svcClient.Write(svc2, clients.WriteOpts{Ctx: ctx})
+	Context("xds port", func() {
+		It("returns xds port", func() {
+			svc := skkube.NewService(defaults.GlooSystem, kubeutils.GlooServiceName)
+			svc.Spec = corev1.ServiceSpec{
+				Ports: []corev1.ServicePort{
+					{
+						Name: kubeutils.GlooXdsPortName,
+						Port: 1111,
+					},
+				},
+			}
+
+			port, err := setup.GetControlPlaneXdsPort(svc)
 			Expect(err).NotTo(HaveOccurred())
+			Expect(port).To(Equal(int32(1111)))
+		})
+
+		It("returns error when the expected port name is not found", func() {
+			svc := skkube.NewService(defaults.GlooSystem, kubeutils.GlooServiceName)
+			svc.Spec = corev1.ServiceSpec{
+				Ports: []corev1.ServicePort{
+					{
+						Name: "not-the-right-name",
+						Port: 1111,
+					},
+				},
+			}
 
-			_, err = setup.GetControlPlaneXdsPort(ctx, svcClient)
+			_, err := setup.GetControlPlaneXdsPort(svc)
 			Expect(err).To(HaveOccurred())
 			Expect(err).To(MatchError(setup.NoXdsPortFoundError))
 		})

projects/gloo/pkg/syncer/setup/setup_syncer.go

@@ -338,11 +338,19 @@ func (s *setupSyncer) Setup(ctx context.Context, kubeCache kube.SharedCache, mem
 	var xdsPort int32
 	switch settings.GetConfigSource().(type) {
 	case *v1.Settings_KubernetesConfigSource:
-		xdsHost = GetControlPlaneXdsHost()
-		xdsPort, err = GetControlPlaneXdsPort(ctx, opts.KubeServiceClient)
+		glooService, err := GetControlPlaneService(ctx, writeNamespace, opts.KubeServiceClient)
 		if err != nil {
 			return err
 		}
+
+		xdsHost = GetControlPlaneXdsHost(glooService)
+		xdsPort, err = GetControlPlaneXdsPort(glooService)
+		if err != nil {
+			return err
+		}
+
+		logger := contextutils.LoggerFrom(ctx)
+		logger.Infof("using xds host %v and xds port %v", xdsHost, xdsPort)
 	}
 
 	// process grpcserver options to understand if any servers will need a restart

test/kubernetes/e2e/features/deployer/suite.go

@@ -331,7 +331,12 @@ func xdsClusterAssertion(testInstallation *e2e.TestInstallation) func(ctx contex
 				Namespace: testInstallation.Metadata.InstallNamespace,
 			})), "xds socket address points to gloo service, in installation namespace")
 
-			xdsPort, err := setup.GetNamespacedControlPlaneXdsPort(ctx, testInstallation.Metadata.InstallNamespace, testInstallation.ResourceClients.ServiceClient())
+			service, err := setup.GetControlPlaneService(ctx, testInstallation.Metadata.InstallNamespace,
+				testInstallation.ResourceClients.ServiceClient())
+			g.Expect(err).NotTo(gomega.HaveOccurred())
+			g.Expect(service).NotTo(gomega.BeNil())
+
+			xdsPort, err := setup.GetControlPlaneXdsPort(service)
 			g.Expect(err).NotTo(gomega.HaveOccurred())
 			g.Expect(xdsSocketAddress.GetPortValue()).To(gomega.Equal(uint32(xdsPort)), "xds socket port points to gloo service, in installation namespace")
 		}).
@@ -383,8 +388,8 @@ func awsStsClusterAssertion(testInstallation *e2e.TestInstallation) func(ctx con
 			g.Expect(socketAddr.GetAddress()).To(gomega.Equal(expectedStsUri))
 		}).
 			WithContext(ctx).
-			WithTimeout(time.Second * 10).
-			WithPolling(time.Millisecond * 200).
+			WithTimeout(time.Second * 30).
+			WithPolling(time.Second * 1).
 			Should(gomega.Succeed())
 	}
 }