diff --git a/internal/adc/translator/l4route_serverport_test.go b/internal/adc/translator/l4route_serverport_test.go new file mode 100644 index 00000000..28415646 --- /dev/null +++ b/internal/adc/translator/l4route_serverport_test.go @@ -0,0 +1,221 @@ +// 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/controller/config" + "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: port, + } +} + +func udpListener(name string, port int32) gatewayv1.Listener { + return gatewayv1.Listener{ + Name: gatewayv1.SectionName(name), + Protocol: gatewayv1.UDPProtocolType, + Port: 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) { + // listener_port_match_mode defaults to off; these cases assert the + // injection behavior, so exercise the translator in auto mode. + translator := NewTranslator(logr.Discard(), config.ListenerPortMatchModeAuto) + 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) { + // listener_port_match_mode defaults to off; these cases assert the + // injection behavior, so exercise the translator in auto mode. + translator := NewTranslator(logr.Discard(), config.ListenerPortMatchModeAuto) + 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 81e12a34..32f0d95d 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,68 @@ func newDefaultUpstreamWithoutScheme() *adctypes.Upstream { } } +// 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[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. +// +// Whether to inject server_port is gated by shouldInjectServerPortVars (the same +// listener_port_match_mode used by HTTPRoute/GRPCRoute): 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 (explicit sectionName/port targeting, +// or more than one listener port). When it is not injected we keep the previous +// 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); t.shouldInjectServerPortVars(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 +213,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 cc7b6361..3fdbd817 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/tcproute_controller.go b/internal/controller/tcproute_controller.go index a2360157..d1380ac8 100644 --- a/internal/controller/tcproute_controller.go +++ b/internal/controller/tcproute_controller.go @@ -293,6 +293,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 17fb5f6a..f5eebbc5 100644 --- a/internal/controller/udproute_controller.go +++ b/internal/controller/udproute_controller.go @@ -293,6 +293,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