Improve Kubernetes GatewayAPI TCPRoute and TLSRoute support
This commit is contained in:
Romain
GitHub
parent
0b34e0cdcb
commit
3eb7ecce19
11 changed files with 690 additions and 484 deletions
|
|
@ -6,6 +6,7 @@ import (
|
|||
"fmt"
|
||||
"net"
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
"github.com/rs/zerolog/log"
|
||||
"github.com/traefik/traefik/v3/pkg/config/dynamic"
|
||||
|
|
@ -108,182 +109,241 @@ func (p *Provider) loadTCPRoute(listener gatewayListener, route *gatev1alpha2.TC
|
|||
Reason: string(gatev1.RouteConditionResolvedRefs),
|
||||
}
|
||||
|
||||
router := dynamic.TCPRouter{
|
||||
Rule: "HostSNI(`*`)",
|
||||
EntryPoints: []string{listener.EPName},
|
||||
RuleSyntax: "v3",
|
||||
}
|
||||
|
||||
if listener.Protocol == gatev1.TLSProtocolType && listener.TLS != nil {
|
||||
// TODO support let's encrypt
|
||||
router.TLS = &dynamic.RouterTCPTLSConfig{
|
||||
Passthrough: listener.TLS.Mode != nil && *listener.TLS.Mode == gatev1.TLSModePassthrough,
|
||||
}
|
||||
}
|
||||
|
||||
// Adding the gateway desc and the entryPoint desc prevents overlapping of routers build from the same routes.
|
||||
routerName := provider.Normalize(route.Namespace + "-" + route.Name + "-" + listener.GWName + "-" + listener.EPName)
|
||||
|
||||
var ruleServiceNames []string
|
||||
for i, rule := range route.Spec.Rules {
|
||||
for ri, rule := range route.Spec.Rules {
|
||||
if rule.BackendRefs == nil {
|
||||
// Should not happen due to validation.
|
||||
// https://github.com/kubernetes-sigs/gateway-api/blob/v0.4.0/apis/v1alpha2/tcproute_types.go#L76
|
||||
continue
|
||||
}
|
||||
|
||||
wrrService, subServices, err := p.loadTCPServices(route.Namespace, rule.BackendRefs)
|
||||
if err != nil {
|
||||
return conf, metav1.Condition{
|
||||
Type: string(gatev1.RouteConditionResolvedRefs),
|
||||
Status: metav1.ConditionFalse,
|
||||
ObservedGeneration: route.Generation,
|
||||
LastTransitionTime: metav1.Now(),
|
||||
Reason: string(gatev1.RouteReasonBackendNotFound),
|
||||
Message: fmt.Sprintf("Cannot load TCPRoute service %s/%s: %v", route.Namespace, route.Name, err),
|
||||
router := dynamic.TCPRouter{
|
||||
Rule: "HostSNI(`*`)",
|
||||
EntryPoints: []string{listener.EPName},
|
||||
RuleSyntax: "v3",
|
||||
}
|
||||
|
||||
if listener.Protocol == gatev1.TLSProtocolType && listener.TLS != nil {
|
||||
router.TLS = &dynamic.RouterTCPTLSConfig{
|
||||
Passthrough: listener.TLS.Mode != nil && *listener.TLS.Mode == gatev1.TLSModePassthrough,
|
||||
}
|
||||
}
|
||||
|
||||
for svcName, svc := range subServices {
|
||||
conf.TCP.Services[svcName] = svc
|
||||
// Adding the gateway desc and the entryPoint desc prevents overlapping of routers build from the same routes.
|
||||
routeKey := provider.Normalize(fmt.Sprintf("%s-%s-%s-%s-%d", route.Namespace, route.Name, listener.GWName, listener.EPName, ri))
|
||||
// Routing criteria should be introduced at some point.
|
||||
routerName := makeRouterName("", routeKey)
|
||||
|
||||
if len(rule.BackendRefs) == 1 && isInternalService(rule.BackendRefs[0]) {
|
||||
router.Service = string(rule.BackendRefs[0].Name)
|
||||
conf.TCP.Routers[routerName] = &router
|
||||
continue
|
||||
}
|
||||
|
||||
serviceName := fmt.Sprintf("%s-wrr-%d", routerName, i)
|
||||
conf.TCP.Services[serviceName] = wrrService
|
||||
var serviceCondition *metav1.Condition
|
||||
router.Service, serviceCondition = p.loadTCPWRRService(conf, routerName, rule.BackendRefs, route)
|
||||
if serviceCondition != nil {
|
||||
condition = *serviceCondition
|
||||
}
|
||||
|
||||
ruleServiceNames = append(ruleServiceNames, serviceName)
|
||||
}
|
||||
|
||||
if len(ruleServiceNames) == 1 {
|
||||
router.Service = ruleServiceNames[0]
|
||||
conf.TCP.Routers[routerName] = &router
|
||||
return conf, condition
|
||||
}
|
||||
|
||||
routeServiceKey := routerName + "-wrr"
|
||||
routeService := &dynamic.TCPService{Weighted: &dynamic.TCPWeightedRoundRobin{}}
|
||||
|
||||
for _, name := range ruleServiceNames {
|
||||
service := dynamic.TCPWRRService{Name: name}
|
||||
service.SetDefaults()
|
||||
|
||||
routeService.Weighted.Services = append(routeService.Weighted.Services, service)
|
||||
}
|
||||
|
||||
conf.TCP.Services[routeServiceKey] = routeService
|
||||
|
||||
router.Service = routeServiceKey
|
||||
conf.TCP.Routers[routerName] = &router
|
||||
|
||||
return conf, condition
|
||||
}
|
||||
|
||||
// loadTCPServices is generating a WRR service, even when there is only one target.
|
||||
func (p *Provider) loadTCPServices(namespace string, backendRefs []gatev1.BackendRef) (*dynamic.TCPService, map[string]*dynamic.TCPService, error) {
|
||||
services := map[string]*dynamic.TCPService{}
|
||||
|
||||
wrrSvc := &dynamic.TCPService{
|
||||
Weighted: &dynamic.TCPWeightedRoundRobin{
|
||||
Services: []dynamic.TCPWRRService{},
|
||||
},
|
||||
// loadTCPWRRService is generating a WRR service, even when there is only one target.
|
||||
func (p *Provider) loadTCPWRRService(conf *dynamic.Configuration, routeKey string, backendRefs []gatev1.BackendRef, route *gatev1alpha2.TCPRoute) (string, *metav1.Condition) {
|
||||
name := routeKey + "-wrr"
|
||||
if _, ok := conf.TCP.Services[name]; ok {
|
||||
return name, nil
|
||||
}
|
||||
|
||||
var wrr dynamic.TCPWeightedRoundRobin
|
||||
var condition *metav1.Condition
|
||||
for _, backendRef := range backendRefs {
|
||||
if backendRef.Group == nil || backendRef.Kind == nil {
|
||||
// Should not happen as this is validated by kubernetes
|
||||
svcName, svc, errCondition := p.loadTCPService(route, backendRef)
|
||||
weight := ptr.To(int(ptr.Deref(backendRef.Weight, 1)))
|
||||
|
||||
if errCondition != nil {
|
||||
condition = errCondition
|
||||
|
||||
errName := routeKey + "-err-lb"
|
||||
conf.TCP.Services[errName] = &dynamic.TCPService{
|
||||
LoadBalancer: &dynamic.TCPServersLoadBalancer{
|
||||
Servers: []dynamic.TCPServer{},
|
||||
},
|
||||
}
|
||||
|
||||
wrr.Services = append(wrr.Services, dynamic.TCPWRRService{
|
||||
Name: errName,
|
||||
Weight: weight,
|
||||
})
|
||||
continue
|
||||
}
|
||||
|
||||
if isInternalService(backendRef) {
|
||||
return nil, nil, fmt.Errorf("traefik internal service %s is not allowed in a WRR loadbalancer", backendRef.Name)
|
||||
if svc != nil {
|
||||
conf.TCP.Services[svcName] = svc
|
||||
}
|
||||
|
||||
weight := int(ptr.Deref(backendRef.Weight, 1))
|
||||
wrr.Services = append(wrr.Services, dynamic.TCPWRRService{
|
||||
Name: svcName,
|
||||
Weight: weight,
|
||||
})
|
||||
}
|
||||
|
||||
if isTraefikService(backendRef) {
|
||||
wrrSvc.Weighted.Services = append(wrrSvc.Weighted.Services, dynamic.TCPWRRService{Name: string(backendRef.Name), Weight: &weight})
|
||||
continue
|
||||
conf.TCP.Services[name] = &dynamic.TCPService{Weighted: &wrr}
|
||||
return name, condition
|
||||
}
|
||||
|
||||
func (p *Provider) loadTCPService(route *gatev1alpha2.TCPRoute, backendRef gatev1.BackendRef) (string, *dynamic.TCPService, *metav1.Condition) {
|
||||
kind := ptr.Deref(backendRef.Kind, kindService)
|
||||
|
||||
group := groupCore
|
||||
if backendRef.Group != nil && *backendRef.Group != "" {
|
||||
group = string(*backendRef.Group)
|
||||
}
|
||||
|
||||
namespace := route.Namespace
|
||||
if backendRef.Namespace != nil && *backendRef.Namespace != "" {
|
||||
namespace = string(*backendRef.Namespace)
|
||||
}
|
||||
|
||||
serviceName := provider.Normalize(namespace + "-" + string(backendRef.Name))
|
||||
|
||||
if err := p.isReferenceGranted(kindTCPRoute, route.Namespace, group, string(kind), string(backendRef.Name), namespace); err != nil {
|
||||
return serviceName, nil, &metav1.Condition{
|
||||
Type: string(gatev1.RouteConditionResolvedRefs),
|
||||
Status: metav1.ConditionFalse,
|
||||
ObservedGeneration: route.Generation,
|
||||
LastTransitionTime: metav1.Now(),
|
||||
Reason: string(gatev1.RouteReasonRefNotPermitted),
|
||||
Message: fmt.Sprintf("Cannot load TCPRoute BackendRef %s/%s/%s/%s: %s", group, kind, namespace, backendRef.Name, err),
|
||||
}
|
||||
}
|
||||
|
||||
if *backendRef.Group != "" && *backendRef.Group != groupCore && *backendRef.Kind != "Service" {
|
||||
return nil, nil, fmt.Errorf("unsupported BackendRef %s/%s/%s", *backendRef.Group, *backendRef.Kind, backendRef.Name)
|
||||
}
|
||||
|
||||
if backendRef.Port == nil {
|
||||
return nil, nil, errors.New("port is required for Kubernetes Service reference")
|
||||
}
|
||||
|
||||
service, exists, err := p.client.GetService(namespace, string(backendRef.Name))
|
||||
if group != groupCore || kind != kindService {
|
||||
name, err := p.loadTCPBackendRef(backendRef)
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("getting service: %w", err)
|
||||
}
|
||||
if !exists {
|
||||
return nil, nil, errors.New("service not found")
|
||||
return serviceName, nil, &metav1.Condition{
|
||||
Type: string(gatev1.RouteConditionResolvedRefs),
|
||||
Status: metav1.ConditionFalse,
|
||||
ObservedGeneration: route.Generation,
|
||||
LastTransitionTime: metav1.Now(),
|
||||
Reason: string(gatev1.RouteReasonInvalidKind),
|
||||
Message: fmt.Sprintf("Cannot load TCPRoute BackendRef %s/%s/%s/%s: %s", group, kind, namespace, backendRef.Name, err),
|
||||
}
|
||||
}
|
||||
|
||||
var svcPort *corev1.ServicePort
|
||||
for _, p := range service.Spec.Ports {
|
||||
if p.Port == int32(*backendRef.Port) {
|
||||
svcPort = &p
|
||||
return name, nil, nil
|
||||
}
|
||||
|
||||
port := ptr.Deref(backendRef.Port, gatev1.PortNumber(0))
|
||||
if port == 0 {
|
||||
return serviceName, nil, &metav1.Condition{
|
||||
Type: string(gatev1.RouteConditionResolvedRefs),
|
||||
Status: metav1.ConditionFalse,
|
||||
ObservedGeneration: route.Generation,
|
||||
LastTransitionTime: metav1.Now(),
|
||||
Reason: string(gatev1.RouteReasonUnsupportedProtocol),
|
||||
Message: fmt.Sprintf("Cannot load TCPRoute BackendRef %s/%s/%s/%s port is required", group, kind, namespace, backendRef.Name),
|
||||
}
|
||||
}
|
||||
|
||||
portStr := strconv.FormatInt(int64(port), 10)
|
||||
serviceName = provider.Normalize(serviceName + "-" + portStr)
|
||||
|
||||
lb, err := p.loadTCPServers(namespace, backendRef)
|
||||
if err != nil {
|
||||
return serviceName, nil, &metav1.Condition{
|
||||
Type: string(gatev1.RouteConditionResolvedRefs),
|
||||
Status: metav1.ConditionFalse,
|
||||
ObservedGeneration: route.Generation,
|
||||
LastTransitionTime: metav1.Now(),
|
||||
Reason: string(gatev1.RouteReasonBackendNotFound),
|
||||
Message: fmt.Sprintf("Cannot load TCPRoute BackendRef %s/%s/%s/%s: %s", group, kind, namespace, backendRef.Name, err),
|
||||
}
|
||||
}
|
||||
|
||||
return serviceName, &dynamic.TCPService{LoadBalancer: lb}, nil
|
||||
}
|
||||
|
||||
func (p *Provider) loadTCPServers(namespace string, backendRef gatev1.BackendRef) (*dynamic.TCPServersLoadBalancer, error) {
|
||||
if backendRef.Port == nil {
|
||||
return nil, errors.New("port is required for Kubernetes Service reference")
|
||||
}
|
||||
|
||||
service, exists, err := p.client.GetService(namespace, string(backendRef.Name))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("getting service: %w", err)
|
||||
}
|
||||
if !exists {
|
||||
return nil, errors.New("service not found")
|
||||
}
|
||||
|
||||
var svcPort *corev1.ServicePort
|
||||
for _, p := range service.Spec.Ports {
|
||||
if p.Port == int32(*backendRef.Port) {
|
||||
svcPort = &p
|
||||
break
|
||||
}
|
||||
}
|
||||
if svcPort == nil {
|
||||
return nil, fmt.Errorf("service port %d not found", *backendRef.Port)
|
||||
}
|
||||
|
||||
endpointSlices, err := p.client.ListEndpointSlicesForService(namespace, string(backendRef.Name))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("getting endpointslices: %w", err)
|
||||
}
|
||||
if len(endpointSlices) == 0 {
|
||||
return nil, errors.New("endpointslices not found")
|
||||
}
|
||||
|
||||
lb := &dynamic.TCPServersLoadBalancer{}
|
||||
|
||||
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 svcPort == nil {
|
||||
return nil, nil, fmt.Errorf("service port %d not found", *backendRef.Port)
|
||||
if port == 0 {
|
||||
continue
|
||||
}
|
||||
|
||||
endpointSlices, err := p.client.ListEndpointSlicesForService(namespace, string(backendRef.Name))
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("getting endpointslices: %w", err)
|
||||
}
|
||||
if len(endpointSlices) == 0 {
|
||||
return nil, nil, errors.New("endpointslices not found")
|
||||
}
|
||||
|
||||
svc := dynamic.TCPService{LoadBalancer: &dynamic.TCPServersLoadBalancer{}}
|
||||
|
||||
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 {
|
||||
for _, endpoint := range endpointSlice.Endpoints {
|
||||
if endpoint.Conditions.Ready == nil || !*endpoint.Conditions.Ready {
|
||||
continue
|
||||
}
|
||||
|
||||
for _, endpoint := range endpointSlice.Endpoints {
|
||||
if endpoint.Conditions.Ready == nil || !*endpoint.Conditions.Ready {
|
||||
for _, address := range endpoint.Addresses {
|
||||
if _, ok := addresses[address]; ok {
|
||||
continue
|
||||
}
|
||||
|
||||
for _, address := range endpoint.Addresses {
|
||||
if _, ok := addresses[address]; ok {
|
||||
continue
|
||||
}
|
||||
|
||||
addresses[address] = struct{}{}
|
||||
svc.LoadBalancer.Servers = append(svc.LoadBalancer.Servers, dynamic.TCPServer{
|
||||
Address: net.JoinHostPort(address, strconv.Itoa(int(port))),
|
||||
})
|
||||
}
|
||||
addresses[address] = struct{}{}
|
||||
lb.Servers = append(lb.Servers, dynamic.TCPServer{
|
||||
// TODO determine whether the servers needs TLS, from the port?
|
||||
Address: net.JoinHostPort(address, strconv.Itoa(int(port))),
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
serviceName := provider.Normalize(service.Namespace + "-" + service.Name + "-" + strconv.Itoa(int(svcPort.Port)))
|
||||
services[serviceName] = &svc
|
||||
|
||||
wrrSvc.Weighted.Services = append(wrrSvc.Weighted.Services, dynamic.TCPWRRService{Name: serviceName, Weight: &weight})
|
||||
}
|
||||
|
||||
if len(wrrSvc.Weighted.Services) == 0 {
|
||||
return nil, nil, errors.New("no service has been created")
|
||||
return lb, nil
|
||||
}
|
||||
|
||||
func (p *Provider) loadTCPBackendRef(backendRef gatev1.BackendRef) (string, error) {
|
||||
// Support for cross-provider references (e.g: api@internal).
|
||||
// This provides the same behavior as for IngressRoutes.
|
||||
if *backendRef.Kind == "TraefikService" && strings.Contains(string(backendRef.Name), "@") {
|
||||
return string(backendRef.Name), nil
|
||||
}
|
||||
|
||||
return wrrSvc, services, nil
|
||||
return "", fmt.Errorf("unsupported BackendRef %s/%s/%s", *backendRef.Group, *backendRef.Kind, backendRef.Name)
|
||||
}
|
||||
|
||||
func mergeTCPConfiguration(from, to *dynamic.Configuration) {
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue