Add Proxy mode for FlowAggregator

In Proxy mode, connection data is not buffered in the FlowAggregator and records from source and destination Nodes are not correlated. Instead, records are proxied directly to the destination collector. In Proxy mode, only the IPFIX exporter is supported. A simple e2e test is added to test this new mode. Fixes antrea-io#6773 Signed-off-by: Antonin Bas <[email protected]>
antoninbas · Jan 14, 2025 · 8166339 · 8166339
1 parent 5ee28ec
commit 8166339
Show file tree

Hide file tree

Showing 14 changed files with 683 additions and 168 deletions.
diff --git a/build/charts/flow-aggregator/conf/flow-aggregator.conf b/build/charts/flow-aggregator/conf/flow-aggregator.conf
@@ -1,3 +1,8 @@
+# Mode in which to run the flow aggregator. Must be one of "Aggregate" (flow records received
+# from source and destination are aggregated and sent as one flow record) or "Proxy" (flow records
+# are enhanced with some additional information, then sent directly without buffering or aggregation).
+mode: {{ .Values.mode }}
+
 # Provide the active flow record timeout as a duration string. This determines
 # how often the flow aggregator exports the active flow records to the flow
 # collector. Thus, for flows with a continuous stream of packets, a flow record

diff --git a/build/charts/flow-aggregator/values.yaml b/build/charts/flow-aggregator/values.yaml
@@ -4,6 +4,10 @@ image:
   pullPolicy: "IfNotPresent"
   tag: ""
 
+# -- Mode in which to run the flow aggregator. Must be one of "Aggregate" (flow records received
+# from source and destination are aggregated and sent as one flow record) or "Proxy" (flow records
+# are enhanced with some additional information, then sent directly without buffering or aggregation).
+mode: "Aggregate"
 # -- Provide the active flow record timeout as a duration string.
 # Valid time units are "ns", "us" (or "µs"), "ms", "s", "m", "h".
 activeFlowRecordTimeout: 60s

diff --git a/build/yamls/flow-aggregator.yml b/build/yamls/flow-aggregator.yml
@@ -154,6 +154,11 @@ subjects:
 apiVersion: v1
 data:
   flow-aggregator.conf: |
+    # Mode in which to run the flow aggregator. Must be one of "Aggregate" (flow records received
+    # from source and destination are aggregated and sent as one flow record) or "Proxy" (flow records
+    # are enhanced with some additional information, then sent directly without buffering or aggregation).
+    mode: Aggregate
+
     # Provide the active flow record timeout as a duration string. This determines
     # how often the flow aggregator exports the active flow records to the flow
     # collector. Thus, for flows with a continuous stream of packets, a flow record

diff --git a/pkg/config/flowaggregator/config.go b/pkg/config/flowaggregator/config.go
@@ -22,7 +22,15 @@ const (
 	AggregatorTransportProtocolUDP AggregatorTransportProtocol = "UDP"
 )
 
