From d80419f449ace8c11c4ac56173db6896d985d03b Mon Sep 17 00:00:00 2001 From: AlinsRan Date: Wed, 22 Jul 2026 08:47:09 +0800 Subject: [PATCH] fix: honor TCPRoute/UDPRoute sectionName and listener port for StreamRoute matching TranslateTCPRoute/TranslateUDPRoute emitted a single StreamRoute with no match criteria, so every TCP/UDP connection on any stream listener matched it. All L4 routes collided onto one backend, ignoring parentRefs.sectionName and per-listener port routing. The controllers now populate tctx.Listeners from the listeners ParseRouteParentRefs matched for each parentRef (honoring sectionName, port, protocol and allowedRoutes). The translator derives StreamRoute.server_port from the matched listener port(s), emitting one StreamRoute per port with a port-qualified name/ID so listeners no longer collide. Injection is gated locally (explicit sectionName/port targeting, or more than one distinct listener port); otherwise the previous single portless StreamRoute is kept for backward compatibility. L4 e2e listeners are updated to APISIX's real stream ports (TCP 9100, UDP 9200) so server_port isolation is exercised. Sync from apache/apisix-ingress-controller#2818 (fixes #2802) --- .../adc/translator/l4route_serverport_test.go | 216 ++++++++++++++++++ internal/adc/translator/tcproute.go | 101 +++++++- internal/adc/translator/udproute.go | 13 +- internal/controller/context.go | 5 + internal/controller/tcproute_controller.go | 7 + internal/controller/udproute_controller.go | 7 + internal/controller/utils.go | 9 + test/e2e/gatewayapi/tcproute.go | 8 +- test/e2e/gatewayapi/udproute.go | 4 +- 9 files changed, 345 insertions(+), 25 deletions(-) create mode 100644 internal/adc/translator/l4route_serverport_test.go diff --git a/internal/adc/translator/l4route_serverport_test.go b/internal/adc/translator/l4route_serverport_test.go new file mode 100644 index 00000000..550570eb --- /dev/null +++ b/internal/adc/translator/l4route_serverport_test.go @@ -0,0 +1,216 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package translator + +import ( + "context" + "testing" + + "github.com/go-logr/logr" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/utils/ptr" + gatewayv1 "sigs.k8s.io/gateway-api/apis/v1" + gatewayv1alpha2 "sigs.k8s.io/gateway-api/apis/v1alpha2" + + "github.com/apache/apisix-ingress-controller/internal/provider" +) + +func tcpListener(name string, port int32) gatewayv1.Listener { + return gatewayv1.Listener{ + Name: gatewayv1.SectionName(name), + Protocol: gatewayv1.TCPProtocolType, + Port: gatewayv1.PortNumber(port), + } +} + +func udpListener(name string, port int32) gatewayv1.Listener { + return gatewayv1.Listener{ + Name: gatewayv1.SectionName(name), + Protocol: gatewayv1.UDPProtocolType, + Port: gatewayv1.PortNumber(port), + } +} + +// sectionParentRefs builds the parentRefs a controller would set on tctx when a +// route explicitly targets a listener by sectionName. +func sectionParentRefs(section string) []gatewayv1.ParentReference { + return []gatewayv1.ParentReference{ + { + Name: "gw", + SectionName: ptr.To(gatewayv1.SectionName(section)), + }, + } +} + +func TestTranslateTCPRouteServerPort(t *testing.T) { + tests := []struct { + name string + // listeners the controller would have matched for this route's parentRefs + listeners []gatewayv1.Listener + // parentRefs the controller stored on tctx (drives server_port injection) + parentRefs []gatewayv1.ParentReference + wantPorts []int32 + wantNoMatch bool + }{ + { + name: "explicit sectionName injects the matching listener port", + listeners: []gatewayv1.Listener{tcpListener("tcp-a", 9100)}, + parentRefs: sectionParentRefs("tcp-a"), + wantPorts: []int32{9100}, + }, + { + name: "multiple listener ports fan out even without explicit targeting", + listeners: []gatewayv1.Listener{tcpListener("tcp-a", 9100), tcpListener("tcp-b", 9101)}, + wantPorts: []int32{9100, 9101}, + }, + { + name: "duplicate ports across gateways are de-duplicated", + listeners: []gatewayv1.Listener{tcpListener("tcp-a", 9100), tcpListener("tcp-a2", 9100)}, + parentRefs: sectionParentRefs("tcp-a"), + wantPorts: []int32{9100}, + }, + { + name: "single listener without explicit targeting keeps a portless StreamRoute", + listeners: []gatewayv1.Listener{tcpListener("tcp-a", 9100)}, + wantNoMatch: true, + }, + { + name: "no matched listener falls back to a single portless StreamRoute", + listeners: nil, + parentRefs: sectionParentRefs("tcp-a"), + wantNoMatch: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + translator := NewTranslator(logr.Discard()) + tctx := provider.NewDefaultTranslateContext(context.Background()) + tctx.Listeners = tt.listeners + tctx.RouteParentRefs = tt.parentRefs + + route := &gatewayv1alpha2.TCPRoute{ + ObjectMeta: metav1.ObjectMeta{Name: "my-tcp", Namespace: "default"}, + Spec: gatewayv1alpha2.TCPRouteSpec{ + Rules: []gatewayv1alpha2.TCPRouteRule{ + {BackendRefs: []gatewayv1alpha2.BackendRef{}}, + }, + }, + } + + result, err := translator.TranslateTCPRoute(tctx, route) + require.NoError(t, err) + require.Len(t, result.Services, 1) + streamRoutes := result.Services[0].StreamRoutes + + if tt.wantNoMatch { + require.Len(t, streamRoutes, 1) + assert.Zero(t, streamRoutes[0].ServerPort) + return + } + + require.Len(t, streamRoutes, len(tt.wantPorts)) + gotPorts := make([]int32, 0, len(streamRoutes)) + ids := make(map[string]struct{}) + names := make(map[string]struct{}) + for _, sr := range streamRoutes { + gotPorts = append(gotPorts, sr.ServerPort) + ids[sr.ID] = struct{}{} + names[sr.Name] = struct{}{} + } + assert.ElementsMatch(t, tt.wantPorts, gotPorts) + // Distinct name/ID per listener port so StreamRoutes do not collide. + assert.Len(t, ids, len(streamRoutes)) + assert.Len(t, names, len(streamRoutes)) + }) + } +} + +func TestTranslateUDPRouteServerPort(t *testing.T) { + tests := []struct { + name string + listeners []gatewayv1.Listener + parentRefs []gatewayv1.ParentReference + wantPorts []int32 + wantNoMatch bool + }{ + { + name: "explicit sectionName injects the matching listener port", + listeners: []gatewayv1.Listener{udpListener("udp-a", 9200)}, + parentRefs: sectionParentRefs("udp-a"), + wantPorts: []int32{9200}, + }, + { + name: "two listeners on different ports produce distinct StreamRoutes", + listeners: []gatewayv1.Listener{udpListener("udp-a", 9200), udpListener("udp-b", 9201)}, + wantPorts: []int32{9200, 9201}, + }, + { + name: "single listener without explicit targeting keeps a portless StreamRoute", + listeners: []gatewayv1.Listener{udpListener("udp-a", 9200)}, + wantNoMatch: true, + }, + { + name: "no matched listener falls back to a single portless StreamRoute", + listeners: nil, + parentRefs: sectionParentRefs("udp-a"), + wantNoMatch: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + translator := NewTranslator(logr.Discard()) + tctx := provider.NewDefaultTranslateContext(context.Background()) + tctx.Listeners = tt.listeners + tctx.RouteParentRefs = tt.parentRefs + + route := &gatewayv1alpha2.UDPRoute{ + ObjectMeta: metav1.ObjectMeta{Name: "my-udp", Namespace: "default"}, + Spec: gatewayv1alpha2.UDPRouteSpec{ + Rules: []gatewayv1alpha2.UDPRouteRule{ + {BackendRefs: []gatewayv1alpha2.BackendRef{}}, + }, + }, + } + + result, err := translator.TranslateUDPRoute(tctx, route) + require.NoError(t, err) + require.Len(t, result.Services, 1) + streamRoutes := result.Services[0].StreamRoutes + + if tt.wantNoMatch { + require.Len(t, streamRoutes, 1) + assert.Zero(t, streamRoutes[0].ServerPort) + return + } + + require.Len(t, streamRoutes, len(tt.wantPorts)) + gotPorts := make([]int32, 0, len(streamRoutes)) + ids := make(map[string]struct{}) + for _, sr := range streamRoutes { + gotPorts = append(gotPorts, sr.ServerPort) + ids[sr.ID] = struct{}{} + } + assert.ElementsMatch(t, tt.wantPorts, gotPorts) + assert.Len(t, ids, len(streamRoutes)) + }) + } +} diff --git a/internal/adc/translator/tcproute.go b/internal/adc/translator/tcproute.go index 36c43880..e88ad8fc 100644 --- a/internal/adc/translator/tcproute.go +++ b/internal/adc/translator/tcproute.go @@ -19,6 +19,7 @@ package translator import ( "fmt" + "sort" gatewayv1 "sigs.k8s.io/gateway-api/apis/v1" gatewayv1alpha2 "sigs.k8s.io/gateway-api/apis/v1alpha2" @@ -42,6 +43,93 @@ func newDefaultUpstreamWithoutScheme() *adctypes.Upstream { } } +// hasExplicitListenerTarget reports whether any parentRef explicitly targets a +// listener by sectionName or port. Non-Gateway parentRefs are ignored. +func hasExplicitListenerTarget(parentRefs []gatewayv1.ParentReference) bool { + for _, parentRef := range parentRefs { + if parentRef.Kind != nil && *parentRef.Kind != "Gateway" { + continue + } + if parentRef.SectionName != nil && *parentRef.SectionName != "" { + return true + } + if parentRef.Port != nil { + return true + } + } + return false +} + +// shouldInjectL4ServerPort decides whether to set StreamRoute.server_port from +// the matched listener port(s). The listener port is a logical Gateway value +// that must equal APISIX's physical stream listen port for the match to work, +// so injection is opt-in ("auto" behavior): inject when the route explicitly +// targets a listener (sectionName/port) or when more than one distinct listener +// port matched. Otherwise keep a single portless StreamRoute (backward compatible). +func shouldInjectL4ServerPort(parentRefs []gatewayv1.ParentReference, ports map[int32]struct{}) bool { + if len(ports) == 0 { + return false + } + return hasExplicitListenerTarget(parentRefs) || len(ports) > 1 +} + +// listenerPortSet returns the de-duplicated set of ports of the listeners the +// route attaches to. tctx.Listeners is populated by the controller from the +// listeners that ParseRouteParentRefs already matched against the route's +// parentRefs (honoring sectionName, port, protocol and allowedRoutes). +func listenerPortSet(tctx *provider.TranslateContext) map[int32]struct{} { + portSet := make(map[int32]struct{}, len(tctx.Listeners)) + for _, listener := range tctx.Listeners { + portSet[int32(listener.Port)] = struct{}{} + } + return portSet +} + +// buildL4StreamRoutes builds the StreamRoutes for one L4 route rule. +// +// A StreamRoute without a server_port match matches every connection on any +// stream listener, so multiple L4 routes collide onto one backend (#2802). To +// isolate them we set server_port from the matched listener port(s), emitting +// one StreamRoute per port with a port-qualified name/ID. When injection is not +// warranted (see shouldInjectL4ServerPort) we keep a single portless StreamRoute, +// preserving backward compatibility. +func (t *Translator) buildL4StreamRoutes(tctx *provider.TranslateContext, namespace, name string, ruleIndex int, typ, routeKind string, labels map[string]string) []*adctypes.StreamRoute { + var ports []int32 + if portSet := listenerPortSet(tctx); shouldInjectL4ServerPort(tctx.RouteParentRefs, portSet) { + ports = make([]int32, 0, len(portSet)) + for port := range portSet { + ports = append(ports, port) + } + sort.Slice(ports, func(i, j int) bool { return ports[i] < ports[j] }) + } + if len(ports) == 0 { + // No server_port isolation: a single StreamRoute that matches all + // connections on the stream listener, as before. + ports = []int32{0} + } + streamRoutes := make([]*adctypes.StreamRoute, 0, len(ports)) + for _, port := range ports { + streamRoute := adctypes.NewDefaultStreamRoute() + ruleKey := fmt.Sprintf("%d", ruleIndex) + if port != 0 { + // Include the port in the name key so multiple listeners produce + // distinct StreamRoute names/IDs instead of colliding. + ruleKey = fmt.Sprintf("%d-%d", ruleIndex, port) + streamRoute.ServerPort = port + } + streamRouteName := adctypes.ComposeStreamRouteName(namespace, name, ruleKey, typ) + streamRoute.Name = streamRouteName + streamRoute.ID = id.GenID(streamRouteName) + streamRoute.Labels = labels + // Attach L4RoutePolicy plugins at the stream_route level: the APISIX stream proxy + // applies plugins from the stream_route, not from the service. + streamRoute.Plugins = make(adctypes.Plugins) + t.AttachL4RoutePolicyPlugins(tctx.L4RoutePolicies, namespace, name, routeKind, streamRoute.Plugins) + streamRoutes = append(streamRoutes, streamRoute) + } + return streamRoutes +} + func (t *Translator) TranslateTCPRoute(tctx *provider.TranslateContext, tcpRoute *gatewayv1alpha2.TCPRoute) (*TranslateResult, error) { result := &TranslateResult{} rules := tcpRoute.Spec.Rules @@ -150,17 +238,8 @@ func (t *Translator) TranslateTCPRoute(tctx *provider.TranslateContext, tcpRoute } } } - streamRoute := adctypes.NewDefaultStreamRoute() - streamRouteName := adctypes.ComposeStreamRouteName(tcpRoute.Namespace, tcpRoute.Name, fmt.Sprintf("%d", ruleIndex), "TCP") - streamRoute.Name = streamRouteName - streamRoute.ID = id.GenID(streamRouteName) - streamRoute.Labels = labels - // TODO: support remote_addr, server_addr, sni, server_port - // Attach L4RoutePolicy plugins at the stream_route level: the APISIX stream proxy - // applies plugins from the stream_route, not from the service. - streamRoute.Plugins = make(adctypes.Plugins) - t.AttachL4RoutePolicyPlugins(tctx.L4RoutePolicies, tcpRoute.Namespace, tcpRoute.Name, "TCPRoute", streamRoute.Plugins) - service.StreamRoutes = append(service.StreamRoutes, streamRoute) + // TODO: support remote_addr, server_addr, sni + service.StreamRoutes = t.buildL4StreamRoutes(tctx, tcpRoute.Namespace, tcpRoute.Name, ruleIndex, "TCP", "TCPRoute", labels) result.Services = append(result.Services, service) } diff --git a/internal/adc/translator/udproute.go b/internal/adc/translator/udproute.go index 650c4256..b9524e0e 100644 --- a/internal/adc/translator/udproute.go +++ b/internal/adc/translator/udproute.go @@ -139,17 +139,8 @@ func (t *Translator) TranslateUDPRoute(tctx *provider.TranslateContext, udpRoute } } } - streamRoute := adctypes.NewDefaultStreamRoute() - streamRouteName := adctypes.ComposeStreamRouteName(udpRoute.Namespace, udpRoute.Name, fmt.Sprintf("%d", ruleIndex), "UDP") - streamRoute.Name = streamRouteName - streamRoute.ID = id.GenID(streamRouteName) - streamRoute.Labels = labels - // TODO: support remote_addr, server_addr, sni, server_port - // Attach L4RoutePolicy plugins at the stream_route level: the APISIX stream proxy - // applies plugins from the stream_route, not from the service. - streamRoute.Plugins = make(adctypes.Plugins) - t.AttachL4RoutePolicyPlugins(tctx.L4RoutePolicies, udpRoute.Namespace, udpRoute.Name, "UDPRoute", streamRoute.Plugins) - service.StreamRoutes = append(service.StreamRoutes, streamRoute) + // TODO: support remote_addr, server_addr, sni + service.StreamRoutes = t.buildL4StreamRoutes(tctx, udpRoute.Namespace, udpRoute.Name, ruleIndex, "UDP", "UDPRoute", labels) result.Services = append(result.Services, service) } diff --git a/internal/controller/context.go b/internal/controller/context.go index 5398f044..45b8bde1 100644 --- a/internal/controller/context.go +++ b/internal/controller/context.go @@ -27,6 +27,11 @@ type RouteParentRefContext struct { ListenerName string Listener *gatewayv1.Listener + // Listeners holds all listeners this parentRef matched (honoring + // sectionName, port, protocol and allowedRoutes). Listener keeps the first + // match for backward compatibility; Listeners is used to derive the + // StreamRoute server_port for L4 routes. + Listeners []gatewayv1.Listener Conditions []metav1.Condition } diff --git a/internal/controller/tcproute_controller.go b/internal/controller/tcproute_controller.go index f3487547..65c2f1a0 100644 --- a/internal/controller/tcproute_controller.go +++ b/internal/controller/tcproute_controller.go @@ -289,6 +289,13 @@ func (r *TCPRouteReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c acceptStatus.status = false acceptStatus.msg = err.Error() } + // Populate the matched listeners so the translator can derive the + // StreamRoute server_port from the listener the route attaches to. + if len(gateway.Listeners) > 0 { + tctx.Listeners = appendListeners(tctx.Listeners, gateway.Listeners...) + } else if gateway.Listener != nil { + tctx.Listeners = appendListeners(tctx.Listeners, *gateway.Listener) + } } var backendRefErr error diff --git a/internal/controller/udproute_controller.go b/internal/controller/udproute_controller.go index 070cee76..8f79e87a 100644 --- a/internal/controller/udproute_controller.go +++ b/internal/controller/udproute_controller.go @@ -289,6 +289,13 @@ func (r *UDPRouteReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c acceptStatus.status = false acceptStatus.msg = err.Error() } + // Populate the matched listeners so the translator can derive the + // StreamRoute server_port from the listener the route attaches to. + if len(gateway.Listeners) > 0 { + tctx.Listeners = appendListeners(tctx.Listeners, gateway.Listeners...) + } else if gateway.Listener != nil { + tctx.Listeners = appendListeners(tctx.Listeners, *gateway.Listener) + } } var backendRefErr error diff --git a/internal/controller/utils.go b/internal/controller/utils.go index 04400609..e4114db1 100644 --- a/internal/controller/utils.go +++ b/internal/controller/utils.go @@ -374,6 +374,7 @@ func ParseRouteParentRefs( reason := gatewayv1.RouteReasonNoMatchingParent var listenerName string var matchedListener gatewayv1.Listener + var matchedListeners []gatewayv1.Listener for _, listener := range gateway.Spec.Listeners { if parentRef.SectionName != nil { @@ -414,6 +415,7 @@ func ParseRouteParentRefs( matched = true matchedListener = listener + matchedListeners = appendListeners(matchedListeners, listener) break } @@ -422,6 +424,7 @@ func ParseRouteParentRefs( Gateway: &gateway, ListenerName: listenerName, Listener: &matchedListener, + Listeners: matchedListeners, Conditions: []metav1.Condition{{ Type: string(gatewayv1.RouteConditionAccepted), Status: metav1.ConditionTrue, @@ -434,6 +437,7 @@ func ParseRouteParentRefs( Gateway: &gateway, ListenerName: listenerName, Listener: &matchedListener, + Listeners: matchedListeners, Conditions: []metav1.Condition{{ Type: string(gatewayv1.RouteConditionAccepted), Status: metav1.ConditionFalse, @@ -1266,6 +1270,11 @@ func isListenerHostnameEffective(listener gatewayv1.Listener) bool { listener.Protocol == gatewayv1.TLSProtocolType } +// appendListeners appends listeners without de-duplication. +func appendListeners(target []gatewayv1.Listener, source ...gatewayv1.Listener) []gatewayv1.Listener { + return append(target, source...) +} + func isRouteAccepted(gateways []RouteParentRefContext) bool { for _, gateway := range gateways { for _, condition := range gateway.Conditions { diff --git a/test/e2e/gatewayapi/tcproute.go b/test/e2e/gatewayapi/tcproute.go index 080bfdb6..c2773c18 100644 --- a/test/e2e/gatewayapi/tcproute.go +++ b/test/e2e/gatewayapi/tcproute.go @@ -43,7 +43,9 @@ spec: listeners: - name: tcp protocol: TCP - port: 80 + # APISIX stream_proxy listens on 9100; the listener port must match it so the + # StreamRoute server_port (derived from this listener) isolates traffic (#2802). + port: 9100 allowedRoutes: kinds: - kind: TCPRoute @@ -119,7 +121,9 @@ spec: listeners: - name: tcp protocol: TCP - port: 80 + # APISIX stream_proxy listens on 9100; the listener port must match it so the + # StreamRoute server_port (derived from this listener) isolates traffic (#2802). + port: 9100 allowedRoutes: kinds: - kind: TCPRoute diff --git a/test/e2e/gatewayapi/udproute.go b/test/e2e/gatewayapi/udproute.go index f59737f2..0920dfd8 100644 --- a/test/e2e/gatewayapi/udproute.go +++ b/test/e2e/gatewayapi/udproute.go @@ -40,7 +40,9 @@ spec: listeners: - name: udp protocol: UDP - port: 80 + # APISIX stream_proxy listens UDP on 9200; the listener port must match it so + # the StreamRoute server_port (derived from this listener) isolates traffic (#2802). + port: 9200 allowedRoutes: kinds: - kind: UDPRoute