Connectivity test concurrent run

.github/external-targets/certs.yaml

@@ -2,14 +2,14 @@
 apiVersion: cert-manager.io/v1
 kind: Certificate
 metadata:
- name: ca-cert
+  name: ca-cert
 spec:
- secretName: ca
- isCA: true
- issuerRef:
+  secretName: ca
+  isCA: true
+  issuerRef:
     name: selfsigned-issuer
     kind: Issuer
- commonName: "cilium.io"
+  commonName: "cilium.io"
 ---
 apiVersion: cert-manager.io/v1
 kind: Certificate
@@ -38,5 +38,5 @@ kind: Issuer
 metadata:
   name: ca-cert-issuer
 spec:
- ca:
-   secretName: ca
+  ca:
+    secretName: ca

.github/workflows/kind.yaml

@@ -24,6 +24,13 @@ jobs:
   installation-and-connectivity:
     name: Kind Installation and Connectivity Test
     runs-on: ubuntu-22.04
+    strategy:
+      matrix:
+        include:
+          # run connectivity tests explicitly without concurrency
+          - test-concurrency: 1
+          # run connectivity tests concurrently
+          - test-concurrency: 5
     timeout-minutes: 50
     steps:
       - name: Checkout
@@ -106,13 +113,15 @@ jobs:
           #
           # Dispatch interval is set to 100ms, b/c otherwise (default is 0), the flow validation might time out.
           cilium connectivity test --debug --test-namespace test-namespace \
+            --test-concurrency=${{ matrix.test-concurrency }} \
             --conn-disrupt-dispatch-interval 100ms \
             --include-conn-disrupt-test --conn-disrupt-test-setup
 
           # Run the connectivity test in non-default namespace (i.e. not cilium-test)
           cilium connectivity test --debug --all-flows --test-namespace test-namespace \
+            --test-concurrency=${{ matrix.test-concurrency }} \
             --include-unsafe-tests --include-conn-disrupt-test \
-            --collect-sysdump-on-failure --junit-file cilium-junit-1.xml \
+            --collect-sysdump-on-failure --junit-file cilium-junit-1-concurrency-${{ matrix.test-concurrency }}.xml \
             --junit-property type=no-tunnel \
             --curl-insecure \
             --external-target chart-testing-worker2 \
@@ -152,8 +161,9 @@ jobs:
       - name: Connectivity test
         run: |
           cilium connectivity test --debug --force-deploy --all-flows --test-namespace test-namespace \
+            --test-concurrency=${{ matrix.test-concurrency }} \
             --include-unsafe-tests \
-            --collect-sysdump-on-failure --junit-file cilium-junit-2.xml \
+            --collect-sysdump-on-failure --junit-file cilium-junit-2-concurrency-${{ matrix.test-concurrency }}.xml \
             --junit-property type=ipsec \
             --curl-insecure \
             --external-target chart-testing-worker2 \
@@ -167,7 +177,7 @@ jobs:
         if: ${{ always() }}
         uses: actions/upload-artifact@65462800fd760344b1a7b4382951275a0abb4808 # v4.3.3
         with:
-          name: cilium-junits
+          name: cilium-junits-concurrency-${{ matrix.test-concurrency }}
           path: cilium-junit*.xml
           retention-days: 2
 
@@ -176,20 +186,27 @@ jobs:
         run: |
           cilium status
           kubectl get pods --all-namespaces -o wide
-          cilium sysdump --output-filename cilium-sysdump-out --hubble-flows-count 10000
+          cilium sysdump --output-filename cilium-sysdump-out-concurrency-${{ matrix.test-concurrency }} --hubble-flows-count 10000
         shell: bash {0} # Disable default fail-fast behaviour so that all commands run independently
 
       - name: Upload sysdump
         if: ${{ !success() }}
         uses: actions/upload-artifact@65462800fd760344b1a7b4382951275a0abb4808 # v4.3.3
         with:
-          name: cilium-sysdumps
+          name: cilium-sysdumps-concurrency-${{ matrix.test-concurrency }}
           path: cilium-sysdump-*.zip
           retention-days: 5
 
   helm-upgrade-clustermesh:
     name: Kind Helm Upgrade Clustermesh
     runs-on: ubuntu-22.04
+    strategy:
+      matrix:
+        include:
+          # run connectivity tests explicitly without concurrency
+          - test-concurrency: 1
+          # run connectivity tests concurrently
+          - test-concurrency: 5
     timeout-minutes: 50
 
     env:
@@ -315,19 +332,21 @@ jobs:
           #
           # Dispatch interval is set to 100ms, b/c otherwise (default is 0), the flow validation might time out.
           cilium connectivity test --context $CLUSTER1 --multi-cluster $CLUSTER2 --debug \
