// Copyright 2020 Antrea Authors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//      http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package e2e

import (
	"context"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"net"
	"slices"
	"strconv"
	"strings"
	"testing"
	"time"

	"github.com/google/uuid"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	ipfixregistry "github.com/vmware/go-ipfix/pkg/registry"
	corev1 "k8s.io/api/core/v1"
	networkingv1 "k8s.io/api/networking/v1"
	"k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/util/wait"
	"k8s.io/utils/ptr"

	"antrea.io/antrea/pkg/agent/config"
	antreaagenttypes "antrea.io/antrea/pkg/agent/types"
	"antrea.io/antrea/pkg/antctl"
	"antrea.io/antrea/pkg/antctl/runtime"
	secv1beta1 "antrea.io/antrea/pkg/apis/crd/v1beta1"
	flowaggregatorconfig "antrea.io/antrea/pkg/config/flowaggregator"
	"antrea.io/antrea/pkg/features"
	"antrea.io/antrea/pkg/flowaggregator/apis"
	"antrea.io/antrea/test/e2e/utils"
)

/* Sample output from the collector:
IPFIX-HDR:
  version: 10,  Message Length: 617
  Exported Time: 1637706974 (2021-11-23 22:36:14 +0000 UTC)
  Sequence No.: 27,  Observation Domain ID: 2569248951
DATA SET:
  DATA RECORD-0:
    flowStartSeconds: 1637706961
    flowEndSeconds: 1637706973
    flowEndReason: 3
    sourceTransportPort: 44752
    destinationTransportPort: 5201
    protocolIdentifier: 6
    packetTotalCount: 823188
    octetTotalCount: 30472817041
    packetDeltaCount: 241333
    octetDeltaCount: 8982624938
    sourceIPv4Address: 10.10.0.79
    destinationIPv4Address: 10.10.0.80
    reversePacketTotalCount: 471111
    reverseOctetTotalCount: 24500996
    reversePacketDeltaCount: 136211
    reverseOctetDeltaCount: 7083284
    sourcePodName: perftest-a
    sourcePodNamespace: testflowaggregator-b6mjmbpl
    sourceNodeName: k8s-node-control-plane
    destinationPodName: perftest-b
    destinationPodNamespace: testflowaggregator-b6mjmbpl
    destinationNodeName: k8s-node-control-plane
    destinationServicePort: 0
    destinationServicePortName:
    ingressNetworkPolicyName: test-flow-aggregator-networkpolicy-ingress-allow
    ingressNetworkPolicyNamespace: testflowaggregator-b6mjmbpl
    ingressNetworkPolicyType: 1
    ingressNetworkPolicyRuleName:
    ingressNetworkPolicyRuleAction: 1
    egressNetworkPolicyName: test-flow-aggregator-networkpolicy-egress-allow
    egressNetworkPolicyNamespace: testflowaggregator-b6mjmbpl
    egressNetworkPolicyType: 1
    egressNetworkPolicyRuleName:
    egressNetworkPolicyRuleAction: 1
    tcpState: TIME_WAIT
    flowType: 1
    egressName: test-egressbkclk
    egressIP: 172.18.0.2
    appProtocolName: http
    httpVals: mockHttpString
    egressNodeName: k8s-node-worker
    destinationClusterIPv4: 0.0.0.0
    octetDeltaCountFromSourceNode: 8982624938
    octetDeltaCountFromDestinationNode: 8982624938
    octetTotalCountFromSourceNode: 30472817041
    octetTotalCountFromDestinationNode: 30472817041
    packetDeltaCountFromSourceNode: 241333
    packetDeltaCountFromDestinationNode: 241333
    packetTotalCountFromSourceNode: 823188
    packetTotalCountFromDestinationNode: 823188
    reverseOctetDeltaCountFromSourceNode: 7083284
    reverseOctetDeltaCountFromDestinationNode: 7083284
    reverseOctetTotalCountFromSourceNode: 24500996
    reverseOctetTotalCountFromDestinationNode: 24500996
    reversePacketDeltaCountFromSourceNode: 136211
    reversePacketDeltaCountFromDestinationNode: 136211
    reversePacketTotalCountFromSourceNode: 471111
    reversePacketTotalCountFromDestinationNode: 471111
    flowEndSecondsFromSourceNode: 1637706973
    flowEndSecondsFromDestinationNode: 1637706973
    throughput: 15902813472
    throughputFromSourceNode: 15902813472
    throughputFromDestinationNode: 15902813472
    reverseThroughput: 12381344
    reverseThroughputFromSourceNode: 12381344
    reverseThroughputFromDestinationNode: 12381344
    sourcePodLabels: {"antrea-e2e":"perftest-a","app":"iperf"}
    destinationPodLabels: {"antrea-e2e":"perftest-b","app":"iperf"}
Intra-Node: Flow record information is complete for source and destination e.g. sourcePodName, destinationPodName
Inter-Node: Flow record from destination Node is ignored, so only flow record from the source Node has its K8s info e.g., sourcePodName, sourcePodNamespace, sourceNodeName etc.
AntreaProxy enabled (Intra-Node): Flow record information is complete for source and destination along with K8s service info such as destinationClusterIP, destinationServicePort, destinationServicePortName etc.
AntreaProxy enabled (Inter-Node): Flow record from destination Node is ignored, so only flow record from the source Node has its K8s info like in Inter-Node case along with K8s Service info such as destinationClusterIP, destinationServicePort, destinationServicePortName etc.
*/

const (
	ingressAllowNetworkPolicyName  = "test-flow-aggregator-networkpolicy-ingress-allow"
	ingressRejectANPName           = "test-flow-aggregator-anp-ingress-reject"
	ingressDropANPName             = "test-flow-aggregator-anp-ingress-drop"
	ingressDenyNPName              = "test-flow-aggregator-np-ingress-deny"
	egressAllowNetworkPolicyName   = "test-flow-aggregator-networkpolicy-egress-allow"
	egressAllowANPName             = "test-flow-aggregator-anp-egress-allow"
	egressRejectANPName            = "test-flow-aggregator-anp-egress-reject"
	egressDropANPName              = "test-flow-aggregator-anp-egress-drop"
	egressDenyNPName               = "test-flow-aggregator-np-egress-deny"
	ingressAntreaNetworkPolicyName = "test-flow-aggregator-antrea-networkpolicy-ingress"
	egressAntreaNetworkPolicyName  = "test-flow-aggregator-antrea-networkpolicy-egress"
	testIngressRuleName            = "test-ingress-rule-name"
	testEgressRuleName             = "test-egress-rule-name"
	clickHousePodName              = "chi-clickhouse-clickhouse-0-0-0"
	iperfTimeSec                   = 12
	protocolIdentifierTCP          = 6
	// Set target bandwidth(bits/sec) of iPerf traffic to a relatively small value
	// (default unlimited for TCP), to reduce the variances caused by network performance
	// during 12s, and make the throughput test more stable.
	iperfBandwidth  = "10m"
	serverPodPort   = int32(80)
	customClusterID = "custom-cluster-id"
)

var (
	// Single iperf run results in two connections with separate ports (control connection and actual data connection).
	// As 2s is the export active timeout of flow exporter and iperf traffic runs for 12s, we expect totally 12 records
	// exporting to the flow aggregator at time 2s, 4s, 6s, 8s, 10s, and 12s after iperf traffic begins.
	// Since flow aggregator will aggregate records based on 5-tuple connection key and active timeout is 3.5 seconds,
	// we expect 3 records at time 5.5s, 9s, and 12.5s after iperf traffic begins.
	expectedNumDataRecords                      = 3
	podAIPs, podBIPs, podCIPs, podDIPs, podEIPs *PodIPs
	serviceNames                                = []string{"perftest-a", "perftest-b", "perftest-c", "perftest-d", "perftest-e"}
	podNames                                    = serviceNames
	// We use a global variable for this to avoid having to pass it down to all helper functions.
	// It will be initialized the first time setupFlowAggregatorTest is called.
	antreaClusterUUID = ""

	// In the ToExternalFlows test, flow record will arrive 5.5s (exporterActiveFlowExportTimeout+aggregatorActiveFlowRecordTimeout) after executing wget command
	// We set the timeout to 9s (5.5s plus one more aggregatorActiveFlowRecordTimeout) to make the ToExternalFlows test more stable
	getCollectorOutputDefaultTimeout = exporterActiveFlowExportTimeout + 2*aggregatorActiveFlowRecordTimeout
)

type testFlow struct {
	srcIP                   string
	dstIP                   string
	srcPodName              string
	dstPodName              string
	svcIP                   string
	srcNodeInfoNotAvailable bool
}

type flowRecord struct {
	Data string `json:"data"`
}

type IPFIXCollectorResponse struct {
	FlowRecords []flowRecord `json:"flowRecords"`
}

func setupFlowAggregatorTest(t *testing.T, options flowVisibilityTestOptions) (*TestData, bool, bool) {
	teardownFuncs := make([]func(), 0)
	t.Cleanup(func() {
		for _, fn := range teardownFuncs {
			fn()
		}
	})
	data, err := setupTest(t)
	if err != nil {
		t.Fatalf("Error when setting up test: %v", err)
	}
	teardownFuncs = append(teardownFuncs, func() { teardownTest(t, data) })
	// Make sure that antreaClusterUUID is set if this function is called for the first time.
	if antreaClusterUUID == "" {
		if uuid, err := data.getAntreaClusterUUID(10 * time.Second); err != nil {
			t.Fatalf("Error when retrieving Antrea Cluster UUID: %v", err)
		} else {
			antreaClusterUUID = uuid.String()
		}
	}
	if err := setupFlowAggregator(t, data, options); err != nil {
		t.Fatalf("Error when setting up FlowAggregator: %v", err)
	}
	// Execute teardownFlowAggregator later than teardownTest to ensure that the logs of Flow
	// Aggregator has been exported.
	teardownFuncs = append(teardownFuncs, func() { teardownFlowAggregator(t, data) })
	return data, isIPv4Enabled(), isIPv6Enabled()
}

func TestFlowAggregatorSecureConnection(t *testing.T) {
	skipIfNotFlowVisibilityTest(t)
	skipIfHasWindowsNodes(t)

	testCases := []struct {
		flowVisibilityTestOptions
		name string
	}{
		{
			flowVisibilityTestOptions: flowVisibilityTestOptions{
				databaseURL: "tcp://clickhouse-clickhouse.flow-visibility.svc:9000",
			},
			name: "clickhouse-tcp",
		},
		{
			flowVisibilityTestOptions: flowVisibilityTestOptions{
				databaseURL: "http://clickhouse-clickhouse.flow-visibility.svc:8123",
			},
			name: "clickhouse-http",
		},
		{
			flowVisibilityTestOptions: flowVisibilityTestOptions{
				databaseURL:              "tls://clickhouse-clickhouse.flow-visibility.svc:9440",
				databaseSecureConnection: true,
			},
			name: "clickhouse-tls",
		},
		{
			flowVisibilityTestOptions: flowVisibilityTestOptions{
				databaseURL:              "https://clickhouse-clickhouse.flow-visibility.svc:8443",
				databaseSecureConnection: true,
			},
			name: "clickhouse-https",
		},
		{
			flowVisibilityTestOptions: flowVisibilityTestOptions{
				databaseURL: "tcp://clickhouse-clickhouse.flow-visibility.svc:9000",
				ipfixCollector: flowVisibilityIPFIXTestOptions{
					tls: true,
				},
			},
			name: "ipfix-tls",
		},
		{
			flowVisibilityTestOptions: flowVisibilityTestOptions{
				databaseURL: "tcp://clickhouse-clickhouse.flow-visibility.svc:9000",
				ipfixCollector: flowVisibilityIPFIXTestOptions{
					tls:        true,
					clientAuth: true,
				},
			},
			name: "ipfix-mtls",
		},
	}
	for _, o := range testCases {
		t.Run(o.name, func(t *testing.T) {
			var err error
			data, v4Enabled, v6Enabled := setupFlowAggregatorTest(t, o.flowVisibilityTestOptions)
			podAIPs, podBIPs, _, _, _, err = createPerftestPods(data)
			if err != nil {
				t.Fatalf("Error when creating perftest Pods: %v", err)
			}
			if v4Enabled {
				t.Run("IPv4", func(t *testing.T) { checkIntraNodeFlows(t, data, podAIPs, podBIPs, false, "") })
			}
			if v6Enabled {
				t.Run("IPv6", func(t *testing.T) { checkIntraNodeFlows(t, data, podAIPs, podBIPs, true, "") })
			}
		})
	}
}

func TestFlowAggregator(t *testing.T) {
	skipIfNotFlowVisibilityTest(t)
	skipIfHasWindowsNodes(t)

	var err error
	data, v4Enabled, v6Enabled := setupFlowAggregatorTest(t, flowVisibilityTestOptions{
		databaseURL: defaultCHDatabaseURL,
	})
	if err := getAndCheckFlowAggregatorMetrics(t, data, true); err != nil {
		t.Fatalf("Error when checking metrics of Flow Aggregator: %v", err)
	}

	k8sUtils, err = NewKubernetesUtils(data)
	if err != nil {
		t.Fatalf("Error when creating Kubernetes utils client: %v", err)
	}

	podAIPs, podBIPs, podCIPs, podDIPs, podEIPs, err = createPerftestPods(data)
	if err != nil {
		t.Fatalf("Error when creating perftest Pods: %v", err)
	}

	if v4Enabled {
		t.Run("IPv4", func(t *testing.T) { testHelper(t, data, false) })
		t.Run("L7FlowExporterController_IPv4", func(t *testing.T) {
			testL7FlowExporterController(t, data, false)
		})
	}

	if v6Enabled {
		t.Run("IPv6", func(t *testing.T) { testHelper(t, data, true) })
		t.Run("L7FlowExporterController_IPv6", func(t *testing.T) {
			testL7FlowExporterController(t, data, true)
		})
	}

}

func TestFlowAggregatorProxyMode(t *testing.T) {
	skipIfNotFlowVisibilityTest(t)
	skipIfHasWindowsNodes(t)

	runTest := func(t *testing.T, tls bool, clientAuth bool, k8sUIDsInsteadOfNames bool) {
		var err error
		var includeK8sUIDs, includeK8sNames *bool
		if k8sUIDsInsteadOfNames {
			includeK8sUIDs = ptr.To(true)
			includeK8sNames = ptr.To(false)
		}
		data, v4Enabled, v6Enabled := setupFlowAggregatorTest(t, flowVisibilityTestOptions{
			mode:      flowaggregatorconfig.AggregatorModeProxy,
			clusterID: customClusterID,
			ipfixCollector: flowVisibilityIPFIXTestOptions{
				tls:             tls,
				clientAuth:      clientAuth,
				includeK8sUIDs:  includeK8sUIDs,
				includeK8sNames: includeK8sNames,
			},
		})
		require.NoError(t, getAndCheckFlowAggregatorMetrics(t, data, false), "Error when checking metrics of Flow Aggregator")

		// UIDs are only supported when using gRPC between FE and FA.
		if k8sUIDsInsteadOfNames {
			skipIfFlowExportProtocolIsNotGRPC(t, data)
		}

		k8sUtils, err = NewKubernetesUtils(data)
		require.NoError(t, err, "Error when creating Kubernetes utils client")

		podAIPs, _, podCIPs, _, _, err = createPerftestPods(data)
		require.NoError(t, err, "Error when creating perftest Pods")

		if v4Enabled {
			t.Run("IPv4", func(t *testing.T) { testHelperProxyMode(t, data, false, k8sUIDsInsteadOfNames) })
		}

		if v6Enabled {
			t.Run("IPv6", func(t *testing.T) { testHelperProxyMode(t, data, true, k8sUIDsInsteadOfNames) })
		}
	}

	t.Run("plaintext", func(t *testing.T) { runTest(t, false, false, false) })
	t.Run("plaintext-with-K8s-UIDs", func(t *testing.T) { runTest(t, false, false, true) })
	t.Run("TLS", func(t *testing.T) { runTest(t, true, false, false) })
	t.Run("mTLS", func(t *testing.T) { runTest(t, true, true, false) })
}

