Add flowDirection IE when proxying records in FlowAggregator (#6979)

This can help identify if the flow was exported from the source or
destination Node in a more "standard" way. Note that we use a "special"
value (0xff) for intra-Node flows.

Signed-off-by: Antonin Bas <[email protected]>

Author: antoninbas

pkg/flowaggregator/exporter/ipfix.go

@@ -255,22 +255,22 @@ func (e *IPFIXExporter) sendTemplateSet(isIPv6 bool) (int, error) {
 		antreaInfoElements = infoelements.AntreaInfoElementsIPv6
 		templateID = e.templateIDv6
 	}
-	for _, ie := range ianaInfoElements {
-		ie, err := e.createInfoElementForTemplateSet(ie, ipfixregistry.IANAEnterpriseID)
+	for _, ieName := range ianaInfoElements {
+		ie, err := e.createInfoElementForTemplateSet(ieName, ipfixregistry.IANAEnterpriseID)
 		if err != nil {
 			return 0, err
 		}
 		elements = append(elements, ie)
 	}
-	for _, ie := range infoelements.IANAReverseInfoElements {
-		ie, err := e.createInfoElementForTemplateSet(ie, ipfixregistry.IANAReversedEnterpriseID)
+	for _, ieName := range infoelements.IANAReverseInfoElements {
+		ie, err := e.createInfoElementForTemplateSet(ieName, ipfixregistry.IANAReversedEnterpriseID)
 		if err != nil {
 			return 0, err
 		}
 		elements = append(elements, ie)
 	}
-	for _, ie := range antreaInfoElements {
-		ie, err := e.createInfoElementForTemplateSet(ie, ipfixregistry.AntreaEnterpriseID)
+	for _, ieName := range antreaInfoElements {
+		ie, err := e.createInfoElementForTemplateSet(ieName, ipfixregistry.AntreaEnterpriseID)
 		if err != nil {
 			return 0, err
 		}
@@ -295,8 +295,8 @@ func (e *IPFIXExporter) sendTemplateSet(isIPv6 bool) (int, error) {
 			}
 			elements = append(elements, ie)
 		}
-		for _, ie := range infoelements.AntreaFlowEndSecondsElementList {
-			ie, err := e.createInfoElementForTemplateSet(ie, ipfixregistry.AntreaEnterpriseID)
+		for _, ieName := range infoelements.AntreaFlowEndSecondsElementList {
+			ie, err := e.createInfoElementForTemplateSet(ieName, ipfixregistry.AntreaEnterpriseID)
 			if err != nil {
 				return 0, err
 			}
@@ -326,8 +326,8 @@ func (e *IPFIXExporter) sendTemplateSet(isIPv6 bool) (int, error) {
 			elements = append(elements, ie)
 		}
 	}
-	for _, ie := range infoelements.AntreaLabelsElementList {
-		ie, err := e.createInfoElementForTemplateSet(ie, ipfixregistry.AntreaEnterpriseID)
+	for _, ieName := range infoelements.AntreaLabelsElementList {
+		ie, err := e.createInfoElementForTemplateSet(ieName, ipfixregistry.AntreaEnterpriseID)
 		if err != nil {
 			return 0, err
 		}
@@ -339,21 +339,13 @@ func (e *IPFIXExporter) sendTemplateSet(isIPv6 bool) (int, error) {
 	}
 	elements = append(elements, ie)
 	if e.aggregatorMode == flowaggregatorconfig.AggregatorModeProxy {
-		ie, err := e.createInfoElementForTemplateSet("originalObservationDomainId", ipfixregistry.IANAEnterpriseID)
-		if err != nil {
-			return 0, err
-		}
-		elements = append(elements, ie)
-		ie, err = e.createInfoElementForTemplateSet("originalExporterIPv4Address", ipfixregistry.IANAEnterpriseID)
-		if err != nil {
-			return 0, err
-		}
-		elements = append(elements, ie)
-		ie, err = e.createInfoElementForTemplateSet("originalExporterIPv6Address", ipfixregistry.IANAEnterpriseID)
-		if err != nil {
-			return 0, err
+		for _, ieName := range infoelements.IANAProxyModeElementList {
+			ie, err := e.createInfoElementForTemplateSet(ieName, ipfixregistry.IANAEnterpriseID)
+			if err != nil {
+				return 0, err
+			}
+			elements = append(elements, ie)
 		}
-		elements = append(elements, ie)
 	}
 	e.set.ResetSet()
 	if err := e.set.PrepareSet(ipfixentities.Template, templateID); err != nil {

pkg/flowaggregator/flowaggregator.go

@@ -275,8 +275,7 @@ func (fa *flowAggregator) InitCollectingProcess() error {
 		cpInput.NumExtraElements += len(infoelements.AntreaSourceStatsElementList) + len(infoelements.AntreaDestinationStatsElementList) +
 			len(infoelements.AntreaFlowEndSecondsElementList) + len(infoelements.AntreaThroughputElementList) + len(infoelements.AntreaSourceThroughputElementList) + len(infoelements.AntreaDestinationThroughputElementList)
 	} else {
-		// originalObservationDomainId, originalExporterIPv4Address, originalExporterIPv6Address
-		cpInput.NumExtraElements += 3
+		cpInput.NumExtraElements += len(infoelements.IANAProxyModeElementList)
 	}
 	// Tell the collector to accept IEs which are not part of the IPFIX registry (hardcoded in
 	// the go-ipfix library). The preprocessor will take care of removing these elements.
@@ -505,7 +504,36 @@ func (fa *flowAggregator) proxyRecord(record ipfixentities.Record, obsDomainID u
 	if err != nil {
 		return fmt.Errorf("cannot find record start time: %w", err)
 	}
-	if getFlowType(record) == ipfixregistry.FlowTypeInterNode {
+	flowType := getFlowType(record)
+	var withSource, withDestination bool
+	if sourcePodName, _, exist := record.GetInfoElementWithValue("sourcePodName"); exist {
+		withSource = sourcePodName.GetStringValue() != ""
+	}
+	if destinationPodName, _, exist := record.GetInfoElementWithValue("destinationPodName"); exist {
+		withDestination = destinationPodName.GetStringValue() != ""
+	}
+	var direction uint8
+	switch {
+	// !withDestination should be redundant here
+	case flowType == ipfixregistry.FlowTypeInterNode && withSource && !withDestination:
+		// egress
+		direction = 0x01
+	// !withSource should be redundant here
+	case flowType == ipfixregistry.FlowTypeInterNode && !withSource && withDestination:
+		// ingress
+		direction = 0x00
+	case flowType == ipfixregistry.FlowTypeToExternal && withSource:
+		// egress
+		direction = 0x01
+	case flowType == ipfixregistry.FlowTypeFromExternal && withDestination:
+		// ingress
+		direction = 0x00
+	default:
+		// not a valid value for the IE, we use it as a reserved value (unknown)
+		// this covers the IntraNode case
+		direction = 0xff
+	}
+	if flowType == ipfixregistry.FlowTypeInterNode {
 		// This is the only case where K8s metadata could be missing
 		fa.fillK8sMetadata(sourceAddress, destinationAddress, record, startTime)
 	}
@@ -523,6 +551,9 @@ func (fa *flowAggregator) proxyRecord(record ipfixentities.Record, obsDomainID u
 	if err := fa.addOriginalExporterIPv6Address(record, originalExporterAddress); err != nil {
 		klog.ErrorS(err, "Failed to add originalExporterIPv6Address")
 	}
+	if err := fa.addFlowDirection(record, direction); err != nil {
+		klog.ErrorS(err, "Failed to add flowDirection")
+	}
 	return fa.sendRecord(record, isIPv6)
 }
 
@@ -824,6 +855,17 @@ func (fa *flowAggregator) addOriginalExporterIPv6Address(record ipfixentities.Re
 	return nil
 }
 
+func (fa *flowAggregator) addFlowDirection(record ipfixentities.Record, direction uint8) error {
+	ie, err := fa.registry.GetInfoElement("flowDirection", ipfixregistry.IANAEnterpriseID)
+	if err != nil {
+		return fmt.Errorf("error when getting flowDirection InfoElement: %w", err)
+	}
+	if err := record.AddInfoElement(ipfixentities.NewUnsigned8InfoElement(ie, direction)); err != nil {
+		return fmt.Errorf("error when adding flowDirection InfoElement with value: %w", err)
+	}
+	return nil
+}
+
 func (fa *flowAggregator) GetFlowRecords(flowKey *ipfixintermediate.FlowKey) []map[string]interface{} {
 	if fa.aggregationProcess != nil {
 		return fa.aggregationProcess.GetRecords(flowKey)

pkg/flowaggregator/flowaggregator_test.go

@@ -356,6 +356,9 @@ func TestFlowAggregator_proxyRecord(t *testing.T) {
 			originalExporterIPv4AddressIE := ipfixentities.NewInfoElement("originalExporterIPv4Address", 0, 0, ipfixregistry.IANAEnterpriseID, 4)
 			mockIPFIXRegistry.EXPECT().GetInfoElement("originalExporterIPv4Address", ipfixregistry.IANAEnterpriseID).Return(originalExporterIPv4AddressIE, nil)
 			mockRecord.EXPECT().AddInfoElement(ipfixentities.NewIPAddressInfoElement(originalExporterIPv4AddressIE, exporterAddressIPv4))
+			flowDirectionIE := ipfixentities.NewInfoElement("flowDirection", 0, 0, ipfixregistry.IANAEnterpriseID, 1)
+			mockIPFIXRegistry.EXPECT().GetInfoElement("flowDirection", ipfixregistry.IANAEnterpriseID).Return(flowDirectionIE, nil)
+			mockRecord.EXPECT().AddInfoElement(ipfixentities.NewUnsigned8InfoElement(flowDirectionIE, uint8(0xff)))
 
 			err := fa.proxyRecord(mockRecord, obsDomainID, exporterAddress)
 			assert.NoError(t, err, "Error when proxying flow record")

pkg/flowaggregator/infoelements/elements.go

@@ -123,4 +123,11 @@ var (
 		"throughputFromDestinationNode",
 		"reverseThroughputFromDestinationNode",
 	}
+
+	IANAProxyModeElementList = []string{
+		"originalObservationDomainId",
+		"originalExporterIPv4Address",
+		"originalExporterIPv6Address",
+		"flowDirection",
+	}
 )