feat(status) add UDP publish service

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/knative/knative.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/dataplane/address_finder.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_test.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{"support@konghq.com"}
 	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/manager/config.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.)`)