func getPodUID(t *testing.T, data *TestData, namespace, name string) string {
	pod, err := data.clientset.CoreV1().Pods(namespace).Get(context.TODO(), name, metav1.GetOptions{})
	require.NoError(t, err)
	return string(pod.UID)
}

func k8sUIDAsHexString(uid string) string {
	v := uuid.MustParse(uid)
	return hex.EncodeToString(v[:])
}

func testHelperProxyMode(t *testing.T, data *TestData, isIPv6 bool, k8sUIDsInsteadOfNames bool) {
	label := "Proxy-InterNodeFlows"
	addLabelToTestPods(t, data, label, []string{"perftest-a", "perftest-b"})

	var srcIP, dstIP string
	var cmd []string
	if !isIPv6 {
		srcIP = podAIPs.IPv4.String()
		dstIP = podCIPs.IPv4.String()
		cmd = []string{"iperf3", "-c", dstIP, "-t", "5"}
	} else {
		srcIP = podAIPs.IPv6.String()
		dstIP = podCIPs.IPv6.String()
		cmd = []string{"iperf3", "-6", "-c", dstIP, "-t", "5"}
	}
	stdout, _, err := data.RunCommandFromPod(data.testNamespace, "perftest-a", "iperf", cmd)
	require.NoError(t, err, "Error when running iperf3 client")
	_, srcPort, _ := getBandwidthAndPorts(stdout)

	// should be larger than exporterActiveFlowExportTimeout (+ safety margin).
	const timeout = 10 * time.Second
	records := getCollectorOutput(t, srcIP, dstIP, srcPort, false /* isDstService */, true /* lookForFlowEnd */, isIPv6, data, label /* labelFilter */, timeout)
	require.NotEmpty(t, records)
	record := records[len(records)-1]
	if k8sUIDsInsteadOfNames {
		assert.NotContains(t, record, "sourcePodNamespace: ")
		assert.NotContains(t, record, "destinationPodNamespace: ")
		assert.NotContains(t, record, "sourcePodName: ")
		assert.NotContains(t, record, "destinationPodName: ")
		assert.Contains(t, record, fmt.Sprintf("sourcePodUUID: %s", k8sUIDAsHexString(getPodUID(t, data, data.testNamespace, "perftest-a"))), "Record does not have correct sourcePodUUID")
		assert.Contains(t, record, fmt.Sprintf("destinationPodUUID: %s", k8sUIDAsHexString(getPodUID(t, data, data.testNamespace, "perftest-c"))), "Record does not have correct destinationPodUUID")
		assert.Contains(t, record, "sourceNodeUUID: ", "Record does not have sourceNodeUUID")
		assert.Contains(t, record, "destinationNodeUUID: ", "Record does not have destinationNodeUUID")
	} else {
		assert.Contains(t, record, fmt.Sprintf("sourcePodNamespace: %s", data.testNamespace), "Record does not have correct sourcePodNamespace")
		assert.Contains(t, record, fmt.Sprintf("destinationPodNamespace: %s", data.testNamespace), "Record does not have correct destinationPodNamespace")
		assert.Contains(t, record, fmt.Sprintf("sourcePodName: %s", "perftest-a"), "Record does not have correct sourcePodName")
		assert.Contains(t, record, fmt.Sprintf("destinationPodName: %s", "perftest-c"), "Record does not have correct destinationPodName")
		assert.NotContains(t, record, "sourcePodUUID: ")
		assert.NotContains(t, record, "destinationPodUUID: ")
		assert.NotContains(t, record, "sourceNodeUUID: ")
		assert.NotContains(t, record, "destinationNodeUUID: ")
	}
	// Check the clusterId field, which should match the customClusterID set in the flowVisibilityTestOptions
	assert.Contains(t, record, fmt.Sprintf("clusterId: %s", customClusterID), "Record does not have the correct clusterId")
	assert.Contains(t, record, "originalObservationDomainId", "Record does not have originalObservationDomainId")
	assert.Contains(t, record, "originalExporterIPv4Address", "Record does not have originalExporterIPv4Address")
	assert.Contains(t, record, "originalExporterIPv6Address", "Record does not have originalExporterIPv6Address")
}

func checkIntraNodeFlows(t *testing.T, data *TestData, podAIPs, podBIPs *PodIPs, isIPv6 bool, labelFilter string) {
	deployK8sNetworkPolicies(t, data, "perftest-a", "perftest-b")
	if !isIPv6 {
		checkRecordsForFlows(t, data, podAIPs.IPv4.String(), podBIPs.IPv4.String(), isIPv6, true, false, true, false, labelFilter)
	} else {
		checkRecordsForFlows(t, data, podAIPs.IPv6.String(), podBIPs.IPv6.String(), isIPv6, true, false, true, false, labelFilter)
	}
}

