fix: honor TCPRoute/UDPRoute sectionName and listener port for StreamRoute matching

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))
+		})
+	}
+}

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)
 	}

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)
 	}

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
 }

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

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