Add TCPRoute and UDPRoute Support for L4 Load Balancing

Author: Skcey

internal/controller/manager.go

@@ -528,6 +528,18 @@ func registerControllers(
 					controller.WithK8sPredicate(k8spredicate.GenerationChangedPredicate{}),
 				},
 			},
+			{
+				objectType: &gatewayv1alpha2.TCPRoute{},
+				options: []controller.Option{
+					controller.WithK8sPredicate(k8spredicate.GenerationChangedPredicate{}),
+				},
+			},
+			{
+				objectType: &gatewayv1alpha2.UDPRoute{},
+				options: []controller.Option{
+					controller.WithK8sPredicate(k8spredicate.GenerationChangedPredicate{}),
+				},
+			},
 		}
 		controllerRegCfgs = append(controllerRegCfgs, gwExpFeatures...)
 	}
@@ -754,6 +766,8 @@ func prepareFirstEventBatchPreparerArgs(cfg config.Config) ([]client.Object, []c
 			&gatewayv1alpha3.BackendTLSPolicyList{},
 			&apiv1.ConfigMapList{},
 			&gatewayv1alpha2.TLSRouteList{},
+			&gatewayv1alpha2.TCPRouteList{},
+			&gatewayv1alpha2.UDPRouteList{},
 		)
 	}
 

internal/controller/nginx/config/stream/config.go

@@ -11,6 +11,12 @@ type Server struct {
 	RewriteClientIP shared.RewriteClientIPSettings
 	SSLPreread      bool
 	IsSocket        bool
+	Protocol        string
+	UDPConfig       *UDPConfig
+}
+
+type UDPConfig struct {
+	ProxyTimeout string
 }
 
 // Upstream holds all configuration for a stream upstream.

internal/controller/nginx/config/stream_servers.go

