feat: allow configuring cluster domain and use FQDN for upstream serv…

…ice targets

CHANGELOG.md

@@ -125,6 +125,12 @@ Adding a new version? You'll need three changes:
   - `konghq.com/publish-service`
   - `konghq.com/tags`
   [#6729](https://github.com/Kong/kubernetes-ingress-controller/pull/6729)
+- From now on, upstreams produced by KIC from `Service`s that are configured as
+  upstream services (either by `ingress.kubernetes.io/service-upstream` annotation
+  or through `IngressClassNamespacedParameters`'s `serviceUpstream` field, will use
+  a FQDN with a default cluster domain of `cluster.local`.
+  Users can override the default by setting the `cluster-domain` flag.
+  [#6697](https://github.com/Kong/kubernetes-ingress-controller/pull/6697)
 
 ### Fixed
 

internal/dataplane/kong_client_golden_test.go

@@ -259,7 +259,7 @@ func runKongClientGoldenTest(t *testing.T, tc kongClientGoldenTestCase) {
 	// Create the translator.
 	logger := zapr.NewLogger(zap.NewNop())
 	s := store.New(cacheStores, "kong", logger)
-	p, err := translator.NewTranslator(logger, s, "", tc.featureFlags, fakeSchemaServiceProvier{})
+	p, err := translator.NewTranslator(logger, s, "", tc.featureFlags, fakeSchemaServiceProvier{}, "cluster.local")
 	require.NoError(t, err, "failed creating translator")
 
 	// Start a mock Admin API server and create an Admin API client for inspecting the configuration.

internal/dataplane/translator/translate_upstreams.go

@@ -78,7 +78,7 @@ func (t *Translator) getUpstreams(serviceMap map[string]kongstate.Service) ([]ko
 				serviceMap[serviceName] = service
 
 				// get the new targets for this backend service
-				newTargets := getServiceEndpoints(t.logger, t.storer, k8sService, port)
+				newTargets := getServiceEndpoints(t.logger, t.storer, k8sService, port, t.clusterDomain)
 
 				if len(newTargets) == 0 {
 					t.logger.V(logging.InfoLevel).Info("No targets could be found for kubernetes service",
@@ -195,6 +195,7 @@ func getServiceEndpoints(
 	s store.Storer,
 	svc *corev1.Service,
 	servicePort *corev1.ServicePort,
+	clusterDomain string,
 ) []kongstate.Target {
 	logger = logger.WithValues(
 		"service_name", svc.Name,
@@ -219,7 +220,7 @@ func getServiceEndpoints(
 	// Check all protocols for associated endpoints.
 	endpoints := []util.Endpoint{}
 	for protocol := range protocols {
-		newEndpoints := getEndpoints(logger, svc, servicePort, protocol, s.GetEndpointSlicesForService, isSvcUpstream)
+		newEndpoints := getEndpoints(logger, svc, servicePort, protocol, s.GetEndpointSlicesForService, isSvcUpstream, clusterDomain)
 		endpoints = append(endpoints, newEndpoints...)
 	}
 	if len(endpoints) == 0 {
@@ -257,6 +258,7 @@ func getEndpoints(
 	proto corev1.Protocol,
 	getEndpointSlices func(string, string) ([]*discoveryv1.EndpointSlice, error),
 	isSvcUpstream bool,
+	clusterDomain string,
 ) []util.Endpoint {
 	if service == nil || port == nil {
 		return []util.Endpoint{}
@@ -267,7 +269,7 @@ func getEndpoints(
 		// ... return its address as the only endpoint.
 		return []util.Endpoint{
 			{
-				Address: service.Name + "." + service.Namespace + ".svc",
+				Address: service.Name + "." + service.Namespace + ".svc." + clusterDomain,
 				Port:    fmt.Sprint(port.Port),
 			},
 		}

internal/dataplane/translator/translator.go

@@ -92,6 +92,8 @@ type Translator struct {
 
 	failuresCollector          *failures.ResourceFailuresCollector
 	translatedObjectsCollector *ObjectsCollector
+
+	clusterDomain string
 }
 
 // NewTranslator produces a new Translator object provided a logging mechanism
@@ -102,6 +104,7 @@ func NewTranslator(
 	workspace string,
 	featureFlags FeatureFlags,
 	schemaServiceProvider SchemaServiceProvider,
+	clusterDomain string,
 ) (*Translator, error) {
 	failuresCollector := failures.NewResourceFailuresCollector(logger)
 
@@ -119,6 +122,7 @@ func NewTranslator(
 		schemaServiceProvider:      schemaServiceProvider,
 		failuresCollector:          failuresCollector,
 		translatedObjectsCollector: translatedObjectsCollector,
+		clusterDomain:              clusterDomain,
 	}, nil
 }
 

internal/dataplane/translator/translator_test.go

@@ -3763,8 +3763,9 @@ func TestGetEndpoints(t *testing.T) {
 		port              *corev1.ServicePort
 		proto             corev1.Protocol
 		fn                func(string, string) ([]*discoveryv1.EndpointSlice, error)
-		result            []util.Endpoint
 		isServiceUpstream bool
+		clusterDomain     string
+		result            []util.Endpoint
 	}{
 		{
 			name:  "no service should return 0 endpoints",
@@ -3857,7 +3858,7 @@ func TestGetEndpoints(t *testing.T) {
 			},
 			result: []util.Endpoint{
 				{
-					Address: "foo.bar.svc",
+					Address: "foo.bar.svc.cluster.local",
 					Port:    "2080",
 				},
 			},
@@ -3891,12 +3892,50 @@ func TestGetEndpoints(t *testing.T) {
 			},
 			result: []util.Endpoint{
 				{
-					Address: "foo.bar.svc",
+					Address: "foo.bar.svc.cluster.local",
 					Port:    "2080",
 				},
 			},
 			isServiceUpstream: true,
 		},
+		{
+			name: "a service with ingress.kubernetes.io/service-upstream annotation should return one endpoint properly mapping to provided custom domain",
+			svc: &corev1.Service{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "foo",
+					Namespace: "bar",
+					Annotations: map[string]string{
+						"ingress.kubernetes.io/service-upstream": "true",
+					},
+				},
+				Spec: corev1.ServiceSpec{
+					Type: corev1.ServiceTypeClusterIP,
+					Ports: []corev1.ServicePort{
+						{
+							Name:       "default",
+							TargetPort: intstr.FromInt(80),
+							Port:       2080,
+						},
+					},
+				},
+			},
+			port: &corev1.ServicePort{
+				Name:       "default",
+				TargetPort: intstr.FromInt(80),
+				Port:       2080,
+			},
+			proto: corev1.ProtocolTCP,
+			fn: func(string, string) ([]*discoveryv1.EndpointSlice, error) {
+				return []*discoveryv1.EndpointSlice{}, nil
+			},
+			clusterDomain: "acme.com",
+			result: []util.Endpoint{
+				{
+					Address: "foo.bar.svc.acme.com",
+					Port:    "2080",
+				},
+			},
+		},
 		{
 			name: "should return no endpoints when there is an error searching for endpoints",
 			svc: &corev1.Service{
@@ -4123,8 +4162,11 @@ func TestGetEndpoints(t *testing.T) {
 
 	for _, testCase := range tests {
 		t.Run(testCase.name, func(t *testing.T) {
-			result := getEndpoints(zapr.NewLogger(zap.NewNop()), testCase.svc, testCase.port, testCase.proto, testCase.fn,
-				testCase.isServiceUpstream)
+			clusterDomain := "cluster.local"
+			if testCase.clusterDomain != "" {
+				clusterDomain = testCase.clusterDomain
+			}
+			result := getEndpoints(zapr.NewLogger(zap.NewNop()), testCase.svc, testCase.port, testCase.proto, testCase.fn, testCase.isServiceUpstream, clusterDomain)
 			require.Equal(t, testCase.result, result)
 		})
 	}
@@ -5060,12 +5102,17 @@ func (p fakeSchemaServiceProvier) GetSchemaService() kong.AbstractSchemaService
 }
 
 func mustNewTranslator(t *testing.T, storer store.Storer) *Translator {
-	p, err := NewTranslator(zapr.NewLogger(zap.NewNop()), storer, "", FeatureFlags{
-		// We'll assume these are true for all tests.
-		FillIDs:                           true,
-		ReportConfiguredKubernetesObjects: true,
-		KongServiceFacade:                 true,
-	}, fakeSchemaServiceProvier{},
+	logger := zapr.NewLogger(zap.NewNop())
+	clusterDomain := "cluster.local"
+	p, err := NewTranslator(logger, storer, "",
+		FeatureFlags{
+			// We'll assume these are true for all tests.
+			FillIDs:                           true,
+			ReportConfiguredKubernetesObjects: true,
+			KongServiceFacade:                 true,
+		},
+		fakeSchemaServiceProvier{},
+		clusterDomain,
 	)
 	require.NoError(t, err)
 	return p

internal/manager/config.go

@@ -96,6 +96,7 @@ type Config struct {
 	GatewayAPIControllerName string
 	Impersonate              string
 	EmitKubernetesEvents     bool
+	ClusterDomain            string
 
 	// Ingress status
 	PublishServiceUDP       OptionalNamespacedName
@@ -241,6 +242,7 @@ func (c *Config) FlagSet() *pflag.FlagSet {
 	flagSet.StringSliceVar(&c.WatchNamespaces, "watch-namespace", nil,
 		`Namespace(s) in comma-separated format (or specify this flag multiple times) to watch for Kubernetes resources. Defaults to all namespaces.`)
 	flagSet.BoolVar(&c.EmitKubernetesEvents, "emit-kubernetes-events", true, `Emit Kubernetes events for successful configuration applies, translation failures and configuration apply failures on managed objects.`)
+	flagSet.StringVar(&c.ClusterDomain, "cluster-domain", DefaultClusterDomain, `The cluster domain. This is used e.g. in generating addresses for upstream services.`)
 
 	// Ingress status
 	flagSet.Var(flags.NewValidatedValue(&c.PublishService, namespacedNameFromFlagValue, nnTypeNameOverride), "publish-service",

internal/manager/consts.go

@@ -22,3 +22,6 @@ const DiagnosticsPort = 10256
 
 // KongClientEventRecorderComponentName is a KongClient component name used to identify the events recording component.
 const KongClientEventRecorderComponentName = "kong-client"
+
+// DefaultClusterDomain is the default cluster domain used by the controller.
+const DefaultClusterDomain = "cluster.local"

internal/manager/run.go

@@ -181,7 +181,8 @@ func Run(
 	referenceIndexers := ctrlref.NewCacheIndexers(setupLog.WithName("reference-indexers"))
 	cache := store.NewCacheStores()
 	storer := store.New(cache, c.IngressClassName, logger)
-	configTranslator, err := translator.NewTranslator(logger, storer, c.KongWorkspace, translatorFeatureFlags, NewSchemaServiceGetter(clientsManager))
+
+	configTranslator, err := translator.NewTranslator(logger, storer, c.KongWorkspace, translatorFeatureFlags, NewSchemaServiceGetter(clientsManager), c.ClusterDomain)
 	if err != nil {
 		return fmt.Errorf("failed to create translator: %w", err)
 	}