func testHelper(t *testing.T, data *TestData, isIPv6 bool) {
	_, svcB, svcC, svcD, svcE, err := createPerftestServices(data, isIPv6)
	if err != nil {
		t.Fatalf("Error when creating perftest Services: %v", err)
	}
	defer deletePerftestServices(t, data)
	// Wait for the Service to be realized.
	time.Sleep(3 * time.Second)

	// IntraNodeFlows tests the case, where Pods are deployed on same Node
	// and their flow information is exported as IPFIX flow records.
	// K8s network policies are being tested here.
	t.Run("IntraNodeFlows", func(t *testing.T) {
		label := "IntraNodeFlows"
		// As we use the same perftest Pods to generate traffic across all test cases, there's a potential for collecting
		// records from previous subtests. To mitigate this, we add a different label to perftest Pods during each subtest
		// before initiating traffic. This label is then employed as a filter when collecting records from either the
		// ClickHouse or the IPFIX collector Pod.
		addLabelToTestPods(t, data, label, podNames)
		checkIntraNodeFlows(t, data, podAIPs, podBIPs, isIPv6, label)
	})

	// IntraNodeDenyConnIngressANP tests the case, where Pods are deployed on same Node with an Antrea ingress deny policy rule
	// applied to destination Pod (one reject rule, one drop rule) and their flow information is exported as IPFIX flow records.
	// perftest-a -> perftest-b (Ingress reject), perftest-a -> perftest-d (Ingress drop)
	t.Run("IntraNodeDenyConnIngressANP", func(t *testing.T) {
		skipIfAntreaPolicyDisabled(t)
		label := "IntraNodeDenyConnIngressANP"
		addLabelToTestPods(t, data, label, podNames)
		deployDenyAntreaNetworkPolicies(t, data, "perftest-a", "perftest-b", "perftest-d", controlPlaneNodeName(), controlPlaneNodeName(), true)
		testFlow1 := testFlow{
			srcPodName: "perftest-a",
			dstPodName: "perftest-b",
		}
		testFlow2 := testFlow{
			srcPodName: "perftest-a",
			dstPodName: "perftest-d",
		}
		if !isIPv6 {
			testFlow1.srcIP, testFlow1.dstIP, testFlow2.srcIP, testFlow2.dstIP = podAIPs.IPv4.String(), podBIPs.IPv4.String(), podAIPs.IPv4.String(), podDIPs.IPv4.String()
			checkRecordsForDenyFlows(t, data, testFlow1, testFlow2, isIPv6, true, true, false, label)
		} else {
			testFlow1.srcIP, testFlow1.dstIP, testFlow2.srcIP, testFlow2.dstIP = podAIPs.IPv6.String(), podBIPs.IPv6.String(), podAIPs.IPv6.String(), podDIPs.IPv6.String()
			checkRecordsForDenyFlows(t, data, testFlow1, testFlow2, isIPv6, true, true, false, label)
		}
	})

	// IntraNodeDenyConnEgressANP tests the case, where Pods are deployed on same Node with an Antrea egress deny policy rule
	// applied to source Pods (one reject rule, one drop rule) and their flow information is exported as IPFIX flow records.
	// perftest-a (Egress reject) -> perftest-b , perftest-a (Egress drop) -> perftest-d
	t.Run("IntraNodeDenyConnEgressANP", func(t *testing.T) {
		skipIfAntreaPolicyDisabled(t)
		label := "IntraNodeDenyConnEgressANP"
		addLabelToTestPods(t, data, label, podNames)
		deployDenyAntreaNetworkPolicies(t, data, "perftest-a", "perftest-b", "perftest-d", controlPlaneNodeName(), controlPlaneNodeName(), false)
		testFlow1 := testFlow{
			srcPodName: "perftest-a",
			dstPodName: "perftest-b",
		}
		testFlow2 := testFlow{
			srcPodName: "perftest-a",
			dstPodName: "perftest-d",
		}
		if !isIPv6 {
			testFlow1.srcIP, testFlow1.dstIP, testFlow2.srcIP, testFlow2.dstIP = podAIPs.IPv4.String(), podBIPs.IPv4.String(), podAIPs.IPv4.String(), podDIPs.IPv4.String()
			checkRecordsForDenyFlows(t, data, testFlow1, testFlow2, isIPv6, true, true, false, label)
		} else {
			testFlow1.srcIP, testFlow1.dstIP, testFlow2.srcIP, testFlow2.dstIP = podAIPs.IPv6.String(), podBIPs.IPv6.String(), podAIPs.IPv6.String(), podDIPs.IPv6.String()
			checkRecordsForDenyFlows(t, data, testFlow1, testFlow2, isIPv6, true, true, false, label)
		}
	})

	// IntraNodeDenyConnNP tests the case, where Pods are deployed on same Node with an ingress and an egress deny policy rule
	// applied to one destination Pod, one source Pod, respectively and their flow information is exported as IPFIX flow records.
	// perftest-a -> perftest-b (Ingress deny), perftest-d (Egress deny) -> perftest-a
	t.Run("IntraNodeDenyConnNP", func(t *testing.T) {
		skipIfAntreaPolicyDisabled(t)
		label := "IntraNodeDenyConnNP"
		addLabelToTestPods(t, data, label, podNames)
		deployDenyNetworkPolicies(t, data, "perftest-b", "perftest-d", controlPlaneNodeName(), controlPlaneNodeName())
		testFlow1 := testFlow{
			srcPodName: "perftest-a",
			dstPodName: "perftest-b",
		}
		testFlow2 := testFlow{
			srcPodName: "perftest-d",
			dstPodName: "perftest-a",
		}
		if !isIPv6 {
			testFlow1.srcIP, testFlow1.dstIP, testFlow2.srcIP, testFlow2.dstIP = podAIPs.IPv4.String(), podBIPs.IPv4.String(), podDIPs.IPv4.String(), podAIPs.IPv4.String()
			checkRecordsForDenyFlows(t, data, testFlow1, testFlow2, isIPv6, true, false, false, label)
		} else {
			testFlow1.srcIP, testFlow1.dstIP, testFlow2.srcIP, testFlow2.dstIP = podAIPs.IPv6.String(), podBIPs.IPv6.String(), podDIPs.IPv6.String(), podAIPs.IPv6.String()
			checkRecordsForDenyFlows(t, data, testFlow1, testFlow2, isIPv6, true, false, false, label)
		}
	})

	// IntraNodeDenyConnIngressANPThroughSvc tests the case, where Pods are deployed on same Node with an Antrea
	// ingress deny policy rule applied to destination Pod (one reject rule, one drop rule) and their flow information
	// is exported as IPFIX flow records. The test also verify if the service information is well filled in the record.
	// perftest-a -> svcB -> perftest-b (Ingress reject), perftest-a -> svcD ->perftest-d (Ingress drop)
	t.Run("IntraNodeDenyConnIngressANPThroughSvc", func(t *testing.T) {
		skipIfAntreaPolicyDisabled(t)
		label := "IntraNodeDenyConnIngressANPThroughSvc"
		addLabelToTestPods(t, data, label, podNames)
		deployDenyAntreaNetworkPolicies(t, data, "perftest-a", "perftest-b", "perftest-d", controlPlaneNodeName(), controlPlaneNodeName(), true)
		testFlow1 := testFlow{
			srcPodName: "perftest-a",
			dstPodName: "perftest-b",
			svcIP:      svcB.Spec.ClusterIP,
		}
		testFlow2 := testFlow{
			srcPodName: "perftest-a",
			dstPodName: "perftest-d",
			svcIP:      svcD.Spec.ClusterIP,
		}
		if !isIPv6 {
			testFlow1.srcIP, testFlow1.dstIP, testFlow2.srcIP, testFlow2.dstIP = podAIPs.IPv4.String(), podBIPs.IPv4.String(), podAIPs.IPv4.String(), podDIPs.IPv4.String()
			checkRecordsForDenyFlows(t, data, testFlow1, testFlow2, isIPv6, true, true, true, label)
		} else {
			testFlow1.srcIP, testFlow1.dstIP, testFlow2.srcIP, testFlow2.dstIP = podAIPs.IPv6.String(), podBIPs.IPv6.String(), podAIPs.IPv6.String(), podDIPs.IPv6.String()
			checkRecordsForDenyFlows(t, data, testFlow1, testFlow2, isIPv6, true, true, true, label)
		}
	})

	// IntraNodeDenyConnEgressANPThroughSvc tests the case, where Pods are deployed on same Node with an Antrea
	// egress deny policy rule applied to source Pod (one reject rule, one drop rule) and their flow information
	// is exported as IPFIX flow records. The test also verify if the service information is well filled in the record.
	// perftest-a (Egress reject) -> svcB ->perftest-b, perftest-a (Egress drop) -> svcD -> perftest-d
	t.Run("IntraNodeDenyConnEgressANPThroughSvc", func(t *testing.T) {
		skipIfAntreaPolicyDisabled(t)
		label := "IntraNodeDenyConnEgressANPThroughSvc"
		addLabelToTestPods(t, data, label, podNames)
		deployDenyAntreaNetworkPolicies(t, data, "perftest-a", "perftest-b", "perftest-d", controlPlaneNodeName(), controlPlaneNodeName(), false)
		testFlow1 := testFlow{
			srcPodName: "perftest-a",
			dstPodName: "perftest-b",
			svcIP:      svcB.Spec.ClusterIP,
		}
		testFlow2 := testFlow{
			srcPodName: "perftest-a",
			dstPodName: "perftest-d",
			svcIP:      svcD.Spec.ClusterIP,
		}
		if !isIPv6 {
			testFlow1.srcIP, testFlow1.dstIP, testFlow2.srcIP, testFlow2.dstIP = podAIPs.IPv4.String(), podBIPs.IPv4.String(), podAIPs.IPv4.String(), podDIPs.IPv4.String()
			checkRecordsForDenyFlows(t, data, testFlow1, testFlow2, isIPv6, true, true, true, label)
		} else {
			testFlow1.srcIP, testFlow1.dstIP, testFlow2.srcIP, testFlow2.dstIP = podAIPs.IPv6.String(), podBIPs.IPv6.String(), podAIPs.IPv6.String(), podDIPs.IPv6.String()
			checkRecordsForDenyFlows(t, data, testFlow1, testFlow2, isIPv6, true, true, true, label)
		}
	})

	// InterNodeFlows tests the case, where Pods are deployed on different Nodes
	// and their flow information is exported as IPFIX flow records.
	// Antrea network policies are being tested here.
	t.Run("InterNodeFlows", func(t *testing.T) {
		skipIfAntreaPolicyDisabled(t)
		label := "InterNodeFlows"
		addLabelToTestPods(t, data, label, podNames)
		deployAntreaNetworkPolicies(t, data, "perftest-a", "perftest-c", controlPlaneNodeName(), workerNodeName(1))
		if !isIPv6 {
			checkRecordsForFlows(t, data, podAIPs.IPv4.String(), podCIPs.IPv4.String(), isIPv6, false, false, false, true, label)
		} else {
			checkRecordsForFlows(t, data, podAIPs.IPv6.String(), podCIPs.IPv6.String(), isIPv6, false, false, false, true, label)
		}
	})

	// InterNodeDenyConnIngressANP tests the case, where Pods are deployed on different Nodes with an Antrea ingress deny policy rule
	// applied to destination Pod (one reject rule, one drop rule) and their flow information is exported as IPFIX flow records.
	// perftest-a -> perftest-c (Ingress reject), perftest-a -> perftest-e (Ingress drop)
	t.Run("InterNodeDenyConnIngressANP", func(t *testing.T) {
		skipIfAntreaPolicyDisabled(t)
		label := "InterNodeDenyConnIngressANP"
		addLabelToTestPods(t, data, label, podNames)
		deployDenyAntreaNetworkPolicies(t, data, "perftest-a", "perftest-c", "perftest-e", controlPlaneNodeName(), workerNodeName(1), true)
		// For the reject action at the destination Node (i.e., by an ingress policy rule), it is not possible
		// to retrieve information that is exclusive to the source Node (service information / egress policy
		// information) as the connection will be reset and removed from conntrack.
		testFlow1 := testFlow{
			srcPodName:              "perftest-a",
			dstPodName:              "perftest-c",
			srcNodeInfoNotAvailable: true,
		}
		testFlow2 := testFlow{
			srcPodName: "perftest-a",
			dstPodName: "perftest-e",
		}
		if !isIPv6 {
			testFlow1.srcIP, testFlow1.dstIP, testFlow2.srcIP, testFlow2.dstIP = podAIPs.IPv4.String(), podCIPs.IPv4.String(), podAIPs.IPv4.String(), podEIPs.IPv4.String()
			checkRecordsForDenyFlows(t, data, testFlow1, testFlow2, isIPv6, false, true, false, label)
		} else {
			testFlow1.srcIP, testFlow1.dstIP, testFlow2.srcIP, testFlow2.dstIP = podAIPs.IPv6.String(), podCIPs.IPv6.String(), podAIPs.IPv6.String(), podEIPs.IPv6.String()
			checkRecordsForDenyFlows(t, data, testFlow1, testFlow2, isIPv6, false, true, false, label)
		}
	})

	// InterNodeDenyConnEgressANP tests the case, where Pods are deployed on different Nodes with an Antrea egress deny policy rule
	// applied to source Pod (one reject rule, one drop rule) and their flow information is exported as IPFIX flow records.
	// perftest-a (Egress reject) -> perftest-c, perftest-a (Egress drop)-> perftest-e
	t.Run("InterNodeDenyConnEgressANP", func(t *testing.T) {
		skipIfAntreaPolicyDisabled(t)
		label := "InterNodeDenyConnEgressANP"
		addLabelToTestPods(t, data, label, podNames)
		deployDenyAntreaNetworkPolicies(t, data, "perftest-a", "perftest-c", "perftest-e", controlPlaneNodeName(), workerNodeName(1), false)
		testFlow1 := testFlow{
			srcPodName: "perftest-a",
			dstPodName: "perftest-c",
		}
		testFlow2 := testFlow{
			srcPodName: "perftest-a",
			dstPodName: "perftest-e",
		}
		if !isIPv6 {
			testFlow1.srcIP, testFlow1.dstIP, testFlow2.srcIP, testFlow2.dstIP = podAIPs.IPv4.String(), podCIPs.IPv4.String(), podAIPs.IPv4.String(), podEIPs.IPv4.String()
			checkRecordsForDenyFlows(t, data, testFlow1, testFlow2, isIPv6, false, true, false, label)
		} else {
			testFlow1.srcIP, testFlow1.dstIP, testFlow2.srcIP, testFlow2.dstIP = podAIPs.IPv6.String(), podCIPs.IPv6.String(), podAIPs.IPv6.String(), podEIPs.IPv6.String()
			checkRecordsForDenyFlows(t, data, testFlow1, testFlow2, isIPv6, false, true, false, label)
		}
	})

	// InterNodeDenyConnNP tests the case, where Pods are deployed on different Nodes with an ingress and an egress deny policy rule
	// applied to one destination Pod, one source Pod, respectively and their flow information is exported as IPFIX flow records.
	// perftest-a -> perftest-c (Ingress deny), perftest-b (Egress deny) -> perftest-e
	t.Run("InterNodeDenyConnNP", func(t *testing.T) {
		skipIfAntreaPolicyDisabled(t)
		label := "InterNodeDenyConnNP"
		addLabelToTestPods(t, data, label, podNames)
		deployDenyNetworkPolicies(t, data, "perftest-c", "perftest-b", workerNodeName(1), controlPlaneNodeName())
		testFlow1 := testFlow{
			srcPodName: "perftest-a",
			dstPodName: "perftest-c",
		}
		testFlow2 := testFlow{
			srcPodName: "perftest-b",
			dstPodName: "perftest-e",
		}
		if !isIPv6 {
			testFlow1.srcIP, testFlow1.dstIP, testFlow2.srcIP, testFlow2.dstIP = podAIPs.IPv4.String(), podCIPs.IPv4.String(), podBIPs.IPv4.String(), podEIPs.IPv4.String()
			checkRecordsForDenyFlows(t, data, testFlow1, testFlow2, isIPv6, false, false, false, label)
		} else {
			testFlow1.srcIP, testFlow1.dstIP, testFlow2.srcIP, testFlow2.dstIP = podAIPs.IPv6.String(), podCIPs.IPv6.String(), podBIPs.IPv6.String(), podEIPs.IPv6.String()
			checkRecordsForDenyFlows(t, data, testFlow1, testFlow2, isIPv6, false, false, false, label)
		}
	})

	// InterNodeDenyConnIngressANPThroughSvc tests the case, where Pods are deployed on different Node with an Antrea
	// ingress deny policy rule applied to destination Pod (one reject rule, one drop rule) and their flow information
	// is exported as IPFIX flow records. The test also verify if the service information is well filled in the record.
	// perftest-a -> svcC -> perftest-c (Ingress reject), perftest-a -> svcE -> perftest-e (Ingress drop)
	t.Run("InterNodeDenyConnIngressANPThroughSvc", func(t *testing.T) {
		skipIfAntreaPolicyDisabled(t)
		label := "InterNodeDenyConnIngressANPThroughSvc"
		addLabelToTestPods(t, data, label, podNames)
		deployDenyAntreaNetworkPolicies(t, data, "perftest-a", "perftest-c", "perftest-e", controlPlaneNodeName(), workerNodeName(1), true)
		// For the reject action at the destination Node (i.e., by an ingress policy rule), it is not possible
		// to retrieve information that is exclusive to the source Node (service information / egress policy
		// information) as the connection will be reset and removed from conntrack.
		testFlow1 := testFlow{
			srcPodName:              "perftest-a",
			dstPodName:              "perftest-c",
			svcIP:                   svcC.Spec.ClusterIP,
			srcNodeInfoNotAvailable: true,
		}
		testFlow2 := testFlow{
			srcPodName: "perftest-a",
			dstPodName: "perftest-e",
			svcIP:      svcE.Spec.ClusterIP,
		}
		if !isIPv6 {
			testFlow1.srcIP, testFlow1.dstIP, testFlow2.srcIP, testFlow2.dstIP = podAIPs.IPv4.String(), podCIPs.IPv4.String(), podAIPs.IPv4.String(), podEIPs.IPv4.String()
			checkRecordsForDenyFlows(t, data, testFlow1, testFlow2, isIPv6, false, true, true, label)
		} else {
			testFlow1.srcIP, testFlow1.dstIP, testFlow2.srcIP, testFlow2.dstIP = podAIPs.IPv6.String(), podCIPs.IPv6.String(), podAIPs.IPv6.String(), podEIPs.IPv6.String()
			checkRecordsForDenyFlows(t, data, testFlow1, testFlow2, isIPv6, false, true, true, label)
		}
	})

	// InterNodeDenyConnEgressANPThroughSvc tests the case, where Pods are deployed on different Node with an Antrea
	// egress deny policy rule applied to source Pod (one reject rule, one drop rule) and their flow information
	// is exported as IPFIX flow records. The test also verify if the service information is well filled in the record.
	// perftest-a (Egress reject) -> svcC -> perftest-c, perftest-a (Egress drop) -> svcE -> perftest-e
	t.Run("InterNodeDenyConnEgressANPThroughSvc", func(t *testing.T) {
		skipIfAntreaPolicyDisabled(t)
		label := "InterNodeDenyConnEgressANPThroughSvc"
		addLabelToTestPods(t, data, label, podNames)
		deployDenyAntreaNetworkPolicies(t, data, "perftest-a", "perftest-c", "perftest-e", controlPlaneNodeName(), workerNodeName(1), false)
		testFlow1 := testFlow{
			srcPodName: "perftest-a",
			dstPodName: "perftest-c",
			svcIP:      svcC.Spec.ClusterIP,
		}
		testFlow2 := testFlow{
			srcPodName: "perftest-a",
			dstPodName: "perftest-e",
			svcIP:      svcE.Spec.ClusterIP,
		}
		if !isIPv6 {
			testFlow1.srcIP, testFlow1.dstIP, testFlow2.srcIP, testFlow2.dstIP = podAIPs.IPv4.String(), podCIPs.IPv4.String(), podAIPs.IPv4.String(), podEIPs.IPv4.String()
			checkRecordsForDenyFlows(t, data, testFlow1, testFlow2, isIPv6, false, true, true, label)
		} else {
			testFlow1.srcIP, testFlow1.dstIP, testFlow2.srcIP, testFlow2.dstIP = podAIPs.IPv6.String(), podCIPs.IPv6.String(), podAIPs.IPv6.String(), podEIPs.IPv6.String()
			checkRecordsForDenyFlows(t, data, testFlow1, testFlow2, isIPv6, false, true, true, label)
		}
	})

	// Creating test server Pod for ToExternal tests
	serverIPs := createToExternalTestServer(t, data)

	// ToExternalEgressOnSourceNode tests the export of IPFIX flow records when
	// a source Pod sends traffic to an external IP and an Egress is applied on
	// the source Pod. In this case, the Egress Node is the same as the Source Node.
	t.Run("ToExternalEgressOnSourceNode", func(t *testing.T) {
		// Skip the test if Egress doesn't work on the cluster
		// Reference: TestEgress function in test/e2e/egress_test.go
		skipIfNumNodesLessThan(t, 2)
		skipIfEgressDisabled(t)
		skipIfEncapModeIsNot(t, data, config.TrafficEncapModeEncap)

		// Deploy the client Pod on the control-plane node
		clientName, clientIPs, clientCleanupFunc := createAndWaitForPod(t, data, data.createToolboxPodOnNode, "test-client-", nodeName(0), data.testNamespace, false)
		defer clientCleanupFunc()
		label := "ToExternalEgressOnSourceNode"
		addLabelToTestPods(t, data, label, []string{clientName})

		// Create an Egress and the Egress IP is assigned to the Node running the client Pods
		var egressNodeIP, egressNodeName string
		egressNodeName = nodeName(0)
		if !isIPv6 {
			egressNodeIP = nodeIPv4(0)
		} else {
			egressNodeIP = nodeIPv6(0)
		}
		egress := data.createEgress(t, "test-egress", nil, map[string]string{"app": "toolbox"}, "", egressNodeIP, nil)
		egress, err := data.waitForEgressRealized(egress)
		if err != nil {
			t.Fatalf("Error when waiting for Egress to be realized: %v", err)
		}
		t.Logf("Egress %s is realized with Egress IP %s", egress.Name, egressNodeIP)
		defer data.CRDClient.CrdV1beta1().Egresses().Delete(context.TODO(), egress.Name, metav1.DeleteOptions{})
		if !isIPv6 {
			if clientIPs.IPv4 != nil && serverIPs.IPv4 != nil {
				checkRecordsForToExternalFlows(t, data, nodeName(0), clientName, clientIPs.IPv4.String(), serverIPs.IPv4.String(), serverPodPort, isIPv6, egress.Name, egressNodeIP, egressNodeName, label)
			}
		} else {
			if clientIPs.IPv6 != nil && serverIPs.IPv6 != nil {
				checkRecordsForToExternalFlows(t, data, nodeName(0), clientName, clientIPs.IPv6.String(), serverIPs.IPv6.String(), serverPodPort, isIPv6, egress.Name, egressNodeIP, egressNodeName, label)
			}
		}
	})

	// ToExternalEgressOnOtherNode tests the export of IPFIX flow records when
	// a source Pod sends traffic to an external IP and an Egress applied on
	// the source Pod. In this case, the Egress Node is different from the Source Node.
	t.Run("ToExternalEgressOnOtherNode", func(t *testing.T) {
		// Skip the test if Egress doesn't work on the cluster
		// Reference: TestEgress function in test/e2e/egress_test.go
		skipIfNumNodesLessThan(t, 2)
		skipIfEgressDisabled(t)
		skipIfEncapModeIsNot(t, data, config.TrafficEncapModeEncap)

		// Deploy the client Pod on the control-plane node
		clientName, clientIPs, clientCleanupFunc := createAndWaitForPod(t, data, data.createToolboxPodOnNode, "test-client-", nodeName(0), data.testNamespace, false)
		defer clientCleanupFunc()
		label := "ToExternalEgressOnOtherNode"
		addLabelToTestPods(t, data, label, []string{clientName})

		// Create an Egress and the Egress IP is assigned to the Node not running the client Pods
		var egressNodeIP, egressNodeName string
		egressNodeName = nodeName(1)
		if !isIPv6 {
			egressNodeIP = nodeIPv4(1)
		} else {
			egressNodeIP = nodeIPv6(1)
		}
		egress := data.createEgress(t, "test-egress", nil, map[string]string{"app": "toolbox"}, "", egressNodeIP, nil)
		egress, err := data.waitForEgressRealized(egress)
		if err != nil {
			t.Fatalf("Error when waiting for Egress to be realized: %v", err)
		}
		t.Logf("Egress %s is realized with Egress IP %s", egress.Name, egressNodeIP)
		defer data.CRDClient.CrdV1beta1().Egresses().Delete(context.TODO(), egress.Name, metav1.DeleteOptions{})
		if !isIPv6 {
			if clientIPs.IPv4 != nil && serverIPs.IPv4 != nil {
				checkRecordsForToExternalFlows(t, data, nodeName(0), clientName, clientIPs.IPv4.String(), serverIPs.IPv4.String(), serverPodPort, isIPv6, egress.Name, egressNodeIP, egressNodeName, label)
			}
		} else {
			if clientIPs.IPv6 != nil && serverIPs.IPv6 != nil {
				checkRecordsForToExternalFlows(t, data, nodeName(0), clientName, clientIPs.IPv6.String(), serverIPs.IPv6.String(), serverPodPort, isIPv6, egress.Name, egressNodeIP, egressNodeName, label)
			}
		}
	})

	// ToExternalFlows tests the export of IPFIX flow records when a source Pod
	// sends traffic to an external IP
	t.Run("ToExternalFlows", func(t *testing.T) {
		// Deploy the client Pod on the control-plane node
		clientName, clientIPs, clientCleanupFunc := createAndWaitForPod(t, data, data.createToolboxPodOnNode, "test-client-", nodeName(0), data.testNamespace, false)
		defer clientCleanupFunc()
		label := "ToExternalFlows"
		addLabelToTestPods(t, data, label, []string{clientName})
		if !isIPv6 {
			if clientIPs.IPv4 != nil && serverIPs.IPv4 != nil {
				checkRecordsForToExternalFlows(t, data, nodeName(0), clientName, clientIPs.IPv4.String(), serverIPs.IPv4.String(), serverPodPort, isIPv6, "", "", "", label)
			}
		} else {
			if clientIPs.IPv6 != nil && serverIPs.IPv6 != nil {
				checkRecordsForToExternalFlows(t, data, nodeName(0), clientName, clientIPs.IPv6.String(), serverIPs.IPv6.String(), serverPodPort, isIPv6, "", "", "", label)
			}
		}
	})

	// LocalServiceAccess tests the case, where Pod and Service are deployed on the same Node and their flow information is exported as IPFIX flow records.
	t.Run("LocalServiceAccess", func(t *testing.T) {
		skipIfProxyDisabled(t, data)
		label := "LocalServiceAccess"
		addLabelToTestPods(t, data, label, podNames)
		// In dual stack cluster, Service IP can be assigned as different IP family from specified.
		// In that case, source IP and destination IP will align with IP family of Service IP.
		// For IPv4-only and IPv6-only cluster, IP family of Service IP will be same as Pod IPs.
		isServiceIPv6 := net.ParseIP(svcB.Spec.ClusterIP).To4() == nil
		if isServiceIPv6 {
			checkRecordsForFlows(t, data, podAIPs.IPv6.String(), svcB.Spec.ClusterIP, isServiceIPv6, true, true, false, false, label)
		} else {
			checkRecordsForFlows(t, data, podAIPs.IPv4.String(), svcB.Spec.ClusterIP, isServiceIPv6, true, true, false, false, label)
		}
	})

	// RemoteServiceAccess tests the case, where Pod and Service are deployed on different Nodes and their flow information is exported as IPFIX flow records.
	t.Run("RemoteServiceAccess", func(t *testing.T) {
		skipIfProxyDisabled(t, data)
		label := "RemoteServiceAccess"
		addLabelToTestPods(t, data, label, podNames)
		// In dual stack cluster, Service IP can be assigned as different IP family from specified.
		// In that case, source IP and destination IP will align with IP family of Service IP.
		// For IPv4-only and IPv6-only cluster, IP family of Service IP will be same as Pod IPs.
		isServiceIPv6 := net.ParseIP(svcC.Spec.ClusterIP).To4() == nil
		if isServiceIPv6 {
			checkRecordsForFlows(t, data, podAIPs.IPv6.String(), svcC.Spec.ClusterIP, isServiceIPv6, false, true, false, false, label)
		} else {
			checkRecordsForFlows(t, data, podAIPs.IPv4.String(), svcC.Spec.ClusterIP, isServiceIPv6, false, true, false, false, label)
		}
	})

	// Antctl tests ensure antctl is available in a Flow Aggregator Pod
	// and check the output of antctl commands.
	t.Run("Antctl", func(t *testing.T) {
		skipIfNotRequired(t, "mode-irrelevant")
		flowAggPod, err := data.getFlowAggregator()
		if err != nil {
			t.Fatalf("Error when getting flow-aggregator Pod: %v", err)
		}
		podName := flowAggPod.Name
		for _, args := range antctl.CommandList.GetDebugCommands(runtime.ModeFlowAggregator) {
			command := append([]string{"antctl"}, args...)
			t.Logf("Run command: %s", command)

			t.Run(strings.Join(command, " "), func(t *testing.T) {
				stdout, stderr, err := runAntctl(podName, command, data)
				require.NoErrorf(t, err, "Error when running 'antctl %s' from %s: %v\n%s", args, podName, err, antctlOutput(stdout, stderr))
			})
		}
		t.Run("GetFlowRecordsJson", func(t *testing.T) {
			checkAntctlGetFlowRecordsJson(t, data, podName, podAIPs, podBIPs, isIPv6)
		})
	})
}

