feat(status) add UDP publish service (#3325)

Add CLI arguments for setting a UDP publish service and address
overrides. If set, these will be used to populate status information for
UDPIngresses. If not set, UDPIngress status will fall back to the
default publish service/override, with the expectation that it is a
dual-transport LoadBalancer.

Author: rainest

CHANGELOG.md

@@ -75,7 +75,10 @@ Adding a new version? You'll need three changes:
   configuration.
   [#3359](https://github.com/Kong/kubernetes-ingress-controller/pull/3359)
 - Added `version` command
-  [#3379](https://github.com/Kong/kubernetes-ingress-controller/pull/3379)
+  [#3379](https://github.com/Kong/kubernetes-ingress-controller/pull/3379)  
+- Added `--publish-service-udp` to indicate the Service that handles inbound
+  UDP traffic.
+  [#3325](https://github.com/Kong/kubernetes-ingress-controller/pull/3325)
 - Added possibility to configure multiple Kong Gateways through the
   `--kong-admin-url` CLI flag (which can be specified multiple times) or through
   a corresponding environment variable `CONTROLLER_KONG_ADMIN_URL` (which can

Makefile

@@ -432,6 +432,7 @@ test.istio: gotestsum
 KUBECONFIG ?= "${HOME}/.kube/config"
 KONG_NAMESPACE ?= kong-system
 KONG_PROXY_SERVICE ?= ingress-controller-kong-proxy
+KONG_PROXY_UDP_SERVICE ?= ingress-controller-kong-udp-proxy
 KONG_ADMIN_SERVICE ?= ingress-controller-kong-admin
 KONG_ADMIN_PORT ?= 8001
 KONG_ADMIN_URL ?= "http://$(shell kubectl -n $(KONG_NAMESPACE) get service $(KONG_ADMIN_SERVICE) -o=go-template='{{range .status.loadBalancer.ingress}}{{.ip}}{{end}}'):$(KONG_ADMIN_PORT)"
@@ -446,6 +447,7 @@ debug: install _ensure-namespace
 		--anonymous-reports=false \
 		--kong-admin-url $(KONG_ADMIN_URL) \
 		--publish-service $(KONG_NAMESPACE)/$(KONG_PROXY_SERVICE) \
+		--publish-service-udp $(KONG_NAMESPACE)/$(KONG_PROXY_UDP_SERVICE) \
 		--kubeconfig $(KUBECONFIG) \
 		--feature-gates=$(KONG_CONTROLLER_FEATURE_GATES)
 
@@ -494,6 +496,7 @@ _run:
 		--anonymous-reports=false \
 		--kong-admin-url $(KONG_ADMIN_URL) \
 		--publish-service $(KONG_NAMESPACE)/$(KONG_PROXY_SERVICE) \
+		--publish-service-udp $(KONG_NAMESPACE)/$(KONG_PROXY_UDP_SERVICE) \
 		--kubeconfig $(KUBECONFIG) \
 		--feature-gates=$(KONG_CONTROLLER_FEATURE_GATES)
 

hack/generators/controllers/networking/main.go

@@ -206,6 +206,7 @@ var inputControllersNeeded = &typesNeeded{
 		Package:                           kongv1beta1,
 		Plural:                            "udpingresses",
 		CacheType:                         "UDPIngress",
+		IsUDP:                             true,
 		NeedsStatusPermissions:            true,
 		CapableOfStatusUpdates:            true,
 		AcceptsIngressClassNameAnnotation: true,
@@ -333,6 +334,7 @@ type typeNeeded struct {
 	Plural             string
 	CacheType          string
 	RBACVerbs          []string
+	IsUDP              bool
 
 	// AcceptsIngressClassNameAnnotation indicates that the object accepts (and the controller will listen to)
 	// the "kubernetes.io/ingress.class" annotation to decide whether or not the object is supported.
@@ -627,7 +629,11 @@ func (r *{{.PackageAlias}}{{.Kind}}Reconciler) Reconcile(ctx context.Context, re
 		}
 
 		log.V(util.DebugLevel).Info("determining gateway addresses for object status updates", "namespace", req.Namespace, "name", req.Name)
-		addrs, err := r.DataplaneAddressFinder.GetLoadBalancerAddresses()
+		{{- if .IsUDP }}
+		addrs, err := r.DataplaneAddressFinder.GetUDPLoadBalancerAddresses(ctx)
+		{{- else }}
+		addrs, err := r.DataplaneAddressFinder.GetLoadBalancerAddresses(ctx)
+		{{- end }}
 		if err != nil {
 			return ctrl.Result{}, err
 		}

internal/controllers/configuration/zz_generated_controllers.go

@@ -208,7 +208,7 @@ func (r *Knativev1alpha1IngressReconciler) Reconcile(ctx context.Context, req ct
 		}
 
 		log.V(util.DebugLevel).Info("determining gateway addresses for object status updates", "namespace", req.Namespace, "name", req.Name)
-		addrs, err := r.DataplaneAddressFinder.GetLoadBalancerAddresses()
+		addrs, err := r.DataplaneAddressFinder.GetLoadBalancerAddresses(ctx)
 		if err != nil {
 			return ctrl.Result{}, err
 		}

internal/controllers/knative/knative.go

@@ -1,6 +1,7 @@
 package dataplane
 
 import (
+	"context"
 	"fmt"
 	"net"
 	"strings"
@@ -16,13 +17,14 @@ import (
 
 // AddressGetter is a function which can dynamically retrieve the list of IPs
 // that the data-plane is listening on for ingress network traffic.
-type AddressGetter func() ([]string, error)
+type AddressGetter func(ctx context.Context) ([]string, error)
 
 // AddressFinder is a threadsafe metadata object which can provide the current
 // live addresses in use by the dataplane at any point in time.
 type AddressFinder struct {
-	overrideAddresses []string
-	addressGetter     AddressGetter
+	overrideAddresses    []string
+	overrideAddressesUDP []string
+	addressGetter        AddressGetter
 
 	lock sync.RWMutex
 }
@@ -54,10 +56,19 @@ func (a *AddressFinder) SetOverrides(addrs []string) {
 	a.overrideAddresses = addrs
 }
 
+// SetUDPOverrides hard codes a specific list of addresses to be the UDP addresses
+// that this finder produces for the data-plane. To disable overrides, call
+// this method again with an empty list.
+func (a *AddressFinder) SetUDPOverrides(addrs []string) {
+	a.lock.Lock()
+	defer a.lock.Unlock()
+	a.overrideAddressesUDP = addrs
+}
+
 // GetAddresses provides a list of the addresses which the data-plane is
 // listening on for ingress network traffic. Addresses can either be IP
 // addresses or hostnames.
-func (a *AddressFinder) GetAddresses() ([]string, error) {
+func (a *AddressFinder) GetAddresses(ctx context.Context) ([]string, error) {
 	a.lock.RLock()
 	defer a.lock.RUnlock()
 
@@ -66,22 +77,50 @@ func (a *AddressFinder) GetAddresses() ([]string, error) {
 	}
 
 	if a.addressGetter != nil {
-		return a.addressGetter()
+		return a.addressGetter(ctx)
 	}
 
 	return nil, fmt.Errorf("data-plane addresses can't be retrieved: no valid method available")
 }
 
+// GetUDPAddresses provides a list of the UDP addresses which the data-plane is
+// listening on for ingress network traffic. Addresses can either be IP
+// addresses or hostnames. If UDP settings are not configured, falls back to GetAddresses().
+func (a *AddressFinder) GetUDPAddresses(ctx context.Context) ([]string, error) {
+	a.lock.RLock()
+	defer a.lock.RUnlock()
+
+	if len(a.overrideAddressesUDP) > 0 {
+		return a.overrideAddressesUDP, nil
+	}
+
+	if len(a.overrideAddresses) > 0 && a.addressGetter == nil {
+		return a.overrideAddresses, nil
+	}
+
+	if a.addressGetter != nil {
+		return a.addressGetter(ctx)
+	}
+
+	return a.GetAddresses(ctx)
+}
+
 // GetLoadBalancerAddresses provides a list of the addresses which the
 // data-plane is listening on for ingress network traffic, but provides the
 // addresses in Kubernetes corev1.LoadBalancerIngress format. Addresses can be
 // IP addresses or hostnames.
-func (a *AddressFinder) GetLoadBalancerAddresses() ([]netv1.IngressLoadBalancerIngress, error) {
-	addrs, err := a.GetAddresses()
+func (a *AddressFinder) GetLoadBalancerAddresses(ctx context.Context) ([]netv1.IngressLoadBalancerIngress, error) {
+	addrs, err := a.GetAddresses(ctx)
 	if err != nil {
 		return nil, err
 	}
+	return getAddressHelper(addrs)
+}
 
+// getAddressHelper converts a string slice of addresses (IPs or hostnames) into an IngressLoadBalancerIngress
+// (https://pkg.go.dev/k8s.io/api/networking/v1#IngressLoadBalancerIngress), or an error if one of the given strings
+// is neither a valid IP nor a valid hostname.
+func getAddressHelper(addrs []string) ([]netv1.IngressLoadBalancerIngress, error) {
 	var loadBalancerAddresses []netv1.IngressLoadBalancerIngress
 	for _, addr := range addrs {
 		ing := netv1.IngressLoadBalancerIngress{}
@@ -99,6 +138,18 @@ func (a *AddressFinder) GetLoadBalancerAddresses() ([]netv1.IngressLoadBalancerI
 	return loadBalancerAddresses, nil
 }
 
+// GetUDPLoadBalancerAddresses provides a list of the addresses which the
+// data-plane is listening on for UDP network traffic, but provides the
+// addresses in Kubernetes corev1.LoadBalancerIngress format. Addresses can be
+// IP addresses or hostnames.
+func (a *AddressFinder) GetUDPLoadBalancerAddresses(ctx context.Context) ([]netv1.IngressLoadBalancerIngress, error) {
+	addrs, err := a.GetUDPAddresses(ctx)
+	if err != nil {
+		return nil, err
+	}
+	return getAddressHelper(addrs)
+}
+
 // -----------------------------------------------------------------------------
 //
 // -----------------------------------------------------------------------------

internal/dataplane/address_finder.go

@@ -1,6 +1,7 @@
 package dataplane
 
 import (
+	"context"
 	"fmt"
 	"testing"
 
@@ -15,43 +16,44 @@ func TestAddressFinder(t *testing.T) {
 	require.Nil(t, finder.addressGetter)
 
 	t.Log("verifying that a finder with no overrides or getter produces an error")
-	addrs, err := finder.GetAddresses()
+	ctx := context.Background()
+	addrs, err := finder.GetAddresses(ctx)
 	require.Error(t, err)
 	require.Empty(t, addrs)
 	require.Equal(t, "data-plane addresses can't be retrieved: no valid method available", err.Error())
 
 	t.Log("generating a fake AddressGetter")
 	defaultAddrs := []string{"127.0.0.1", "127.0.0.2"}
 	overrideAddrs := []string{"192.168.1.1", "192.168.1.2", "192.168.1.3"}
-	fakeGetter := func() ([]string, error) { return defaultAddrs, nil }
+	fakeGetter := func(ctx context.Context) ([]string, error) { return defaultAddrs, nil }
 
 	t.Log("verifying getting a list of addresses from the finder after a getter function is provided")
 	finder.SetGetter(fakeGetter)
-	addrs, err = finder.GetAddresses()
+	addrs, err = finder.GetAddresses(ctx)
 	require.NoError(t, err)
 	require.Equal(t, defaultAddrs, addrs)
 
 	t.Log("verifying that overrides take precedent over the getter")
 	finder.SetOverrides(overrideAddrs)
-	addrs, err = finder.GetAddresses()
+	addrs, err = finder.GetAddresses(ctx)
 	require.NoError(t, err)
 	require.Equal(t, overrideAddrs, addrs)
 
 	t.Log("verifying disabling overrides")
 	finder.SetOverrides(nil)
-	addrs, err = finder.GetAddresses()
+	addrs, err = finder.GetAddresses(ctx)
 	require.NoError(t, err)
 	require.Equal(t, defaultAddrs, addrs)
 
 	t.Log("verifying k8s load balancer formatted version of the addresses")
-	lbs, err := finder.GetLoadBalancerAddresses()
+	lbs, err := finder.GetLoadBalancerAddresses(ctx)
 	require.NoError(t, err)
 	require.Equal(t, []netv1.IngressLoadBalancerIngress{{IP: defaultAddrs[0]}, {IP: defaultAddrs[1]}}, lbs)
 
 	t.Log("verifying valid DNS names are formatting properly")
 	dnsAddrs := []string{"127.0.0.1", "example1.konghq.com", "example2.konghq.com"}
 	finder.SetOverrides(dnsAddrs)
-	lbs, err = finder.GetLoadBalancerAddresses()
+	lbs, err = finder.GetLoadBalancerAddresses(ctx)
 	require.NoError(t, err)
 	require.Equal(t, []netv1.IngressLoadBalancerIngress{
 		{IP: dnsAddrs[0]},
@@ -61,16 +63,55 @@ func TestAddressFinder(t *testing.T) {
 
 	t.Log("verifying empty addresses return an error")
 	finder.SetOverrides([]string{""})
-	lbs, err = finder.GetLoadBalancerAddresses()
+	lbs, err = finder.GetLoadBalancerAddresses(ctx)
 	require.Error(t, err)
 	require.Empty(t, lbs)
 	require.Equal(t, "empty address found", err.Error())
 
 	t.Log("verifying invalid DNS names return an error")
 	invalidDNSAddrs := []string{"[email protected]"}
 	finder.SetOverrides(invalidDNSAddrs)
-	lbs, err = finder.GetLoadBalancerAddresses()
+	lbs, err = finder.GetLoadBalancerAddresses(ctx)
 	require.Error(t, err)
 	require.Empty(t, lbs)
 	require.Equal(t, fmt.Sprintf("%s is not a valid DNS hostname", invalidDNSAddrs[0]), err.Error())
 }
+
+func TestUDPAddressFinder(t *testing.T) {
+	t.Log("generating a new AddressFinder")
+	finder := NewAddressFinder()
+	require.NotNil(t, finder)
+	require.Nil(t, finder.addressGetter)
+
+	t.Log("generating fake AddressGetters")
+	ctx := context.Background()
+	defaultAddrs := []string{"127.0.0.1", "127.0.0.2"}
+	overrideAddrs := []string{"192.168.1.1", "192.168.1.2", "192.168.1.3"}
+	fakeGetter := func(ctx context.Context) ([]string, error) { return defaultAddrs, nil }
+
+	defaultUDPAddrs := []string{"127.1.0.1", "127.1.0.2"}
+	overrideUDPAddrs := []string{"192.168.2.1", "192.168.2.2", "192.168.2.3"}
+	fakeUDPGetter := func(ctx context.Context) ([]string, error) { return defaultUDPAddrs, nil }
+
+	t.Log("verifying getting a list of addresses from the finder after a getter function is provided")
+	finder.SetGetter(fakeGetter)
+	addrs, err := finder.GetUDPAddresses(ctx)
+	require.NoError(t, err)
+	require.Equal(t, defaultAddrs, addrs)
+
+	t.Log("verifying that overrides take precedent over the getter")
+	finder.SetOverrides(overrideAddrs)
+	addrs, err = finder.GetAddresses(ctx)
+	require.NoError(t, err)
+	require.Equal(t, overrideAddrs, addrs)
+
+	finder.SetGetter(fakeUDPGetter)
+	addrs, err = finder.GetUDPAddresses(ctx)
+	require.NoError(t, err)
+	require.Equal(t, defaultUDPAddrs, addrs)
+
+	finder.SetUDPOverrides(overrideUDPAddrs)
+	addrs, err = finder.GetUDPAddresses(ctx)
+	require.NoError(t, err)
+	require.Equal(t, overrideUDPAddrs, addrs)
+}

internal/dataplane/address_finder_test.go

@@ -67,9 +67,11 @@ type Config struct {
 	GatewayAPIControllerName string
 
 	// Ingress status
-	PublishService       types.NamespacedName
-	PublishStatusAddress []string
-	UpdateStatus         bool
+	PublishServiceUDP       types.NamespacedName
+	PublishService          types.NamespacedName
+	PublishStatusAddress    []string
+	PublishStatusAddressUDP []string
+	UpdateStatus            bool
 
 	// Kubernetes API toggling
 	IngressExtV1beta1Enabled      bool
@@ -180,6 +182,11 @@ func (c *Config) FlagSet() *pflag.FlagSet {
 	flagSet.StringSliceVar(&c.PublishStatusAddress, "publish-status-address", []string{},
 		`User-provided addresses in comma-separated string format, for use in lieu of "publish-service" `+
 			`when that Service lacks useful address information (for example, in bare-metal environments).`)
+	flagSet.Var(NewValidatedValue(&c.PublishServiceUDP, namespacedNameFromFlagValue), "publish-service-udp", `Service fronting UDP routing resources in
+			"namespace/name" format. The controller will update UDP route status information with this Service's
+			endpoints. If omitted, the same Service will be used for both TCP and UDP routes.`)
+	flagSet.StringSliceVar(&c.PublishStatusAddressUDP, "publish-status-address-udp", []string{}, `User-provided
+			address CSV, for use in lieu of "publish-service-udp" when that Service lacks useful address information.`)
 	flagSet.BoolVar(&c.UpdateStatus, "update-status", true,
 		`Indicates if the ingress controller should update the status of resources (e.g. IP/Hostname for v1.Ingress, e.t.c.)`)