Add UDP support in kubernetesCRD provider

Co-authored-by: Mathieu Lonjaret <mathieu.lonjaret@gmail.com>

pkg/provider/kubernetes/crd/kubernetes_udp.go

@@ -0,0 +1,161 @@
+package crd
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/containous/traefik/v2/pkg/config/dynamic"
+	"github.com/containous/traefik/v2/pkg/log"
+	"github.com/containous/traefik/v2/pkg/provider/kubernetes/crd/traefik/v1alpha1"
+	corev1 "k8s.io/api/core/v1"
+)
+
+func (p *Provider) loadIngressRouteUDPConfiguration(ctx context.Context, client Client) *dynamic.UDPConfiguration {
+	conf := &dynamic.UDPConfiguration{
+		Routers:  map[string]*dynamic.UDPRouter{},
+		Services: map[string]*dynamic.UDPService{},
+	}
+
+	for _, ingressRouteUDP := range client.GetIngressRouteUDPs() {
+		logger := log.FromContext(log.With(ctx, log.Str("ingress", ingressRouteUDP.Name), log.Str("namespace", ingressRouteUDP.Namespace)))
+
+		if !shouldProcessIngress(p.IngressClass, ingressRouteUDP.Annotations[annotationKubernetesIngressClass]) {
+			continue
+		}
+
+		ingressName := ingressRouteUDP.Name
+		if len(ingressName) == 0 {
+			ingressName = ingressRouteUDP.GenerateName
+		}
+
+		for i, route := range ingressRouteUDP.Spec.Routes {
+			key := fmt.Sprintf("%s-%d", ingressName, i)
+			serviceName := makeID(ingressRouteUDP.Namespace, key)
+
+			for _, service := range route.Services {
+				balancerServerUDP, err := createLoadBalancerServerUDP(client, ingressRouteUDP.Namespace, service)
+				if err != nil {
+					logger.
+						WithField("serviceName", service.Name).
+						WithField("servicePort", service.Port).
+						Errorf("Cannot create service: %v", err)
+					continue
+				}
+
+				// If there is only one service defined, we skip the creation of the load balancer of services,
+				// i.e. the service on top is directly a load balancer of servers.
+				if len(route.Services) == 1 {
+					conf.Services[serviceName] = balancerServerUDP
+					break
+				}
+
+				serviceKey := fmt.Sprintf("%s-%s-%d", serviceName, service.Name, service.Port)
+				conf.Services[serviceKey] = balancerServerUDP
+
+				srv := dynamic.UDPWRRService{Name: serviceKey}
+				srv.SetDefaults()
+				if service.Weight != nil {
+					srv.Weight = service.Weight
+				}
+
+				if conf.Services[serviceName] == nil {
+					conf.Services[serviceName] = &dynamic.UDPService{Weighted: &dynamic.UDPWeightedRoundRobin{}}
+				}
+				conf.Services[serviceName].Weighted.Services = append(conf.Services[serviceName].Weighted.Services, srv)
+			}
+
+			conf.Routers[serviceName] = &dynamic.UDPRouter{
+				EntryPoints: ingressRouteUDP.Spec.EntryPoints,
+				Service:     serviceName,
+			}
+		}
+	}
+
+	return conf
+}
+
+func createLoadBalancerServerUDP(client Client, namespace string, service v1alpha1.ServiceUDP) (*dynamic.UDPService, error) {
+	ns := namespace
+	if len(service.Namespace) > 0 {
+		ns = service.Namespace
+	}
+
+	servers, err := loadUDPServers(client, ns, service)
+	if err != nil {
+		return nil, err
+	}
+
+	udpService := &dynamic.UDPService{
+		LoadBalancer: &dynamic.UDPServersLoadBalancer{
+			Servers: servers,
+		},
+	}
+
+	return udpService, nil
+}
+
+func loadUDPServers(client Client, namespace string, svc v1alpha1.ServiceUDP) ([]dynamic.UDPServer, error) {
+	service, exists, err := client.GetService(namespace, svc.Name)
+	if err != nil {
+		return nil, err
+	}
+
+	if !exists {
+		return nil, errors.New("service not found")
+	}
+
+	var portSpec *corev1.ServicePort
+	for _, p := range service.Spec.Ports {
+		if svc.Port == p.Port {
+			portSpec = &p
+			break
+		}
+	}
+
+	if portSpec == nil {
+		return nil, errors.New("service port not found")
+	}
+
+	var servers []dynamic.UDPServer
+	if service.Spec.Type == corev1.ServiceTypeExternalName {
+		servers = append(servers, dynamic.UDPServer{
+			Address: fmt.Sprintf("%s:%d", service.Spec.ExternalName, portSpec.Port),
+		})
+	} else {
+		endpoints, endpointsExists, endpointsErr := client.GetEndpoints(namespace, svc.Name)
+		if endpointsErr != nil {
+			return nil, endpointsErr
+		}
+
+		if !endpointsExists {
+			return nil, errors.New("endpoints not found")
+		}
+
+		if len(endpoints.Subsets) == 0 {
+			return nil, errors.New("subset not found")
+		}
+
+		var port int32
+		for _, subset := range endpoints.Subsets {
+			for _, p := range subset.Ports {
+				if portSpec.Name == p.Name {
+					port = p.Port
+					break
+				}
+			}
+
+			if port == 0 {
+				return nil, errors.New("cannot define a port")
+			}
+
+			for _, addr := range subset.Addresses {
+				servers = append(servers, dynamic.UDPServer{
+					Address: fmt.Sprintf("%s:%d", addr.IP, port),
+				})
+			}
+		}
+	}
+
+	return servers, nil
+}