ci: add connectivity e2e test

Signed-off-by: zhangzujian <[email protected]>

.github/workflows/build-x86-image.yaml

@@ -2836,6 +2836,118 @@ jobs:
         if: ${{ success() || (failure() && (steps.install.conclusion == 'failure' || steps.kube-ovn-ipsec-e2e.conclusion == 'failure')) }}
         run: make check-kube-ovn-pod-restarts
 
+  kube-ovn-connectivity-test:
+    name: Kube-OVN Connectivity E2E
+    needs:
+      - build-kube-ovn
+      - build-e2e-binaries
+    runs-on: ubuntu-24.04
+    timeout-minutes: 15
+    strategy:
+      fail-fast: false
+      matrix:
+        mode:
+          - overlay
+          - underlay
+    steps:
+      - uses: jlumbroso/[email protected]
+        with:
+          android: true
+          dotnet: true
+          haskell: true
+          docker-images: false
+          large-packages: false
+          tool-cache: false
+          swap-storage: false
+
+      - uses: actions/checkout@v4
+
+      - name: Create the default branch directory
+        if: (github.base_ref || github.ref_name) != github.event.repository.default_branch
+        run: mkdir -p test/e2e/source
+
+      - name: Check out the default branch
+        if: (github.base_ref || github.ref_name) != github.event.repository.default_branch
+        uses: actions/checkout@v4
+        with:
+          ref: ${{ github.event.repository.default_branch }}
+          fetch-depth: 1
+          path: test/e2e/source
+
+      - name: Export E2E directory
+        run: |
+          if [ '${{ github.base_ref || github.ref_name }}' = '${{ github.event.repository.default_branch }}' ]; then
+            echo "E2E_DIR=." >> "$GITHUB_ENV"
+          else
+            echo "E2E_DIR=test/e2e/source" >> "$GITHUB_ENV"
+          fi
+
+      - uses: actions/setup-go@v5
+        id: setup-go
+        with:
+          go-version-file: ${{ env.E2E_DIR }}/go.mod
+          check-latest: true
+          cache: false
+
+      - name: Export Go full version
+        run: echo "GO_VERSION=${{ steps.setup-go.outputs.go-version }}" >> "$GITHUB_ENV"
+
+      - name: Go cache
+        uses: actions/cache/restore@v4
+        with:
+          path: |
+            ~/.cache/go-build
+            ~/go/pkg/mod
+          key: ${{ runner.os }}-e2e-go-${{ env.GO_VERSION }}-x86-${{ hashFiles(format('{0}/**/go.sum', env.E2E_DIR)) }}
+          restore-keys: ${{ runner.os }}-e2e-go-${{ env.GO_VERSION }}-x86-
+
+      - name: Install kind
+        uses: helm/[email protected]
+        with:
+          version: ${{ env.KIND_VERSION }}
+          install_only: true
+
+      - name: Install ginkgo
+        working-directory: ${{ env.E2E_DIR }}
+        run: go install -v -mod=mod github.com/onsi/ginkgo/v2/ginkgo
+
+      - name: Download kube-ovn image
+        uses: actions/download-artifact@v4
+        with:
+          name: kube-ovn
+
+      - name: Load images
+        run: docker load -i kube-ovn.tar
+
+      - name: Create kind cluster
+        run: |
+          pipx install jinjanator
+          make kind-init
+
+      - name: Install Kube-OVN
+        id: install
+        run: make kind-install-${{ matrix.mode }}
+
+      - name: Run E2E
+        id: e2e
+        working-directory: ${{ env.E2E_DIR }}
+        env:
+          E2E_BRANCH: ${{ github.base_ref || github.ref_name }}
+        run: make kube-ovn-connectivity-e2e
+
+      - name: kubectl ko log
+        if: failure() && (steps.e2e.conclusion == 'failure')
+        run: |
+          make kubectl-ko-log
+          mv kubectl-ko-log.tar.gz kube-ovn-connectivity-e2e-${{ matrix.mode }}-ko-log.tar.gz
+
+      - name: upload kubectl ko log
+        uses: actions/upload-artifact@v4
+        if: failure() && (steps.kube-ovn-connectivity-e2e.conclusion == 'failure')
+        with:
+          name: kube-ovn-connectivity-e2e-${{ matrix.mode }}-ko-log
+          path: kube-ovn-connectivity-e2e-${{ matrix.mode }}-ko-log.tar.gz
+
   push:
     name: Push Images
     needs:

Makefile

@@ -559,6 +559,9 @@ kind-install-ipv6:
 kind-install-dual:
 	@DUAL_STACK=true $(MAKE) kind-install
 
+.PHONY: kind-install-overlay
+kind-install-overlay: kind-install-overlay-ipv4
+
 .PHONY: kind-install-overlay-%
 kind-install-overlay-%:
 	@$(MAKE) kind-install-$*

test/e2e/connectivity/e2e_test.go