+            --test-concurrency=${{ matrix.test-concurrency }} \
             --conn-disrupt-dispatch-interval 100ms \
             --include-conn-disrupt-test --conn-disrupt-test-setup
 
           cilium connectivity test --context $CLUSTER1 --multi-cluster $CLUSTER2 --debug \
+            --test-concurrency=${{ matrix.test-concurrency }} \
             --include-unsafe-tests --include-conn-disrupt-test \
-            --collect-sysdump-on-failure --junit-file cilium-junit-clustermesh-1.xml \
+            --collect-sysdump-on-failure --junit-file cilium-junit-clustermesh-1-concurrency-${{ matrix.test-concurrency }}.xml \
             --junit-property mode=clustermesh --junit-property type=ipsec
 
       - name: Upload JUnit
         if: ${{ always() }}
         uses: actions/upload-artifact@65462800fd760344b1a7b4382951275a0abb4808 # v4.3.3
         with:
-          name: cilium-junits-helm-upgrade-clustermesh
+          name: cilium-junits-helm-upgrade-clustermesh-concurrency-${{ matrix.test-concurrency }}
           path: cilium-junit*.xml
           retention-days: 2
 
@@ -336,24 +355,24 @@ jobs:
         run: |
           cilium --context $CLUSTER1 status
           kubectl --context $CLUSTER1 get pods --all-namespaces -o wide
-          cilium --context $CLUSTER1 sysdump --output-filename cilium-sysdump-out-c1
+          cilium --context $CLUSTER1 sysdump --output-filename cilium-sysdump-out-c1-concurrency-${{ matrix.test-concurrency }}
           cilium --context $CLUSTER2 status
           kubectl --context $CLUSTER2 get pods --all-namespaces -o wide
-          cilium --context $CLUSTER2 sysdump --output-filename cilium-sysdump-out-c2
+          cilium --context $CLUSTER2 sysdump --output-filename cilium-sysdump-out-c2-concurrency-${{ matrix.test-concurrency }}
         shell: bash {0} # Disable default fail-fast behaviour so that all commands run independently
 
       - name: Upload sysdump from cluster 1
         if: ${{ !success() }}
         uses: actions/upload-artifact@65462800fd760344b1a7b4382951275a0abb4808 # v4.3.3
         with:
-          name: cilium-sysdump-out-c1.zip
-          path: cilium-sysdump-out-c1.zip
+          name: cilium-sysdump-out-c1-concurrency-${{ matrix.test-concurrency }}.zip
+          path: cilium-sysdump-out-c1-concurrency-${{ matrix.test-concurrency }}.zip
           retention-days: 5
 
       - name: Upload sysdump from cluster 2
         if: ${{ !success() }}
         uses: actions/upload-artifact@65462800fd760344b1a7b4382951275a0abb4808 # v4.3.3
         with:
-          name: cilium-sysdump-out-c2.zip
-          path: cilium-sysdump-out-c2.zip
+          name: cilium-sysdump-out-c2-concurrency-${{ matrix.test-concurrency }}.zip
+          path: cilium-sysdump-out-c2-concurrency-${{ matrix.test-concurrency }}.zip
           retention-days: 5

cli/connectivity.go

@@ -5,7 +5,6 @@ package cli
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"os"
 	"os/signal"
@@ -24,8 +23,6 @@ import (
 	"github.com/cilium/cilium-cli/sysdump"
 )
 