func checkAntctlGetFlowRecordsJson(t *testing.T, data *TestData, podName string, podAIPs, podBIPs *PodIPs, isIPv6 bool) {
	// A shorter iperfTime that provides stable test results, at which the first record ready in the AggregationProcess but not sent.
	const iperfTimeSecShort = 5
	var cmdStr, srcIP, dstIP string
	// trigger a flow with iperf
	if !isIPv6 {
		srcIP = podAIPs.IPv4.String()
		dstIP = podBIPs.IPv4.String()
		cmdStr = fmt.Sprintf("iperf3 -c %s -t %d", dstIP, iperfTimeSecShort)
	} else {
		srcIP = podAIPs.IPv6.String()
		dstIP = podBIPs.IPv6.String()
		cmdStr = fmt.Sprintf("iperf3 -6 -c %s -t %d", dstIP, iperfTimeSecShort)
	}
	stdout, _, err := data.RunCommandFromPod(data.testNamespace, "perftest-a", "iperf", []string{"bash", "-c", cmdStr})
	require.NoErrorf(t, err, "Error when running iperf3 client: %v", err)
	_, srcPort, dstPort := getBandwidthAndPorts(stdout)

	// run antctl command on flow aggregator to get flow records
	args := []string{"get", "flowrecords", "-o", "json", "--srcip", srcIP, "--srcport", srcPort}
	command := append([]string{"antctl"}, args...)
	t.Logf("Run command: %s", command)
	stdout, stderr, err := runAntctl(podName, command, data)
	require.NoErrorf(t, err, "Error when running 'antctl get flowrecords -o json' from %s: %v\n%s", podName, err, antctlOutput(stdout, stderr))

	var records []map[string]interface{}
	err = json.Unmarshal([]byte(stdout), &records)
	require.NoErrorf(t, err, "Error when parsing flow records from antctl: %v", err)
	require.Len(t, records, 1)

	checkAntctlRecord(t, records[0], srcIP, dstIP, srcPort, dstPort, isIPv6, data.testNamespace)
}

func checkAntctlRecord(t *testing.T, record map[string]interface{}, srcIP, dstIP, srcPort, dstPort string, isIPv6 bool, namespace string) {
	assert := assert.New(t)
	if isIPv6 {
		assert.Equal(srcIP, record["sourceIPv6Address"], "The record from antctl does not have correct sourceIPv6Address")
		assert.Equal(dstIP, record["destinationIPv6Address"], "The record from antctl does not have correct destinationIPv6Address")
	} else {
		assert.Equal(srcIP, record["sourceIPv4Address"], "The record from antctl does not have correct sourceIPv4Address")
		assert.Equal(dstIP, record["destinationIPv4Address"], "The record from antctl does not have correct destinationIPv4Address")
	}
	srcPortNum, err := strconv.Atoi(srcPort)
	require.NoErrorf(t, err, "error when converting the iperf srcPort to int type: %s", srcPort)
	assert.EqualValues(srcPortNum, record["sourceTransportPort"], "The record from antctl does not have correct sourceTransportPort")
	assert.Equal("perftest-a", record["sourcePodName"], "The record from antctl does not have correct sourcePodName")
	assert.Equal(namespace, record["sourcePodNamespace"], "The record from antctl does not have correct sourcePodNamespace")
	assert.Equal(controlPlaneNodeName(), record["sourceNodeName"], "The record from antctl does not have correct sourceNodeName")

	dstPortNum, err := strconv.Atoi(dstPort)
	require.NoErrorf(t, err, "error when converting the iperf dstPort to int type: %s", dstPort)
	assert.EqualValues(dstPortNum, record["destinationTransportPort"], "The record from antctl does not have correct destinationTransportPort")
	assert.Equal("perftest-b", record["destinationPodName"], "The record from antctl does not have correct destinationPodName")
	assert.Equal(namespace, record["destinationPodNamespace"], "The record from antctl does not have correct destinationPodNamespace")
	assert.Equal(controlPlaneNodeName(), record["destinationNodeName"], "The record from antctl does not have correct destinationNodeName")

	assert.EqualValues(ipfixregistry.FlowTypeIntraNode, record["flowType"], "The record from antctl does not have correct flowType")
	assert.EqualValues(protocolIdentifierTCP, record["protocolIdentifier"], "The record from antctl does not have correct protocolIdentifier")
}

func checkRecordsForFlows(t *testing.T, data *TestData, srcIP string, dstIP string, isIPv6 bool, isIntraNode bool, checkService bool, checkK8sNetworkPolicy bool, checkAntreaNetworkPolicy bool, labelFilter string) {
	var cmdStr string
	if !isIPv6 {
		cmdStr = fmt.Sprintf("iperf3 -c %s -t %d -b %s", dstIP, iperfTimeSec, iperfBandwidth)
	} else {
		cmdStr = fmt.Sprintf("iperf3 -6 -c %s -t %d -b %s", dstIP, iperfTimeSec, iperfBandwidth)
	}
	if checkService {
		cmdStr += fmt.Sprintf(" -p %d", iperfSvcPort)
	}
	stdout, _, err := data.RunCommandFromPod(data.testNamespace, "perftest-a", "iperf", []string{"bash", "-c", cmdStr})
	require.NoErrorf(t, err, "Error when running iperf3 client: %v", err)
	bwSlice, srcPort, _ := getBandwidthAndPorts(stdout)
	require.Equal(t, 2, len(bwSlice), "bandwidth value and / or bandwidth unit are not available")
	// bandwidth from iperf output
	bandwidthInFloat, err := strconv.ParseFloat(bwSlice[0], 64)
	require.NoErrorf(t, err, "Error when converting iperf bandwidth %s to float64 type", bwSlice[0])
	var bandwidthInMbps float64
	if strings.Contains(bwSlice[1], "Mbits") {
		bandwidthInMbps = bandwidthInFloat
	} else {
		t.Fatalf("Unit of the traffic bandwidth reported by iperf should be Mbits.")
	}

	checkRecordsForFlowsCollector(t, data, srcIP, dstIP, srcPort, isIPv6, isIntraNode, checkService, checkK8sNetworkPolicy, checkAntreaNetworkPolicy, bandwidthInMbps, labelFilter)
	checkRecordsForFlowsClickHouse(t, data, srcIP, dstIP, srcPort, isIntraNode, checkService, checkK8sNetworkPolicy, checkAntreaNetworkPolicy, bandwidthInMbps, labelFilter)
}

func checkRecordsForFlowsCollector(t *testing.T, data *TestData, srcIP, dstIP, srcPort string, isIPv6, isIntraNode, checkService, checkK8sNetworkPolicy, checkAntreaNetworkPolicy bool, bandwidthInMbps float64, labelFilter string) {
	records := getCollectorOutput(t, srcIP, dstIP, srcPort, checkService, true, isIPv6, data, labelFilter, getCollectorOutputDefaultTimeout)
	// Checking only data records as data records cannot be decoded without template
	// record.
	assert.GreaterOrEqualf(t, len(records), expectedNumDataRecords, "IPFIX collector should receive expected number of flow records, filtered records: %v", records)
	// Iterate over recordSlices and build some results to test with expected results
	for _, record := range records {
		// Check if record has both Pod name of source and destination Pod.
		if isIntraNode {
			checkPodAndNodeData(t, record, "perftest-a", controlPlaneNodeName(), "perftest-b", controlPlaneNodeName(), data.testNamespace)
			checkFlowType(t, record, ipfixregistry.FlowTypeIntraNode)
		} else {
			checkPodAndNodeData(t, record, "perftest-a", controlPlaneNodeName(), "perftest-c", workerNodeName(1), data.testNamespace)
			checkFlowType(t, record, ipfixregistry.FlowTypeInterNode)
		}
		assert := assert.New(t)
		// Check the clusterId field, which should match the ClusterUUID generated by the Antrea Controller.
		assert.Contains(record, fmt.Sprintf("clusterId: %s", antreaClusterUUID), "Record does not have the correct clusterId")
		if checkService {
			if isIntraNode {
				assert.Contains(record, data.testNamespace+"/perftest-b", "Record with ServiceIP does not have Service name")
			} else {
				assert.Contains(record, data.testNamespace+"/perftest-c", "Record with ServiceIP does not have Service name")
			}
		}
		if checkK8sNetworkPolicy {
			// Check if records have both ingress and egress network policies.
			assert.Contains(record, fmt.Sprintf("ingressNetworkPolicyName: %s", ingressAllowNetworkPolicyName), "Record does not have the correct NetworkPolicy name with the ingress rule")
			assert.Contains(record, fmt.Sprintf("ingressNetworkPolicyNamespace: %s", data.testNamespace), "Record does not have the correct NetworkPolicy Namespace with the ingress rule")
			assert.Contains(record, fmt.Sprintf("ingressNetworkPolicyType: %d", ipfixregistry.PolicyTypeK8sNetworkPolicy), "Record does not have the correct NetworkPolicy Type with the ingress rule")
			assert.Contains(record, fmt.Sprintf("egressNetworkPolicyName: %s", egressAllowNetworkPolicyName), "Record does not have the correct NetworkPolicy name with the egress rule")
			assert.Contains(record, fmt.Sprintf("egressNetworkPolicyNamespace: %s", data.testNamespace), "Record does not have the correct NetworkPolicy Namespace with the egress rule")
			assert.Contains(record, fmt.Sprintf("egressNetworkPolicyType: %d", ipfixregistry.PolicyTypeK8sNetworkPolicy), "Record does not have the correct NetworkPolicy Type with the egress rule")
		}
		if checkAntreaNetworkPolicy {
			// Check if records have both ingress and egress network policies.
			assert.Contains(record, fmt.Sprintf("ingressNetworkPolicyName: %s", ingressAntreaNetworkPolicyName), "Record does not have the correct NetworkPolicy name with the ingress rule")
			assert.Contains(record, fmt.Sprintf("ingressNetworkPolicyNamespace: %s", data.testNamespace), "Record does not have the correct NetworkPolicy Namespace with the ingress rule")
			assert.Contains(record, fmt.Sprintf("ingressNetworkPolicyType: %d", ipfixregistry.PolicyTypeAntreaNetworkPolicy), "Record does not have the correct NetworkPolicy Type with the ingress rule")
			assert.Contains(record, fmt.Sprintf("ingressNetworkPolicyRuleName: %s", testIngressRuleName), "Record does not have the correct NetworkPolicy RuleName with the ingress rule")
			assert.Contains(record, fmt.Sprintf("ingressNetworkPolicyRuleAction: %d", ipfixregistry.NetworkPolicyRuleActionAllow), "Record does not have the correct NetworkPolicy RuleAction with the ingress rule")
			assert.Contains(record, fmt.Sprintf("egressNetworkPolicyName: %s", egressAntreaNetworkPolicyName), "Record does not have the correct NetworkPolicy name with the egress rule")
			assert.Contains(record, fmt.Sprintf("egressNetworkPolicyNamespace: %s", data.testNamespace), "Record does not have the correct NetworkPolicy Namespace with the egress rule")
			assert.Contains(record, fmt.Sprintf("egressNetworkPolicyType: %d", ipfixregistry.PolicyTypeAntreaNetworkPolicy), "Record does not have the correct NetworkPolicy Type with the egress rule")
			assert.Contains(record, fmt.Sprintf("egressNetworkPolicyRuleName: %s", testEgressRuleName), "Record does not have the correct NetworkPolicy RuleName with the egress rule")
			assert.Contains(record, fmt.Sprintf("egressNetworkPolicyRuleAction: %d", ipfixregistry.NetworkPolicyRuleActionAllow), "Record does not have the correct NetworkPolicy RuleAction with the egress rule")
		}

		// Skip the bandwidth check for the iperf control flow records which have 0 throughput.
		if !strings.Contains(record, "throughput: 0") {
			flowStartTime := int64(getUint64FieldFromRecord(t, record, "flowStartSeconds"))
			exportTime := int64(getUint64FieldFromRecord(t, record, "flowEndSeconds"))
			flowEndReason := int64(getUint64FieldFromRecord(t, record, "flowEndReason"))
			var recBandwidth float64
			// flowEndReason == 3 means the end of flow detected
			if flowEndReason == 3 {
				// Check average bandwidth on the last record.
				octetTotalCount := getUint64FieldFromRecord(t, record, "octetTotalCount")
				recBandwidth = float64(octetTotalCount) * 8 / float64(iperfTimeSec) / 1000000
			} else {
				// Check bandwidth with the field "throughput" except for the last record,
				// as their throughput may be significantly lower than the average Iperf throughput.
				throughput := getUint64FieldFromRecord(t, record, "throughput")
				recBandwidth = float64(throughput) / 1000000
			}
			t.Logf("Throughput check on record with flowEndSeconds-flowStartSeconds: %v, Iperf throughput: %.2f Mbits/s, IPFIX record throughput: %.2f Mbits/s", exportTime-flowStartTime, bandwidthInMbps, recBandwidth)
			assert.InDeltaf(recBandwidth, bandwidthInMbps, bandwidthInMbps*0.15, "Difference between Iperf bandwidth and IPFIX record bandwidth should be lower than 15%%, record: %s", record)
		}
	}
}

