Upgrade go-ipfix to 0.13.0

As part of this upgrade, we make the following changes:
* Switch to buffered IPFIX exporter in the Flow Aggregator. This
  exporter has better performance for UDP IPFIX messages, by ensuring
  that multiple data records can be batched together in a single
  message.
* Provide Path MTU (PMTU) when creating the IPFIX exporter in the Flow
  Aggregator. The value is used by the new buffered exporter to
  determine how many IPFIX records can fit in a single message while
  avoiding IP fragmentation. In our case, we "approximate" the Path MTU
  by looking up the MTU of the Flow Aggregator Pod's eth0 interface.
* Add a MaxMsgSize configuration parameter to the Flow Aggregator as a
  way to override the default behavior, which is to use the MTU (minus
  header overhead) when the UDP protocol is used.
* Add periodic flushing when exporting IPFIX records, which is necessary
  after switching to the buffered exporter. In Aggregation mode,
  flushing happens after processing a given batch of expired records. In
  Proxy mode, flushing happens every second.
* Use updated reference IPFIX collector in e2e tests. The updated
  collector handles the case where multiple data records are included in
  the same IPFIX message more gracefully, which leads to some
  simplification in the test code.

Signed-off-by: Antonin Bas <antonin.bas@broadcom.com>

build/charts/flow-aggregator/README.md

@@ -34,6 +34,7 @@ Kubernetes: `>= 1.19.0-0`
 | flowAggregatorAddress | string | `""` | Provide an extra DNS name or IP address of flow aggregator for generating TLS certificate. |
 | flowCollector.address | string | `""` | Provide the flow collector address as string with format <IP>:<port>[:<proto>],  where proto is tcp or udp. If no L4 transport proto is given, we consider tcp as default. |
 | flowCollector.enable | bool | `false` | Determine whether to enable exporting flow records to external flow collector. |
+| flowCollector.maxIPFIXMsgSize | int | `0` | Maximum message size to use for IPFIX records. If set to 0 (recommended), a reasonable default value will be used based on the protocol (tcp or udp) used to connect to the collector. |
 | flowCollector.observationDomainID | string | `""` | Provide the 32-bit Observation Domain ID which will uniquely identify this instance of the flow aggregator to an external flow collector. If omitted, an Observation Domain ID will be generated from the persistent cluster UUID generated by Antrea. |
 | flowCollector.recordFormat | string | `"IPFIX"` | Provide format for records sent to the configured flow collector. Supported formats are IPFIX and JSON. |
 | flowCollector.templateRefreshTimeout | string | `"600s"` | Template retransmission interval when using the udp protocol to export records. The value must be provided as a duration string. Valid time units are "ns", "us" (or "µs"), "ms", "s", "m", "h". |

build/charts/flow-aggregator/conf/flow-aggregator.conf

@@ -75,6 +75,10 @@ flowCollector:
   # Valid time units are "ns", "us" (or "µs"), "ms", "s", "m", "h".
   templateRefreshTimeout: {{ .Values.flowCollector.templateRefreshTimeout | quote }}
 
+  # Maximum message size to use for IPFIX records. If set to 0 (recommended), a reasonable default
+  # value will be used based on the protocol (tcp or udp) used to connect to the collector.
+  maxIPFIXMsgSize: {{ .Values.flowCollector.maxIPFIXMsgSize }}
+
 # clickHouse contains ClickHouse related configuration options.
 clickHouse:
   # Enable is the switch to enable exporting flow records to ClickHouse.

build/charts/flow-aggregator/values.yaml

@@ -52,6 +52,9 @@ flowCollector:
   # -- Template retransmission interval when using the udp protocol to export records.
   # The value must be provided as a duration string. Valid time units are "ns", "us" (or "µs"), "ms", "s", "m", "h".
   templateRefreshTimeout: "600s"
+  # -- Maximum message size to use for IPFIX records. If set to 0 (recommended), a reasonable
+  # default value will be used based on the protocol (tcp or udp) used to connect to the collector.
+  maxIPFIXMsgSize: 0
 # clickHouse contains ClickHouse related configuration options.
 clickHouse:
   # -- Determine whether to enable exporting flow records to ClickHouse.

build/yamls/flow-aggregator.yml

@@ -227,6 +227,10 @@ data:
       # Valid time units are "ns", "us" (or "µs"), "ms", "s", "m", "h".
       templateRefreshTimeout: "600s"
 
+      # Maximum message size to use for IPFIX records. If set to 0 (recommended), a reasonable default
+      # value will be used based on the protocol (tcp or udp) used to connect to the collector.
+      maxIPFIXMsgSize: 0
+
     # clickHouse contains ClickHouse related configuration options.
     clickHouse:
       # Enable is the switch to enable exporting flow records to ClickHouse.
@@ -401,7 +405,7 @@ spec:
   template:
     metadata:
       annotations:
-        checksum/config: 5ba1a6d1b9d3b40e2ea26e37aa2bea38fda2558c20564873936472136651de37
+        checksum/config: 96acfb574fbfb758e6388d677cbc8359c0375031fd68875d4ec2d03f34d2e49c
       labels:
         app: flow-aggregator
     spec:

