traefik/pkg/provider/kubernetes/crd/kubernetes_udp.go
Jean-Baptiste Doumenjou 665aeb34b2
Add UDP support in kubernetesCRD provider
Co-authored-by: Mathieu Lonjaret <mathieu.lonjaret@gmail.com>
2020-02-26 12:28:05 +01:00

161 lines
4.2 KiB
Go

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
}