func checkRecordsForFlowsClickHouse(t *testing.T, data *TestData, srcIP, dstIP, srcPort string, isIntraNode, checkService, checkK8sNetworkPolicy, checkAntreaNetworkPolicy bool, bandwidthInMbps float64, labelFilter string) {
	// Check the source port along with source and destination IPs as there
	// are flow records for control flows during the iperf with same IPs
	// and destination port.
	clickHouseRecords := getClickHouseOutput(t, data, srcIP, dstIP, srcPort, checkService, true, labelFilter)

	for _, record := range clickHouseRecords {
		// Check if record has both Pod name of source and destination Pod.
		if isIntraNode {
			checkPodAndNodeDataClickHouse(data, t, record, "perftest-a", controlPlaneNodeName(), "perftest-b", controlPlaneNodeName())
			checkFlowTypeClickHouse(t, record, ipfixregistry.FlowTypeIntraNode)
		} else {
			checkPodAndNodeDataClickHouse(data, t, record, "perftest-a", controlPlaneNodeName(), "perftest-c", workerNodeName(1))
			checkFlowTypeClickHouse(t, record, ipfixregistry.FlowTypeInterNode)
		}
		assert := assert.New(t)
		if checkService {
			if isIntraNode {
				assert.Contains(record.DestinationServicePortName, data.testNamespace+"/perftest-b", "Record with ServiceIP does not have Service name")
			} else {
				assert.Contains(record.DestinationServicePortName, data.testNamespace+"/perftest-c", "Record with ServiceIP does not have Service name")
			}
		}
		if checkK8sNetworkPolicy {
			// Check if records have both ingress and egress network policies.
			assert.Equal(record.IngressNetworkPolicyName, ingressAllowNetworkPolicyName, "Record does not have the correct NetworkPolicy name with the ingress rule")
			assert.Equal(record.IngressNetworkPolicyNamespace, data.testNamespace, "Record does not have the correct NetworkPolicy Namespace with the ingress rule")
			assert.Equal(record.IngressNetworkPolicyType, ipfixregistry.PolicyTypeK8sNetworkPolicy, "Record does not have the correct NetworkPolicy Type with the ingress rule")
			assert.Equal(record.EgressNetworkPolicyName, egressAllowNetworkPolicyName, "Record does not have the correct NetworkPolicy name with the egress rule")
			assert.Equal(record.EgressNetworkPolicyNamespace, data.testNamespace, "Record does not have the correct NetworkPolicy Namespace with the egress rule")
			assert.Equal(record.EgressNetworkPolicyType, ipfixregistry.PolicyTypeK8sNetworkPolicy, "Record does not have the correct NetworkPolicy Type with the egress rule")
		}
		if checkAntreaNetworkPolicy {
			// Check if records have both ingress and egress network policies.
			assert.Equal(record.IngressNetworkPolicyName, ingressAntreaNetworkPolicyName, "Record does not have the correct NetworkPolicy name with the ingress rule")
			assert.Equal(record.IngressNetworkPolicyNamespace, data.testNamespace, "Record does not have the correct NetworkPolicy Namespace with the ingress rule")
			assert.Equal(record.IngressNetworkPolicyType, ipfixregistry.PolicyTypeAntreaNetworkPolicy, "Record does not have the correct NetworkPolicy Type with the ingress rule")
			assert.Equal(record.IngressNetworkPolicyRuleName, testIngressRuleName, "Record does not have the correct NetworkPolicy RuleName with the ingress rule")
			assert.Equal(record.IngressNetworkPolicyRuleAction, ipfixregistry.NetworkPolicyRuleActionAllow, "Record does not have the correct NetworkPolicy RuleAction with the ingress rule")
			assert.Equal(record.EgressNetworkPolicyName, egressAntreaNetworkPolicyName, "Record does not have the correct NetworkPolicy name with the egress rule")
			assert.Equal(record.EgressNetworkPolicyNamespace, data.testNamespace, "Record does not have the correct NetworkPolicy Namespace with the egress rule")
			assert.Equal(record.EgressNetworkPolicyType, ipfixregistry.PolicyTypeAntreaNetworkPolicy, "Record does not have the correct NetworkPolicy Type with the egress rule")
			assert.Equal(record.EgressNetworkPolicyRuleName, testEgressRuleName, "Record does not have the correct NetworkPolicy RuleName with the egress rule")
			assert.Equal(record.EgressNetworkPolicyRuleAction, ipfixregistry.NetworkPolicyRuleActionAllow, "Record does not have the correct NetworkPolicy RuleAction with the egress rule")
		}

		// Skip the bandwidth check for the iperf control flow records which have 0 throughput.
		if record.Throughput > 0 {
			flowStartTime := record.FlowStartSeconds.Unix()
			exportTime := record.FlowEndSeconds.Unix()
			var recBandwidth float64
			// flowEndReason == 3 means the end of flow detected
			if record.FlowEndReason == 3 {
				octetTotalCount := record.OctetTotalCount
				recBandwidth = float64(octetTotalCount) * 8 / float64(exportTime-flowStartTime) / 1000000
			} else {
				// Check bandwidth with the field "throughput" except for the last record,
				// as their throughput may be significantly lower than the average Iperf throughput.
				throughput := record.Throughput
				recBandwidth = float64(throughput) / 1000000
			}
			t.Logf("Throughput check on record with flowEndSeconds-flowStartSeconds: %v, Iperf throughput: %.2f Mbits/s, ClickHouse record throughput: %.2f Mbits/s", exportTime-flowStartTime, bandwidthInMbps, recBandwidth)
			assert.InDeltaf(recBandwidth, bandwidthInMbps, bandwidthInMbps*0.15, "Difference between Iperf bandwidth and ClickHouse record bandwidth should be lower than 15%%, record: %v", record)
		}

	}
	// Checking only data records as data records cannot be decoded without template record.
	assert.GreaterOrEqualf(t, len(clickHouseRecords), expectedNumDataRecords, "ClickHouse should receive expected number of flow records. Considered records: %s", clickHouseRecords)
}

func checkRecordsForToExternalFlows(t *testing.T, data *TestData, srcNodeName string, srcPodName string, srcIP string, dstIP string, dstPort int32, isIPv6 bool, egressName, egressIP, egressNodeName, labelFilter string) {
	var cmd string
	if !isIPv6 {
		cmd = fmt.Sprintf("wget -O- %s:%d", dstIP, dstPort)
	} else {
		cmd = fmt.Sprintf("wget -O- http://[%s]:%d", dstIP, dstPort)
	}
	stdout, stderr, err := data.RunCommandFromPod(data.testNamespace, srcPodName, toolboxContainerName, strings.Fields(cmd))
	require.NoErrorf(t, err, "Error when running wget command, stdout: %s, stderr: %s", stdout, stderr)
	records := getCollectorOutput(t, srcIP, dstIP, "", false, false, isIPv6, data, labelFilter, getCollectorOutputDefaultTimeout)
	for _, record := range records {
		checkPodAndNodeData(t, record, srcPodName, srcNodeName, "", "", data.testNamespace)
		checkFlowType(t, record, ipfixregistry.FlowTypeToExternal)
		if egressName != "" {
			checkEgressInfo(t, record, egressName, egressIP, egressNodeName)
		}
	}

	clickHouseRecords := getClickHouseOutput(t, data, srcIP, dstIP, "", false, false, labelFilter)
	for _, record := range clickHouseRecords {
		checkPodAndNodeDataClickHouse(data, t, record, srcPodName, srcNodeName, "", "")
		checkFlowTypeClickHouse(t, record, ipfixregistry.FlowTypeToExternal)
		if egressName != "" {
			checkEgressInfoClickHouse(t, record, egressName, egressIP, egressNodeName)
		}
	}
}

func checkRecordsForDenyFlows(t *testing.T, data *TestData, testFlow1, testFlow2 testFlow, isIPv6, isIntraNode, isANP, useSvcIP bool, labelFilter string) {
	var cmdStr1, cmdStr2 string
	if !isIPv6 {
		if useSvcIP {
			cmdStr1 = fmt.Sprintf("iperf3 -c %s -p %d -n 1", testFlow1.svcIP, iperfSvcPort)
			cmdStr2 = fmt.Sprintf("iperf3 -c %s -p %d -n 1", testFlow2.svcIP, iperfSvcPort)
		} else {
			cmdStr1 = fmt.Sprintf("iperf3 -c %s -n 1", testFlow1.dstIP)
			cmdStr2 = fmt.Sprintf("iperf3 -c %s -n 1", testFlow2.dstIP)
		}

	} else {
		if useSvcIP {
			cmdStr1 = fmt.Sprintf("iperf3 -6 -c %s -p %d -n 1", testFlow1.svcIP, iperfSvcPort)
			cmdStr2 = fmt.Sprintf("iperf3 -6 -c %s -p %d -n 1", testFlow2.svcIP, iperfSvcPort)
		} else {
			cmdStr1 = fmt.Sprintf("iperf3 -6 -c %s -n 1", testFlow1.dstIP)
			cmdStr2 = fmt.Sprintf("iperf3 -6 -c %s -n 1", testFlow2.dstIP)
		}
	}
	_, _, err := data.RunCommandFromPod(data.testNamespace, testFlow1.srcPodName, "", []string{"timeout", "2", "bash", "-c", cmdStr1})
	assert.Error(t, err)
	_, _, err = data.RunCommandFromPod(data.testNamespace, testFlow2.srcPodName, "", []string{"timeout", "2", "bash", "-c", cmdStr2})
	assert.Error(t, err)

	checkRecordsForDenyFlowsCollector(t, data, testFlow1, testFlow2, isIPv6, isIntraNode, isANP, labelFilter)
	checkRecordsForDenyFlowsClickHouse(t, data, testFlow1, testFlow2, isIPv6, isIntraNode, isANP, labelFilter)
}

func checkRecordsForDenyFlowsCollector(t *testing.T, data *TestData, testFlow1, testFlow2 testFlow, isIPv6, isIntraNode, isANP bool, labelFilter string) {
	records1 := getCollectorOutput(t, testFlow1.srcIP, testFlow1.dstIP, "", false, false, isIPv6, data, labelFilter, getCollectorOutputDefaultTimeout)
	records2 := getCollectorOutput(t, testFlow2.srcIP, testFlow2.dstIP, "", false, false, isIPv6, data, labelFilter, getCollectorOutputDefaultTimeout)
	records := append(records1, records2...)
	src_flow1, dst_flow1 := matchSrcAndDstAddress(testFlow1.srcIP, testFlow1.dstIP, false, isIPv6)
	src_flow2, dst_flow2 := matchSrcAndDstAddress(testFlow2.srcIP, testFlow2.dstIP, false, isIPv6)
	// Iterate over records and build some results to test with expected results
	for _, record := range records {
		var srcPodName, dstPodName string
		var svcIP string
		var checkSrcNodeInfo bool
		if strings.Contains(record, src_flow1) && strings.Contains(record, dst_flow1) {
			srcPodName = testFlow1.srcPodName
			dstPodName = testFlow1.dstPodName
			svcIP = testFlow1.svcIP
			checkSrcNodeInfo = !testFlow1.srcNodeInfoNotAvailable
		} else if strings.Contains(record, src_flow2) && strings.Contains(record, dst_flow2) {
			srcPodName = testFlow2.srcPodName
			dstPodName = testFlow2.dstPodName
			svcIP = testFlow2.svcIP
			checkSrcNodeInfo = !testFlow2.srcNodeInfoNotAvailable
		}
		require.True(t, !isIntraNode || checkSrcNodeInfo, "source Node info should always be available for intra-Node flows")

		if strings.Contains(record, src_flow1) && strings.Contains(record, dst_flow1) || strings.Contains(record, src_flow2) && strings.Contains(record, dst_flow2) {
			ingressRejectStr := fmt.Sprintf("ingressNetworkPolicyRuleAction: %d", ipfixregistry.NetworkPolicyRuleActionReject)
			ingressDropStr := fmt.Sprintf("ingressNetworkPolicyRuleAction: %d", ipfixregistry.NetworkPolicyRuleActionDrop)
			egressRejectStr := fmt.Sprintf("egressNetworkPolicyRuleAction: %d", ipfixregistry.NetworkPolicyRuleActionReject)
			egressDropStr := fmt.Sprintf("egressNetworkPolicyRuleAction: %d", ipfixregistry.NetworkPolicyRuleActionDrop)

			if isIntraNode {
				checkPodAndNodeData(t, record, srcPodName, controlPlaneNodeName(), dstPodName, controlPlaneNodeName(), data.testNamespace)
				checkFlowType(t, record, ipfixregistry.FlowTypeIntraNode)
			} else {
				checkPodAndNodeData(t, record, srcPodName, controlPlaneNodeName(), dstPodName, workerNodeName(1), data.testNamespace)
				checkFlowType(t, record, ipfixregistry.FlowTypeInterNode)
			}
			assert := assert.New(t)
			if !isANP { // K8s Network Policies
				if strings.Contains(record, ingressDropStr) && !strings.Contains(record, ingressDropANPName) {
					assert.Contains(record, testFlow1.dstIP)
				} else if strings.Contains(record, egressDropStr) && !strings.Contains(record, egressDropANPName) {
					assert.Contains(record, testFlow2.dstIP)
				}
			} else { // Antrea Network Policies
				if strings.Contains(record, ingressRejectStr) {
					assert.Contains(record, ingressRejectANPName, "Record does not have Antrea NetworkPolicy name with ingress reject rule")
					assert.Contains(record, fmt.Sprintf("ingressNetworkPolicyNamespace: %s", data.testNamespace), "Record does not have correct ingressNetworkPolicyNamespace")
					assert.Contains(record, fmt.Sprintf("ingressNetworkPolicyType: %d", ipfixregistry.PolicyTypeAntreaNetworkPolicy), "Record does not have the correct NetworkPolicy Type with the ingress reject rule")
					assert.Contains(record, fmt.Sprintf("ingressNetworkPolicyRuleName: %s", testIngressRuleName), "Record does not have the correct NetworkPolicy RuleName with the ingress reject rule")
				} else if strings.Contains(record, ingressDropStr) {
					assert.Contains(record, ingressDropANPName, "Record does not have Antrea NetworkPolicy name with ingress drop rule")
					assert.Contains(record, fmt.Sprintf("ingressNetworkPolicyNamespace: %s", data.testNamespace), "Record does not have correct ingressNetworkPolicyNamespace")
					assert.Contains(record, fmt.Sprintf("ingressNetworkPolicyType: %d", ipfixregistry.PolicyTypeAntreaNetworkPolicy), "Record does not have the correct NetworkPolicy Type with the ingress drop rule")
					assert.Contains(record, fmt.Sprintf("ingressNetworkPolicyRuleName: %s", testIngressRuleName), "Record does not have the correct NetworkPolicy RuleName with the ingress drop rule")
				} else if strings.Contains(record, egressRejectStr) {
					assert.Contains(record, egressRejectANPName, "Record does not have Antrea NetworkPolicy name with egress reject rule")
					assert.Contains(record, fmt.Sprintf("egressNetworkPolicyNamespace: %s", data.testNamespace), "Record does not have correct egressNetworkPolicyNamespace")
					assert.Contains(record, fmt.Sprintf("egressNetworkPolicyType: %d", ipfixregistry.PolicyTypeAntreaNetworkPolicy), "Record does not have the correct NetworkPolicy Type with the egress reject rule")
					assert.Contains(record, fmt.Sprintf("egressNetworkPolicyRuleName: %s", testEgressRuleName), "Record does not have the correct NetworkPolicy RuleName with the egress reject rule")
				} else if strings.Contains(record, egressDropStr) {
					assert.Contains(record, egressDropANPName, "Record does not have Antrea NetworkPolicy name with egress drop rule")
					assert.Contains(record, fmt.Sprintf("egressNetworkPolicyNamespace: %s", data.testNamespace), "Record does not have correct egressNetworkPolicyNamespace")
					assert.Contains(record, fmt.Sprintf("egressNetworkPolicyType: %d", ipfixregistry.PolicyTypeAntreaNetworkPolicy), "Record does not have the correct NetworkPolicy Type with the egress drop rule")
					assert.Contains(record, fmt.Sprintf("egressNetworkPolicyRuleName: %s", testEgressRuleName), "Record does not have the correct NetworkPolicy RuleName with the egress drop rule")
				}
				if checkSrcNodeInfo && (strings.Contains(record, ingressRejectStr) || strings.Contains(record, ingressDropStr)) {
					// For ingress deny rules, we also define an egress allow policy rule to ensure
					// that the information is reported correctly.
					assert.Contains(record, egressAllowANPName, "Record does not have Antrea NetworkPolicy name with egress allow rule")
					assert.Contains(record, fmt.Sprintf("egressNetworkPolicyType: %d", ipfixregistry.PolicyTypeAntreaNetworkPolicy), "Record does not have the correct NetworkPolicy Type with the egress allow rule")
				}
			}
			if checkSrcNodeInfo && svcIP != "" {
				destinationServicePortName := data.testNamespace + "/" + dstPodName
				assert.Contains(record, fmt.Sprintf("destinationServicePortName: %s", destinationServicePortName), "Record does not have correct destinationServicePortName")
				assert.Contains(record, fmt.Sprintf("destinationServicePort: %d", iperfSvcPort), "Record does not have correct destinationServicePort")
			} else {
				assert.Contains(record, "destinationServicePortName:  \n", "Record does not have correct destinationServicePortName")
				assert.Contains(record, "destinationServicePort: 0 \n", "Record does not have correct destinationServicePort")
			}
		}
	}
}

