Add single stack IPv6 support in Flow aggregator and ELK flow collect…

…or (#1819)

In this commit, we add single stack IPv6 support in flow aggregator
 and elk flow collector.

build/yamls/elk-flow-collector/elk-flow-collector.yml

@@ -82,7 +82,9 @@ spec:
             - name: bootstrap.memory_lock
               value: "false"
             - name: network.host
-              value: "0.0.0.0"
+              valueFrom:
+                fieldRef:
+                  fieldPath: status.podIP
             - name: http.port
               value: "9200"
             - name: discovery.type
@@ -155,7 +157,9 @@ spec:
             - name: action.destructive_requires_name
               value: "true"
             - name: SERVER_HOST
-              value: "0.0.0.0"
+              valueFrom:
+                fieldRef:
+                  fieldPath: status.podIP
             - name: SERVER_PORT
               value: "5601"
             - name: ELASTICSEARCH_URL
@@ -201,6 +205,11 @@ spec:
       containers:
         - name: logstash
           image: docker.elastic.co/logstash/logstash-oss:7.8.0
+          env:
+            - name: POD_IP
+              valueFrom:
+                fieldRef:
+                  fieldPath: status.podIP
           volumeMounts:
             - name: logstash-definition-volume
               mountPath: /usr/share/logstash/definitions
@@ -231,4 +240,4 @@ spec:
             name: logstash-configmap
             items:
               - key: logstash.conf
-                path: logstash.conf
+                path: logstash.conf

build/yamls/elk-flow-collector/logstash/filter.rb

@@ -20,6 +20,16 @@ def filter(event)
         event.remove("[ipfix][protocolIdentifier]")
         event.set("[ipfix][protocolIdentifier]", "UDP")
     end
+    if event.get("[ipfix][destinationIPv6Address]").nil?
+        event.set("[ipfix][destinationIP]", event.get("[ipfix][destinationIPv4Address]"))
+    else
+        event.set("[ipfix][destinationIP]", event.get("[ipfix][destinationIPv6Address]"))
+    end
+    if event.get("[ipfix][sourceIPv6Address]").nil?
+        event.set("[ipfix][sourceIP]", event.get("[ipfix][sourceIPv4Address]"))
+    else
+        event.set("[ipfix][sourceIP]", event.get("[ipfix][sourceIPv6Address]"))
+    end
     if event.get("[ipfix][sourcePodName]") != ""
         if event.get("[ipfix][destinationServicePortName]") != ""
             flowkey = ""
@@ -51,7 +61,7 @@ def filter(event)
             flowkey << ":"
             flowkey << event.get("[ipfix][sourceTransportPort]").to_s
             flowkey << "->"
-            flowkey << event.get("[ipfix][destinationIPv4Address]")
+            flowkey << event.get("[ipfix][destinationIP]")
             flowkey << ":"
             flowkey << event.get("[ipfix][destinationTransportPort]").to_s
             flowkey << " "

build/yamls/elk-flow-collector/logstash/logstash.conf

@@ -4,7 +4,7 @@
 
 input {
   udp {
-    host => "0.0.0.0"
+    host => "${POD_IP}"
     port => "4739"
     workers => "4"
     queue_size => "2048"

build/yamls/elk-flow-collector/logstash/logstash.yml

@@ -1,2 +1,2 @@
-http.host: "0.0.0.0"
-path.config: /usr/share/logstash/pipeline
+http.host: "${POD_IP}"
+path.config: /usr/share/logstash/pipeline

ci/jenkins/test.sh

@@ -289,6 +289,7 @@ function deliver_antrea {
         docker tag "${DOCKER_REGISTRY}/antrea/sonobuoy-systemd-logs:v0.3" "sonobuoy/systemd-logs:v0.3"
     fi
     DOCKER_REGISTRY=${DOCKER_REGISTRY} make
+    DOCKER_REGISTRY=${DOCKER_REGISTRY} make flow-aggregator-ubuntu
 
     echo "====== Delivering Antrea to all the Nodes ======"
     echo "=== Fill serviceCIDRv6 and serviceCIDR ==="
@@ -311,10 +312,13 @@ function deliver_antrea {
 
     cp -f build/yamls/*.yml $WORKDIR
     docker save -o antrea-ubuntu.tar projects.registry.vmware.com/antrea/antrea-ubuntu:latest
+    docker save -o flow-aggregator.tar projects.registry.vmware.com/antrea/flow-aggregator:latest
 
     kubectl get nodes -o wide --no-headers=true | awk -v role="$CONTROL_PLANE_NODE_ROLE" '$3 != role {print $6}' | while read IP; do
         rsync -avr --progress --inplace -e "ssh -o StrictHostKeyChecking=no" antrea-ubuntu.tar jenkins@[${IP}]:${WORKDIR}/antrea-ubuntu.tar
+        rsync -avr --progress --inplace -e "ssh -o StrictHostKeyChecking=no" flow-aggregator.tar jenkins@[${IP}]:${WORKDIR}/flow-aggregator.tar
         ssh -o StrictHostKeyChecking=no -n jenkins@${IP} "docker images | grep 'antrea-ubuntu' | awk '{print \$3}' | xargs -r docker rmi ; docker load -i ${WORKDIR}/antrea-ubuntu.tar ; docker images | grep '<none>' | awk '{print \$3}' | xargs -r docker rmi" || true
+        ssh -o StrictHostKeyChecking=no -n jenkins@${IP} "docker images | grep 'flow-aggregator' | awk '{print \$3}' | xargs -r docker rmi ; docker load -i ${WORKDIR}/flow-aggregator.tar ; docker images | grep '<none>' | awk '{print \$3}' | xargs -r docker rmi" || true
         if [[ "${DOCKER_REGISTRY}" != "" ]]; then
             ssh -o StrictHostKeyChecking=no -n jenkins@${IP} "docker pull ${DOCKER_REGISTRY}/antrea/sonobuoy-systemd-logs:v0.3 ; docker tag ${DOCKER_REGISTRY}/antrea/sonobuoy-systemd-logs:v0.3 sonobuoy/systemd-logs:v0.3"
         fi

pkg/flowaggregator/flowaggregator.go

@@ -16,6 +16,7 @@ package flowaggregator
 
 import (
 	"fmt"
+	"net"
 	"time"
 
 	"github.com/vmware/go-ipfix/pkg/collector"
@@ -30,7 +31,7 @@ import (
 )
 
 var (
-	ianaInfoElements = []string{
+	ianaInfoElementsCommon = []string{
 		"flowStartSeconds",
 		"flowEndSeconds",
 		"flowEndReason",
@@ -41,16 +42,16 @@ var (
 		"octetTotalCount",
 		"packetDeltaCount",
 		"octetDeltaCount",
-		"sourceIPv4Address",
-		"destinationIPv4Address",
 	}
+	ianaInfoElementsIPv4    = append(ianaInfoElementsCommon, []string{"sourceIPv4Address", "destinationIPv4Address"}...)
+	ianaInfoElementsIPv6    = append(ianaInfoElementsCommon, []string{"sourceIPv6Address", "destinationIPv6Address"}...)
 	ianaReverseInfoElements = []string{
 		"reversePacketTotalCount",
 		"reverseOctetTotalCount",
 		"reversePacketDeltaCount",
 		"reverseOctetDeltaCount",
 	}
-	antreaInfoElements = []string{
+	antreaInfoElementsCommon = []string{
 		"sourcePodName",
 		"sourcePodNamespace",
 		"sourceNodeName",
@@ -63,12 +64,14 @@ var (
 		"ingressNetworkPolicyNamespace",
 		"egressNetworkPolicyName",
 		"egressNetworkPolicyNamespace",
-		"destinationClusterIPv4",
 	}
-	aggregatorElements = []string{
-		"originalExporterIPv4Address",
+	antreaInfoElementsIPv4   = append(antreaInfoElementsCommon, []string{"destinationClusterIPv4"}...)
+	antreaInfoElementsIPv6   = append(antreaInfoElementsCommon, []string{"destinationClusterIPv6"}...)
+	aggregatorElementsCommon = []string{
 		"originalObservationDomainId",
 	}
+	aggregatorElementsIPv4 = append([]string{"originalExporterIPv4Address"}, aggregatorElementsCommon...)
+	aggregatorElementsIPv6 = append([]string{"originalExporterIPv6Address"}, aggregatorElementsCommon...)
 
 	nonStatsElementList = []string{
 		"flowEndSeconds",
@@ -118,6 +121,7 @@ var (
 		"destinationPodNamespace",
 		"destinationNodeName",
 		"destinationClusterIPv4",
+		"destinationClusterIPv6",
 		"destinationServicePort",
 		"destinationServicePortName",
 		"ingressNetworkPolicyName",
@@ -150,7 +154,8 @@ type flowAggregator struct {
 	aggregationProcess          ipfix.IPFIXAggregationProcess
 	exportInterval              time.Duration
 	exportingProcess            ipfix.IPFIXExportingProcess
-	templateID                  uint16
+	templateIDv4                uint16
+	templateIDv6                uint16
 	registry                    ipfix.IPFIXRegistry
 	set                         ipfix.IPFIXSet
 	flowAggregatorAddress       string
@@ -178,6 +183,7 @@ func NewFlowAggregator(
 		exportInterval,
 		nil,
 		0,
+		0,
 		registry,
 		ipfix.NewSet(false),
 		flowAggregatorAddress,
@@ -281,20 +287,36 @@ func (fa *flowAggregator) initExportingProcess() error {
 		return fmt.Errorf("got error when initializing IPFIX exporting process: %v", err)
 	}
 	fa.exportingProcess = ep
-	fa.templateID = fa.exportingProcess.NewTemplateID()
-	if err := fa.set.PrepareSet(ipfixentities.Template, fa.templateID); err != nil {
-		return fmt.Errorf("error when preparing set: %v", err)
+	// Currently, we send two templates for IPv4 and IPv6 regardless of the IP families supported by cluster
+	fa.templateIDv4 = fa.exportingProcess.NewTemplateID()
+	if err := fa.set.PrepareSet(ipfixentities.Template, fa.templateIDv4); err != nil {
+		return fmt.Errorf("error when preparing IPv4 template set: %v", err)
 	}
+	bytesSent, err := fa.sendTemplateSet(fa.set, false)
+	if err != nil {
+		fa.exportingProcess.CloseConnToCollector()
+		fa.exportingProcess = nil
+		fa.templateIDv4 = 0
+		fa.set.ResetSet()
+		return fmt.Errorf("sending IPv4 template set failed, err: %v", err)
+	}
+	klog.V(2).Infof("Initialized exporting process and sent %d bytes size of IPv4 template set", bytesSent)
 
-	bytesSent, err := fa.sendTemplateSet(fa.set)
+	fa.set.ResetSet()
+	fa.templateIDv6 = fa.exportingProcess.NewTemplateID()
+	if err := fa.set.PrepareSet(ipfixentities.Template, fa.templateIDv6); err != nil {
+		return fmt.Errorf("error when preparing IPv6 template set: %v", err)
+	}
+	bytesSent, err = fa.sendTemplateSet(fa.set, true)
 	if err != nil {
 		fa.exportingProcess.CloseConnToCollector()
 		fa.exportingProcess = nil
-		fa.templateID = 0
+		fa.templateIDv6 = 0
 		fa.set.ResetSet()
-		return fmt.Errorf("sending template set failed, err: %v", err)
+		return fmt.Errorf("sending IPv6 template set failed, err: %v", err)
 	}
-	klog.V(2).Infof("Initialized exporting process and sent %d bytes size of template set", bytesSent)
+	klog.V(2).Infof("Initialized exporting process and sent %d bytes size of IPv6 template set", bytesSent)
+
 	return nil
 }
 
@@ -339,12 +361,16 @@ func (fa *flowAggregator) sendFlowKeyRecord(key ipfixintermediate.FlowKey, recor
 		klog.V(4).Info("Skip sending record that is not correlated.")
 		return nil
 	}
+	templateID := fa.templateIDv4
+	if net.ParseIP(key.SourceAddress).To4() == nil || net.ParseIP(key.DestinationAddress).To4() == nil {
+		templateID = fa.templateIDv6
+	}
 	// TODO: more records per data set will be supported when go-ipfix supports size check when adding records
 	fa.set.ResetSet()
-	if err := fa.set.PrepareSet(ipfixentities.Data, fa.templateID); err != nil {
+	if err := fa.set.PrepareSet(ipfixentities.Data, templateID); err != nil {
 		return fmt.Errorf("error when preparing set: %v", err)
 	}
-	err := fa.set.AddRecord(record.Record.GetOrderedElementList(), fa.templateID)
+	err := fa.set.AddRecord(record.Record.GetOrderedElementList(), templateID)
 	if err != nil {
 		return fmt.Errorf("error when adding the record to the set: %v", err)
 	}
@@ -356,8 +382,18 @@ func (fa *flowAggregator) sendFlowKeyRecord(key ipfixintermediate.FlowKey, recor
 	return nil
 }
 
-func (fa *flowAggregator) sendTemplateSet(templateSet ipfix.IPFIXSet) (int, error) {
+func (fa *flowAggregator) sendTemplateSet(templateSet ipfix.IPFIXSet, isIPv6 bool) (int, error) {
 	elements := make([]*ipfixentities.InfoElementWithValue, 0)
+	ianaInfoElements := ianaInfoElementsIPv4
+	antreaInfoElements := antreaInfoElementsIPv4
+	aggregatorElements := aggregatorElementsIPv4
+	templateID := fa.templateIDv4
+	if isIPv6 {
+		ianaInfoElements = ianaInfoElementsIPv6
+		antreaInfoElements = antreaInfoElementsIPv6
+		aggregatorElements = aggregatorElementsIPv6
+		templateID = fa.templateIDv6
+	}
 	for _, ie := range ianaInfoElements {
 		element, err := fa.registry.GetInfoElement(ie, ipfixregistry.IANAEnterpriseID)
 		if err != nil {
@@ -406,7 +442,7 @@ func (fa *flowAggregator) sendTemplateSet(templateSet ipfix.IPFIXSet) (int, erro
 		ie := ipfixentities.NewInfoElementWithValue(element, nil)
 		elements = append(elements, ie)
 	}
-	err := templateSet.AddRecord(elements, fa.templateID)
+	err := templateSet.AddRecord(elements, templateID)
 	if err != nil {
 		return 0, fmt.Errorf("error when adding record to set, error: %v", err)
 	}

pkg/flowaggregator/flowaggregator_test.go

@@ -28,7 +28,8 @@ import (
 )
 
 const (
-	testTemplateID          = uint16(256)
+	testTemplateIDv4        = uint16(256)
+	testTemplateIDv6        = uint16(257)
 	testExportInterval      = 60 * time.Second
 	testObservationDomainID = 0xabcd
 )
@@ -52,47 +53,59 @@ func TestFlowAggregator_sendTemplateSet(t *testing.T) {
 		nil,
 		testExportInterval,
 		mockIPFIXExpProc,
-		testTemplateID,
+		testTemplateIDv4,
+		testTemplateIDv6,
 		mockIPFIXRegistry,
 		ipfixtest.NewMockIPFIXSet(ctrl),
 		"",
 		nil,
 		testObservationDomainID,
 	}
 
-	// Following consists of all elements that are in ianaInfoElements and antreaInfoElements (globals)
-	// Only the element name is needed, other arguments have dummy values.
-	elemList := make([]*ipfixentities.InfoElementWithValue, 0)
-	for i, ie := range ianaInfoElements {
-		elemList = append(elemList, ipfixentities.NewInfoElementWithValue(ipfixentities.NewInfoElement(ie, 0, 0, ipfixregistry.IANAEnterpriseID, 0), nil))
-		mockIPFIXRegistry.EXPECT().GetInfoElement(ie, ipfixregistry.IANAEnterpriseID).Return(elemList[i].Element, nil)
-	}
-	for i, ie := range ianaReverseInfoElements {
-		elemList = append(elemList, ipfixentities.NewInfoElementWithValue(ipfixentities.NewInfoElement(ie, 0, 0, ipfixregistry.IANAReversedEnterpriseID, 0), nil))
-		mockIPFIXRegistry.EXPECT().GetInfoElement(ie, ipfixregistry.IANAReversedEnterpriseID).Return(elemList[i+len(ianaInfoElements)].Element, nil)
-	}
-	for i, ie := range antreaInfoElements {
-		elemList = append(elemList, ipfixentities.NewInfoElementWithValue(ipfixentities.NewInfoElement(ie, 0, 0, ipfixregistry.AntreaEnterpriseID, 0), nil))
-		mockIPFIXRegistry.EXPECT().GetInfoElement(ie, ipfixregistry.AntreaEnterpriseID).Return(elemList[i+len(ianaInfoElements)+len(ianaReverseInfoElements)].Element, nil)
-	}
-	for i, ie := range aggregatorElements {
-		elemList = append(elemList, ipfixentities.NewInfoElementWithValue(ipfixentities.NewInfoElement(ie, 0, 0, ipfixregistry.IANAEnterpriseID, 0), nil))
-		mockIPFIXRegistry.EXPECT().GetInfoElement(ie, ipfixregistry.IANAEnterpriseID).Return(elemList[i+len(ianaInfoElements)+len(ianaReverseInfoElements)+len(antreaInfoElements)].Element, nil)
-	}
-	for i, ie := range antreaSourceStatsElementList {
-		elemList = append(elemList, ipfixentities.NewInfoElementWithValue(ipfixentities.NewInfoElement(ie, 0, 0, ipfixregistry.AntreaEnterpriseID, 0), nil))
-		mockIPFIXRegistry.EXPECT().GetInfoElement(ie, ipfixregistry.AntreaEnterpriseID).Return(elemList[i+len(ianaInfoElements)+len(ianaReverseInfoElements)+len(antreaInfoElements)+len(aggregatorElements)].Element, nil)
-	}
+	for _, isIPv6 := range []bool{false, true} {
+		ianaInfoElements := ianaInfoElementsIPv4
+		antreaInfoElements := antreaInfoElementsIPv4
+		aggregatorElements := aggregatorElementsIPv4
+		testTemplateID := fa.templateIDv4
+		if isIPv6 {
+			ianaInfoElements = ianaInfoElementsIPv6
+			antreaInfoElements = antreaInfoElementsIPv6
+			aggregatorElements = aggregatorElementsIPv6
+			testTemplateID = fa.templateIDv6
+		}
+		// Following consists of all elements that are in ianaInfoElements and antreaInfoElements (globals)
+		// Only the element name is needed, other arguments have dummy values.
+		elemList := make([]*ipfixentities.InfoElementWithValue, 0)
+		for i, ie := range ianaInfoElements {
+			elemList = append(elemList, ipfixentities.NewInfoElementWithValue(ipfixentities.NewInfoElement(ie, 0, 0, ipfixregistry.IANAEnterpriseID, 0), nil))
+			mockIPFIXRegistry.EXPECT().GetInfoElement(ie, ipfixregistry.IANAEnterpriseID).Return(elemList[i].Element, nil)
+		}
+		for i, ie := range ianaReverseInfoElements {
+			elemList = append(elemList, ipfixentities.NewInfoElementWithValue(ipfixentities.NewInfoElement(ie, 0, 0, ipfixregistry.IANAReversedEnterpriseID, 0), nil))
+			mockIPFIXRegistry.EXPECT().GetInfoElement(ie, ipfixregistry.IANAReversedEnterpriseID).Return(elemList[i+len(ianaInfoElements)].Element, nil)
+		}
+		for i, ie := range antreaInfoElements {
+			elemList = append(elemList, ipfixentities.NewInfoElementWithValue(ipfixentities.NewInfoElement(ie, 0, 0, ipfixregistry.AntreaEnterpriseID, 0), nil))
+			mockIPFIXRegistry.EXPECT().GetInfoElement(ie, ipfixregistry.AntreaEnterpriseID).Return(elemList[i+len(ianaInfoElements)+len(ianaReverseInfoElements)].Element, nil)
+		}
+		for i, ie := range aggregatorElements {
+			elemList = append(elemList, ipfixentities.NewInfoElementWithValue(ipfixentities.NewInfoElement(ie, 0, 0, ipfixregistry.IANAEnterpriseID, 0), nil))
+			mockIPFIXRegistry.EXPECT().GetInfoElement(ie, ipfixregistry.IANAEnterpriseID).Return(elemList[i+len(ianaInfoElements)+len(ianaReverseInfoElements)+len(antreaInfoElements)].Element, nil)
+		}
+		for i, ie := range antreaSourceStatsElementList {
+			elemList = append(elemList, ipfixentities.NewInfoElementWithValue(ipfixentities.NewInfoElement(ie, 0, 0, ipfixregistry.AntreaEnterpriseID, 0), nil))
+			mockIPFIXRegistry.EXPECT().GetInfoElement(ie, ipfixregistry.AntreaEnterpriseID).Return(elemList[i+len(ianaInfoElements)+len(ianaReverseInfoElements)+len(antreaInfoElements)+len(aggregatorElements)].Element, nil)
+		}
+		for i, ie := range antreaDestinationStatsElementList {
+			elemList = append(elemList, ipfixentities.NewInfoElementWithValue(ipfixentities.NewInfoElement(ie, 0, 0, ipfixregistry.AntreaEnterpriseID, 0), nil))
+			mockIPFIXRegistry.EXPECT().GetInfoElement(ie, ipfixregistry.AntreaEnterpriseID).Return(elemList[i+len(ianaInfoElements)+len(ianaReverseInfoElements)+len(antreaInfoElements)+len(aggregatorElements)+len(antreaSourceStatsElementList)].Element, nil)
+		}
+		mockTempSet.EXPECT().AddRecord(elemList, testTemplateID).Return(nil)
+		// Passing 0 for sentBytes as it is not used anywhere in the test. If this not a call to mock, the actual sentBytes
+		// above elements: ianaInfoElements, ianaReverseInfoElements and antreaInfoElements.
+		mockIPFIXExpProc.EXPECT().SendSet(mockTempSet).Return(0, nil)
 
-	for i, ie := range antreaDestinationStatsElementList {
-		elemList = append(elemList, ipfixentities.NewInfoElementWithValue(ipfixentities.NewInfoElement(ie, 0, 0, ipfixregistry.AntreaEnterpriseID, 0), nil))
-		mockIPFIXRegistry.EXPECT().GetInfoElement(ie, ipfixregistry.AntreaEnterpriseID).Return(elemList[i+len(ianaInfoElements)+len(ianaReverseInfoElements)+len(antreaInfoElements)+len(aggregatorElements)+len(antreaSourceStatsElementList)].Element, nil)
+		_, err := fa.sendTemplateSet(mockTempSet, isIPv6)
+		assert.NoErrorf(t, err, "Error in sending template record: %v, isIPv6: %v", err, isIPv6)
 	}
-	mockTempSet.EXPECT().AddRecord(elemList, testTemplateID).Return(nil)
-	// Passing 0 for sentBytes as it is not used anywhere in the test. If this not a call to mock, the actual sentBytes
-	// above elements: ianaInfoElements, ianaReverseInfoElements and antreaInfoElements.
-	mockIPFIXExpProc.EXPECT().SendSet(mockTempSet).Return(0, nil)
-
-	_, err := fa.sendTemplateSet(mockTempSet)
-	assert.NoErrorf(t, err, "Error in sending template record: %v", err)
 }

pkg/util/flowexport/flowexport.go

@@ -33,7 +33,7 @@ func ParseFlowCollectorAddr(addr string, defaultPort string, defaultProtocol str
 	}
 	if match {
 		idx := strings.Index(addr, "]")
-		strSlice = append(strSlice, addr[:idx+1])
+		strSlice = append(strSlice, addr[1:idx])
 		strSlice = append(strSlice, strings.Split(addr[idx+2:], ":")...)
 	} else {
 		strSlice = strings.Split(addr, ":")

pkg/util/flowexport/flowexport_test.go

@@ -55,7 +55,7 @@ func TestParseFlowCollectorAddr(t *testing.T) {
 		},
 		{
 			addr:          "[fe80:ffff:ffff:ffff:ffff:ffff:ffff:ffff]:80:tcp",
-			expectedHost:  "[fe80:ffff:ffff:ffff:ffff:ffff:ffff:ffff]",
+			expectedHost:  "fe80:ffff:ffff:ffff:ffff:ffff:ffff:ffff",
 			expectedPort:  "80",
 			expectedProto: "tcp",
 			expectedError: nil,