Skip to content

Commit

Permalink
add support for multiple flow monitoring protocols and for IPv6 endpo…
Browse files Browse the repository at this point in the history 
…int in existing netflow test

- Existing netflow export test now supports netflow_v5, ipfix, sflow.
- The original test only validated the export of netflow v5 (legacy netflow) data, which does not support IPv4 by design. In order to validate flow data export in an IPv6 cluster, we can now choose between IPFIX and sflow; also, the test now supports an IPv6 endpoint;
- For the time being we're only testing netflow v5 in IPv4 (as before) and sflow in IPv6, because:
  (1) The test itself is long (~5 minutes);
  (2) The current implementation of the flow collector container (cloudfare/goflow) doesn't support IPFIX data generated by OVS (cloudflare/goflow#99)

Signed-off-by: Riccardo Ravaioli <[email protected]>
  • Loading branch information
@ricky-rav
ricky-rav committed Oct 19, 2021
1 parent ac5ce1c commit 3dc4efe
Show file tree
Hide file tree
Showing 3 changed files with 150 additions and 90 deletions.
206 changes: 121 additions & 85 deletions test/e2e/e2e.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2541,114 +2541,150 @@ var _ = ginkgo.Describe("e2e ingress gateway traffic validation", func() {
})
})

// This test validates OVS exports NetFlow data from br-int to an external collector
var _ = ginkgo.Describe("e2e br-int NetFlow export validation", func() {
// This test validates that OVS exports flow monitoring data from br-int to an external collector
var _ = ginkgo.Describe("e2e br-int flow monitoring export validation", func() {
type flowMonitoringProtocol string

const (
svcname string = "netflow-test"
ovnNs string = "ovn-kubernetes"
netFlowCollectorContainer string = "netflow-collector"
ciNetworkName string = "kind"
netflow_v5 flowMonitoringProtocol = "netflow"
ipfix flowMonitoringProtocol = "ipfix"
sflow flowMonitoringProtocol = "sflow"

svcname string = "netflow-test"
ovnNs string = "ovn-kubernetes"
collectorContainer string = "netflow-collector"
ciNetworkName string = "kind"
)

f := framework.NewDefaultFramework(svcname)
keywordInLogs := map[flowMonitoringProtocol]string{
netflow_v5: "NETFLOW_V5", ipfix: "IPFIX", sflow: "SFLOW_5"}

f := framework.NewDefaultFramework(svcname)
ginkgo.AfterEach(func() {
// tear down the NetFlow container
if cid, _ := runCommand("docker", "ps", "-qaf", fmt.Sprintf("name=%s", netFlowCollectorContainer)); cid != "" {
if _, err := runCommand("docker", "rm", "-f", netFlowCollectorContainer); err != nil {
framework.Logf("failed to delete the netFlow collector test container %s %v", netFlowCollectorContainer, err)
// tear down the collector container
if cid, _ := runCommand("docker", "ps", "-qaf", fmt.Sprintf("name=%s", collectorContainer)); cid != "" {
if _, err := runCommand("docker", "rm", "-f", collectorContainer); err != nil {
framework.Logf("failed to delete the collector test container %s %v",
collectorContainer, err)
}
}
})

ginkgo.It("Should validate NetFlow data of br-int is sent to an external gateway and unset NetFlow Targets", func() {
var (
ciNetworkFlag = "{{ .NetworkSettings.Networks.kind.IPAddress }}"
)
ginkgo.By("Starting a netflow collector container")
// start the NetFlow collector container that will receive data
_, err := runCommand("docker", "run", "-itd", "--privileged", "--network", ciNetworkName, "--name", netFlowCollectorContainer, "cloudflare/goflow", "-kafka=false")
if err != nil {
framework.Failf("failed to start NetFlow collector test container %s: %v", netFlowCollectorContainer, err)
}
// retrieve the container ip of the NetFlow collector container
netFlowCollectorIp, err := runCommand("docker", "inspect", "-f", ciNetworkFlag, netFlowCollectorContainer)
if err != nil {
framework.Failf("failed to start NetFlow collector test container: %v", err)
}
// trim newline from the inspect output
netFlowCollectorIp = strings.TrimSuffix(netFlowCollectorIp, "\n")
if ip := net.ParseIP(netFlowCollectorIp); ip == nil {
framework.Failf("Unable to retrieve a valid address from container %s with inspect output of %s", netFlowCollectorContainer, netFlowCollectorIp)
}
table.DescribeTable("Should validate flow data of br-int is sent to an external gateway",
func(protocol flowMonitoringProtocol, collectorPort uint16) {
protocol_str := string(protocol)
ipField := "IPAddress"
isIpv6 := IsIPv6Cluster(f.ClientSet)
if isIpv6 {
ipField = "GlobalIPv6Address"
}
ciNetworkFlag := fmt.Sprintf("{{ .NetworkSettings.Networks.kind.%s }}", ipField)

ginkgo.By("Configuring ovnkube-node to use the new netflow collector target")
framework.Logf("Setting OVN_NETFLOW_TARGETS environment variable value to NetFlow collector IP %s", netFlowCollectorIp)
framework.RunKubectlOrDie(ovnNs, "set", "env", "daemonset/ovnkube-node", "-c", "ovnkube-node", "OVN_NETFLOW_TARGETS="+netFlowCollectorIp+":2056")
ginkgo.By("Starting a flow collector container")
// start the collector container that will receive data
_, err := runCommand("docker", "run", "-itd", "--privileged", "--network", ciNetworkName,
"--name", collectorContainer, "cloudflare/goflow", "-kafka=false")
if err != nil {
framework.Failf("failed to start flow collector container %s: %v", collectorContainer, err)
}
ovn_env_var := fmt.Sprintf("OVN_%s_TARGETS", strings.ToUpper(protocol_str))
// retrieve the ip of the collector container
collectorIP, err := runCommand("docker", "inspect", "-f", ciNetworkFlag, collectorContainer)
if err != nil {
framework.Failf("could not retrieve IP address of collector container: %v", err)
}
// trim newline from the inspect output
collectorIP = strings.TrimSpace(collectorIP)
if net.ParseIP(collectorIP) == nil {
framework.Failf("Unable to retrieve a valid address from container %s with inspect output of %s",
collectorContainer, collectorIP)
}
addressAndPort, err1 := formatAddressAndPort(collectorIP, collectorPort)
if err1 != nil {
framework.Failf("Unable to correctly format value for input env variable %s with "+
"IP address %s and port %d", ovn_env_var, collectorIP, collectorPort)
}
ginkgo.By(fmt.Sprintf("Configuring ovnkube-node to use the new %s collector target", protocol_str))
framework.Logf("Setting %s environment variable to %s",
ovn_env_var, addressAndPort)

// Make sure the updated daemonset has rolled out, verify it's completion 10 times
// TODO (Change this to use the exported upstream function)
err = waitForDaemonSetUpdate(f.ClientSet, ovnNs, "ovnkube-node", 0, dsRestartTimeout)
framework.ExpectNoError(err)
framework.RunKubectlOrDie(ovnNs, "set", "env", "daemonset/ovnkube-node", "-c", "ovnkube-node",
fmt.Sprintf("%s=%s", ovn_env_var, addressAndPort))

ginkgo.By("Checking that the collector container received netflow")
netFlowCollectorContainerLogsTest := func() wait.ConditionFunc {
return func() (bool, error) {
netFlowCollectorContainerLogs, err := runCommand("docker", "logs", netFlowCollectorContainer)
if err != nil {
framework.Logf("failed to inspect logs in test container: %v", err)
// Make sure the updated daemonset has rolled out, verify it's completion 10 times
// TODO (Change this to use the exported upstream function)
err = waitForDaemonSetUpdate(f.ClientSet, ovnNs, "ovnkube-node", 0, dsRestartTimeout)
framework.ExpectNoError(err)
ginkgo.By(fmt.Sprintf("Checking that the collector container received %s data", protocol_str))
collectorContainerLogsTest := func() wait.ConditionFunc {
return func() (bool, error) {
collectorContainerLogs, err := runCommand("docker", "logs", collectorContainer)
if err != nil {
framework.Logf("failed to inspect logs in test container: %v", err)
return false, nil
}
collectorContainerLogs = strings.TrimSuffix(collectorContainerLogs, "\n")
logLines := strings.Split(collectorContainerLogs, "\n")
lastLine := logLines[len(logLines)-1]
// check that flow monitoring traffic has been logged
if strings.Contains(lastLine, keywordInLogs[protocol]) {
framework.Logf("Successfully found string %s in last log line of"+
" the collector: %s", keywordInLogs[protocol], lastLine)
return true, nil
}
return false, nil
}
netFlowCollectorContainerLogs = strings.TrimSuffix(netFlowCollectorContainerLogs, "\n")
logLines := strings.Split(netFlowCollectorContainerLogs, "\n")
lastLine := logLines[len(logLines)-1]
// check that NetFlow traffic has been logged.
if strings.Contains(lastLine, "NETFLOW_V5") {
framework.Logf("the NetFlow collector received NetFlow data, last logs: %s", logLines[len(logLines)-1])
return true, nil
}
return false, nil
}
}

err = wait.PollImmediate(retryInterval, retryTimeout, netFlowCollectorContainerLogsTest())
framework.ExpectNoError(err, "failed to verify that NetFlow collector container received NetFlow data from br-int")
// retryTimeout of 40s is short sometimes, so let's increase it to 60s
err = wait.PollImmediate(retryInterval, 60*time.Second, collectorContainerLogsTest())
framework.ExpectNoError(err, fmt.Sprintf("failed to verify that collector container "+
"received %s data from br-int: string %s not found in logs",
protocol_str, keywordInLogs[protocol]))

ginkgo.By("Unsetting the OVN_NETFLOW_TARGETS variable in the ovnkube-node daemonset")
framework.RunKubectlOrDie(ovnNs, "set", "env", "daemonset/ovnkube-node", "-c", "ovnkube-node", "OVN_NETFLOW_TARGETS-")
ginkgo.By(fmt.Sprintf("Unsetting %s variable in ovnkube-node daemonset", ovn_env_var))
framework.RunKubectlOrDie(ovnNs, "set", "env", "daemonset/ovnkube-node", "-c", "ovnkube-node",
fmt.Sprintf("%s-", ovn_env_var))

// Make sure the updated daemonset has rolled out, verify it's completion 10 times
// TODO (Change this to use the exported upstream function)
err = waitForDaemonSetUpdate(f.ClientSet, ovnNs, "ovnkube-node", 0, dsRestartTimeout)
framework.ExpectNoError(err)
// Make sure the updated daemonset has rolled out, verify its completion 10 times
// TODO (Change this to use the exported upstream function)
err = waitForDaemonSetUpdate(f.ClientSet, ovnNs, "ovnkube-node", 0, dsRestartTimeout)
framework.ExpectNoError(err)

ovnKubeNodePods, err := f.ClientSet.CoreV1().Pods(ovnNs).List(context.TODO(), metav1.ListOptions{
LabelSelector: "name=ovnkube-node",
})
if err != nil {
framework.Failf("could not get ovnkube-node pods: %v", err)
}
ovnKubeNodePods, err := f.ClientSet.CoreV1().Pods(ovnNs).List(context.TODO(), metav1.ListOptions{
LabelSelector: "name=ovnkube-node",
})
if err != nil {
framework.Failf("could not get ovnkube-node pods: %v", err)
}

for _, ovnKubeNodePod := range ovnKubeNodePods.Items {
for _, ovnKubeNodePod := range ovnKubeNodePods.Items {

execOptions := framework.ExecOptions{
Command: []string{"ovs-vsctl", "find", "netflow"},
Namespace: ovnNs,
PodName: ovnKubeNodePod.Name,
ContainerName: "ovnkube-node",
CaptureStdout: true,
CaptureStderr: true,
}
execOptions := framework.ExecOptions{
Command: []string{"ovs-vsctl", "find", strings.ToLower(protocol_str)},
Namespace: ovnNs,
PodName: ovnKubeNodePod.Name,
ContainerName: "ovnkube-node",
CaptureStdout: true,
CaptureStderr: true,
}

targets, stderr, _ := f.ExecWithOptions(execOptions)
framework.Logf("execOptions are %v", execOptions)
if err != nil {
framework.Failf("could not lookup ovs netflow targets: %v", stderr)
targets, stderr, _ := f.ExecWithOptions(execOptions)
framework.Logf("execOptions are %v", execOptions)
if err != nil {
framework.Failf("could not lookup ovs %s targets: %v", protocol_str, stderr)
}
framework.ExpectEmpty(targets)
}
framework.ExpectEmpty(targets)
}
},
// This is a long test (~5 minutes per run), so let's just validate netflow v5
// in an IPv4 cluster and sflow in IPv6 cluster
table.Entry("with netflow v5", netflow_v5, uint16(2056)),
// goflow doesn't currently support OVS ipfix:
// https://github.com/cloudflare/goflow/issues/99
// table.Entry("ipfix", ipfix, uint16(2055)),
table.Entry("with sflow", sflow, uint16(6343)),
)

})
})

func getNodePodCIDR(nodeName string) (string, error) {
Expand Down
15 changes: 15 additions & 0 deletions test/e2e/util.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -463,6 +463,21 @@ func addressIsIP(address v1.NodeAddress) bool {
return true
}

func formatAddressAndPort(addr string, port uint16) (string, error) {
var (
err error
addrAndPort string
)
if utilnet.IsIPv4String(addr) {
addrAndPort = fmt.Sprintf("%s:%d", addr, port)
} else if utilnet.IsIPv6String(addr) {
addrAndPort = fmt.Sprintf("[%s]:%d", addr, port)
} else {
err = error(fmt.Errorf("IP address %s is not valid", addr))
}
return addrAndPort, err
}

// Returns pod's ipv4 and ipv6 addresses IN ORDER
func getPodAddresses(pod *v1.Pod) (string, string) {
var ipv4Res, ipv6Res string
Expand Down
19 changes: 14 additions & 5 deletions test/scripts/e2e-cp.sh
View file
Original file line number Diff line number Diff line change
Expand Up @@ -9,16 +9,25 @@ export KUBECONFIG=${HOME}/admin.conf
# Skip tests which are not IPv6 ready yet (see description of https://github.com/ovn-org/ovn-kubernetes/pull/2276)
IPV6_SKIPPED_TESTS="Should be allowed by externalip services|\
should provide connection to external host by DNS name from a pod|\
Should validate NetFlow data of br-int is sent to an external gateway|\
recovering from deleting db files while maintain connectivity|\
test tainting a node according to its defaults interface MTU size"
test tainting a node according to its defaults interface MTU size|\
Should validate flow data of br-int is sent to an external gateway with netflow v5" # netflow v5 is ipv4 only


SKIPPED_TESTS=""
if [ "$KIND_IPV4_SUPPORT" == true ] && [ "$KIND_IPV6_SUPPORT" == true ]; then
# No support for these features in dual-stack yet
SKIPPED_TESTS="hybrid.overlay|external.gateway"

if [ "$KIND_IPV4_SUPPORT" == true ]; then
if [ "$KIND_IPV6_SUPPORT" == true ]; then
# No support for these features in dual-stack yet
SKIPPED_TESTS="hybrid.overlay|external.gateway"
else
# Skip sflow in IPv4 since it's a long test (~5 minutes)
# We're validating netflow v5 with an ipv4 cluster, sflow with an ipv6 cluster
SKIPPED_TESTS="Should validate flow data of br-int is sent to an external gateway with sflow"
fi
fi


if [ "$OVN_HA" == false ]; then
if [ "$SKIPPED_TESTS" != "" ]; then
SKIPPED_TESTS+="|"
Expand Down

0 comments on commit 3dc4efe

Please sign in to comment.