func checkRecordsForDenyFlowsClickHouse(t *testing.T, data *TestData, testFlow1, testFlow2 testFlow, isIPv6, isIntraNode, isANP bool, labelFilter string) {
	clickHouseRecords1 := getClickHouseOutput(t, data, testFlow1.srcIP, testFlow1.dstIP, "", false, false, labelFilter)
	clickHouseRecords2 := getClickHouseOutput(t, data, testFlow2.srcIP, testFlow2.dstIP, "", false, false, labelFilter)
	records := append(clickHouseRecords1, clickHouseRecords2...)
	// Iterate over records and build some results to test with expected results
	for _, record := range records {
		var srcPodName, dstPodName string
		var svcIP string
		var checkSrcNodeInfo bool
		if record.SourceIP == testFlow1.srcIP && (record.DestinationIP == testFlow1.dstIP || record.DestinationClusterIP == testFlow1.dstIP) {
			srcPodName = testFlow1.srcPodName
			dstPodName = testFlow1.dstPodName
			svcIP = testFlow1.svcIP
			checkSrcNodeInfo = !testFlow1.srcNodeInfoNotAvailable
		} else if record.SourceIP == testFlow2.srcIP && (record.DestinationIP == testFlow2.dstIP || record.DestinationClusterIP == testFlow2.dstIP) {
			srcPodName = testFlow2.srcPodName
			dstPodName = testFlow2.dstPodName
			svcIP = testFlow2.svcIP
			checkSrcNodeInfo = !testFlow2.srcNodeInfoNotAvailable
		}
		require.True(t, !isIntraNode || checkSrcNodeInfo, "checkSrcNodeInfo should always be true for intra-Node flows")

		if isIntraNode {
			checkPodAndNodeDataClickHouse(data, t, record, srcPodName, controlPlaneNodeName(), dstPodName, controlPlaneNodeName())
			checkFlowTypeClickHouse(t, record, ipfixregistry.FlowTypeIntraNode)
		} else {
			checkPodAndNodeDataClickHouse(data, t, record, srcPodName, controlPlaneNodeName(), dstPodName, workerNodeName(1))
			checkFlowTypeClickHouse(t, record, ipfixregistry.FlowTypeInterNode)
		}
		if checkSrcNodeInfo && svcIP != "" {
			destinationServicePortName := data.testNamespace + "/" + dstPodName
			assert.Contains(t, record.DestinationServicePortName, destinationServicePortName)
			assert.Equal(t, iperfSvcPort, int(record.DestinationServicePort))
		} else {
			assert.Equal(t, "", record.DestinationServicePortName)
			assert.Equal(t, 0, int(record.DestinationServicePort))
		}
		assert := assert.New(t)
		if !isANP { // K8s Network Policies
			if (record.IngressNetworkPolicyRuleAction == ipfixregistry.NetworkPolicyRuleActionDrop) && (record.IngressNetworkPolicyName != ingressDropANPName) {
				assert.Equal(record.DestinationIP, testFlow1.dstIP)
			} else if (record.EgressNetworkPolicyRuleAction == ipfixregistry.NetworkPolicyRuleActionDrop) && (record.EgressNetworkPolicyName != egressDropANPName) {
				assert.Equal(record.DestinationIP, testFlow2.dstIP)
			}
		} else { // Antrea Network Policies
			if record.IngressNetworkPolicyRuleAction == ipfixregistry.NetworkPolicyRuleActionReject {
				assert.Equal(record.IngressNetworkPolicyName, ingressRejectANPName, "Record does not have Antrea NetworkPolicy name with ingress reject rule")
				assert.Equal(record.IngressNetworkPolicyNamespace, data.testNamespace, "Record does not have correct ingressNetworkPolicyNamespace")
				assert.Equal(record.IngressNetworkPolicyType, ipfixregistry.PolicyTypeAntreaNetworkPolicy, "Record does not have the correct NetworkPolicy Type with the ingress reject rule")
				assert.Equal(record.IngressNetworkPolicyRuleName, testIngressRuleName, "Record does not have the correct NetworkPolicy RuleName with the ingress reject rule")
			} else if record.IngressNetworkPolicyRuleAction == ipfixregistry.NetworkPolicyRuleActionDrop {
				assert.Equal(record.IngressNetworkPolicyName, ingressDropANPName, "Record does not have Antrea NetworkPolicy name with ingress drop rule")
				assert.Equal(record.IngressNetworkPolicyNamespace, data.testNamespace, "Record does not have correct ingressNetworkPolicyNamespace")
				assert.Equal(record.IngressNetworkPolicyType, ipfixregistry.PolicyTypeAntreaNetworkPolicy, "Record does not have the correct NetworkPolicy Type with the ingress drop rule")
				assert.Equal(record.IngressNetworkPolicyRuleName, testIngressRuleName, "Record does not have the correct NetworkPolicy RuleName with the ingress drop rule")
			} else if record.EgressNetworkPolicyRuleAction == ipfixregistry.NetworkPolicyRuleActionReject {
				assert.Equal(record.EgressNetworkPolicyName, egressRejectANPName, "Record does not have Antrea NetworkPolicy name with egress reject rule")
				assert.Equal(record.EgressNetworkPolicyNamespace, data.testNamespace, "Record does not have correct egressNetworkPolicyNamespace")
				assert.Equal(record.EgressNetworkPolicyType, ipfixregistry.PolicyTypeAntreaNetworkPolicy, "Record does not have the correct NetworkPolicy Type with the egress reject rule")
				assert.Equal(record.EgressNetworkPolicyRuleName, testEgressRuleName, "Record does not have the correct NetworkPolicy RuleName with the egress reject rule")
			} else if record.EgressNetworkPolicyRuleAction == ipfixregistry.NetworkPolicyRuleActionDrop {
				assert.Equal(record.EgressNetworkPolicyName, egressDropANPName, "Record does not have Antrea NetworkPolicy name with egress drop rule")
				assert.Equal(record.EgressNetworkPolicyNamespace, data.testNamespace, "Record does not have correct egressNetworkPolicyNamespace")
				assert.Equal(record.EgressNetworkPolicyType, ipfixregistry.PolicyTypeAntreaNetworkPolicy, "Record does not have the correct NetworkPolicy Type with the egress drop rule")
				assert.Equal(record.EgressNetworkPolicyRuleName, testEgressRuleName, "Record does not have the correct NetworkPolicy RuleName with the egress drop rule")
			}
			if checkSrcNodeInfo && (record.IngressNetworkPolicyRuleAction == ipfixregistry.NetworkPolicyRuleActionReject || record.IngressNetworkPolicyRuleAction == ipfixregistry.NetworkPolicyRuleActionDrop) {
				// For ingress deny rules, we also define an egress allow policy rule to ensure
				// that the information is reported correctly.
				assert.Equal(record.EgressNetworkPolicyName, egressAllowANPName, "Record does not have Antrea NetworkPolicy name with egress allow rule")
				assert.Equal(record.EgressNetworkPolicyType, ipfixregistry.PolicyTypeAntreaNetworkPolicy, "Record does not have the correct NetworkPolicy Type with the egress allow rule")
			}
		}
	}
}

func checkPodAndNodeData(t *testing.T, record, srcPod, srcNode, dstPod, dstNode string, namespace string) {
	assert := assert.New(t)
	assert.Contains(record, srcPod, "Record with srcIP does not have Pod name: %s", srcPod)
	assert.Contains(record, fmt.Sprintf("sourcePodNamespace: %s", namespace), "Record does not have correct sourcePodNamespace: %s", namespace)
	assert.Contains(record, fmt.Sprintf("sourceNodeName: %s", srcNode), "Record does not have correct sourceNodeName: %s", srcNode)
	// For Pod-To-External flow type, we send traffic to an external address,
	// so we skip the verification of destination Pod info.
	// Also, source Pod labels are different for Pod-To-External flow test.
	if dstPod != "" {
		assert.Contains(record, dstPod, "Record with dstIP does not have Pod name: %s", dstPod)
		assert.Contains(record, fmt.Sprintf("destinationPodNamespace: %s", namespace), "Record does not have correct destinationPodNamespace: %s", namespace)
		assert.Contains(record, fmt.Sprintf("destinationNodeName: %s", dstNode), "Record does not have correct destinationNodeName: %s", dstNode)
		assert.Contains(record, fmt.Sprintf("\"antrea-e2e\":\"%s\",\"app\":\"iperf\"", srcPod), "Record does not have correct label for source Pod")
		assert.Contains(record, fmt.Sprintf("\"antrea-e2e\":\"%s\",\"app\":\"iperf\"", dstPod), "Record does not have correct label for destination Pod")
	} else {
		assert.Contains(record, fmt.Sprintf("\"antrea-e2e\":\"%s\",\"app\":\"toolbox\"", srcPod), "Record does not have correct label for source Pod")
	}
}

func checkPodAndNodeDataClickHouse(data *TestData, t *testing.T, record *ClickHouseFullRow, srcPod, srcNode, dstPod, dstNode string) {
	assert := assert.New(t)
	assert.Equal(record.SourcePodName, srcPod, "Record with srcIP does not have Pod name: %s", srcPod)
	assert.Equal(record.SourcePodNamespace, data.testNamespace, "Record does not have correct sourcePodNamespace: %s", data.testNamespace)
	assert.Equal(record.SourceNodeName, srcNode, "Record does not have correct sourceNodeName: %s", srcNode)
	// For Pod-To-External flow type, we send traffic to an external address,
	// so we skip the verification of destination Pod info.
	// Also, source Pod labels are different for Pod-To-External flow test.
	if dstPod != "" {
		assert.Equal(record.DestinationPodName, dstPod, "Record with dstIP does not have Pod name: %s", dstPod)
		assert.Equal(record.DestinationPodNamespace, data.testNamespace, "Record does not have correct destinationPodNamespace: %s", data.testNamespace)
		assert.Equal(record.DestinationNodeName, dstNode, "Record does not have correct destinationNodeName: %s", dstNode)
		assert.Contains(record.SourcePodLabels, fmt.Sprintf("\"antrea-e2e\":\"%s\",\"app\":\"iperf\"", srcPod), "Record does not have correct label for source Pod")
		assert.Contains(record.DestinationPodLabels, fmt.Sprintf("\"antrea-e2e\":\"%s\",\"app\":\"iperf\"", dstPod), "Record does not have correct label for destination Pod")
	} else {
		assert.Contains(record.SourcePodLabels, fmt.Sprintf("\"antrea-e2e\":\"%s\",\"app\":\"toolbox\"", srcPod), "Record does not have correct label for source Pod")
	}
}

func checkFlowType(t *testing.T, record string, flowType uint8) {
	assert.Containsf(t, record, fmt.Sprintf("flowType: %d", flowType), "Record does not have correct flowType")
}

func checkFlowTypeClickHouse(t *testing.T, record *ClickHouseFullRow, flowType uint8) {
	assert.Equal(t, record.FlowType, flowType, "Record does not have correct flowType")
}

func checkEgressInfo(t *testing.T, record, egressName, egressIP, egressNodeName string) {
	assert.Containsf(t, record, fmt.Sprintf("egressName: %s", egressName), "Record does not have correct egressName")
	assert.Containsf(t, record, fmt.Sprintf("egressIP: %s", egressIP), "Record does not have correct egressIP")
	assert.Containsf(t, record, fmt.Sprintf("egressNodeName: %s", egressNodeName), "Record does not have correct egressNodeName")
}

func checkEgressInfoClickHouse(t *testing.T, record *ClickHouseFullRow, egressName, egressIP, egressNodeName string) {
	assert.Equal(t, egressName, record.EgressName, "Record does not have correct egressName")
	assert.Equal(t, egressIP, record.EgressIP, "Record does not have correct egressIP")
	assert.Equal(t, egressNodeName, record.EgressNodeName, "Record does not have correct egressNodeName")
}

func checkL7FlowExporterData(t *testing.T, record, appProtocolName string) {
	assert.Containsf(t, record, fmt.Sprintf("appProtocolName: %s", appProtocolName), "Record does not have correct Layer 7 protocol Name")
}

func checkL7FlowExporterDataClickHouse(t *testing.T, record *ClickHouseFullRow, appProtocolName string) {
	assert.Equal(t, record.AppProtocolName, appProtocolName, "Record does not have correct Layer 7 protocol Name")
	assert.NotEmpty(t, record.HttpVals, "Record does not have httpVals")
}

func getUint64FieldFromRecord(t require.TestingT, record string, field string) uint64 {
	splitLines := strings.Split(record, "\n")
	for _, line := range splitLines {
		if strings.Contains(line, field) {
			lineSlice := strings.Split(line, ":")
			value, err := strconv.ParseUint(strings.TrimSpace(lineSlice[1]), 10, 64)
			require.NoError(t, err, "Error when converting %s to uint64 type", field)
			return value
		}
	}
	return 0
}

