Add e2e test

Signed-off-by: Yanjun Zhou <[email protected]>

test/e2e/antctl_test.go

@@ -39,7 +39,7 @@ func TestAntctl(t *testing.T) {
 	skipIfHasWindowsNodes(t)
 	skipIfNotRequired(t, "mode-irrelevant")
 
-	data, err := setupTest(t)
+	data, _, _, err := setupTestWithIPFIXCollector(t)
 	if err != nil {
 		t.Fatalf("Error when setting up test: %v", err)
 	}
@@ -51,6 +51,9 @@ func TestAntctl(t *testing.T) {
 	t.Run("testAntctlControllerRemoteAccess", func(t *testing.T) {
 		testAntctlControllerRemoteAccess(t, data)
 	})
+	t.Run("testAntctlFlowAggregatorLocalAccess", func(t *testing.T) {
+		testAntctlFlowAggregatorLocalAccess(t, data)
+	})
 	t.Run("testAntctlVerboseMode", func(t *testing.T) {
 		testAntctlVerboseMode(t, data)
 	})
@@ -67,12 +70,18 @@ func antctlOutput(stdout, stderr string) string {
 // runAntctl runs antctl commands on antrea Pods, the controller, or agents.
 func runAntctl(podName string, cmds []string, data *TestData) (string, string, error) {
 	var containerName string
+	var namespace string
 	if strings.Contains(podName, "agent") {
 		containerName = "antrea-agent"
+		namespace = antreaNamespace
+	} else if strings.Contains(podName, "flow-aggregator") {
+		containerName = "flow-aggregator"
+		namespace = flowAggregatorNamespace
 	} else {
 		containerName = "antrea-controller"
+		namespace = antreaNamespace
 	}
-	stdout, stderr, err := data.runCommandFromPod(antreaNamespace, podName, containerName, cmds)
+	stdout, stderr, err := data.runCommandFromPod(namespace, podName, containerName, cmds)
 	// remove Bincover metadata if needed
 	if err == nil {
 		index := strings.Index(stdout, "START_BINCOVER_METADATA")
@@ -115,6 +124,32 @@ func testAntctlAgentLocalAccess(t *testing.T, data *TestData) {
 	}
 }
 
+// testAntctlAgentLocalAccess ensures antctl is accessible in a flow aggregator Pod.
+func testAntctlFlowAggregatorLocalAccess(t *testing.T, data *TestData) {
+	podName, err := data.getFlowAggregator(controlPlaneNodeName())
+	if err != nil {
+		t.Fatalf("Error when getting flow-aggregator pod name: %v", err)
+	}
+	for _, c := range antctl.CommandList.GetDebugCommands(runtime.ModeFlowAggregator) {
+		args := []string{}
+		if testOptions.enableCoverage {
+			antctlCovArgs := antctlCoverageArgs("antctl-coverage")
+			args = append(antctlCovArgs, c...)
+		} else {
+			args = append([]string{"antctl", "-v"}, c...)
+		}
+		t.Logf("args: %s", args)
+
+		cmd := strings.Join(args, " ")
+		t.Run(cmd, func(t *testing.T) {
+			stdout, stderr, err := runAntctl(podName, args, data)
+			if err != nil {
+				t.Fatalf("Error when running `antctl %s` from %s: %v\n%s", c, podName, err, antctlOutput(stdout, stderr))
+			}
+		})
+	}
+}
+
 func copyAntctlToNode(data *TestData, nodeName string, antctlName string, nodeAntctlPath string) error {
 	pod, err := data.getAntreaController()
 	if err != nil {

test/e2e/framework.go

@@ -1296,6 +1296,21 @@ func (data *TestData) getAntreaPodOnNode(nodeName string) (podName string, err e
 	return pods.Items[0].Name, nil
 }
 
+// getFlowAggregator retrieves the name of the Flow-Aggregator Pod (flow-aggregator-*) running on a specific Node.
+func (data *TestData) getFlowAggregator(nodeName string) (podName string, err error) {
+	listOptions := metav1.ListOptions{
+		LabelSelector: "app=flow-aggregator",
+	}
+	pods, err := data.clientset.CoreV1().Pods(flowAggregatorNamespace).List(context.TODO(), listOptions)
+	if err != nil {
+		return "", fmt.Errorf("failed to list Antrea Pods: %v", err)
+	}
+	if len(pods.Items) != 1 {
+		return "", fmt.Errorf("expected *exactly* one Pod")
+	}
+	return pods.Items[0].Name, nil
+}
+
 // getAntreaController retrieves the name of the Antrea Controller (antrea-controller-*) running in the k8s cluster.
 func (data *TestData) getAntreaController() (*corev1.Pod, error) {
 	listOptions := metav1.ListOptions{