@@ -72,7 +72,7 @@ func TestE2E(t *testing.T) {
 
 type suiteContext struct {
 	Node     string
-	HostIP   string
+	NodeIP   string
 	NodePort int32
 }
 
@@ -108,9 +108,28 @@ var _ = ginkgo.SynchronizedBeforeSuite(func() []byte {
 	framework.ExpectNoError(err)
 	framework.ExpectNotNil(pods)
 	framework.ExpectNotEmpty(pods.Items, "no pod found in deployment "+deploymentName)
-	suiteCtx.HostIP = pods.Items[0].Status.HostIP
+	suiteCtx.NodeIP = pods.Items[0].Status.HostIP
 	suiteCtx.Node = pods.Items[0].Spec.NodeName
 
+	ginkgo.By("Getting all nodes")
+	nodes, err := cs.CoreV1().Nodes().List(context.TODO(), metav1.ListOptions{})
+	framework.ExpectNoError(err)
+	framework.ExpectNotNil(nodes)
+	framework.ExpectNotEmpty(nodes.Items)
+
+	// use the internal IP of the node that is not the same as the pod node
+	for _, node := range nodes.Items {
+		if node.Name != suiteCtx.Node {
+			ipv4, ipv6 := util.GetNodeInternalIP(node)
+			if ipv4 != "" {
+				suiteCtx.NodeIP = ipv4
+			} else {
+				suiteCtx.NodeIP = ipv6
+			}
+			break
+		}
+	}
+
 	ginkgo.By("Creating service " + serviceName)
 	ports := []corev1.ServicePort{{
 		Name:       "tcp",
@@ -157,13 +176,13 @@ var _ = framework.Describe("[group:connectivity]", func() {
 	})
 	ginkgo.AfterEach(ginkgo.OncePerOrdered, func() {
 		ginkgo.By("Closing RPC server")
-		framework.ExpectNoError(server.Close())
+		framework.ExpectNoError(server.Shutdown(context.Background()))
 	})
 
 	ginkgo.It("Continuous NodePort HTTP testing", func() {
 		u := url.URL{
 			Scheme: "http",
-			Host:   util.JoinHostPort(suiteCtx.HostIP, suiteCtx.NodePort),
+			Host:   util.JoinHostPort(suiteCtx.NodeIP, suiteCtx.NodePort),
 			Path:   "/clientip",
 		}
 		ginkgo.By("GET " + u.String())
@@ -263,7 +282,7 @@ var _ = framework.OrderedDescribe("[group:disaster]", func() {
 		framework.ExpectNotEmpty(pods.Items, "no pod found in deployment ovn-central")
 
 		for _, pod := range pods.Items {
-			framework.Logf("pod %s is running on %s", pod.Name, pod.Spec.NodeName)
+			ginkgo.By("Deleting pod " + pod.Name + " running on " + pod.Spec.NodeName)
 			err = cs.CoreV1().Pods(pod.Namespace).Delete(context.TODO(), pod.Name, metav1.DeleteOptions{})
 			framework.ExpectNoError(err, "failed to delete pod "+pod.Name)
 		}
@@ -281,6 +300,90 @@ var _ = framework.OrderedDescribe("[group:disaster]", func() {
 		}
 	})
 
+	framework.DisruptiveIt("Stop ovn sb process", func() {
+		ginkgo.By("Getting deployment ovn-central")
+		deploymentClient := framework.NewDeploymentClient(cs, framework.KubeOvnNamespace)
+		deploy := deploymentClient.Get("ovn-central")
+
+		ginkgo.By("Getting pods of deployment ovn-central")
+		pods, err := deploymentClient.GetPods(deploy)
+		framework.ExpectNoError(err)
+		framework.ExpectNotEmpty(pods.Items, "no pod found in deployment ovn-central")
+
+		for _, pod := range pods.Items {
+			ginkgo.By("Getting ovn sb pid of pod " + pod.Name + " running on " + pod.Spec.NodeName)
+			cmd := []string{"cat", "/run/ovn/ovnsb_db.pid"}
+			stdout, stderr, err := framework.KubectlExec(pod.Namespace, pod.Name, cmd...)
+			framework.ExpectNoError(err, "failed to get ovn sb pid: %v, %s", err, string(stderr))
+			pid := string(bytes.TrimSpace(stdout))
+			framework.Logf("ovn sb pid: %s", pid)
+
+			ginkgo.By("Stopping ovn sb process by sending a STOP signal")
+			cmd = []string{"sh", "-c", fmt.Sprintf(`"kill -STOP %s"`, pid)}
+			_, stderr, err = framework.KubectlExec(pod.Namespace, pod.Name, cmd...)
+			framework.ExpectNoError(err, "failed to send STOP signal to ovn sb process: %v, %s", err, string(stderr))
+		}
+
+		ginkgo.By("Waiting 60s")
+		time.Sleep(60 * time.Second)
+
+		for _, pod := range pods.Items {
+			ginkgo.By("Getting ovn sb pid of pod " + pod.Name + " running on " + pod.Spec.NodeName)
+			cmd := []string{"cat", "/run/ovn/ovnsb_db.pid"}
+			stdout, stderr, err := framework.KubectlExec(pod.Namespace, pod.Name, cmd...)
+			framework.ExpectNoError(err, "failed to get ovn sb pid: %v, %s", err, string(stderr))
+			pid := string(bytes.TrimSpace(stdout))
+			framework.Logf("ovn sb pid: %s", pid)
+
+			ginkgo.By("Stopping ovn sb process by sending a CONT signal")
+			cmd = []string{"sh", "-c", fmt.Sprintf(`"kill -CONT %s"`, pid)}
+			_, stderr, err = framework.KubectlExec(pod.Namespace, pod.Name, cmd...)
+			framework.ExpectNoError(err, "failed to send CONT signal to ovn sb process: %v, %s", err, string(stderr))
+		}
+	})
+
+	framework.DisruptiveIt("Stop ovn-controller process", func() {
+		ginkgo.By("Getting DaemonSet ovs-ovn")
+		dsClient := framework.NewDaemonSetClient(cs, framework.KubeOvnNamespace)
+		ds := dsClient.Get("ovs-ovn")
+
+		ginkgo.By("Getting pods of DaemonSet ovs-ovn")
+		pods, err := dsClient.GetPods(ds)
+		framework.ExpectNoError(err)
+		framework.ExpectNotEmpty(pods.Items, "no pod found in DaemonSet ovs-ovn")
+
+		ginkgo.By("Getting ovs-ovn pod running on node " + suiteCtx.Node)
+		var pod *corev1.Pod
+		for i := range pods.Items {
+			framework.Logf("pod %s is running on %s", pods.Items[i].Name, pods.Items[i].Spec.NodeName)
+			if pods.Items[i].Spec.NodeName == suiteCtx.Node {
+				pod = &pods.Items[i]
+				break
+			}
+		}
+		framework.ExpectNotNil(pod, "no ovs-ovn pod running on node "+suiteCtx.Node)
+
+		ginkgo.By("Getting ovn-controller pid")
+		cmd := []string{"sh", "-c", `"pidof -s ovn-controller"`}
+		stdout, stderr, err := framework.KubectlExec(pod.Namespace, pod.Name, cmd...)
+		framework.ExpectNoError(err, "failed to get ovn-controller pid: %v, %s", err, string(stderr))
+		pid := string(bytes.TrimSpace(stdout))
+		framework.Logf("ovn-controller pid: %s", pid)
+
+		ginkgo.By("Stopping ovn-controller process by sending a STOP signal")
+		cmd = []string{"sh", "-c", fmt.Sprintf(`"kill -STOP %s"`, pid)}
+		_, stderr, err = framework.KubectlExec(pod.Namespace, pod.Name, cmd...)
+		framework.ExpectNoError(err, "failed to send STOP signal to ovn-controller process: %v, %s", err, string(stderr))
+
+		ginkgo.By("Waiting 60s")
+		time.Sleep(60 * time.Second)
+
+		ginkgo.By("Continuing the stopped ovn-controller process by sending a CONT signal")
+		cmd = []string{"sh", "-c", fmt.Sprintf(`"kill -CONT %s"`, pid)}
+		_, stderr, err = framework.KubectlExec(pod.Namespace, pod.Name, cmd...)
+		framework.ExpectNoError(err, "failed to send CONT signal to ovn-controller process: %v, %s", err, string(stderr))
+	})
+
 	framework.DisruptiveIt("Stop ovs-vswitchd process", func() {
 		ginkgo.By("Getting DaemonSet ovs-ovn")
 		dsClient := framework.NewDaemonSetClient(cs, framework.KubeOvnNamespace)

test/e2e/framework/rpc/server.go

@@ -1,24 +1,16 @@
 package rpc
 
 import (
+	"errors"
 	"fmt"
-	"net"
 	"net/http"
 	"net/rpc"
 
 	"github.com/kubeovn/kube-ovn/test/e2e/framework"
 )
 
 type Server struct {
-	listener net.Listener
-}
-
-func (s *Server) Addr() string {
-	return s.listener.Addr().String()
-}
-
-func (s *Server) Close() error {
-	return s.listener.Close()
+	*http.Server
 }
 
 func NewServer(addr string, rcvr any) (*Server, error) {
@@ -27,16 +19,12 @@ func NewServer(addr string, rcvr any) (*Server, error) {
 	}
 
 	rpc.HandleHTTP()
-	listener, err := net.Listen("tcp", addr)
-	if err != nil {
-		return nil, fmt.Errorf("failed to listen on %q: %w", addr, err)
-	}
-
+	svr := &http.Server{Addr: addr}
 	go func() {
-		if err := http.Serve(listener, nil); err != nil {
-			framework.Failf("failed to serve rpc: %v", err)
+		if err := svr.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
+			framework.Failf("failed to listen and serve rpc on %q: %v", addr, err)
 		}
 	}()
 
-	return &Server{listener: listener}, nil
+	return &Server{Server: svr}, nil
 }