fix(test): wss related tests are unstable

test/e2e/crds/v2/route.go

@@ -2145,22 +2145,12 @@ spec:
 			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"}}
-			dialer := websocket.Dialer{
-				TLSClientConfig: &tls.Config{
-					InsecureSkipVerify: true,
-					ServerName:         "api6.com",
-				},
-			}
-
-			conn, resp, err := dialer.Dial(u.String(), headers)
-			Expect(err).ShouldNot(HaveOccurred(), "WebSocket handshake")
-			Expect(resp.StatusCode).Should(Equal(http.StatusSwitchingProtocols))
+			hostname := "api6.com"
+			conn, err := s.NewWebsocketClient(&tls.Config{
+				InsecureSkipVerify: true,
+				ServerName:         hostname,
+			}, "/ws", http.Header{"Host": []string{hostname}})
+			Expect(err).ShouldNot(HaveOccurred(), "creating WebSocket client")
 
 			defer func() {
 				_ = conn.Close()

test/e2e/framework/manifests/nginx.yaml

@@ -111,13 +111,19 @@ spec:
       containers:
         - livenessProbe:
             failureThreshold: 3
-            initialDelaySeconds: 1
-            periodSeconds: 5
+            initialDelaySeconds: 10
+            periodSeconds: 15
             successThreshold: 1
             httpGet:
               path: /healthz
               port: 80
             timeoutSeconds: 2
+          readinessProbe:
+            httpGet:
+              path: /healthz
+              port: 80
+            initialDelaySeconds: 5
+            periodSeconds: 5
           image: "openresty/openresty:1.27.1.2-4-bullseye-fat"
           imagePullPolicy: IfNotPresent
           name: nginx

test/e2e/gatewayapi/httproute.go

@@ -22,7 +22,6 @@ import (
 	"crypto/tls"
 	"fmt"
 	"net/http"
-	"net/url"
 	"strings"
 	"time"
 
@@ -2504,28 +2503,14 @@ spec:
 
 		It("WSS 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))
+			conn, err := s.NewWebsocketClient(&tls.Config{
+				InsecureSkipVerify: true,
+				ServerName:         hostname,
+			}, "/ws", http.Header{"Host": []string{hostname}})
+			Expect(err).ShouldNot(HaveOccurred(), "creating WebSocket client")
 
 			defer func() {
 				_ = conn.Close()

test/e2e/ingress/ingress.go

@@ -23,7 +23,6 @@ import (
 	"encoding/base64"
 	"fmt"
 	"net/http"
-	"net/url"
 	"strings"
 	"time"
 
@@ -1070,25 +1069,14 @@ spec:
 			createSecret(s, _secretName)
 			By("create Ingress")
 			Expect(s.CreateResourceFromString(fmt.Sprintf(ingressWithWSS, s.Namespace()))).ShouldNot(HaveOccurred(), "creating Ingress")
-			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"}}
-			dialer := websocket.Dialer{
-				TLSClientConfig: &tls.Config{
-					InsecureSkipVerify: true,
-					ServerName:         "api6.com",
-				},
-			}
-
-			conn, resp, err := dialer.Dial(u.String(), headers)
-			Expect(err).ShouldNot(HaveOccurred(), "WebSocket handshake")
-			Expect(resp.StatusCode).Should(Equal(http.StatusSwitchingProtocols))
+			hostname := "api6.com"
+			conn, err := s.NewWebsocketClient(&tls.Config{
+				InsecureSkipVerify: true,
+				ServerName:         hostname,
+			}, "/ws", http.Header{"Host": []string{hostname}})
+			Expect(err).ShouldNot(HaveOccurred(), "creating WebSocket client")
 
 			defer func() {
 				_ = conn.Close()

test/e2e/scaffold/scaffold.go

@@ -25,15 +25,18 @@ import (
 	"net/url"
 	"os"
 	"strings"
+	"time"
 
 	"github.com/api7/gopkg/pkg/log"
 	"github.com/gavv/httpexpect/v2"
+	"github.com/gorilla/websocket"
 	"github.com/gruntwork-io/terratest/modules/k8s"
 	"github.com/gruntwork-io/terratest/modules/testing"
 	. "github.com/onsi/ginkgo/v2" //nolint:staticcheck
 	. "github.com/onsi/gomega"    //nolint:staticcheck
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/wait"
 
 	apiv2 "github.com/apache/apisix-ingress-controller/api/v2"
 	"github.com/apache/apisix-ingress-controller/test/e2e/framework"
@@ -553,3 +556,42 @@ func (s *Scaffold) GetMetricsEndpoint() string {
 	s.addFinalizers(tunnel.Close)
 	return fmt.Sprintf("http://%s/metrics", tunnel.Endpoint())
 }
+
+func (s *Scaffold) NewWebsocketClient(tls *tls.Config, path string, headers http.Header) (*websocket.Conn, error) {
+	var host = s.ApisixHTTPEndpoint()
+	var scheme = "ws"
+	if tls != nil {
+		scheme = "wss"
+		host = s.GetAPISIXHTTPSEndpoint()
+	}
+
+	dialer := websocket.Dialer{
+		TLSClientConfig: tls,
+	}
+
+	u := url.URL{
+		Scheme: scheme,
+		Host:   host,
+		Path:   path,
+	}
+	var conn *websocket.Conn
+
+	err := wait.PollUntilContextTimeout(context.Background(), 2*time.Second, 12*time.Second, true, func(ctx context.Context) (bool, error) {
+		c, resp, err := dialer.Dial(u.String(), headers)
+		if err != nil {
+			return false, err
+		}
+		if resp == nil || resp.StatusCode != http.StatusSwitchingProtocols {
+			_ = c.Close()
+			return false, fmt.Errorf("unexpected status code: %d", resp.StatusCode)
+		}
+		conn = c
+		return true, nil
+	})
+
+	if conn == nil {
+		return nil, fmt.Errorf("failed to connect websocket after retries: %v", err)
+	}
+
+	return conn, nil
+}