[IPv6] Support flow exporter (#1541)

Co-authored-by: Antonin Bas <[email protected]>
Co-authored-by: srikartati <[email protected]>

build/yamls/antrea-aks.yml

@@ -1200,9 +1200,10 @@ data:
     # Enable metrics exposure via Prometheus. Initializes Prometheus metrics listener.
     #enablePrometheusMetrics: true
 
-    # Provide flow collector address as string with format <IP>:<port>[:<proto>], where proto is tcp or udp. This also enables
-    # the flow exporter that sends IPFIX flow records of conntrack flows on OVS bridge. If no L4 transport proto is given,
-    # we consider tcp as default.
+    # Provide flow collector address as string with format <IP>:<port>[:<proto>], where proto is tcp or udp.
+    # IP can be either IPv4 or IPv6. However, IPv6 address should be wrapped with [].
+    # This also enables the flow exporter that sends IPFIX flow records of conntrack flows on OVS bridge.
+    # If no L4 transport proto is given, we consider tcp as default.
     #flowCollectorAddr: ""
 
     # Provide flow poll interval as a duration string. This determines how often the flow exporter dumps connections from the conntrack module.
@@ -1270,7 +1271,7 @@ metadata:
   annotations: {}
   labels:
     app: antrea
-  name: antrea-config-t4t2mdfhkc
+  name: antrea-config-mdmtkcfh59
   namespace: kube-system
 ---
 apiVersion: v1
@@ -1377,7 +1378,7 @@ spec:
         key: node-role.kubernetes.io/master
       volumes:
       - configMap:
-          name: antrea-config-t4t2mdfhkc
+          name: antrea-config-mdmtkcfh59
         name: antrea-config
       - name: antrea-controller-tls
         secret:
@@ -1641,7 +1642,7 @@ spec:
         operator: Exists
       volumes:
       - configMap:
-          name: antrea-config-t4t2mdfhkc
+          name: antrea-config-mdmtkcfh59
         name: antrea-config
       - hostPath:
           path: /etc/cni/net.d

build/yamls/antrea-eks.yml

@@ -1200,9 +1200,10 @@ data:
     # Enable metrics exposure via Prometheus. Initializes Prometheus metrics listener.
     #enablePrometheusMetrics: true
 
-    # Provide flow collector address as string with format <IP>:<port>[:<proto>], where proto is tcp or udp. This also enables
-    # the flow exporter that sends IPFIX flow records of conntrack flows on OVS bridge. If no L4 transport proto is given,
-    # we consider tcp as default.
+    # Provide flow collector address as string with format <IP>:<port>[:<proto>], where proto is tcp or udp.
+    # IP can be either IPv4 or IPv6. However, IPv6 address should be wrapped with [].
+    # This also enables the flow exporter that sends IPFIX flow records of conntrack flows on OVS bridge.
+    # If no L4 transport proto is given, we consider tcp as default.
     #flowCollectorAddr: ""
 
     # Provide flow poll interval as a duration string. This determines how often the flow exporter dumps connections from the conntrack module.
@@ -1270,7 +1271,7 @@ metadata:
   annotations: {}
   labels:
     app: antrea
-  name: antrea-config-t4t2mdfhkc
+  name: antrea-config-mdmtkcfh59
   namespace: kube-system
 ---
 apiVersion: v1
@@ -1377,7 +1378,7 @@ spec:
         key: node-role.kubernetes.io/master
       volumes:
       - configMap:
-          name: antrea-config-t4t2mdfhkc
+          name: antrea-config-mdmtkcfh59
         name: antrea-config
       - name: antrea-controller-tls
         secret:
@@ -1643,7 +1644,7 @@ spec:
         operator: Exists
       volumes:
       - configMap:
-          name: antrea-config-t4t2mdfhkc
+          name: antrea-config-mdmtkcfh59
         name: antrea-config
       - hostPath:
           path: /etc/cni/net.d

build/yamls/antrea-gke.yml

@@ -1200,9 +1200,10 @@ data:
     # Enable metrics exposure via Prometheus. Initializes Prometheus metrics listener.
     #enablePrometheusMetrics: true
 
-    # Provide flow collector address as string with format <IP>:<port>[:<proto>], where proto is tcp or udp. This also enables
-    # the flow exporter that sends IPFIX flow records of conntrack flows on OVS bridge. If no L4 transport proto is given,
-    # we consider tcp as default.
+    # Provide flow collector address as string with format <IP>:<port>[:<proto>], where proto is tcp or udp.
+    # IP can be either IPv4 or IPv6. However, IPv6 address should be wrapped with [].
+    # This also enables the flow exporter that sends IPFIX flow records of conntrack flows on OVS bridge.
+    # If no L4 transport proto is given, we consider tcp as default.
     #flowCollectorAddr: ""
 
     # Provide flow poll interval as a duration string. This determines how often the flow exporter dumps connections from the conntrack module.
@@ -1270,7 +1271,7 @@ metadata:
   annotations: {}
   labels:
     app: antrea
-  name: antrea-config-gmt86d9t68
+  name: antrea-config-b5dkk776t2
   namespace: kube-system
 ---
 apiVersion: v1
@@ -1377,7 +1378,7 @@ spec:
         key: node-role.kubernetes.io/master
       volumes:
       - configMap:
-          name: antrea-config-gmt86d9t68
+          name: antrea-config-b5dkk776t2
         name: antrea-config
       - name: antrea-controller-tls
         secret:
@@ -1641,7 +1642,7 @@ spec:
         operator: Exists
       volumes:
       - configMap:
-          name: antrea-config-gmt86d9t68
+          name: antrea-config-b5dkk776t2
         name: antrea-config
       - hostPath:
           path: /etc/cni/net.d

build/yamls/antrea-ipsec.yml

@@ -1205,9 +1205,10 @@ data:
     # Enable metrics exposure via Prometheus. Initializes Prometheus metrics listener.
     #enablePrometheusMetrics: true
 
-    # Provide flow collector address as string with format <IP>:<port>[:<proto>], where proto is tcp or udp. This also enables
-    # the flow exporter that sends IPFIX flow records of conntrack flows on OVS bridge. If no L4 transport proto is given,
-    # we consider tcp as default.
+    # Provide flow collector address as string with format <IP>:<port>[:<proto>], where proto is tcp or udp.
+    # IP can be either IPv4 or IPv6. However, IPv6 address should be wrapped with [].
+    # This also enables the flow exporter that sends IPFIX flow records of conntrack flows on OVS bridge.
+    # If no L4 transport proto is given, we consider tcp as default.
     #flowCollectorAddr: ""
 
     # Provide flow poll interval as a duration string. This determines how often the flow exporter dumps connections from the conntrack module.
@@ -1275,7 +1276,7 @@ metadata:
   annotations: {}
   labels:
     app: antrea
-  name: antrea-config-2k6g59bdkg
+  name: antrea-config-6kg9kdbg49
   namespace: kube-system
 ---
 apiVersion: v1
@@ -1391,7 +1392,7 @@ spec:
         key: node-role.kubernetes.io/master
       volumes:
       - configMap:
-          name: antrea-config-2k6g59bdkg
+          name: antrea-config-6kg9kdbg49
         name: antrea-config
       - name: antrea-controller-tls
         secret:
@@ -1690,7 +1691,7 @@ spec:
         operator: Exists
       volumes:
       - configMap:
-          name: antrea-config-2k6g59bdkg
+          name: antrea-config-6kg9kdbg49
         name: antrea-config
       - hostPath:
           path: /etc/cni/net.d

build/yamls/antrea-windows.yml

@@ -56,9 +56,10 @@ data:
     # Enable metrics exposure via Prometheus. Initializes Prometheus metrics listener.
     #enablePrometheusMetrics: true
 
-    # Provide flow collector address as string with format <IP>:<port>[:<proto>], where proto is tcp or udp. This also enables
-    # the flow exporter that sends IPFIX flow records of conntrack flows on OVS bridge. If no L4 transport proto is given,
-    # we consider tcp as default.
+    # Provide flow collector address as string with format <IP>:<port>[:<proto>], where proto is tcp or udp.
+    # IP can be either IPv4 or IPv6. However, IPv6 address should be wrapped with [].
+    # This also enables the flow exporter that sends IPFIX flow records of conntrack flows on OVS bridge.
+    # If no L4 transport proto is given, we consider tcp as default.
     #flowCollectorAddr: ""
 
     # Provide flow poll interval as a duration string. This determines how often the flow exporter dumps connections from the conntrack module.
@@ -88,7 +89,7 @@ kind: ConfigMap
 metadata:
   labels:
     app: antrea
-  name: antrea-windows-config-6d4gc5kdc8
+  name: antrea-windows-config-kc6bfhk4mg
   namespace: kube-system
 ---
 apiVersion: apps/v1
@@ -176,7 +177,7 @@ spec:
         operator: Exists
       volumes:
       - configMap:
-          name: antrea-windows-config-6d4gc5kdc8
+          name: antrea-windows-config-kc6bfhk4mg
         name: antrea-windows-config
       - configMap:
           defaultMode: 420

build/yamls/antrea.yml

@@ -1205,9 +1205,10 @@ data:
     # Enable metrics exposure via Prometheus. Initializes Prometheus metrics listener.
     #enablePrometheusMetrics: true
 
-    # Provide flow collector address as string with format <IP>:<port>[:<proto>], where proto is tcp or udp. This also enables
-    # the flow exporter that sends IPFIX flow records of conntrack flows on OVS bridge. If no L4 transport proto is given,
-    # we consider tcp as default.
+    # Provide flow collector address as string with format <IP>:<port>[:<proto>], where proto is tcp or udp.
+    # IP can be either IPv4 or IPv6. However, IPv6 address should be wrapped with [].
+    # This also enables the flow exporter that sends IPFIX flow records of conntrack flows on OVS bridge.
+    # If no L4 transport proto is given, we consider tcp as default.
     #flowCollectorAddr: ""
 
     # Provide flow poll interval as a duration string. This determines how often the flow exporter dumps connections from the conntrack module.
@@ -1275,7 +1276,7 @@ metadata:
   annotations: {}
   labels:
     app: antrea
-  name: antrea-config-9c7h568bgf
+  name: antrea-config-669cb7d7kt
   namespace: kube-system
 ---
 apiVersion: v1
@@ -1382,7 +1383,7 @@ spec:
         key: node-role.kubernetes.io/master
       volumes:
       - configMap:
-          name: antrea-config-9c7h568bgf
+          name: antrea-config-669cb7d7kt
         name: antrea-config
       - name: antrea-controller-tls
         secret:
@@ -1646,7 +1647,7 @@ spec:
         operator: Exists
       volumes:
       - configMap:
-          name: antrea-config-9c7h568bgf
+          name: antrea-config-669cb7d7kt
         name: antrea-config
       - hostPath:
           path: /etc/cni/net.d

build/yamls/base/conf/antrea-agent.conf

@@ -89,9 +89,10 @@ featureGates:
 # Enable metrics exposure via Prometheus. Initializes Prometheus metrics listener.
 #enablePrometheusMetrics: true
 
-# Provide flow collector address as string with format <IP>:<port>[:<proto>], where proto is tcp or udp. This also enables
-# the flow exporter that sends IPFIX flow records of conntrack flows on OVS bridge. If no L4 transport proto is given,
-# we consider tcp as default.
+# Provide flow collector address as string with format <IP>:<port>[:<proto>], where proto is tcp or udp.
+# IP can be either IPv4 or IPv6. However, IPv6 address should be wrapped with [].
+# This also enables the flow exporter that sends IPFIX flow records of conntrack flows on OVS bridge.
+# If no L4 transport proto is given, we consider tcp as default.
 #flowCollectorAddr: ""
 
 # Provide flow poll interval as a duration string. This determines how often the flow exporter dumps connections from the conntrack module.

build/yamls/windows/base/conf/antrea-agent.conf

@@ -38,9 +38,10 @@ featureGates:
 # Enable metrics exposure via Prometheus. Initializes Prometheus metrics listener.
 #enablePrometheusMetrics: true
 
-# Provide flow collector address as string with format <IP>:<port>[:<proto>], where proto is tcp or udp. This also enables
-# the flow exporter that sends IPFIX flow records of conntrack flows on OVS bridge. If no L4 transport proto is given,
-# we consider tcp as default.
+# Provide flow collector address as string with format <IP>:<port>[:<proto>], where proto is tcp or udp.
+# IP can be either IPv4 or IPv6. However, IPv6 address should be wrapped with [].
+# This also enables the flow exporter that sends IPFIX flow records of conntrack flows on OVS bridge.
+# If no L4 transport proto is given, we consider tcp as default.
 #flowCollectorAddr: ""
 
 # Provide flow poll interval as a duration string. This determines how often the flow exporter dumps connections from the conntrack module.

ci/kind/test-e2e-kind.sh

@@ -91,7 +91,7 @@ if $np; then
     manifest_args="$manifest_args --np --tun vxlan"
 fi
 
-COMMON_IMAGES_LIST=("gcr.io/kubernetes-e2e-test-images/agnhost:2.8" "projects.registry.vmware.com/library/busybox" "projects.registry.vmware.com/antrea/nginx" "projects.registry.vmware.com/antrea/perftool" "projects.registry.vmware.com/antrea/ipfix-collector")
+COMMON_IMAGES_LIST=("gcr.io/kubernetes-e2e-test-images/agnhost:2.8" "projects.registry.vmware.com/library/busybox" "projects.registry.vmware.com/antrea/nginx" "projects.registry.vmware.com/antrea/perftool" "projects.registry.vmware.com/antrea/ipfix-collector:v0.3.1")
 for image in "${COMMON_IMAGES_LIST[@]}"; do
     docker pull $image
 done

cmd/antrea-agent/agent.go

@@ -305,9 +305,14 @@ func run(o *Options) error {
 
 	// Initialize flow exporter to start go routines to poll conntrack flows and export IPFIX flow records
 	if features.DefaultFeatureGate.Enabled(features.FlowExporter) {
+		v4Enabled := config.IsIPv4Enabled(nodeConfig, networkConfig.TrafficEncapMode)
+		v6Enabled := config.IsIPv6Enabled(nodeConfig, networkConfig.TrafficEncapMode)
+
 		connStore := connections.NewConnectionStore(
-			connections.InitializeConnTrackDumper(nodeConfig, serviceCIDRNet, o.config.OVSDatapathType, features.DefaultFeatureGate.Enabled(features.AntreaProxy)),
+			connections.InitializeConnTrackDumper(nodeConfig, serviceCIDRNet, serviceCIDRNetv6, o.config.OVSDatapathType, features.DefaultFeatureGate.Enabled(features.AntreaProxy)),
 			ifaceStore,
+			v4Enabled,
+			v6Enabled,
 			proxier,
 			networkPolicyController,
 			o.pollInterval)
@@ -316,7 +321,9 @@ func run(o *Options) error {
 
 		flowExporter := exporter.NewFlowExporter(
 			flowrecords.NewFlowRecords(connStore),
-			o.config.FlowExportFrequency)
+			o.config.FlowExportFrequency,
+			v4Enabled,
+			v6Enabled)
 		go wait.Until(func() { flowExporter.Export(o.flowCollector, stopCh, pollDone) }, 0, stopCh)
 	}
 

cmd/antrea-agent/options.go

@@ -18,6 +18,7 @@ import (
 	"fmt"
 	"io/ioutil"
 	"net"
+	"regexp"
 	"strings"
 	"time"
 
@@ -194,7 +195,10 @@ func (o *Options) validateFlowExporterConfig() error {
 			return fmt.Errorf("IPFIX flow collector address should be provided")
 		} else {
 			// Check if it is TCP or UDP
-			strSlice := strings.Split(o.config.FlowCollectorAddr, ":")
+			strSlice, err := parseFlowCollectorAddr(o.config.FlowCollectorAddr)
+			if err != nil {
+				return err
+			}
 			var proto string
 			if len(strSlice) == 2 {
 				// If no separator ":" and proto is given, then default to TCP.
@@ -210,7 +214,7 @@ func (o *Options) validateFlowExporterConfig() error {
 
 			// Convert the string input in net.Addr format
 			hostPortAddr := strSlice[0] + ":" + strSlice[1]
-			_, _, err := net.SplitHostPort(hostPortAddr)
+			_, _, err = net.SplitHostPort(hostPortAddr)
 			if err != nil {
 				return fmt.Errorf("IPFIX flow collector is given in invalid format: %v", err)
 			}
@@ -239,3 +243,19 @@ func (o *Options) validateFlowExporterConfig() error {
 	}
 	return nil
 }
+
+func parseFlowCollectorAddr(addr string) ([]string, error) {
+	var strSlice []string
+	match, err := regexp.MatchString("\\[.*\\]:.*", addr)
+	if err != nil {
+		return strSlice, fmt.Errorf("Failed to parse FlowCollectorAddr: %s", addr)
+	}
+	if match {
+		idx := strings.Index(addr, "]")
+		strSlice = append(strSlice, addr[:idx+1])
+		strSlice = append(strSlice, strings.Split(addr[idx+2:], ":")...)
+	} else {
+		strSlice = strings.Split(addr, ":")
+	}
+	return strSlice, nil
+}

cmd/antrea-agent/options_test.go

@@ -51,3 +51,28 @@ func TestOptions_validateFlowExporterConfig(t *testing.T) {
 	}
 
 }
+
+func TestParseFlowCollectorAddr(t *testing.T) {
+	testcases := []struct {
+		addr     string
+		expected []string
+	}{
+		{
+			"1.2.3.4:80:udp",
+			[]string{"1.2.3.4", "80", "udp"},
+		},
+		{
+			"1.2.3.4:80",
+			[]string{"1.2.3.4", "80"},
+		},
+		{
+			"[fe80:ffff:ffff:ffff:ffff:ffff:ffff:ffff]:80:tcp",
+			[]string{"[fe80:ffff:ffff:ffff:ffff:ffff:ffff:ffff]", "80", "tcp"},
+		},
+	}
+	for _, tc := range testcases {
+		res, err := parseFlowCollectorAddr(tc.addr)
+		assert.Nil(t, err)
+		assert.Equal(t, tc.expected, res)
+	}
+}