@@ -32,17 +32,24 @@ func (g GeneratorImpl) executeStreamServers(conf dataplane.Configuration) []exec
 }
 
 func createStreamServers(conf dataplane.Configuration) []stream.Server {
-	if len(conf.TLSPassthroughServers) == 0 {
+	totalServers := len(conf.TLSPassthroughServers) + len(conf.TCPServers) + len(conf.UDPServers)
+	if totalServers == 0 {
 		return nil
 	}
 
-	streamServers := make([]stream.Server, 0, len(conf.TLSPassthroughServers)*2)
+	streamServers := make([]stream.Server, 0, totalServers*2)
 	portSet := make(map[int32]struct{})
 	upstreams := make(map[string]dataplane.Upstream)
 
 	for _, u := range conf.StreamUpstreams {
 		upstreams[u.Name] = u
 	}
+	for _, u := range conf.TCPUpstreams {
+		upstreams[u.Name] = u
+	}
+	for _, u := range conf.UDPUpstreams {
+		upstreams[u.Name] = u
+	}
 
 	for _, server := range conf.TLSPassthroughServers {
 		if u, ok := upstreams[server.UpstreamName]; ok && server.UpstreamName != "" {
@@ -76,6 +83,47 @@ func createStreamServers(conf dataplane.Configuration) []stream.Server {
 		}
 		streamServers = append(streamServers, streamServer)
 	}
+
+	// Process TCP servers
+	for i, server := range conf.TCPServers {
+		if _, inPortSet := portSet[server.Port]; inPortSet {
+			continue // Skip if port already in use
+		}
+
+		if u, ok := upstreams[server.UpstreamName]; ok && server.UpstreamName != "" && len(u.Endpoints) > 0 {
+			streamServer := stream.Server{
+				Listen:     fmt.Sprint(server.Port),
+				StatusZone: fmt.Sprintf("tcp_%d", server.Port),
+				ProxyPass:  server.UpstreamName,
+			}
+			streamServers = append(streamServers, streamServer)
+			portSet[server.Port] = struct{}{}
+		} else {
+			fmt.Printf("DEBUG: createStreamServers - TCP Server %d: Skipped - upstream not found or no endpoints\n", i)
+		}
+	}
+
+	// Process UDP servers
+	for _, server := range conf.UDPServers {
+		if _, inPortSet := portSet[server.Port]; inPortSet {
+			continue // Skip if port already in use
+		}
+
+		if u, ok := upstreams[server.UpstreamName]; ok && server.UpstreamName != "" && len(u.Endpoints) > 0 {
+			streamServer := stream.Server{
+				Listen:     fmt.Sprintf("%d udp", server.Port),
+				StatusZone: fmt.Sprintf("udp_%d", server.Port),
+				ProxyPass:  server.UpstreamName,
+				Protocol:   "udp",
+				UDPConfig: &stream.UDPConfig{
+					ProxyTimeout: "1s",
+				},
+			}
+			streamServers = append(streamServers, streamServer)
+			portSet[server.Port] = struct{}{}
+		}
+	}
+
 	return streamServers
 }
 

internal/controller/nginx/config/stream_servers_template.go

@@ -26,6 +26,10 @@ server {
 	{{- if $s.SSLPreread }}
     ssl_preread on;
 	{{- end }}
+
+    {{- if and (eq $s.Protocol "udp") $s.UDPConfig }}
+    proxy_timeout {{ $s.UDPConfig.ProxyTimeout }};
+    {{- end }}
 }
 {{- end }}
 

internal/controller/nginx/config/upstreams.go

@@ -69,7 +69,13 @@ func executeUpstreams(upstreams []http.Upstream) []executeResult {
 }
 
 func (g GeneratorImpl) executeStreamUpstreams(conf dataplane.Configuration) []executeResult {
-	upstreams := g.createStreamUpstreams(conf.StreamUpstreams)
+	// Combine all stream upstreams: TLS, TCP, and UDP
+	allUpstreams := make([]dataplane.Upstream, 0, len(conf.StreamUpstreams)+len(conf.TCPUpstreams)+len(conf.UDPUpstreams))
+	allUpstreams = append(allUpstreams, conf.StreamUpstreams...)
+	allUpstreams = append(allUpstreams, conf.TCPUpstreams...)
+	allUpstreams = append(allUpstreams, conf.UDPUpstreams...)
+
+	upstreams := g.createStreamUpstreams(allUpstreams)
 
 	result := executeResult{
 		dest: streamConfigFile,

internal/controller/provisioner/objects.go

@@ -39,6 +39,11 @@ const (
 	defaultImagePullPolicy    = corev1.PullIfNotPresent
 )
 
+type PortInfo struct {
+	Port     int32
+	Protocol corev1.Protocol
+}
+
 var emptyDirVolumeSource = corev1.VolumeSource{EmptyDir: &corev1.EmptyDirVolumeSource{}}
 
 func (p *NginxProvisioner) buildNginxResourceObjects(
@@ -127,9 +132,18 @@ func (p *NginxProvisioner) buildNginxResourceObjects(
 		openshiftObjs = p.buildOpenshiftObjects(objectMeta)
 	}
 
-	ports := make(map[int32]struct{})
+	ports := make(map[int32]PortInfo)
 	for _, listener := range gateway.Spec.Listeners {
-		ports[int32(listener.Port)] = struct{}{}
+		var protocol corev1.Protocol
+		switch listener.Protocol {
+		case gatewayv1.TCPProtocolType:
+			protocol = corev1.ProtocolTCP
+		case gatewayv1.UDPProtocolType:
+			protocol = corev1.ProtocolUDP
+		default:
+			protocol = corev1.ProtocolTCP
+		}
+		ports[int32(listener.Port)] = PortInfo{Port: int32(listener.Port), Protocol: protocol}
 	}
 
 	service := buildNginxService(objectMeta, nProxyCfg, ports, selectorLabels)
@@ -418,7 +432,7 @@ func (p *NginxProvisioner) buildOpenshiftObjects(objectMeta metav1.ObjectMeta) [
 func buildNginxService(
 	objectMeta metav1.ObjectMeta,
 	nProxyCfg *graph.EffectiveNginxProxy,
-	ports map[int32]struct{},
+	ports map[int32]PortInfo,
 	selectorLabels map[string]string,
 ) *corev1.Service {
 	var serviceCfg ngfAPIv1alpha2.ServiceSpec
@@ -440,16 +454,17 @@ func buildNginxService(
 	}
 
 	servicePorts := make([]corev1.ServicePort, 0, len(ports))
-	for port := range ports {
+	for _, portInfo := range ports {
 		servicePort := corev1.ServicePort{
-			Name:       fmt.Sprintf("port-%d", port),
-			Port:       port,
-			TargetPort: intstr.FromInt32(port),
+			Name:       fmt.Sprintf("port-%d", portInfo.Port),
+			Port:       portInfo.Port,
+			TargetPort: intstr.FromInt32(portInfo.Port),
+			Protocol:   portInfo.Protocol,
 		}
 
 		if serviceType != corev1.ServiceTypeClusterIP {
 			for _, nodePort := range serviceCfg.NodePorts {
-				if nodePort.ListenerPort == port {
+				if nodePort.ListenerPort == portInfo.Port {
 					servicePort.NodePort = nodePort.Port
 				}
 			}
@@ -506,7 +521,7 @@ func (p *NginxProvisioner) buildNginxDeployment(
 	nProxyCfg *graph.EffectiveNginxProxy,
 	ngxIncludesConfigMapName string,
 	ngxAgentConfigMapName string,
-	ports map[int32]struct{},
+	ports map[int32]PortInfo,
 	selectorLabels map[string]string,
 	agentTLSSecretName string,
 	dockerSecretNames map[string]string,
@@ -567,18 +582,19 @@ func (p *NginxProvisioner) buildNginxPodTemplateSpec(
 	nProxyCfg *graph.EffectiveNginxProxy,
 	ngxIncludesConfigMapName string,
 	ngxAgentConfigMapName string,
-	ports map[int32]struct{},
+	ports map[int32]PortInfo,
 	agentTLSSecretName string,
 	dockerSecretNames map[string]string,
 	jwtSecretName string,
 	caSecretName string,
 	clientSSLSecretName string,
 ) corev1.PodTemplateSpec {
 	containerPorts := make([]corev1.ContainerPort, 0, len(ports))
-	for port := range ports {
+	for _, portInfo := range ports {
 		containerPort := corev1.ContainerPort{
-			Name:          fmt.Sprintf("port-%d", port),
-			ContainerPort: port,
+			Name:          fmt.Sprintf("port-%d", portInfo.Port),
+			ContainerPort: portInfo.Port,
+			Protocol:      portInfo.Protocol,
 		}
 		containerPorts = append(containerPorts, containerPort)
 	}

internal/controller/state/change_processor.go

@@ -96,6 +96,8 @@ func NewChangeProcessorImpl(cfg ChangeProcessorConfig) *ChangeProcessorImpl {
 		NginxProxies:       make(map[types.NamespacedName]*ngfAPIv1alpha2.NginxProxy),
 		GRPCRoutes:         make(map[types.NamespacedName]*v1.GRPCRoute),
 		TLSRoutes:          make(map[types.NamespacedName]*v1alpha2.TLSRoute),
+		TCPRoutes:          make(map[types.NamespacedName]*v1alpha2.TCPRoute),
+		UDPRoutes:          make(map[types.NamespacedName]*v1alpha2.UDPRoute),
 		NGFPolicies:        make(map[graph.PolicyKey]policies.Policy),
 		SnippetsFilters:    make(map[types.NamespacedName]*ngfAPIv1alpha1.SnippetsFilter),
 	}
@@ -211,6 +213,16 @@ func NewChangeProcessorImpl(cfg ChangeProcessorConfig) *ChangeProcessorImpl {
 				store:     newObjectStoreMapAdapter(clusterStore.TLSRoutes),
 				predicate: nil,
 			},
+			{
+				gvk:       cfg.MustExtractGVK(&v1alpha2.TCPRoute{}),
+				store:     newObjectStoreMapAdapter(clusterStore.TCPRoutes),
+				predicate: nil,
+			},
+			{
+				gvk:       cfg.MustExtractGVK(&v1alpha2.UDPRoute{}),
+				store:     newObjectStoreMapAdapter(clusterStore.UDPRoutes),
+				predicate: nil,
+			},
 			{
 				gvk:       cfg.MustExtractGVK(&ngfAPIv1alpha1.SnippetsFilter{}),
 				store:     newObjectStoreMapAdapter(clusterStore.SnippetsFilters),

internal/controller/state/change_processor_test.go

@@ -3776,7 +3776,7 @@ var _ = Describe("ChangeProcessor", func() {
 			},
 			Entry(
 				"an unsupported resource",
-				&v1alpha2.TCPRoute{ObjectMeta: metav1.ObjectMeta{Namespace: "test", Name: "tcp"}},
+				&apiv1.Pod{ObjectMeta: metav1.ObjectMeta{Namespace: "test", Name: "pod"}},
 			),
 			Entry(
 				"nil resource",
@@ -3794,8 +3794,8 @@ var _ = Describe("ChangeProcessor", func() {
 			},
 			Entry(
 				"an unsupported resource",
-				&v1alpha2.TCPRoute{},
-				types.NamespacedName{Namespace: "test", Name: "tcp"},
+				&apiv1.Pod{},
+				types.NamespacedName{Namespace: "test", Name: "pod"},
 			),
 			Entry(
 				"nil resource type",