// getCollectorOutput polls the output of go-ipfix collector and checks if we have
// received all the expected records for a given flow with source IP, destination IP
// and source port. We send source port to ignore the control flows during the
// iperf test.
func getCollectorOutput(t require.TestingT, srcIP, dstIP, srcPort string, isDstService bool, lookForFlowEnd bool, isIPv6 bool, data *TestData, labelFilter string, timeout time.Duration) []string {
	var allRecords, records []string
	err := wait.PollUntilContextTimeout(context.Background(), 500*time.Millisecond, timeout, true, func(ctx context.Context) (bool, error) {
		var rc int
		var err error
		var cmd string
		ipfixCollectorIP, err := testData.podWaitForIPs(defaultTimeout, "ipfix-collector", testData.testNamespace)
		if err != nil || len(ipfixCollectorIP.IPStrings) == 0 {
			require.NoErrorf(t, err, "Should be able to get IP from IPFIX collector Pod")
		}
		if !isIPv6 {
			cmd = fmt.Sprintf("curl http://%s:8080/records", ipfixCollectorIP.IPv4.String())
		} else {
			cmd = fmt.Sprintf("curl http://[%s]:8080/records", ipfixCollectorIP.IPv6.String())
		}
		rc, collectorOutput, _, err := data.RunCommandOnNode(controlPlaneNodeName(), cmd)
		if err != nil || rc != 0 {
			return false, fmt.Errorf("failed to run curl command to retrieve flow records, rc: %d - err: %v", rc, err)
		}
		// Checking that all the data records which correspond to the iperf flow are received
		src, dst := matchSrcAndDstAddress(srcIP, dstIP, isDstService, isIPv6)
		var response IPFIXCollectorResponse
		if err := json.Unmarshal([]byte(collectorOutput), &response); err != nil {
			return false, fmt.Errorf("error when unmarshalling output from IPFIX collector Pod: %w", err)
		}
		allRecords = make([]string, len(response.FlowRecords))
		for idx := range response.FlowRecords {
			allRecords[idx] = response.FlowRecords[idx].Data
		}
		records = filterCollectorRecords(allRecords, labelFilter, src, dst, srcPort)
		if lookForFlowEnd {
			for _, record := range records {
				flowEndReason := int64(getUint64FieldFromRecord(t, record, "flowEndReason"))
				// flowEndReason == 3 means the end of flow detected
				if flowEndReason == 3 {
					return true, nil
				}
			}
			return false, nil
		}
		return len(records) > 0, nil
	})
	// In case of a timeout, print some debug information.
	if err == context.DeadlineExceeded {
		const numRecordsToPrint = 20
		fmt.Printf("Last %d records received by IPFIX collector:\n", numRecordsToPrint)
		for i := 0; i < len(allRecords) && i < numRecordsToPrint; i++ {
			fmt.Println(allRecords[i])
		}
	}
	require.NoErrorf(t, err, "IPFIX collector did not receive the expected records, source IP: %s, dest IP: %s, source port: %s, total records count: %d, filtered records count: %d", srcIP, dstIP, srcPort, len(allRecords), len(records))
	return records
}

// getClickHouseOutput queries clickhouse with built-in client and checks if we have
// received all the expected records for a given flow with source IP, destination IP
// and source port. We send source port to ignore the control flows during the iperf test.
// Polling timeout is coded assuming IPFIX output has been checked first.
func getClickHouseOutput(t *testing.T, data *TestData, srcIP, dstIP, srcPort string, isDstService, lookForFlowEnd bool, labelFilter string) []*ClickHouseFullRow {
	var flowRecords []*ClickHouseFullRow
	var queryOutput string

	query := fmt.Sprintf("SELECT * FROM flows WHERE (sourceIP = '%s') AND (destinationIP = '%s') AND (octetDeltaCount != 0)", srcIP, dstIP)
	if isDstService {
		query = fmt.Sprintf("SELECT * FROM flows WHERE (sourceIP = '%s') AND (destinationClusterIP = '%s') AND (octetDeltaCount != 0)", srcIP, dstIP)
	}
	if len(srcPort) > 0 {
		query = fmt.Sprintf("%s AND (sourceTransportPort = %s)", query, srcPort)
	}
	if labelFilter != "" {
		query = fmt.Sprintf("%s AND (sourcePodLabels LIKE '%%%s%%')", query, labelFilter)
	}
	cmd := []string{
		"clickhouse-client",
		"--date_time_output_format=iso",
		"--format=JSONEachRow",
		fmt.Sprintf("--query=%s", query),
	}
	// ClickHouse output expected to be checked after IPFIX collector.
	// Waiting additional 4x commit interval to be adequate for 3 commit attempts.
	err := wait.PollUntilContextTimeout(context.Background(), 500*time.Millisecond, aggregatorClickHouseCommitInterval*4, true, func(ctx context.Context) (bool, error) {
		queryOutput, _, err := data.RunCommandFromPod(flowVisibilityNamespace, clickHousePodName, "clickhouse", cmd)
		if err != nil {
			return false, err
		}
		rows := strings.Split(queryOutput, "\n")
		flowRecords = make([]*ClickHouseFullRow, 0, len(rows))
		for _, row := range rows {
			row = strings.TrimSpace(row)
			if len(row) == 0 {
				continue
			}
			flowRecord := ClickHouseFullRow{}
			err = json.Unmarshal([]byte(row), &flowRecord)
			if err != nil {
				return false, err
			}
			flowRecords = append(flowRecords, &flowRecord)
		}

		if lookForFlowEnd {
			for _, record := range flowRecords {
				// flowEndReason == 3 means the end of flow detected
				if record.FlowEndReason == 3 {
					return true, nil
				}
			}
			return false, nil
		}
		return len(flowRecords) > 0, nil
	})
	require.NoErrorf(t, err, "ClickHouse did not receive the expected records in query output: %v; query: %s", queryOutput, query)
	return flowRecords
}

func filterCollectorRecords(records []string, filters ...string) []string {
	filteredRecords := []string{}
	match := func(record string) bool {
		// We don't check the last record.
		if strings.Contains(record, "octetDeltaCount: 0") {
			return false
		}
		for _, filter := range filters {
			if filter != "" && !strings.Contains(record, filter) {
				return false
			}
		}
		return true
	}
	for _, record := range records {
		if match(record) {
			filteredRecords = append(filteredRecords, record)
		}
	}
	return filteredRecords
}

func deployK8sNetworkPolicies(t *testing.T, data *TestData, srcPod, dstPod string) {
	// Add K8s NetworkPolicy between two iperf Pods.
	np1, err := data.createNetworkPolicy(ingressAllowNetworkPolicyName, &networkingv1.NetworkPolicySpec{
		PodSelector: metav1.LabelSelector{
			MatchLabels: map[string]string{
				"antrea-e2e": dstPod,
			},
		},
		PolicyTypes: []networkingv1.PolicyType{networkingv1.PolicyTypeIngress},
		Ingress: []networkingv1.NetworkPolicyIngressRule{{
			From: []networkingv1.NetworkPolicyPeer{{
				PodSelector: &metav1.LabelSelector{
					MatchLabels: map[string]string{
						"antrea-e2e": srcPod,
					},
				}},
			},
		}},
	})
	require.NoError(t, err, "Error when creating Network Policy")
	t.Cleanup(func() {
		assert.NoError(t, data.deleteNetworkpolicy(np1), "Error when deleting network policy")
	})
	np2, err := data.createNetworkPolicy(egressAllowNetworkPolicyName, &networkingv1.NetworkPolicySpec{
		PodSelector: metav1.LabelSelector{
			MatchLabels: map[string]string{
				"antrea-e2e": srcPod,
			},
		},
		PolicyTypes: []networkingv1.PolicyType{networkingv1.PolicyTypeEgress},
		Egress: []networkingv1.NetworkPolicyEgressRule{{
			To: []networkingv1.NetworkPolicyPeer{{
				PodSelector: &metav1.LabelSelector{
					MatchLabels: map[string]string{
						"antrea-e2e": dstPod,
					},
				}},
			},
		}},
	})
	require.NoError(t, err, "Error when creating Network Policy")
	t.Cleanup(func() {
		assert.NoError(t, data.deleteNetworkpolicy(np2), "Error when deleting network policy")
	})
	time.Sleep(5 * time.Second)
	t.Log("Network Policies are realized.")
}

func deployAntreaNetworkPolicies(t *testing.T, data *TestData, srcPod, dstPod string, srcNode, dstNode string) {
	// apply anp to dstPod, allow ingress from srcPod
	builder1 := &utils.AntreaNetworkPolicySpecBuilder{}
	builder1.SetName(data.testNamespace, ingressAntreaNetworkPolicyName).
		SetPriority(2.0).
		SetAppliedToGroup([]utils.ANNPAppliedToSpec{{PodSelector: map[string]string{"antrea-e2e": dstPod}}}).
		AddIngress(utils.ANNPRuleBuilder{
			BaseRuleBuilder: utils.BaseRuleBuilder{
				Protoc:      utils.ProtocolTCP,
				PodSelector: map[string]string{"antrea-e2e": srcPod},
				NSSelector:  map[string]string{},
				Action:      secv1beta1.RuleActionAllow,
				Name:        testIngressRuleName,
			}})
	anp1, err := k8sUtils.CreateOrUpdateANNP(builder1.Get())
	require.NoError(t, err, "Error when creating Antrea Network Policy")
	t.Cleanup(func() {
		assert.NoError(t, data.deleteAntreaNetworkpolicy(anp1), "Error when deleting Antrea Network Policy")
	})

	// apply anp to srcPod, allow egress to dstPod
	builder2 := &utils.AntreaNetworkPolicySpecBuilder{}
	builder2.SetName(data.testNamespace, egressAntreaNetworkPolicyName).
		SetPriority(2.0).
		SetAppliedToGroup([]utils.ANNPAppliedToSpec{{PodSelector: map[string]string{"antrea-e2e": srcPod}}}).
		AddEgress(utils.ANNPRuleBuilder{
			BaseRuleBuilder: utils.BaseRuleBuilder{
				Protoc:      utils.ProtocolTCP,
				PodSelector: map[string]string{"antrea-e2e": dstPod},
				NSSelector:  map[string]string{},
				Action:      secv1beta1.RuleActionAllow,
				Name:        testEgressRuleName,
			}})
	anp2, err := k8sUtils.CreateOrUpdateANNP(builder2.Get())
	require.NoError(t, err, "Error when creating Antrea Network Policy")
	t.Cleanup(func() {
		assert.NoError(t, data.deleteAntreaNetworkpolicy(anp2), "Error when deleting Antrea Network Policy")
	})

	// Wait for network policies to be realized.
	time.Sleep(5 * time.Second)
	t.Log("Antrea Network Policies are realized.")
}

func deployDenyAntreaNetworkPolicies(t *testing.T, data *TestData, srcPod, podReject, podDrop string, srcNode, dstNode string, isIngress bool) {
	var err error
	builders := make([]*utils.AntreaNetworkPolicySpecBuilder, 0)
	if isIngress {
		// apply reject and drop ingress rule to destination pods
		builder1 := &utils.AntreaNetworkPolicySpecBuilder{}
		builder1.SetName(data.testNamespace, ingressRejectANPName).
			SetPriority(2.0).
			SetAppliedToGroup([]utils.ANNPAppliedToSpec{{PodSelector: map[string]string{"antrea-e2e": podReject}}}).
			AddIngress(utils.ANNPRuleBuilder{
				BaseRuleBuilder: utils.BaseRuleBuilder{
					Protoc:      utils.ProtocolTCP,
					PodSelector: map[string]string{"antrea-e2e": srcPod},
					NSSelector:  map[string]string{},
					Action:      secv1beta1.RuleActionReject,
					Name:        testIngressRuleName,
				}})
		builder2 := &utils.AntreaNetworkPolicySpecBuilder{}
		builder2.SetName(data.testNamespace, ingressDropANPName).
			SetPriority(2.0).
			SetAppliedToGroup([]utils.ANNPAppliedToSpec{{PodSelector: map[string]string{"antrea-e2e": podDrop}}}).
			AddIngress(utils.ANNPRuleBuilder{
				BaseRuleBuilder: utils.BaseRuleBuilder{
					Protoc:      utils.ProtocolTCP,
					PodSelector: map[string]string{"antrea-e2e": srcPod},
					NSSelector:  map[string]string{},
					Action:      secv1beta1.RuleActionDrop,
					Name:        testIngressRuleName,
				}})
		// add an explicit egress allow policy so we can check that egress policy
		// information is reported correctly when a connection is denied by an ingress
		// policy rule.
		builder3 := &utils.AntreaNetworkPolicySpecBuilder{}
		builder3.SetName(data.testNamespace, egressAllowANPName).
			SetPriority(2.0).
			SetAppliedToGroup([]utils.ANNPAppliedToSpec{{PodSelector: map[string]string{"antrea-e2e": srcPod}}}).
			AddEgress(utils.ANNPRuleBuilder{
				BaseRuleBuilder: utils.BaseRuleBuilder{
					Protoc:      utils.ProtocolTCP,
					PodSelector: map[string]string{},
					NSSelector:  map[string]string{},
					Action:      secv1beta1.RuleActionAllow,
				}})

		builders = append(builders, builder1, builder2, builder3)
	} else {
		// apply reject and drop egress rule to source pod
		builder1 := &utils.AntreaNetworkPolicySpecBuilder{}
		builder1.SetName(data.testNamespace, egressRejectANPName).
			SetPriority(2.0).
			SetAppliedToGroup([]utils.ANNPAppliedToSpec{{PodSelector: map[string]string{"antrea-e2e": srcPod}}}).
			AddEgress(utils.ANNPRuleBuilder{
				BaseRuleBuilder: utils.BaseRuleBuilder{
					Protoc:      utils.ProtocolTCP,
					PodSelector: map[string]string{"antrea-e2e": podReject},
					NSSelector:  map[string]string{},
					Action:      secv1beta1.RuleActionReject,
					Name:        testEgressRuleName,
				}})
		builder2 := &utils.AntreaNetworkPolicySpecBuilder{}
		builder2.SetName(data.testNamespace, egressDropANPName).
			SetPriority(2.0).
			SetAppliedToGroup([]utils.ANNPAppliedToSpec{{PodSelector: map[string]string{"antrea-e2e": srcPod}}}).
			AddEgress(utils.ANNPRuleBuilder{
				BaseRuleBuilder: utils.BaseRuleBuilder{
					Protoc:      utils.ProtocolTCP,
					PodSelector: map[string]string{"antrea-e2e": podDrop},
					NSSelector:  map[string]string{},
					Action:      secv1beta1.RuleActionDrop,
					Name:        testEgressRuleName,
				}})
		builders = append(builders, builder1, builder2)
	}
	for _, b := range builders {
		anp := b.Get()
		anp, err = k8sUtils.CreateOrUpdateANNP(anp)
		require.NoError(t, err, "Error when creating Antrea Network Policy")
		t.Cleanup(func() {
			assert.NoError(t, data.deleteAntreaNetworkpolicy(anp), "Error when deleting Antrea Network Policy")
		})
	}
	time.Sleep(5 * time.Second)
	t.Log("Antrea Network Policies are realized.")
}

func deployDenyNetworkPolicies(t *testing.T, data *TestData, pod1, pod2 string, node1, node2 string) {
	np1, err := data.createNetworkPolicy(ingressDenyNPName, &networkingv1.NetworkPolicySpec{
		PodSelector: metav1.LabelSelector{
			MatchLabels: map[string]string{
				"antrea-e2e": pod1,
			},
		},
		PolicyTypes: []networkingv1.PolicyType{networkingv1.PolicyTypeIngress},
		Ingress:     []networkingv1.NetworkPolicyIngressRule{},
	})
	require.NoError(t, err, "Error when creating Network Policy")
	t.Cleanup(func() {
		assert.NoError(t, data.deleteNetworkpolicy(np1), "Error when deleting network policy")
	})
	np2, err := data.createNetworkPolicy(egressDenyNPName, &networkingv1.NetworkPolicySpec{
		PodSelector: metav1.LabelSelector{
			MatchLabels: map[string]string{
				"antrea-e2e": pod2,
			},
		},
		PolicyTypes: []networkingv1.PolicyType{networkingv1.PolicyTypeEgress},
		Egress:      []networkingv1.NetworkPolicyEgressRule{},
	})
	require.NoError(t, err, "Error when creating Network Policy")
	t.Cleanup(func() {
		assert.NoError(t, data.deleteNetworkpolicy(np2), "Error when deleting network policy")
	})
	// Wait for NetworkPolicy to be realized.
	time.Sleep(5 * time.Second)
	t.Log("Network Policies are realized.")
}

