fix: wss related tests are unstable

internal/controller/gateway_controller.go

@@ -140,6 +140,10 @@ func (r *GatewayReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 		}
 		return ctrl.Result{}, err
 	}
+	if !r.checkGatewayClass(gateway) {
+		return ctrl.Result{}, nil
+	}
+
 	conditionProgrammedStatus, conditionProgrammedMsg := true, "Programmed"
 
 	r.Log.Info("gateway has been accepted", "gateway", gateway.GetName())

internal/controller/indexer/ssl_host.go

@@ -69,9 +69,11 @@ func GatewayTLSHostIndexFunc(rawObj client.Object) []string {
 		}
 	}
 
-	tlsHostIndexLogger.Info("GatewayTLSHostIndexFunc", "hosts", hostSetToSlice(hosts), "len", len(hostSetToSlice(hosts)))
+	hostsSlice := hostSetToSlice(hosts)
 
-	return hostSetToSlice(hosts)
+	tlsHostIndexLogger.V(1).Info("GatewayTLSHostIndexFunc", "hosts", hostsSlice)
+
+	return hostsSlice
 }
 
 // IngressTLSHostIndexFunc indexes Ingresses by their TLS SNI hosts.

test/e2e/crds/v2/route.go

@@ -2145,22 +2145,11 @@ 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 := s.NewWebsocketClient(&tls.Config{
+				InsecureSkipVerify: true,
+				ServerName:         hostname,
+			}, "/ws", http.Header{"Host": []string{hostname}})
 
 			defer func() {
 				_ = conn.Close()

test/e2e/framework/manifests/apisix.yaml

@@ -66,15 +66,18 @@ metadata:
   name: apisix
   labels:
     app.kubernetes.io/name: apisix
+    app: apisix
 spec:
   replicas: {{ default 1 .Replicas }}
   selector:
     matchLabels:
       app.kubernetes.io/name: apisix
+      app: apisix
   template:
     metadata:
       labels:
         app.kubernetes.io/name: apisix
+        app: apisix
     spec:
       initContainers:
         - name: config-setup

test/e2e/framework/manifests/nginx.yaml

@@ -124,13 +124,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,36 +2503,21 @@ 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 := s.NewWebsocketClient(&tls.Config{
+				InsecureSkipVerify: true,
+				ServerName:         hostname,
+			}, "/ws", http.Header{"Host": []string{hostname}})
 
 			defer func() {
 				_ = conn.Close()
 			}()
 
 			By("send and receive message through WebSocket")
 			testMessage := "hello, this is APISIX"
-			err = conn.WriteMessage(websocket.TextMessage, []byte(testMessage))
+			err := conn.WriteMessage(websocket.TextMessage, []byte(testMessage))
 			Expect(err).ShouldNot(HaveOccurred(), "writing WebSocket message")
 
 			// Then our echo

test/e2e/ingress/ingress.go

@@ -23,7 +23,6 @@ import (
 	"encoding/base64"
 	"fmt"
 	"net/http"
-	"net/url"
 	"strings"
 	"time"
 
@@ -944,14 +943,13 @@ spec:
 			By("create Ingress")
 			err := s.CreateResourceFromStringWithNamespace(fmt.Sprintf(ingress, s.Namespace()), s.Namespace())
 			Expect(err).NotTo(HaveOccurred(), "creating Ingress")
-			time.Sleep(5 * time.Second)
 
 			By("verify Ingress works")
-			s.NewAPISIXClient().
-				GET("/get").
-				WithHost("ingress.example.com").
-				Expect().
-				Status(200)
+			s.RequestAssert(&scaffold.RequestAssert{
+				Method: "GET",
+				Host:   "ingress.example.com",
+				Check:  scaffold.WithExpectedStatus(http.StatusOK),
+			})
 
 			By("create additional gateway group to get new admin key")
 			additionalGatewayGroupID, _, err = s.Deployer.CreateAdditionalGateway("gateway-proxy-update")
@@ -1070,33 +1068,21 @@ 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 := s.NewWebsocketClient(&tls.Config{
+				InsecureSkipVerify: true,
+				ServerName:         hostname,
+			}, "/ws", http.Header{"Host": []string{hostname}})
 
 			defer func() {
 				_ = conn.Close()
 			}()
 
 			By("send and receive message through WebSocket")
 			testMessage := "hello, this is APISIX"
-			err = conn.WriteMessage(websocket.TextMessage, []byte(testMessage))
+			err := conn.WriteMessage(websocket.TextMessage, []byte(testMessage))
 			Expect(err).ShouldNot(HaveOccurred(), "writing WebSocket message")
 
 			// Then our echo

test/e2e/scaffold/apisix_deployer.go

@@ -111,6 +111,19 @@ func (s *APISIXDeployer) AfterEach() {
 		if output != "" {
 			_, _ = fmt.Fprintln(GinkgoWriter, output)
 		}
+
+		_, _ = fmt.Fprintln(GinkgoWriter, "Dumping APISIX logs:")
+		output = s.GetDeploymentLogs("apisix")
+		if output != "" {
+			_, _ = fmt.Fprintln(GinkgoWriter, output)
+		}
+		if framework.ProviderType == framework.ProviderTypeAPISIXStandalone && s.adminTunnel != nil {
+			client := NewClient("http", s.adminTunnel.Endpoint())
+			reporter := &ErrorReporter{}
+			body := client.GET("/apisix/admin/configs").WithHeader("X-API-KEY", s.AdminKey()).WithReporter(reporter).Expect().Body().Raw()
+			_, _ = fmt.Fprintln(GinkgoWriter, "Dumping APISIX configs:")
+			_, _ = fmt.Fprintln(GinkgoWriter, body)
+		}
 	}
 
 	// Delete all additional gateways

test/e2e/scaffold/scaffold.go

@@ -30,6 +30,7 @@ import (
 
 	"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
@@ -562,6 +563,40 @@ func (s *Scaffold) GetMetricsEndpoint() string {
 	return fmt.Sprintf("http://%s/metrics", tunnel.Endpoint())
 }
 
+func (s *Scaffold) NewWebsocketClient(tls *tls.Config, path string, headers http.Header) *websocket.Conn {
+	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
+
+	s.RetryAssertion(func() error {
+		c, resp, err := dialer.Dial(u.String(), headers)
+		if err != nil {
+			return err
+		}
+		if resp == nil || resp.StatusCode != http.StatusSwitchingProtocols {
+			_ = c.Close()
+			return fmt.Errorf("unexpected status code: %d", resp.StatusCode)
+		}
+		conn = c
+		return nil
+	}).ShouldNot(HaveOccurred(), "establishing websocket connection")
+	return conn
+}
+
 func (s *Scaffold) ControlAPIClient() (ControlAPIClient, error) {
 	tunnel := k8s.NewTunnel(s.kubectlOptions, k8s.ResourceTypeService, "apisix-control-api", 9090, 9090)
 	if err := tunnel.ForwardPortE(s.t); err != nil {