+type AggregatorMode string
+
+const (
+	AggregatorModeAggregate AggregatorMode = "Aggregate"
+	AggregatorModeProxy     AggregatorMode = "Proxy"
+)
+
 type FlowAggregatorConfig struct {
+	Mode AggregatorMode `yaml:"mode,omitempty"`
 	// Provide the active flow record timeout as a duration string. This determines
 	// how often the flow aggregator exports the active flow records to the flow
 	// collector. Thus, for flows with a continuous stream of packets, a flow record

diff --git a/pkg/config/flowaggregator/default.go b/pkg/config/flowaggregator/default.go
@@ -48,6 +48,9 @@ const (
 )
 
 func SetConfigDefaults(flowAggregatorConf *FlowAggregatorConfig) {
+	if flowAggregatorConf.Mode == "" {
+		flowAggregatorConf.Mode = AggregatorModeAggregate
+	}
 	if flowAggregatorConf.ActiveFlowRecordTimeout == "" {
 		flowAggregatorConf.ActiveFlowRecordTimeout = DefaultActiveFlowRecordTimeout
 	}

diff --git a/pkg/flowaggregator/exporter/ipfix.go b/pkg/flowaggregator/exporter/ipfix.go
@@ -45,6 +45,7 @@ type IPFIXExporter struct {
 	externalFlowCollectorProto string
 	exportingProcess           ipfix.IPFIXExportingProcess
 	sendJSONRecord             bool
+	aggregatorMode             flowaggregatorconfig.AggregatorMode
 	observationDomainID        uint32
 	templateRefreshTimeout     time.Duration
 	templateIDv4               uint16
@@ -88,6 +89,7 @@ func NewIPFIXExporter(
 		externalFlowCollectorAddr:  opt.ExternalFlowCollectorAddr,
 		externalFlowCollectorProto: opt.ExternalFlowCollectorProto,
 		sendJSONRecord:             sendJSONRecord,
+		aggregatorMode:             opt.AggregatorMode,
 		observationDomainID:        observationDomainID,
 		templateRefreshTimeout:     opt.TemplateRefreshTimeout,
 		registry:                   registry,
@@ -274,66 +276,85 @@ func (e *IPFIXExporter) sendTemplateSet(isIPv6 bool) (int, error) {
 		}
 		elements = append(elements, ie)
 	}
-	// The order of source and destination stats elements needs to match the order specified in
-	// addFieldsForStatsAggregation method in go-ipfix aggregation process.
-	for i := range infoelements.StatsElementList {
-		// Add Antrea source stats fields
-		ieName := infoelements.AntreaSourceStatsElementList[i]
-		ie, err := e.createInfoElementForTemplateSet(ieName, ipfixregistry.AntreaEnterpriseID)
-		if err != nil {
-			return 0, err
+	if e.aggregatorMode == flowaggregatorconfig.AggregatorModeAggregate {
+		// The order of source and destination stats elements needs to match the order specified in
+		// addFieldsForStatsAggregation method in go-ipfix aggregation process.
+		for i := range infoelements.StatsElementList {
+			// Add Antrea source stats fields
+			ieName := infoelements.AntreaSourceStatsElementList[i]
+			ie, err := e.createInfoElementForTemplateSet(ieName, ipfixregistry.AntreaEnterpriseID)
+			if err != nil {
+				return 0, err
+			}
+			elements = append(elements, ie)
+			// Add Antrea destination stats fields
+			ieName = infoelements.AntreaDestinationStatsElementList[i]
+			ie, err = e.createInfoElementForTemplateSet(ieName, ipfixregistry.AntreaEnterpriseID)
+			if err != nil {
+				return 0, err
+			}
+			elements = append(elements, ie)
 		}
-		elements = append(elements, ie)
-		// Add Antrea destination stats fields
-		ieName = infoelements.AntreaDestinationStatsElementList[i]
-		ie, err = e.createInfoElementForTemplateSet(ieName, ipfixregistry.AntreaEnterpriseID)
-		if err != nil {
-			return 0, err
+		for _, ie := range infoelements.AntreaFlowEndSecondsElementList {
+			ie, err := e.createInfoElementForTemplateSet(ie, ipfixregistry.AntreaEnterpriseID)
+			if err != nil {
+				return 0, err
+			}
+			elements = append(elements, ie)
+		}
+		for i := range infoelements.AntreaThroughputElementList {
+			// Add common throughput fields
+			ieName := infoelements.AntreaThroughputElementList[i]
+			ie, err := e.createInfoElementForTemplateSet(ieName, ipfixregistry.AntreaEnterpriseID)
+			if err != nil {
+				return 0, err
+			}
+			elements = append(elements, ie)
+			// Add source node specific throughput fields
+			ieName = infoelements.AntreaSourceThroughputElementList[i]
+			ie, err = e.createInfoElementForTemplateSet(ieName, ipfixregistry.AntreaEnterpriseID)
+			if err != nil {
+				return 0, err
+			}
+			elements = append(elements, ie)
+			// Add destination node specific throughput fields
+			ieName = infoelements.AntreaDestinationThroughputElementList[i]
+			ie, err = e.createInfoElementForTemplateSet(ieName, ipfixregistry.AntreaEnterpriseID)
+			if err != nil {
+				return 0, err
+			}
+			elements = append(elements, ie)
 		}
-		elements = append(elements, ie)
 	}
-	for _, ie := range infoelements.AntreaFlowEndSecondsElementList {
+	for _, ie := range infoelements.AntreaLabelsElementList {
 		ie, err := e.createInfoElementForTemplateSet(ie, ipfixregistry.AntreaEnterpriseID)
 		if err != nil {
 			return 0, err
 		}
 		elements = append(elements, ie)
 	}
-	for i := range infoelements.AntreaThroughputElementList {
-		// Add common throughput fields
-		ieName := infoelements.AntreaThroughputElementList[i]
-		ie, err := e.createInfoElementForTemplateSet(ieName, ipfixregistry.AntreaEnterpriseID)
-		if err != nil {
-			return 0, err
-		}
-		elements = append(elements, ie)
-		// Add source node specific throughput fields
-		ieName = infoelements.AntreaSourceThroughputElementList[i]
-		ie, err = e.createInfoElementForTemplateSet(ieName, ipfixregistry.AntreaEnterpriseID)
+	ie, err := e.createInfoElementForTemplateSet("clusterId", ipfixregistry.AntreaEnterpriseID)
+	if err != nil {
+		return 0, err
+	}
+	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)
-		// Add destination node specific throughput fields
-		ieName = infoelements.AntreaDestinationThroughputElementList[i]
-		ie, err = e.createInfoElementForTemplateSet(ieName, ipfixregistry.AntreaEnterpriseID)
+		ie, err = e.createInfoElementForTemplateSet("originalExporterIPv4Address", ipfixregistry.IANAEnterpriseID)
 		if err != nil {
 			return 0, err
 		}
 		elements = append(elements, ie)
-	}
-	for _, ie := range infoelements.AntreaLabelsElementList {
-		ie, err := e.createInfoElementForTemplateSet(ie, ipfixregistry.AntreaEnterpriseID)
+		ie, err = e.createInfoElementForTemplateSet("originalExporterIPv6Address", ipfixregistry.IANAEnterpriseID)
 		if err != nil {
 			return 0, err
 		}
 		elements = append(elements, ie)
 	}
-	ie, err := e.createInfoElementForTemplateSet("clusterId", ipfixregistry.AntreaEnterpriseID)
-	if err != nil {
-		return 0, err
-	}
-	elements = append(elements, ie)
 	e.set.ResetSet()
 	if err := e.set.PrepareSet(ipfixentities.Template, templateID); err != nil {
 		return 0, err

diff --git a/pkg/flowaggregator/exporter/ipfix_test.go b/pkg/flowaggregator/exporter/ipfix_test.go
@@ -67,6 +67,7 @@ func TestIPFIXExporter_sendTemplateSet(t *testing.T) {
 			templateIDv6:               testTemplateIDv6,
 			registry:                   mockIPFIXRegistry,
 			set:                        mockTempSet,
+			aggregatorMode:             flowaggregatorconfig.AggregatorModeAggregate,
 			observationDomainID:        testObservationDomainID,
 		}
 		elemList := createElementList(isIPv6, mockIPFIXRegistry)
@@ -118,13 +119,14 @@ func TestIPFIXExporter_UpdateOptions(t *testing.T) {
 			RecordFormat:        "IPFIX",
 		},
 	}
-	ipfixExporter := IPFIXExporter{
+	ipfixExporter := &IPFIXExporter{
 		config:                     config.FlowCollector,
 		externalFlowCollectorAddr:  "",
 		externalFlowCollectorProto: "",
 		templateIDv4:               testTemplateIDv4,
 		templateIDv6:               testTemplateIDv6,
 		set:                        mockSet,
+		aggregatorMode:             flowaggregatorconfig.AggregatorModeAggregate,
 		observationDomainID:        testObservationDomainID,
 	}
 	testTemplateID := testTemplateIDv4
@@ -182,12 +184,13 @@ func TestIPFIXExporter_AddRecord(t *testing.T) {
 		initIPFIXExportingProcess = initIPFIXExportingProcessSaved
 	}()
 
-	ipfixExporter := IPFIXExporter{
+	ipfixExporter := &IPFIXExporter{
 		externalFlowCollectorAddr:  "",
 		externalFlowCollectorProto: "",
 		templateIDv4:               testTemplateIDv4,
 		templateIDv6:               testTemplateIDv6,
 		set:                        mockSet,
+		aggregatorMode:             flowaggregatorconfig.AggregatorModeAggregate,
 		observationDomainID:        testObservationDomainID,
 	}
 	testTemplateID := testTemplateIDv4
@@ -216,9 +219,10 @@ func TestIPFIXExporter_initIPFIXExportingProcess_Error(t *testing.T) {
 		initIPFIXExportingProcess = initIPFIXExportingProcessSaved
 	}()
 
-	ipfixExporter := IPFIXExporter{
+	ipfixExporter := &IPFIXExporter{
 		externalFlowCollectorAddr:  "",
 		externalFlowCollectorProto: "",
+		aggregatorMode:             flowaggregatorconfig.AggregatorModeAggregate,
 	}
 
 	assert.Error(t, ipfixExporter.AddRecord(mockRecord, false))
@@ -231,13 +235,14 @@ func TestIPFIXExporter_sendRecord_Error(t *testing.T) {
 	mockSet := ipfixentitiestesting.NewMockSet(ctrl)
 	mockRecord := ipfixentitiestesting.NewMockRecord(ctrl)
 
-	ipfixExporter := IPFIXExporter{
+	ipfixExporter := &IPFIXExporter{
 		externalFlowCollectorAddr:  "",
 		externalFlowCollectorProto: "",
 		exportingProcess:           mockIPFIXExpProc,
 		templateIDv4:               testTemplateIDv4,
 		templateIDv6:               testTemplateIDv6,
 		set:                        mockSet,
+		aggregatorMode:             flowaggregatorconfig.AggregatorModeAggregate,
 		observationDomainID:        testObservationDomainID,
 	}
 	testTemplateID := testTemplateIDv4
@@ -308,7 +313,9 @@ func TestInitExportingProcess(t *testing.T) {
 	t.Run("tcp success", func(t *testing.T) {
 		ctrl := gomock.NewController(t)
 		mockIPFIXRegistry := ipfixtesting.NewMockIPFIXRegistry(ctrl)
-		opt := &options.Options{}
+		opt := &options.Options{
+			AggregatorMode: flowaggregatorconfig.AggregatorModeAggregate,
+		}
 		opt.Config = &flowaggregatorconfig.FlowAggregatorConfig{}
 		flowaggregatorconfig.SetConfigDefaults(opt.Config)
 		listener, err := net.Listen("tcp", "127.0.0.1:0")
@@ -326,7 +333,9 @@ func TestInitExportingProcess(t *testing.T) {
 	t.Run("udp success", func(t *testing.T) {
 		ctrl := gomock.NewController(t)
 		mockIPFIXRegistry := ipfixtesting.NewMockIPFIXRegistry(ctrl)
-		opt := &options.Options{}
+		opt := &options.Options{
+			AggregatorMode: flowaggregatorconfig.AggregatorModeAggregate,
+		}
 		opt.Config = &flowaggregatorconfig.FlowAggregatorConfig{}
 		flowaggregatorconfig.SetConfigDefaults(opt.Config)
 		udpAddr, err := net.ResolveUDPAddr("udp", "127.0.0.1:0")
@@ -346,7 +355,9 @@ func TestInitExportingProcess(t *testing.T) {
 	t.Run("tcp failure", func(t *testing.T) {
 		ctrl := gomock.NewController(t)
 		mockIPFIXRegistry := ipfixtesting.NewMockIPFIXRegistry(ctrl)
-		opt := &options.Options{}
+		opt := &options.Options{
+			AggregatorMode: flowaggregatorconfig.AggregatorModeAggregate,
+		}
 		opt.Config = &flowaggregatorconfig.FlowAggregatorConfig{}
 		flowaggregatorconfig.SetConfigDefaults(opt.Config)
 		// dialing this address is guaranteed to fail (we use 0 as the port number)