-var errInternal = errors.New("encountered internal error, exiting")
-
 func newCmdConnectivity(hooks api.Hooks) *cobra.Command {
 	cmd := &cobra.Command{
 		Use:   "connectivity",
@@ -40,7 +37,9 @@ func newCmdConnectivity(hooks api.Hooks) *cobra.Command {
 }
 
 var params = check.Parameters{
-	Writer: os.Stdout,
+	ExternalDeploymentPort: 8080,
+	EchoServerHostPort:     4000,
+	Writer:                 os.Stdout,
 	SysdumpOptions: sysdump.Options{
 		LargeSysdumpAbortTimeout: sysdump.DefaultLargeSysdumpAbortTimeout,
 		LargeSysdumpThreshold:    sysdump.DefaultLargeSysdumpThreshold,
@@ -70,8 +69,7 @@ func RunE(hooks api.Hooks) func(cmd *cobra.Command, args []string) error {
 			}
 		}
 
-		// Instantiate the test harness.
-		cc, err := check.NewConnectivityTest(k8sClient, params, defaults.CLIVersion)
+		connTests, err := newConnectivityTests(params)
 		if err != nil {
 			return err
 		}
@@ -86,33 +84,10 @@ func RunE(hooks api.Hooks) func(cmd *cobra.Command, args []string) error {
 
 		go func() {
 			<-ctx.Done()
-			cc.Logf("Cancellation request (%s) received, cancelling tests...", context.Cause(ctx))
-		}()
-
-		done := make(chan struct{})
-		var finished bool
-
-		// Execute connectivity.Run() in its own goroutine, it might call Fatal()
-		// and end the goroutine without returning.
-		go func() {
-			defer func() { done <- struct{}{} }()
-			err = connectivity.Run(ctx, cc, hooks)
-
-			// If Fatal() was called in the test suite, the statement below won't fire.
-			finished = true
+			connTests[0].Logf("Cancellation request (%s) received, cancelling tests...", context.Cause(ctx))
 		}()
-		<-done
 
-		if !finished {
-			// Exit with a non-zero return code.
-			return errInternal
-		}
-
-		if err != nil {
-			return fmt.Errorf("connectivity test failed: %w", err)
-		}
-
-		return nil
+		return connectivity.Run(ctx, connTests, hooks)
 	}
 }
 
@@ -197,6 +172,9 @@ func newCmdConnectivityTest(hooks api.Hooks) *cobra.Command {
 
 	cmd.Flags().DurationVar(&params.Timeout, "timeout", defaults.ConnectivityTestSuiteTimeout, "Maximum time to allow the connectivity test suite to take")
 
+	cmd.Flags().IntVar(&params.TestConcurrency, "test-concurrency", 1, "Count of namespaces to perform the connectivity tests in parallel (value <= 0 will be treated as 1)")
+	_ = cmd.Flags().MarkHidden("test-concurrency")
+
 	hooks.AddConnectivityTestFlags(cmd.Flags())
 
 	registerCommonFlags(cmd.Flags())
@@ -238,3 +216,31 @@ func registerCommonFlags(flags *pflag.FlagSet) {
 	flags.StringVar(&params.TestNamespace, "test-namespace", defaults.ConnectivityCheckNamespace, "Namespace to perform the connectivity in")
 	flags.Var(&params.DeploymentAnnotations, "deployment-pod-annotations", "Add annotations to the connectivity pods, e.g. '{\"client\":{\"foo\":\"bar\"}}'")
 }
+
+func newConnectivityTests(params check.Parameters) ([]*check.ConnectivityTest, error) {
+	if params.TestConcurrency < 1 {
+		fmt.Printf("--test-concurrency parameter value is invalid [%d], using 1 instead\n", params.TestConcurrency)
+		params.TestConcurrency = 1
+	}
+	if params.TestConcurrency < 2 {
+		cc, err := check.NewConnectivityTest(k8sClient, params, defaults.CLIVersion)
+		if err != nil {
+			return nil, err
+		}
+		return []*check.ConnectivityTest{cc}, nil
+	}
+
+	connTests := make([]*check.ConnectivityTest, 0, params.TestConcurrency)
+	for i := 0; i < params.TestConcurrency; i++ {
+		params := params
+		params.TestNamespace = fmt.Sprintf("%s-%d", params.TestNamespace, i+1)
+		params.ExternalDeploymentPort += i
+		params.EchoServerHostPort += i
+		cc, err := check.NewConnectivityTest(k8sClient, params, defaults.CLIVersion)
+		if err != nil {
+			return nil, err
+		}
+		connTests = append(connTests, cc)
+	}
+	return connTests, nil
+}

cli/connectivity_test.go

@@ -0,0 +1,57 @@
+// SPDX-License-Identifier: Apache-2.0
+// Copyright Authors of Cilium
+
+package cli
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/cilium/cilium-cli/connectivity/check"
+)
+
+func TestNewConnectivityTests(t *testing.T) {
+	testCases := []struct {
+		params                 check.Parameters
+		expectedCount          int
+		expectedTestNamespaces []string
+	}{
+		{
+			params: check.Parameters{
+				FlowValidation: check.FlowValidationModeDisabled,
+				TestNamespace:  "cilium-test",
+			},
+			expectedCount:          1,
+			expectedTestNamespaces: []string{"cilium-test"},
+		},
+		{
+			params: check.Parameters{
+				FlowValidation:  check.FlowValidationModeDisabled,
+				TestNamespace:   "cilium-test",
+				TestConcurrency: -1,
+			},
+			expectedCount:          1,
+			expectedTestNamespaces: []string{"cilium-test"},
+		},
+		{
+			params: check.Parameters{
+				FlowValidation:  check.FlowValidationModeDisabled,
+				TestNamespace:   "cilium-test",
+				TestConcurrency: 3,
+			},
+			expectedCount:          3,
+			expectedTestNamespaces: []string{"cilium-test-1", "cilium-test-2", "cilium-test-3"},
+		},
+	}
+	for _, tt := range testCases {
+		// function to test
+		actual, err := newConnectivityTests(tt.params)
+
+		require.NoError(t, err)
+		require.Equal(t, tt.expectedCount, len(actual))
+		for i, n := range tt.expectedTestNamespaces {
+			require.Equal(t, n, actual[i].Params().TestNamespace)
+		}
+	}
+}