func createPerftestPods(data *TestData) (*PodIPs, *PodIPs, *PodIPs, *PodIPs, *PodIPs, error) {
	cmd := []string{"iperf3", "-s"}
	create := func(name string, nodeName string, ports []corev1.ContainerPort) error {
		return NewPodBuilder(name, data.testNamespace, ToolboxImage).WithContainerName("iperf").WithCommand(cmd).OnNode(nodeName).WithPorts(ports).Create(data)
	}
	var err error
	var podIPsArray [5]*PodIPs
	for i, podName := range podNames {
		var nodeName string
		if slices.Contains([]string{"perftest-a", "perftest-b", "perftest-d"}, podName) {
			nodeName = controlPlaneNodeName()
		} else {
			nodeName = workerNodeName(1)
		}
		if err := create(podName, nodeName, []corev1.ContainerPort{{Protocol: corev1.ProtocolTCP, ContainerPort: iperfPort}}); err != nil {
			return nil, nil, nil, nil, nil, fmt.Errorf("error when creating the perftest client Pod: %v", err)
		}
		podIPsArray[i], err = data.podWaitForIPs(defaultTimeout, podName, data.testNamespace)
		if err != nil {
			return nil, nil, nil, nil, nil, fmt.Errorf("error when waiting for the perftest client Pod: %v", err)
		}
	}
	return podIPsArray[0], podIPsArray[1], podIPsArray[2], podIPsArray[3], podIPsArray[4], nil
}

func createPerftestServices(data *TestData, isIPv6 bool) (*corev1.Service, *corev1.Service, *corev1.Service, *corev1.Service, *corev1.Service, error) {
	svcIPFamily := corev1.IPv4Protocol
	if isIPv6 {
		svcIPFamily = corev1.IPv6Protocol
	}
	var err error
	var services [5]*corev1.Service
	for i, serviceName := range serviceNames {
		services[i], err = data.CreateService(serviceName, data.testNamespace, iperfSvcPort, iperfPort, map[string]string{"antrea-e2e": serviceName}, false, false, corev1.ServiceTypeClusterIP, &svcIPFamily)
		if err != nil {
			return nil, nil, nil, nil, nil, fmt.Errorf("error when creating perftest-b Service: %v", err)
		}
	}
	return services[0], services[1], services[2], services[3], services[4], nil
}

func deletePerftestServices(t *testing.T, data *TestData) {
	for _, serviceName := range serviceNames {
		err := data.deleteService(data.testNamespace, serviceName)
		if err != nil {
			t.Logf("Error when deleting %s Service: %v", serviceName, err)
		}
	}
}

func addLabelToTestPods(t *testing.T, data *TestData, label string, podNames []string) {
	for _, podName := range podNames {
		testPod, err := data.clientset.CoreV1().Pods(data.testNamespace).Get(context.TODO(), podName, metav1.GetOptions{})
		require.NoErrorf(t, err, "Error when getting Pod %s in %s", testPod, data.testNamespace)
		testPod.Labels["targetLabel"] = label
		_, err = data.clientset.CoreV1().Pods(data.testNamespace).Update(context.TODO(), testPod, metav1.UpdateOptions{})
		require.NoErrorf(t, err, "Error when adding label to %s", testPod.Name)
		err = wait.PollUntilContextTimeout(context.Background(), defaultInterval, timeout, false, func(ctx context.Context) (bool, error) {
			pod, err := data.clientset.CoreV1().Pods(data.testNamespace).Get(context.TODO(), testPod.Name, metav1.GetOptions{})
			if err != nil {
				if errors.IsNotFound(err) {
					return false, nil
				}
				return false, fmt.Errorf("error when getting Pod '%s': %w", pod.Name, err)
			}
			return pod.Labels["targetLabel"] == label, nil
		})
		require.NoErrorf(t, err, "Error when verifying the label on %s", testPod.Name)
	}
}

// getBandwidthAndPorts parses iperf commands output and returns bandwidth,
// source port and destination port. Bandwidth is returned as a slice containing
// two strings (bandwidth value and bandwidth unit).
func getBandwidthAndPorts(iperfStdout string) ([]string, string, string) {
	var bandwidth []string
	var srcPort, dstPort string
	outputLines := strings.Split(iperfStdout, "\n")
	for _, line := range outputLines {
		if strings.Contains(line, "sender") {
			fields := strings.Fields(line)
			bandwidth = fields[6:8]
		}
		if strings.Contains(line, "connected") {
			fields := strings.Fields(line)
			srcPort = fields[5]
			dstPort = fields[10]
		}
	}
	return bandwidth, srcPort, dstPort
}

func matchSrcAndDstAddress(srcIP string, dstIP string, isDstService bool, isIPv6 bool) (string, string) {
	srcField := fmt.Sprintf("sourceIPv4Address: %s", srcIP)
	dstField := fmt.Sprintf("destinationIPv4Address: %s", dstIP)
	if isDstService {
		dstField = fmt.Sprintf("destinationClusterIPv4: %s", dstIP)
	}
	if isIPv6 {
		srcField = fmt.Sprintf("sourceIPv6Address: %s", srcIP)
		dstField = fmt.Sprintf("destinationIPv6Address: %s", dstIP)
		if isDstService {
			dstField = fmt.Sprintf("destinationClusterIPv6: %s", dstIP)
		}
	}
	return srcField, dstField
}

func createToExternalTestServer(t *testing.T, data *TestData) *PodIPs {
	// Creating an agnhost server as a hostNetwork Pod
	// Egress will be applied to the traffic when the destination is a hostNetwork Pod.
	_, serverIPs, serverCleanupFunc := createAndWaitForPod(t, data, func(name string, ns string, nodeName string, hostNetwork bool) error {
		return data.createServerPod(name, data.testNamespace, "", serverPodPort, false, true)
	}, "test-server-", "", data.testNamespace, false)

	t.Cleanup(func() {
		serverCleanupFunc()
	})

	return serverIPs
}

func getAndCheckFlowAggregatorMetrics(t *testing.T, data *TestData, withClickHouseExporter bool) error {
	flowAggPod, err := data.getFlowAggregator()
	if err != nil {
		return fmt.Errorf("error when getting flow-aggregator Pod: %w", err)
	}
	podName := flowAggPod.Name
	command := []string{"antctl", "get", "recordmetrics", "-o", "json"}
	if err := wait.PollUntilContextTimeout(context.Background(), defaultInterval, 2*defaultTimeout, false, func(ctx context.Context) (bool, error) {
		stdout, _, err := runAntctl(podName, command, data)
		if err != nil {
			t.Logf("Error when requesting recordmetrics, %v", err)
			return false, nil
		}
		metrics := &apis.RecordMetricsResponse{}
		if err := json.Unmarshal([]byte(stdout), metrics); err != nil {
			return false, fmt.Errorf("error when decoding recordmetrics: %w", err)
		}
		if metrics.NumConnToCollector != int64(clusterInfo.numNodes) || (withClickHouseExporter != metrics.WithClickHouseExporter) || !metrics.WithIPFIXExporter || metrics.NumRecordsExported == 0 {
			t.Logf("Metrics are not correct. Current metrics: NumConnToCollector=%d, ClickHouseExporter=%v, IPFIXExporter=%v, NumRecordsExported=%d", metrics.NumConnToCollector, metrics.WithClickHouseExporter, metrics.WithIPFIXExporter, metrics.NumRecordsExported)
			return false, nil
		}
		return true, nil
	}); err != nil {
		return fmt.Errorf("error when checking recordmetrics for Flow Aggregator: %w", err)
	}
	return nil
}

func testL7FlowExporterController(t *testing.T, data *TestData, isIPv6 bool) {
	skipIfFeatureDisabled(t, features.L7FlowExporter, true, false)
	nodeName := nodeName(1)
	_, serverIPs, cleanupFunc := createAndWaitForPod(t, data, data.createNginxPodOnNode, "l7flowexportertestpodserver", nodeName, data.testNamespace, false)
	defer cleanupFunc()

	clientPodName := "l7flowexportertestpodclient"
	clientPodLabels := map[string]string{"flowexportertest": "l7"}
	clientPodAnnotations := map[string]string{antreaagenttypes.L7FlowExporterAnnotationKey: "both"}
	require.NoError(t, NewPodBuilder(clientPodName, data.testNamespace, ToolboxImage).OnNode(nodeName).WithContainerName("l7flowexporter").WithLabels(clientPodLabels).WithAnnotations(clientPodAnnotations).Create(data))
	clientPodIPs, err := data.podWaitForIPs(defaultTimeout, clientPodName, data.testNamespace)
	require.NoErrorf(t, err, "Error when waiting for IP for Pod '%s': %v", clientPodName, err)
	defer deletePodWrapper(t, data, data.testNamespace, clientPodName)

	// Wait for the Suricata to start.
	time.Sleep(3 * time.Second)

	testFlow1 := testFlow{
		srcPodName: clientPodName,
	}
	if !isIPv6 {
		testFlow1.srcIP = clientPodIPs.IPv4.String()
		testFlow1.dstIP = serverIPs.IPv4.String()
	} else {
		testFlow1.srcIP = clientPodIPs.IPv6.String()
		testFlow1.dstIP = serverIPs.IPv6.String()
	}
	cmd := []string{"curl", getHTTPURLFromIPPort(testFlow1.dstIP, serverPodPort)}
	stdout, stderr, err := data.RunCommandFromPod(data.testNamespace, testFlow1.srcPodName, "l7flowexporter", cmd)
	require.NoErrorf(t, err, "Error when running curl command, stdout: %s, stderr: %s", stdout, stderr)
	records := getCollectorOutput(t, testFlow1.srcIP, testFlow1.dstIP, "", false, true, isIPv6, data, "", getCollectorOutputDefaultTimeout)
	for _, record := range records {
		assert := assert.New(t)
		assert.Contains(record, testFlow1.srcPodName, "Record with srcIP does not have Pod name: %s", testFlow1.srcPodName)
		assert.Contains(record, fmt.Sprintf("sourcePodNamespace: %s", data.testNamespace), "Record does not have correct sourcePodNamespace: %s", data.testNamespace)
		assert.Contains(record, fmt.Sprintf("sourceNodeName: %s", nodeName), "Record does not have correct sourceNodeName: %s", nodeName)
		assert.Contains(record, "\"flowexportertest\":\"l7\"", "Record does not have correct label for source Pod")

		checkL7FlowExporterData(t, record, "http")
	}

	clickHouseRecords := getClickHouseOutput(t, data, testFlow1.srcIP, testFlow1.dstIP, "", false, true, "")
	for _, record := range clickHouseRecords {
		assert := assert.New(t)
		assert.Equal(record.SourcePodName, testFlow1.srcPodName, "Record with srcIP does not have Pod name: %s", testFlow1.srcPodName)
		assert.Equal(record.SourcePodNamespace, data.testNamespace, "Record does not have correct sourcePodNamespace: %s", data.testNamespace)
		assert.Equal(record.SourceNodeName, nodeName, "Record does not have correct sourceNodeName: %s", nodeName)
		assert.Contains(record.SourcePodLabels, "\"flowexportertest\":\"l7\"", "Record does not have correct label for source Pod")

		checkL7FlowExporterDataClickHouse(t, record, "http")
	}

}

type ClickHouseFullRow struct {
	TimeInserted                         time.Time `json:"timeInserted"`
	FlowStartSeconds                     time.Time `json:"flowStartSeconds"`
	FlowEndSeconds                       time.Time `json:"flowEndSeconds"`
	FlowEndSecondsFromSourceNode         time.Time `json:"flowEndSecondsFromSourceNode"`
	FlowEndSecondsFromDestinationNode    time.Time `json:"flowEndSecondsFromDestinationNode"`
	FlowEndReason                        uint8     `json:"flowEndReason"`
	SourceIP                             string    `json:"sourceIP"`
	DestinationIP                        string    `json:"destinationIP"`
	SourceTransportPort                  uint16    `json:"sourceTransportPort"`
	DestinationTransportPort             uint16    `json:"destinationTransportPort"`
	ProtocolIdentifier                   uint8     `json:"protocolIdentifier"`
	PacketTotalCount                     uint64    `json:"packetTotalCount,string"`
	OctetTotalCount                      uint64    `json:"octetTotalCount,string"`
	PacketDeltaCount                     uint64    `json:"packetDeltaCount,string"`
	OctetDeltaCount                      uint64    `json:"octetDeltaCount,string"`
	ReversePacketTotalCount              uint64    `json:"reversePacketTotalCount,string"`
	ReverseOctetTotalCount               uint64    `json:"reverseOctetTotalCount,string"`
	ReversePacketDeltaCount              uint64    `json:"reversePacketDeltaCount,string"`
	ReverseOctetDeltaCount               uint64    `json:"reverseOctetDeltaCount,string"`
	SourcePodName                        string    `json:"sourcePodName"`
	SourcePodNamespace                   string    `json:"sourcePodNamespace"`
	SourceNodeName                       string    `json:"sourceNodeName"`
	DestinationPodName                   string    `json:"destinationPodName"`
	DestinationPodNamespace              string    `json:"destinationPodNamespace"`
	DestinationNodeName                  string    `json:"destinationNodeName"`
	DestinationClusterIP                 string    `json:"destinationClusterIP"`
	DestinationServicePort               uint16    `json:"destinationServicePort"`
	DestinationServicePortName           string    `json:"destinationServicePortName"`
	IngressNetworkPolicyName             string    `json:"ingressNetworkPolicyName"`
	IngressNetworkPolicyNamespace        string    `json:"ingressNetworkPolicyNamespace"`
	IngressNetworkPolicyRuleName         string    `json:"ingressNetworkPolicyRuleName"`
	IngressNetworkPolicyRuleAction       uint8     `json:"ingressNetworkPolicyRuleAction"`
	IngressNetworkPolicyType             uint8     `json:"ingressNetworkPolicyType"`
	EgressNetworkPolicyName              string    `json:"egressNetworkPolicyName"`
	EgressNetworkPolicyNamespace         string    `json:"egressNetworkPolicyNamespace"`
	EgressNetworkPolicyRuleName          string    `json:"egressNetworkPolicyRuleName"`
	EgressNetworkPolicyRuleAction        uint8     `json:"egressNetworkPolicyRuleAction"`
	EgressNetworkPolicyType              uint8     `json:"egressNetworkPolicyType"`
	TcpState                             string    `json:"tcpState"`
	FlowType                             uint8     `json:"flowType"`
	SourcePodLabels                      string    `json:"sourcePodLabels"`
	DestinationPodLabels                 string    `json:"destinationPodLabels"`
	Throughput                           uint64    `json:"throughput,string"`
	ReverseThroughput                    uint64    `json:"reverseThroughput,string"`
	ThroughputFromSourceNode             uint64    `json:"throughputFromSourceNode,string"`
	ThroughputFromDestinationNode        uint64    `json:"throughputFromDestinationNode,string"`
	ReverseThroughputFromSourceNode      uint64    `json:"reverseThroughputFromSourceNode,string"`
	ReverseThroughputFromDestinationNode uint64    `json:"reverseThroughputFromDestinationNode,string"`
	ClusterUUID                          string    `json:"clusterUUID"`
	Trusted                              uint8     `json:"trusted"`
	EgressName                           string    `json:"egressName"`
	EgressIP                             string    `json:"egressIP"`
	AppProtocolName                      string    `json:"appProtocolName"`
	HttpVals                             string    `json:"httpVals"`
	EgressNodeName                       string    `json:"egressNodeName"`
}
