feat: support stream_route for ApisixRoute (#2551)

Author: AlinsRan

api/adc/types.go

@@ -195,15 +195,13 @@ type Timeout struct {
 
 // +k8s:deepcopy-gen=true
 type StreamRoute struct {
-	Description string            `json:"description,omitempty"`
-	ID          string            `json:"id,omitempty"`
-	Labels      map[string]string `json:"labels,omitempty"`
-	Name        string            `json:"name"`
-	Plugins     Plugins           `json:"plugins,omitempty"`
-	RemoteAddr  string            `json:"remote_addr,omitempty"`
-	ServerAddr  string            `json:"server_addr,omitempty"`
-	ServerPort  *int64            `json:"server_port,omitempty"`
-	Sni         string            `json:"sni,omitempty"`
+	Metadata `json:",inline" yaml:",inline"`
+
+	Plugins    Plugins `json:"plugins,omitempty"`
+	RemoteAddr string  `json:"remote_addr,omitempty"`
+	ServerAddr string  `json:"server_addr,omitempty"`
+	ServerPort int32   `json:"server_port,omitempty"`
+	SNI        string  `json:"sni,omitempty"`
 }
 
 // +k8s:deepcopy-gen=true
@@ -537,6 +535,24 @@ func ComposeRouteName(namespace, name string, rule string) string {
 	return buf.String()
 }
 
+// ComposeStreamRouteName uses namespace, name and rule name to compose
+// the stream_route name.
+func ComposeStreamRouteName(namespace, name string, rule string) string {
+	// FIXME Use sync.Pool to reuse this buffer if the upstream
+	// name composing code path is hot.
+	p := make([]byte, 0, len(namespace)+len(name)+len(rule)+6)
+	buf := bytes.NewBuffer(p)
+
+	buf.WriteString(namespace)
+	buf.WriteByte('_')
+	buf.WriteString(name)
+	buf.WriteByte('_')
+	buf.WriteString(rule)
+	buf.WriteString("_tcp")
+
+	return buf.String()
+}
+
 func ComposeServiceNameWithRule(namespace, name string, rule string) string {
 	// FIXME Use sync.Pool to reuse this buffer if the upstream
 	// name composing code path is hot.
@@ -554,6 +570,24 @@ func ComposeServiceNameWithRule(namespace, name string, rule string) string {
 	return buf.String()
 }
 
+func ComposeServiceNameWithStream(namespace, name string, rule string) string {
+	// FIXME Use sync.Pool to reuse this buffer if the upstream
+	// name composing code path is hot.
+	var p []byte
+	plen := len(namespace) + len(name) + 6
+
+	p = make([]byte, 0, plen)
+	buf := bytes.NewBuffer(p)
+	buf.WriteString(namespace)
+	buf.WriteByte('_')
+	buf.WriteString(name)
+	buf.WriteByte('_')
+	buf.WriteString(rule)
+	buf.WriteString("_stream")
+
+	return buf.String()
+}
+
 func ComposeConsumerName(namespace, name string) string {
 	// FIXME Use sync.Pool to reuse this buffer if the upstream
 	// name composing code path is hot.
@@ -604,6 +638,18 @@ func NewDefaultRoute() *Route {
 	}
 }
 
+// NewDefaultStreamRoute returns an empty StreamRoute with default values.
+func NewDefaultStreamRoute() *StreamRoute {
+	return &StreamRoute{
+		Metadata: Metadata{
+			Desc: "Created by apisix-ingress-controller, DO NOT modify it manually",
+			Labels: map[string]string{
+				"managed-by": "apisix-ingress-controller",
+			},
+		},
+	}
+}
+
 const (
 	PluginProxyRewrite    string = "proxy-rewrite"
 	PluginRedirect        string = "redirect"

api/adc/zz_generated.deepcopy.go

@@ -39,7 +39,6 @@ type ApisixRouteSpec struct {
 	HTTP []ApisixRouteHTTP `json:"http,omitempty" yaml:"http,omitempty"`
 	// Stream defines a list of stream route rules.
 	// Each rule specifies conditions to match TCP/UDP traffic and how to forward them.
-	// Stream is currently not supported.
 	Stream []ApisixRouteStream `json:"stream,omitempty" yaml:"stream,omitempty"`
 }
 
@@ -111,7 +110,9 @@ type ApisixRouteHTTP struct {
 type ApisixRouteStream struct {
 	// Name is a unique identifier for the route. This field must not be empty.
 	Name string `json:"name" yaml:"name"`
-	// Protocol specifies the L4 protocol to match. Can be `tcp` or `udp`.
+	// Protocol specifies the L4 protocol to match. Can be `TCP` or `UDP`.
+	//
+	// +kubebuilder:validation:Enum=TCP;UDP
 	Protocol string `json:"protocol" yaml:"protocol"`
 	// Match defines the criteria used to match incoming TCP or UDP connections.
 	Match ApisixRouteStreamMatch `json:"match" yaml:"match"`
@@ -225,6 +226,9 @@ type ApisixRouteAuthentication struct {
 type ApisixRouteStreamMatch struct {
 	// IngressPort is the port on which the APISIX Ingress proxy server listens.
 	// This must be a statically configured port, as APISIX does not support dynamic port binding.
+	//
+	// +kubebuilder:validation:Minimum=0
+	// +kubebuilder:validation:Maximum=65535
 	IngressPort int32 `json:"ingressPort" yaml:"ingressPort"`
 	// Host is the destination host address used to match the incoming TCP/UDP traffic.
 	Host string `json:"host,omitempty" yaml:"host,omitempty"`
@@ -240,9 +244,12 @@ type ApisixRouteStreamBackend struct {
 	// This can be either the port name or port number.
 	ServicePort intstr.IntOrString `json:"servicePort" yaml:"servicePort"`
 	// ResolveGranularity determines how the backend service is resolved.
-	// Valid values are `endpoints` and `service`. When set to `endpoints`,
+	// Valid values are `endpoint` and `service`. When set to `endpoint`,
 	// individual pod IPs will be used; otherwise, the Service's ClusterIP or ExternalIP is used.
-	// The default is `endpoints`.
+	// The default is `endpoint`.
+	//
+	// +kubebuilder:default=endpoint
+	// +kubebuilder:validation:Enum=endpoint;service
 	ResolveGranularity string `json:"resolveGranularity,omitempty" yaml:"resolveGranularity,omitempty"`
 	// Subset specifies a named subset of the target Service.
 	// The subset must be pre-defined in the corresponding ApisixUpstream resource.

api/v2/apisixroute_types.go

@@ -45,6 +45,11 @@ const (
 	DefaultWeight = 100
 )
 
+const (
+	ResolveGranularityService  = "service"
+	ResolveGranularityEndpoint = "endpoint"
+)
+
 const (
 	// OpEqual means the equal ("==") operator in nginxVars.
 	OpEqual = "Equal"

api/v2/shared_types.go

@@ -355,7 +355,6 @@ spec:
                 description: |-
                   Stream defines a list of stream route rules.
                   Each rule specifies conditions to match TCP/UDP traffic and how to forward them.
-                  Stream is currently not supported.
                 items:
                   description: ApisixRouteStream defines the configuration for a Layer
                     4 (TCP/UDP) route. Currently not supported.
@@ -365,11 +364,15 @@ spec:
                         traffic should be forwarded.
                       properties:
                         resolveGranularity:
+                          default: endpoint
                           description: |-
                             ResolveGranularity determines how the backend service is resolved.
-                            Valid values are `endpoints` and `service`. When set to `endpoints`,
+                            Valid values are `endpoint` and `service`. When set to `endpoint`,
                             individual pod IPs will be used; otherwise, the Service's ClusterIP or ExternalIP is used.
-                            The default is `endpoints`.
+                            The default is `endpoint`.
+                          enum:
+                          - endpoint
+                          - service
                           type: string
                         serviceName:
                           description: |-
@@ -407,6 +410,8 @@ spec:
                             IngressPort is the port on which the APISIX Ingress proxy server listens.
                             This must be a statically configured port, as APISIX does not support dynamic port binding.
                           format: int32
+                          maximum: 65535
+                          minimum: 0
                           type: integer
                       required:
                       - ingressPort
@@ -442,7 +447,10 @@ spec:
                       type: array
                     protocol:
                       description: Protocol specifies the L4 protocol to match. Can
-                        be `tcp` or `udp`.
+                        be `TCP` or `UDP`.
+                      enum:
+                      - TCP
+                      - UDP
                       type: string
                   required:
                   - backend

config/crd/bases/apisix.apache.org_apisixroutes.yaml

@@ -1178,7 +1178,7 @@ It defines routing rules for both HTTP and stream traffic.
 | --- | --- |
 | `ingressClassName` _string_ | IngressClassName is the name of the IngressClass this route belongs to. It allows multiple controllers to watch and reconcile different routes. |
 | `http` _[ApisixRouteHTTP](#apisixroutehttp) array_ | HTTP defines a list of HTTP route rules. Each rule specifies conditions to match HTTP requests and how to forward them. |
-| `stream` _[ApisixRouteStream](#apisixroutestream) array_ | Stream defines a list of stream route rules. Each rule specifies conditions to match TCP/UDP traffic and how to forward them. Stream is currently not supported. |
+| `stream` _[ApisixRouteStream](#apisixroutestream) array_ | Stream defines a list of stream route rules. Each rule specifies conditions to match TCP/UDP traffic and how to forward them. |
 
 
 _Appears in:_
@@ -1194,7 +1194,7 @@ ApisixRouteStream defines the configuration for a Layer 4 (TCP/UDP) route. Curre
 | Field | Description |
 | --- | --- |
 | `name` _string_ | Name is a unique identifier for the route. This field must not be empty. |
-| `protocol` _string_ | Protocol specifies the L4 protocol to match. Can be `tcp` or `udp`. |
+| `protocol` _string_ | Protocol specifies the L4 protocol to match. Can be `TCP` or `UDP`. |
 | `match` _[ApisixRouteStreamMatch](#apisixroutestreammatch)_ | Match defines the criteria used to match incoming TCP or UDP connections. |
 | `backend` _[ApisixRouteStreamBackend](#apisixroutestreambackend)_ | Backend specifies the destination service to which traffic should be forwarded. |
 | `plugins` _[ApisixRoutePlugin](#apisixrouteplugin) array_ | Plugins defines a list of plugins to apply to this route. |
@@ -1214,7 +1214,7 @@ ApisixRouteStreamBackend represents the backend service for a TCP or UDP stream
 | --- | --- |
 | `serviceName` _string_ | ServiceName is the name of the Kubernetes Service. Cross-namespace references are not supported—ensure the ApisixRoute and the Service are in the same namespace. |
 | `servicePort` _[IntOrString](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.30/#intorstring-intstr-util)_ | ServicePort is the port of the Kubernetes Service. This can be either the port name or port number. |
-| `resolveGranularity` _string_ | ResolveGranularity determines how the backend service is resolved. Valid values are `endpoints` and `service`. When set to `endpoints`, individual pod IPs will be used; otherwise, the Service's ClusterIP or ExternalIP is used. The default is `endpoints`. |
+| `resolveGranularity` _string_ | ResolveGranularity determines how the backend service is resolved. Valid values are `endpoint` and `service`. When set to `endpoint`, individual pod IPs will be used; otherwise, the Service's ClusterIP or ExternalIP is used. The default is `endpoint`. |
 | `subset` _string_ | Subset specifies a named subset of the target Service. The subset must be pre-defined in the corresponding ApisixUpstream resource. |
 
 

docs/en/latest/reference/api-reference.md

@@ -7,12 +7,13 @@ toolchain go1.24.7
 require (
 	github.com/Masterminds/sprig/v3 v3.2.3
 	github.com/api7/gopkg v0.2.1-0.20230601092738-0f3730f9b57a
+	github.com/eclipse/paho.mqtt.golang v1.5.0
 	github.com/gavv/httpexpect/v2 v2.16.0
 	github.com/go-logr/logr v1.4.2
 	github.com/go-logr/zapr v1.3.0
 	github.com/google/go-cmp v0.7.0
 	github.com/google/uuid v1.6.0
-	github.com/gorilla/websocket v1.5.1
+	github.com/gorilla/websocket v1.5.3
 	github.com/gruntwork-io/terratest v0.50.0
 	github.com/hashicorp/go-memdb v1.3.4
 	github.com/incubator4/go-resty-expr v0.1.1

go.mod

@@ -121,6 +121,8 @@ github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSs
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM=
 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/eclipse/paho.mqtt.golang v1.5.0 h1:EH+bUVJNgttidWFkLLVKaQPGmkTUfQQqjOsyvMGvD6o=
+github.com/eclipse/paho.mqtt.golang v1.5.0/go.mod h1:du/2qNQVqJf/Sqs4MEL77kR8QTqANF7XU7Fk0aOTAgk=
 github.com/emicklei/go-restful/v3 v3.12.0 h1:y2DdzBAURM29NFF94q6RaY4vjIH1rtwDapwQtU84iWk=
 github.com/emicklei/go-restful/v3 v3.12.0/go.mod h1:6n3XBCmQQb25CM2LCACGz8ukIrRry+4bhvbpWn3mrbc=
 github.com/evanphx/json-patch v5.9.0+incompatible h1:fBXyNpNMuTTDdquAq/uisOr2lShz4oaXpDTX2bLe7ls=
@@ -198,8 +200,8 @@ github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+
 github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
 github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1/go.mod h1:wJfORRmW1u3UXTncJ5qlYoELFm8eSnnEO6hX4iZ3EWY=
-github.com/gorilla/websocket v1.5.1 h1:gmztn0JnHVt9JZquRuzLw3g4wouNVzKL15iLr/zn/QY=
-github.com/gorilla/websocket v1.5.1/go.mod h1:x3kM2JMyaluk02fnUJpQuwD2dCS5NDG2ZHL0uE0tcaY=
+github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
+github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/grpc-ecosystem/grpc-gateway/v2 v2.20.0 h1:bkypFPDjIYGfCYD5mRBvpqxfYX1YCS1PXdKYWi8FsN0=
 github.com/grpc-ecosystem/grpc-gateway/v2 v2.20.0/go.mod h1:P+Lt/0by1T8bfcF3z737NnSbmxQAppXMRziHUxPOC8k=
 github.com/gruntwork-io/go-commons v0.8.0 h1:k/yypwrPqSeYHevLlEDmvmgQzcyTwrlZGRaxEM6G0ro=

go.sum

@@ -50,6 +50,14 @@ func (t *Translator) TranslateApisixRoute(tctx *provider.TranslateContext, ar *a
 		}
 		result.Services = append(result.Services, service)
 	}
+
+	for _, part := range ar.Spec.Stream {
+		service, err := t.translateStreamRule(tctx, ar, part)
+		if err != nil {
+			return nil, err
+		}
+		result.Services = append(result.Services, service)
+	}
 	return result, nil
 }
 
@@ -88,7 +96,7 @@ func (t *Translator) buildPlugins(tctx *provider.TranslateContext, ar *apiv2.Api
 	t.loadPluginConfigPlugins(tctx, ar, rule, plugins)
 
 	// Apply plugins from the route itself
-	t.loadRoutePlugins(tctx, ar, rule, plugins)
+	t.loadRoutePlugins(tctx, ar, rule.Plugins, plugins)
 
 	// Add authentication plugins
 	t.addAuthenticationPlugins(rule, plugins)
@@ -121,8 +129,8 @@ func (t *Translator) loadPluginConfigPlugins(tctx *provider.TranslateContext, ar
 	}
 }
 
-func (t *Translator) loadRoutePlugins(tctx *provider.TranslateContext, ar *apiv2.ApisixRoute, rule apiv2.ApisixRouteHTTP, plugins adc.Plugins) {
-	for _, plugin := range rule.Plugins {
+func (t *Translator) loadRoutePlugins(tctx *provider.TranslateContext, ar *apiv2.ApisixRoute, routePlugins []apiv2.ApisixRoutePlugin, plugins adc.Plugins) {
+	for _, plugin := range routePlugins {
 		if !plugin.Enable {
 			continue
 		}
@@ -210,7 +218,7 @@ func (t *Translator) buildUpstream(tctx *provider.TranslateContext, service *adc
 			upstream, _ = t.translateApisixUpstream(tctx, au)
 		}
 
-		if backend.ResolveGranularity == "service" {
+		if backend.ResolveGranularity == apiv2.ResolveGranularityService {
 			upstream.Nodes, backendErr = t.translateApisixRouteBackendResolveGranularityService(tctx, utils.NamespacedName(ar), backend)
 			if backendErr != nil {
 				t.Log.Error(backendErr, "failed to translate ApisixRoute backend with ResolveGranularity Service")
@@ -341,6 +349,20 @@ func (t *Translator) translateApisixRouteBackendResolveGranularityService(tctx *
 	}, nil
 }
 
+func (t *Translator) translateApisixRouteStreamBackendResolveGranularity(tctx *provider.TranslateContext, arNN types.NamespacedName, backend apiv2.ApisixRouteStreamBackend) (adc.UpstreamNodes, error) {
+	tsBackend := apiv2.ApisixRouteHTTPBackend{
+		ServiceName:        backend.ServiceName,
+		ServicePort:        backend.ServicePort,
+		ResolveGranularity: backend.ResolveGranularity,
+		Subset:             backend.Subset,
+	}
+	if backend.ResolveGranularity == apiv2.ResolveGranularityService {
+		return t.translateApisixRouteBackendResolveGranularityService(tctx, arNN, tsBackend)
+	} else {
+		return t.translateApisixRouteBackendResolveGranularityEndpoint(tctx, arNN, tsBackend)
+	}
+}
+
 func (t *Translator) translateApisixRouteBackendResolveGranularityEndpoint(tctx *provider.TranslateContext, arNN types.NamespacedName, backend apiv2.ApisixRouteHTTPBackend) (adc.UpstreamNodes, error) {
 	weight := int32(*cmp.Or(backend.Weight, ptr.To(apiv2.DefaultWeight)))
 	backendRef := gatewayv1.BackendRef{
@@ -355,3 +377,37 @@ func (t *Translator) translateApisixRouteBackendResolveGranularityEndpoint(tctx
 	}
 	return t.translateBackendRef(tctx, backendRef, DefaultEndpointFilter)
 }
+
+func (t *Translator) translateStreamRule(tctx *provider.TranslateContext, ar *apiv2.ApisixRoute, part apiv2.ApisixRouteStream) (*adc.Service, error) {
+	// add stream route plugins
+	plugins := make(adc.Plugins)
+	t.loadRoutePlugins(tctx, ar, part.Plugins, plugins)
+
+	sr := adc.NewDefaultStreamRoute()
+	sr.Name = adc.ComposeStreamRouteName(ar.Namespace, ar.Name, part.Name)
+	sr.ID = id.GenID(sr.Name)
+	sr.ServerPort = part.Match.IngressPort
+	sr.SNI = part.Match.Host
+	sr.Plugins = plugins
+
+	svc := adc.NewDefaultService()
+	svc.Name = adc.ComposeServiceNameWithStream(ar.Namespace, ar.Name, part.Name)
+	svc.ID = id.GenID(svc.Name)
+	svc.StreamRoutes = append(svc.StreamRoutes, sr)
+
+	auNN := types.NamespacedName{Namespace: ar.GetNamespace(), Name: part.Backend.ServiceName}
+	upstream := adc.NewDefaultUpstream()
+	if au, ok := tctx.Upstreams[auNN]; ok {
+		upstream, _ = t.translateApisixUpstream(tctx, au)
+	}
+	nodes, err := t.translateApisixRouteStreamBackendResolveGranularity(tctx, utils.NamespacedName(ar), part.Backend)
+	if err != nil {
+		return nil, err
+	}
+	upstream.Nodes = nodes
+	upstream.ID = ""
+	upstream.Name = ""
+
+	svc.Upstream = upstream
+	return svc, nil
+}