ci/kind/test-e2e-kind.sh

@@ -261,7 +261,7 @@ COMMON_IMAGES_LIST=("registry.k8s.io/e2e-test-images/agnhost:2.40" \
                     "antrea/nginx:1.21.6-alpine" \
                     "antrea/toolbox:1.5-1")
 
-FLOW_VISIBILITY_IMAGE_LIST=("antrea/ipfix-collector:v0.12.0" \
+FLOW_VISIBILITY_IMAGE_LIST=("antrea/ipfix-collector:v0.13.0" \
                             "antrea/clickhouse-operator:0.21.0" \
                             "antrea/metrics-exporter:0.21.0" \
                             "antrea/clickhouse-server:23.4")

go.mod

@@ -53,7 +53,7 @@ require (
 	github.com/stretchr/testify v1.10.0
 	github.com/ti-mo/conntrack v0.5.1
 	github.com/vishvananda/netlink v1.3.0
-	github.com/vmware/go-ipfix v0.12.0
+	github.com/vmware/go-ipfix v0.13.0
 	go.uber.org/mock v0.5.0
 	golang.org/x/crypto v0.33.0
 	golang.org/x/mod v0.23.0
@@ -168,7 +168,7 @@ require (
 	github.com/josharian/native v1.1.0 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/k-sone/critbitgo v1.4.0 // indirect
-	github.com/klauspost/compress v1.17.9 // indirect
+	github.com/klauspost/compress v1.17.11 // indirect
 	github.com/kr/fs v0.1.0 // indirect
 	github.com/kylelemons/godebug v1.1.0 // indirect
 	github.com/liggitt/tabwriter v0.0.0-20181228230101-89fcab3d43de // indirect
@@ -192,7 +192,7 @@ require (
 	github.com/paulmach/orb v0.8.0 // indirect
 	github.com/pelletier/go-toml/v2 v2.0.8 // indirect
 	github.com/peterbourgon/diskv v2.0.1+incompatible // indirect
-	github.com/pierrec/lz4/v4 v4.1.21 // indirect
+	github.com/pierrec/lz4/v4 v4.1.22 // indirect
 	github.com/pion/dtls/v2 v2.2.12 // indirect
 	github.com/pion/logging v0.2.2 // indirect
 	github.com/pion/transport/v2 v2.2.10 // indirect

go.sum

@@ -485,8 +485,8 @@ github.com/kisielk/errcheck v1.2.0/go.mod h1:/BMXB+zMLi60iA8Vv6Ksmxu/1UDYcXs4uQL
 github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
 github.com/kisielk/sqlstruct v0.0.0-20201105191214-5f3e10d3ab46/go.mod h1:yyMNCyc/Ib3bDTKd379tNMpB/7/H5TjM2Y9QJ5THLbE=
-github.com/klauspost/compress v1.17.9 h1:6KIumPrER1LHsvBVuDa0r5xaG0Es51mhhB9BQB2qeMA=
-github.com/klauspost/compress v1.17.9/go.mod h1:Di0epgTjJY877eYKx5yC51cX2A2Vl2ibi7bDH9ttBbw=
+github.com/klauspost/compress v1.17.11 h1:In6xLpyWOi1+C7tXUUWv2ot1QvBjxevKAaI6IXrJmUc=
+github.com/klauspost/compress v1.17.11/go.mod h1:pMDklpSncoRMuLFrf1W9Ss9KT+0rH90U12bZKk7uwG0=
 github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
 github.com/konsorten/go-windows-terminal-sequences v1.0.3/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
 github.com/kr/fs v0.1.0 h1:Jskdu9ieNAYnjxsi0LbQp1ulIKZV1LAFgK1tWhpZgl8=
@@ -622,8 +622,8 @@ github.com/pelletier/go-toml/v2 v2.0.8 h1:0ctb6s9mE31h0/lhu+J6OPmVeDxJn+kYnJc2jZ
 github.com/pelletier/go-toml/v2 v2.0.8/go.mod h1:vuYfssBdrU2XDZ9bYydBu6t+6a6PYNcZljzZR9VXg+4=
 github.com/peterbourgon/diskv v2.0.1+incompatible h1:UBdAOUP5p4RWqPBg048CAvpKN+vxiaj6gdUUzhl4XmI=
 github.com/peterbourgon/diskv v2.0.1+incompatible/go.mod h1:uqqh8zWWbv1HBMNONnaR/tNboyR3/BZd58JJSHlUSCU=
-github.com/pierrec/lz4/v4 v4.1.21 h1:yOVMLb6qSIDP67pl/5F7RepeKYu/VmTyEXvuMI5d9mQ=
-github.com/pierrec/lz4/v4 v4.1.21/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4=
+github.com/pierrec/lz4/v4 v4.1.22 h1:cKFw6uJDK+/gfw5BcDL0JL5aBsAFdsIT18eRtLj7VIU=
+github.com/pierrec/lz4/v4 v4.1.22/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4=
 github.com/pion/dtls/v2 v2.2.12 h1:KP7H5/c1EiVAAKUmXyCzPiQe5+bCJrpOeKg/L05dunk=
 github.com/pion/dtls/v2 v2.2.12/go.mod h1:d9SYc9fch0CqK90mRk1dC7AkzzpwJj6u2GU3u+9pqFE=
 github.com/pion/logging v0.2.2 h1:M9+AIj/+pxNsDfAT64+MAVgJO0rsyLnoJKCqf//DoeY=
@@ -775,8 +775,8 @@ github.com/vishvananda/netlink v1.3.0/go.mod h1:i6NetklAujEcC6fK0JPjT8qSwWyO0HLn
 github.com/vishvananda/netns v0.0.0-20200728191858-db3c7e526aae/go.mod h1:DD4vA1DwXk04H54A1oHXtwZmA0grkVMdPxx/VGLCah0=
 github.com/vishvananda/netns v0.0.4 h1:Oeaw1EM2JMxD51g9uhtC0D7erkIjgmj8+JZc26m1YX8=
 github.com/vishvananda/netns v0.0.4/go.mod h1:SpkAiCQRtJ6TvvxPnOSyH3BMl6unz3xZlaprSwhNNJM=
-github.com/vmware/go-ipfix v0.12.0 h1:a4YXeCWTa251aZO7u7e9dKDOoU2eHJID45SPlq9j+HI=
-github.com/vmware/go-ipfix v0.12.0/go.mod h1:9PiutVWLhQQ6WHncRrGkH0i2Rx82DEOKhu80VSd9jds=
+github.com/vmware/go-ipfix v0.13.0 h1:v3paBzd7oq7LEU1SzDwD5RGoYcGROLQycYyN3EzLvDk=
+github.com/vmware/go-ipfix v0.13.0/go.mod h1:UTIR38AuEePzrWYjQOvnORCYRG33xZJ56E0K75mSosM=
 github.com/wlynxg/anet v0.0.3/go.mod h1:eay5PRQr7fIVAMbTbchTnO9gG65Hg/uYGdc7mguHxoA=
 github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM=
 github.com/x448/float16 v0.8.4/go.mod h1:14CWIYCyZA/cWjXOioeEpHeN/83MdbZDRQHoFcYsOfg=

hack/update-codegen-dockerized.sh

@@ -90,7 +90,7 @@ MOCKGEN_TARGETS=(
   "pkg/controller/networkpolicy EndpointQuerier,PolicyRuleQuerier testing"
   "pkg/controller/querier ControllerQuerier testing"
   "pkg/flowaggregator/exporter Interface testing"
-  "pkg/ipfix IPFIXExportingProcess,IPFIXRegistry,IPFIXCollectingProcess,IPFIXAggregationProcess testing"
+  "pkg/ipfix IPFIXExportingProcess,IPFIXBufferedExporter,IPFIXRegistry,IPFIXCollectingProcess,IPFIXAggregationProcess testing"
   "pkg/ovs/openflow Bridge,Table,Flow,Action,CTAction,FlowBuilder,Group,BucketBuilder,PacketOutBuilder,Meter,MeterBandBuilder testing"
   "pkg/ovs/ovsconfig OVSBridgeClient testing"
   "pkg/ovs/ovsctl OVSCtlClient testing"

pkg/config/flowaggregator/config.go

@@ -105,6 +105,9 @@ type FlowCollectorConfig struct {
 	// The value must be provided as a duration string. Defaults to 600s.
 	// Valid time units are "ns", "us" (or "µs"), "ms", "s", "m", "h".
 	TemplateRefreshTimeout string `yaml:"templateRefreshTimeout,omitempty"`
+	// Maximum message size to use for IPFIX records. If set to 0 (recommended), a reasonable
+	// default value will be used based on the protocol (tcp or udp) used to connect to the collector.
+	MaxIPFIXMsgSize int32 `yaml:"maxIPFIXMsgSize,omitempty"`
 }
 
 type ClickHouseConfig struct {

pkg/flowaggregator/exporter/clickhouse.go

@@ -122,3 +122,7 @@ func (e *ClickHouseExporter) UpdateOptions(opt *options.Options) {
 	}
 	klog.InfoS("New ClickHouse configuration", "database", chConfig.Database, "databaseURL", chConfig.DatabaseURL, "debug", chConfig.Debug, "compress", *chConfig.Compress, "commitInterval", chConfig.CommitInterval)
 }
+
+func (e *ClickHouseExporter) Flush() error {
+	return nil
+}

pkg/flowaggregator/exporter/interface.go

@@ -28,4 +28,7 @@ type Interface interface {
 	Stop()
 	AddRecord(record ipfixentities.Record, isRecordIPv6 bool) error
 	UpdateOptions(opt *options.Options)
+	// Some exporters may be buffered, in which case the FlowAggregator
+	// should call this method periodically.
+	Flush() error
 }