feat: TestLongConnL4Telemetry e2e test

Signed-off-by: Yash Patel <yp969803@gmail.com>

Author: yp969803

test/e2e/baseline_test.go

@@ -1,4 +1,3 @@
-
 /*
  * Copyright The Kmesh Authors.
  *
@@ -26,6 +25,8 @@ import (
 	"fmt"
 	"net/http"
 	"net/netip"
+	"os"
+	"path/filepath"
 	"sort"
 	"strings"
 	"testing"
@@ -817,6 +818,187 @@ func TestL4Telemetry(t *testing.T) {
 	})
 }
 
+func TestLongConnL4Telemetry(t *testing.T) {
+	framework.NewTest(t).Run(func(t framework.TestContext) {
+
+		namespace := apps.Namespace.Name()
+		waypoint := "namespace-waypoint"
+
+		newWaypointProxyOrFail(t, t, apps.Namespace, waypoint, constants.ServiceTraffic)
+		t.Cleanup(func() {
+			deleteWaypointProxyOrFail(t, t, apps.Namespace, waypoint)
+		})
+
+		SetWaypoint(t, namespace, "", waypoint, Namespace)
+		t.Cleanup(func() {
+			UnsetWaypoint(t, namespace, "", Namespace)
+		})
+
+		// deploying websocket server and client for establishing long connection
+		serverPodYAML := `
+apiVersion: v1
+kind: Pod
+metadata:
+  name: ws-server
+  labels:
+    app: ws-server
+spec:
+  containers:
+  - name: ws-server
+    image: node:18
+    command: ["/bin/sh", "-c"]
+    args:
+      - |
+        mkdir server 
+        cd server 
+        npm init -y
+        npm install ws
+        cat << EOF > server.js
+        const WebSocket = require('ws');
+        const server = new WebSocket.Server({ port: 8080 });
+        server.on('listening', () => {
+          console.log('🚀 WebSocket server started on ws://localhost:8080');
+        });
+
+        server.on('connection', socket => {
+          console.log('Client connected');
+          
+          setInterval(() => {
+            socket.send('Hello world');
+            console.log('Sent message to client');
+          }, 1000);
+
+          socket.on('message', message => {
+            console.log('Received');
+          });
+
+          socket.on('close', () => {
+            console.log('Client disconnected');
+          });
+        });
+        EOF
+        node server.js
+    ports:
+    - containerPort: 8080
+`
+
+		tmpFile := filepath.Join(t.TempDir(), "ws-server.yaml")
+		if err := os.WriteFile(tmpFile, []byte(serverPodYAML), 0644); err != nil {
+			t.Fatalf("failed to write WebSocket YAML: %v", err)
+		}
+
+		if _, err := shell.Execute(true,
+			fmt.Sprintf("kubectl apply -f %s -n %s", tmpFile, namespace),
+		); err != nil {
+			t.Fatalf("failed to deploy ws-server: %v", err)
+		}
+		t.Cleanup(func() {
+			if _, err := shell.Execute(true, fmt.Sprintf("kubectl delete -f %s -n %s", tmpFile, namespace)); err != nil {
+				t.Fatalf("failed to delete ws-server: %v", err)
+			}
+		})
+
+		serverServiceYAML := `
+apiVersion: v1
+kind: Service
+metadata:
+  name: ws-server-service
+  labels:
+    app: ws-server
+spec:
+  selector:
+    app: ws-server
+  ports:
+  - protocol: TCP
+    port: 8080        
+    targetPort: 8080  
+`
+		tmpFile = filepath.Join(t.TempDir(), "ws-server-service.yaml")
+		if err := os.WriteFile(tmpFile, []byte(serverServiceYAML), 0644); err != nil {
+			t.Fatalf("failed to write WebSocket server service  YAML: %v", err)
+		}
+
+		if _, err := shell.Execute(true,
+			fmt.Sprintf("kubectl apply -f %s -n %s", tmpFile, namespace),
+		); err != nil {
+			t.Fatalf("failed to deploy ws-server-service: %v", err)
+		}
+
+		t.Cleanup(func() {
+			if _, err := shell.Execute(true, fmt.Sprintf("kubectl delete -f %s -n %s", tmpFile, namespace)); err != nil {
+				t.Fatalf("failed to delete ws-server-service: %v", err)
+			}
+		})
+
+		fetchFn := testKube.NewSinglePodFetch(t.Clusters().Default(), namespace)
+		if _, err := testKube.WaitUntilPodsAreReady(fetchFn); err != nil {
+			t.Fatalf("failed to wait ws-server pod to be ready: %v", err)
+		}
+
+		clientPodYaml := fmt.Sprintf(`
+apiVersion: v1
+kind: Pod
+metadata:
+  name: ws-client
+  labels:
+    app: ws-client
+spec:
+  containers:
+  - name: ws-client
+    image: node:18
+    command: ["/bin/sh", "-c"]
+    args:
+      - |
+        mkdir client
+        cd client
+        npm init -y
+        npm install ws 
+        cat <<EOF > client.js
+        const WebSocket = require('ws');
+        const socket = new WebSocket('ws://ws-server.%s.svc.cluster.local:8080');
+
+        socket.on('open', () => {
+          console.log('Connected to server');
+          socket.send('Hello from client');
+          console.log('Sent message to server');
+        });
+
+        socket.on('message', data => {
+          console.log("Received");
+
+        });
+
+        socket.on('close', () => {
+          console.log('Connection closed');
+        });
+        EOF
+        node client.js
+`, namespace)
+
+		tmpFile = filepath.Join(t.TempDir(), "ws-client.yaml")
+		if err := os.WriteFile(tmpFile, []byte(clientPodYaml), 0644); err != nil {
+			t.Fatalf("failed to write WebSocket client  YAML: %v", err)
+		}
+
+		if _, err := shell.Execute(true,
+			fmt.Sprintf("kubectl apply -f %s -n %s", tmpFile, namespace),
+		); err != nil {
+			t.Fatalf("failed to deploy ws-client: %v", err)
+		}
+
+		t.Cleanup(func() {
+			if _, err := shell.Execute(true, fmt.Sprintf("kubectl delete -f %s -n %s", tmpFile, namespace)); err != nil {
+				t.Fatalf("failed to delete ws-client: %v", err)
+			}
+		})
+		fetchFn = testKube.NewSinglePodFetch(t.Clusters().Default(), namespace)
+		if _, err := testKube.WaitUntilPodsAreReady(fetchFn); err != nil {
+			t.Fatalf("failed to wait ws-client pod to be ready: %v", err)
+		}
+
+	})
+}
+
 func buildL4Query(src, dst echo.Instance) prometheus.Query {
 	query := prometheus.Query{}
 

test/e2e/gateway_test.go

@@ -1,5 +1,3 @@
-
-
 // CODE Copied and modified from https://github.com/istio/istio
 // more specifically: https://github.com/istio/istio/blob/master/pkg/test/framework/components/istio/ingress.go
 //

test/e2e/main_test.go

@@ -1,4 +1,3 @@
-
 /*
  * Copyright The Kmesh Authors.
  *

test/e2e/manage_test.go

@@ -1,4 +1,3 @@
-
 /*
  * Copyright The Kmesh Authors.
  *

test/e2e/restart_test.go

@@ -1,5 +1,3 @@
-
-
 /*
  * Copyright The Kmesh Authors.
  *

test/e2e/run_test.sh

@@ -299,3 +299,4 @@ if [[ -n "${CLEANUP_REGISTRY}" ]]; then
 fi
 
 rm -rf "${TMP}"
+