Add Proxy mode for FlowAggregator (#6920)

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 #6773 Signed-off-by: Antonin Bas <[email protected]>
antrea-io · Jan 27, 2025 · 9e5113d · 9e5113d
1 parent ca8e0ee
commit 9e5113d
Show file tree

Hide file tree

Showing 15 changed files with 695 additions and 169 deletions.
diff --git a/build/charts/flow-aggregator/README.md b/build/charts/flow-aggregator/README.md
@@ -50,6 +50,7 @@ Kubernetes: `>= 1.19.0-0`
 | image | object | `{"pullPolicy":"IfNotPresent","repository":"antrea/flow-aggregator","tag":""}` | Container image used by Flow Aggregator. |
 | inactiveFlowRecordTimeout | string | `"90s"` | Provide the inactive flow record timeout as a duration string. Valid time units are "ns", "us" (or "µs"), "ms", "s", "m", "h". |
 | logVerbosity | int | `0` | Log verbosity switch for Flow Aggregator. |
+| mode | string | `"Aggregate"` | Mode in which to run the flow aggregator. Must be one of "Aggregate" or "Proxy". In Aggregate mode, flow records received from source and destination are aggregated and sent as one flow record. In Proxy mode, flow records are enhanced with some additional information, then sent directly without buffering or aggregation. |
 | recordContents.podLabels | bool | `false` | Determine whether source and destination Pod labels will be included in the flow records. |
 | s3Uploader.awsCredentials | object | `{"aws_access_key_id":"changeme","aws_secret_access_key":"changeme","aws_session_token":""}` | Credentials to authenticate to AWS. They will be stored in a Secret and injected into the Pod as environment variables. |
 | s3Uploader.bucketName | string | `""` | BucketName is the name of the S3 bucket to which flow records will be uploaded. It is required. |

diff --git a/build/charts/flow-aggregator/conf/flow-aggregator.conf b/build/charts/flow-aggregator/conf/flow-aggregator.conf
@@ -1,3 +1,9 @@
+# Mode in which to run the flow aggregator. Must be one of "Aggregate" or "Proxy". In Aggregate
+# mode, flow records received from source and destination are aggregated and sent as one flow
+# record. In Proxy mode, 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,11 @@ image:
   pullPolicy: "IfNotPresent"
   tag: ""
 
+# -- Mode in which to run the flow aggregator. Must be one of "Aggregate" or "Proxy". In Aggregate
+# mode, flow records received from source and destination are aggregated and sent as one flow
+# record. In Proxy mode, 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,12 @@ subjects:
 apiVersion: v1
 data:
   flow-aggregator.conf: |
+    # Mode in which to run the flow aggregator. Must be one of "Aggregate" or "Proxy". In Aggregate
+    # mode, flow records received from source and destination are aggregated and sent as one flow
+    # record. In Proxy mode, 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,20 @@ const (
 	AggregatorTransportProtocolUDP AggregatorTransportProtocol = "UDP"
 )
 
+type AggregatorMode string
+
+const (
+	// In Aggregate mode, flow records received from source and destination are aggregated and
+	// sent as one flow record.
+	AggregatorModeAggregate AggregatorMode = "Aggregate"
+	// In Proxy mode, flow records are enhanced with some additional information, then sent
+	// directly without buffering or aggregation.
+	AggregatorModeProxy AggregatorMode = "Proxy"
+)
+
 type FlowAggregatorConfig struct {
+	// Mode in which to run the flow aggregator. Must be one of "Aggregate" or "Proxy".
+	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)