add test

Author: AlinsRan

internal/adc/translator/apisixroute.go

@@ -368,7 +368,9 @@ func (t *Translator) translateApisixRouteHTTPBackend(tctx *provider.TranslateCon
 		return nil, err
 	}
 	upstream.Nodes = nodes
-	upstream.Scheme = appProtocolToUpstreamScheme(protocol)
+	if upstream.Scheme == "" {
+		upstream.Scheme = appProtocolToUpstreamScheme(protocol)
+	}
 	if protocol == internaltypes.AppProtocolWS || protocol == internaltypes.AppProtocolWSS {
 		*enableWebsocket = ptr.To(true)
 	}

internal/adc/translator/ingress.go

@@ -25,6 +25,7 @@ import (
 	discoveryv1 "k8s.io/api/discovery/v1"
 	networkingv1 "k8s.io/api/networking/v1"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/utils/ptr"
 
 	adctypes "github.com/apache/apisix-ingress-controller/api/adc"
 	"github.com/apache/apisix-ingress-controller/internal/controller/label"
@@ -111,6 +112,7 @@ func (t *Translator) TranslateIngress(tctx *provider.TranslateContext, obj *netw
 			if path.Backend.Service == nil {
 				continue
 			}
+			var enableWebsocket *bool
 
 			service := adctypes.NewDefaultService()
 			service.Labels = labels
@@ -169,6 +171,15 @@ func (t *Translator) TranslateIngress(tctx *provider.TranslateContext, obj *netw
 						break
 					}
 				}
+				if getServicePort != nil && getServicePort.AppProtocol != nil {
+					if upstream.Scheme == "" {
+						upstream.Scheme = appProtocolToUpstreamScheme(*getServicePort.AppProtocol)
+					}
+					if *getServicePort.AppProtocol == internaltypes.AppProtocolWS ||
+						*getServicePort.AppProtocol == internaltypes.AppProtocolWSS {
+						enableWebsocket = ptr.To(true)
+					}
+				}
 				endpointSlices := tctx.EndpointSlices[types.NamespacedName{
 					Namespace: obj.Namespace,
 					Name:      backendService.Name,
@@ -185,6 +196,7 @@ func (t *Translator) TranslateIngress(tctx *provider.TranslateContext, obj *netw
 			route := adctypes.NewDefaultRoute()
 			route.Name = adctypes.ComposeRouteName(obj.Namespace, obj.Name, fmt.Sprintf("%d-%d", i, j))
 			route.ID = id.GenID(route.Name)
+			route.EnableWebsocket = enableWebsocket
 			route.Labels = labels
 
 			uris := []string{path.Path}

test/e2e/framework/manifests/nginx.yaml

@@ -42,6 +42,39 @@ data:
         location / {
           return 200 'Hello, World!';
         }
+
+        location /ws {
+          content_by_lua_block {
+            local server = require "resty.websocket.server"
+            local wb, err = server:new {
+              timeout = 5000,          -- 5 seconds timeout
+              max_payload_len = 65535  -- max message length
+            }
+            if not wb then
+              ngx.log(ngx.ERR, "failed to create websocket: ", err)
+              return ngx.exit(444)
+            end
+
+            while true do
+              local data, typ, err = wb:recv_frame()
+              if wb.fatal then
+                ngx.log(ngx.ERR, "failed to receive frame: ", err)
+                break
+              end
+
+              if typ == "close" then
+                wb:send_close()
+                break
+              elseif typ == "text" then
+                wb:send_text(data)   -- echo text
+              elseif typ == "binary" then
+                wb:send_binary(data) -- echo binary
+              elseif typ == "ping" then
+                wb:send_pong()
+              end
+            end
+          }
+        }
       }
     }
 
@@ -77,7 +110,7 @@ spec:
               path: /healthz
               port: 80
             timeoutSeconds: 2
-          image: "nginx:1.21.4"
+          image: "openresty/openresty:1.27.1.2-4-bullseye-fat"
           imagePullPolicy: IfNotPresent
           name: nginx
           ports:
@@ -88,7 +121,7 @@ spec:
               name: "https"
               protocol: "TCP"
           volumeMounts:
-            - mountPath: /etc/nginx/nginx.conf
+            - mountPath: /usr/local/openresty/nginx/conf/nginx.conf
               name: nginx-config
               subPath: nginx.conf
             - mountPath: /etc/nginx/ssl
@@ -111,6 +144,16 @@ spec:
       protocol: TCP
       targetPort: 443
       appProtocol: https
+    - name: ws
+      port: 8080
+      protocol: TCP
+      targetPort: 80
+      appProtocol: kubernetes.io/ws
+    - name: wss
+      port: 8443
+      protocol: TCP
+      targetPort: 443
+      appProtocol: kubernetes.io/wss
   type: ClusterIP
 ---
 apiVersion: v1

test/e2e/gatewayapi/httproute.go

@@ -19,11 +19,14 @@ package gatewayapi
 
 import (
 	"context"
+	"crypto/tls"
 	"fmt"
 	"net/http"
+	"net/url"
 	"strings"
 	"time"
 
+	"github.com/gorilla/websocket"
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 	"github.com/pkg/errors"
@@ -2440,17 +2443,18 @@ spec:
 
 	})
 
-	Context("Test AppProtocol", func() {
-		var httproute = `
+	Context("Test Service With AppProtocol", func() {
+		var (
+			httproute = `
 apiVersion: gateway.networking.k8s.io/v1
 kind: HTTPRoute
 metadata:
-  name: httpbin
+  name: nginx
 spec:
   parentRefs:
   - name: %s
   hostnames:
-  - httpbin.example
+  - api6.com
   rules:
   - matches:
     - path:
@@ -2460,22 +2464,82 @@ spec:
     - name: nginx
       port: 443
  `
+			httprouteWithWSS = `
+apiVersion: gateway.networking.k8s.io/v1
+kind: HTTPRoute
+metadata:
+  name: nginx-wss
+spec:
+  parentRefs:
+  - name: %s
+  hostnames:
+  - api6.com
+  rules:
+  - matches:
+    - path:
+        type: Exact
+        value: /ws
+    backendRefs:
+    - name: nginx
+      port: 8443
+ `
+		)
 
 		BeforeEach(func() {
-			beforeEachHTTP()
+			beforeEachHTTPS()
 			s.DeployNginx(framework.NginxOptions{
 				Namespace: s.Namespace(),
 				Replicas:  ptr.To(int32(1)),
 			})
 		})
 		It("HTTPS backend", func() {
-			s.ResourceApplied("HTTPRoute", "httpbin", fmt.Sprintf(httproute, s.Namespace()), 1)
+			s.ResourceApplied("HTTPRoute", "nginx", fmt.Sprintf(httproute, s.Namespace()), 1)
 			s.RequestAssert(&scaffold.RequestAssert{
 				Method: "GET",
 				Path:   "/get",
-				Host:   "httpbin.example",
+				Host:   "api6.com",
 				Check:  scaffold.WithExpectedStatus(http.StatusOK),
 			})
 		})
+
+		It("WWS backend", func() {
+			s.ResourceApplied("HTTPRoute", "nginx-wss", fmt.Sprintf(httprouteWithWSS, s.Namespace()), 1)
+			time.Sleep(6 * time.Second)
+
+			By("verify wss connection")
+			u := url.URL{
+				Scheme: "wss",
+				Host:   s.GetAPISIXHTTPSEndpoint(),
+				Path:   "/ws",
+			}
+			headers := http.Header{"Host": []string{"api6.com"}}
+
+			hostname := "api6.com"
+
+			dialer := websocket.Dialer{
+				TLSClientConfig: &tls.Config{
+					InsecureSkipVerify: true,
+					ServerName:         hostname,
+				},
+			}
+
+			conn, resp, err := dialer.Dial(u.String(), headers)
+			Expect(err).ShouldNot(HaveOccurred(), "WebSocket handshake")
+			Expect(resp.StatusCode).Should(Equal(http.StatusSwitchingProtocols))
+
+			defer func() {
+				_ = conn.Close()
+			}()
+
+			By("send and receive message through WebSocket")
+			testMessage := "hello, this is APISIX"
+			err = conn.WriteMessage(websocket.TextMessage, []byte(testMessage))
+			Expect(err).ShouldNot(HaveOccurred(), "writing WebSocket message")
+
+			// Then our echo
+			_, msg, err := conn.ReadMessage()
+			Expect(err).ShouldNot(HaveOccurred(), "reading echo message")
+			Expect(string(msg)).To(Equal(testMessage), "message content verification")
+		})
 	})
 })