traefik/pkg/provider/kubernetes/crd/kubernetes_udp.go
2024-06-21 14:56:03 +02:00

212 lines
5.9 KiB
Go

package crd
import (
"context"
"errors"
"fmt"
"net"
"strconv"
"github.com/rs/zerolog/log"
"github.com/traefik/traefik/v3/pkg/config/dynamic"
traefikv1alpha1 "github.com/traefik/traefik/v3/pkg/provider/kubernetes/crd/traefikio/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.Ctx(ctx).With().Str("ingress", ingressRouteUDP.Name).Str("namespace", ingressRouteUDP.Namespace).Logger()
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 := p.createLoadBalancerServerUDP(client, ingressRouteUDP.Namespace, service)
if err != nil {
logger.Error().
Str("serviceName", service.Name).
Stringer("servicePort", &service.Port).
Err(err).
Msg("Cannot create service")
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-%s", 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 (p *Provider) createLoadBalancerServerUDP(client Client, parentNamespace string, service traefikv1alpha1.ServiceUDP) (*dynamic.UDPService, error) {
ns := parentNamespace
if len(service.Namespace) > 0 {
if !isNamespaceAllowed(p.AllowCrossNamespace, parentNamespace, service.Namespace) {
return nil, fmt.Errorf("udp service %s/%s is not in the parent resource namespace %s", service.Namespace, service.Name, ns)
}
ns = service.Namespace
}
servers, err := p.loadUDPServers(client, ns, service)
if err != nil {
return nil, err
}
udpService := &dynamic.UDPService{
LoadBalancer: &dynamic.UDPServersLoadBalancer{
Servers: servers,
},
}
return udpService, nil
}
func (p *Provider) loadUDPServers(client Client, namespace string, svc traefikv1alpha1.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")
}
if service.Spec.Type == corev1.ServiceTypeExternalName && !p.AllowExternalNameServices {
return nil, fmt.Errorf("externalName services not allowed: %s/%s", namespace, svc.Name)
}
svcPort, err := getServicePort(service, svc.Port)
if err != nil {
return nil, err
}
var servers []dynamic.UDPServer
if service.Spec.Type == corev1.ServiceTypeNodePort && svc.NodePortLB {
nodes, nodesExists, nodesErr := client.GetNodes()
if nodesErr != nil {
return nil, nodesErr
}
if !nodesExists || len(nodes) == 0 {
return nil, fmt.Errorf("nodes not found for NodePort service %s/%s", svc.Namespace, svc.Name)
}
for _, node := range nodes {
for _, addr := range node.Status.Addresses {
if addr.Type == corev1.NodeInternalIP {
servers = append(servers, dynamic.UDPServer{
Address: net.JoinHostPort(addr.Address, strconv.Itoa(int(svcPort.NodePort))),
})
}
}
}
if len(servers) == 0 {
return nil, fmt.Errorf("no servers were generated for service %s/%s", svc.Namespace, svc.Name)
}
return servers, nil
}
if service.Spec.Type == corev1.ServiceTypeExternalName {
servers = append(servers, dynamic.UDPServer{
Address: net.JoinHostPort(service.Spec.ExternalName, strconv.Itoa(int(svcPort.Port))),
})
} else {
nativeLB := p.NativeLBByDefault
if svc.NativeLB != nil {
nativeLB = *svc.NativeLB
}
if nativeLB {
address, err := getNativeServiceAddress(*service, *svcPort)
if err != nil {
return nil, fmt.Errorf("getting native Kubernetes Service address: %w", err)
}
return []dynamic.UDPServer{{Address: address}}, nil
}
endpointSlices, err := client.GetEndpointSlicesForService(namespace, svc.Name)
if err != nil {
return nil, fmt.Errorf("getting endpointslices: %w", err)
}
addresses := map[string]struct{}{}
for _, endpointSlice := range endpointSlices {
var port int32
for _, p := range endpointSlice.Ports {
if svcPort.Name == *p.Name {
port = *p.Port
break
}
}
if port == 0 {
continue
}
for _, endpoint := range endpointSlice.Endpoints {
if endpoint.Conditions.Ready == nil || !*endpoint.Conditions.Ready {
continue
}
for _, address := range endpoint.Addresses {
if _, ok := addresses[address]; ok {
continue
}
addresses[address] = struct{}{}
servers = append(servers, dynamic.UDPServer{
Address: net.JoinHostPort(address, strconv.Itoa(int(port))),
})
}
}
}
}
if len(servers) == 0 && !p.AllowEmptyServices {
return nil, fmt.Errorf("no servers found for %s/%s", namespace, svc.Name)
}
return servers, nil
}