Add Path MTU to Exporter inputs (#407)

It is not straightforward for the consumer of the library to determine
the maximum message size (for UDP) just based on the PMTU, as the amount
to deduct from the PMTU depends on the protocol used for the connection
to the exporter (IPv4 or IPv6). The protocol is only known after
resolving the collector address, which happens in InitExportingProcess.
As a workaround, we add PMTU to the exporter inputs, and let the
exporter compute the maximum message size. It is still possible to
request a specific maximum message size. If both are provided and the
exporter determines that the values are incompatible, an error will be
returned.

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

Author: antoninbas

pkg/exporter/process.go

@@ -48,8 +48,7 @@ type templateValue struct {
 //  3. Supports only TCP and UDP; one session at a time. SCTP is not supported.
 //  4. UDP needs to send PMTU size packets as per RFC7011. In order to guarantee
 //     this, maxMsgSize should be set correctly. maxMsgSize is the maximum
-//     payload (IPFIX message) size, not the maximum packet size. You need to
-//     compute maxMsgSize based on the desired maximum packet size. If
+//     payload (IPFIX message) size, not the maximum packet size. If
 //     maxMsgSize is not set correctly, the message may be fragmented.
 type ExportingProcess struct {
 	connToCollector net.Conn
@@ -94,17 +93,68 @@ type ExporterInput struct {
 	// JSONBufferLen is recommended for sending json records. If not given a
 	// valid value, we use a default of 5000B
 	JSONBufferLen int
-	// For UDP, this should be set by taking into account the PMTU and
-	// header sizes.
-	MaxMsgSize        int
+	// MaxMsgSize can be used to provide a custom maximum IPFIX message
+	// size. If it is omitted, we will use an appropriate default based on
+	// the configured protocol. For UDP, we want to avoid fragmentation, so
+	// the MaxMsgSize should be set by taking into account the PMTU and
+	// header sizes. The recommended approach is to keep MaxMsgSize unset
+	// and provide the correct PMTU value.
+	MaxMsgSize int
+	// PathMTU is used to calculate the maximum message size when the
+	// protocol is UDP. It is ignored for TCP. If both MaxMsgSize and
+	// PathMTU are set, and MaxMsgSize is incompatible with the provided
+	// PathMTU, exporter initialization will fail.
+	PathMTU           int
 	CheckConnInterval time.Duration
 }
 
+func calculateMaxMsgSize(proto string, requestedSize int, pathMTU int, isIPv6 bool) (int, error) {
+	if requestedSize > 0 && (requestedSize < entities.MinSupportedMsgSize || requestedSize > entities.MaxSocketMsgSize) {
+		return 0, fmt.Errorf("requested message size should be between %d and %d", entities.MinSupportedMsgSize, entities.MaxSocketMsgSize)
+	}
+	if proto == "tcp" {
+		if requestedSize == 0 {
+			return entities.MaxSocketMsgSize, nil
+		} else {
+			return requestedSize, nil
+		}
+	}
+	// UDP protocol
+	if pathMTU == 0 {
+		if requestedSize == 0 {
+			klog.InfoS("Neither max IPFIX message size nor PMTU were provided, defaulting to min message size", "messageSize", entities.MinSupportedMsgSize)
+			return entities.MinSupportedMsgSize, nil
+		}
+		klog.InfoS("PMTU was not provided, configured message size may cause fragmentation", "messageSize", requestedSize)
+		return requestedSize, nil
+	}
+	// 20-byte IPv4, 8-byte UDP header
+	mtuDeduction := 28
+	if isIPv6 {
+		// An extra 20 bytes for IPv6
+		mtuDeduction += 20
+	}
+	maxMsgSize := pathMTU - mtuDeduction
+	if maxMsgSize < entities.MinSupportedMsgSize {
+		return 0, fmt.Errorf("provided PMTU %d is not large enough to accommodate min message size %d", pathMTU, entities.MinSupportedMsgSize)
+	}
+	if requestedSize > maxMsgSize {
+		return 0, fmt.Errorf("requested message size %d exceeds max message size %d calculated from provided PMTU", requestedSize, maxMsgSize)
+	}
+	if requestedSize > 0 {
+		return requestedSize, nil
+	}
+	return maxMsgSize, nil
+}
+
 // InitExportingProcess takes in collector address(net.Addr format), obsID(observation ID)
 // and tempRefTimeout(template refresh timeout). tempRefTimeout is applicable only
 // for collectors listening over UDP; unit is seconds. For TCP, you can pass any
 // value and it will be ignored. For UDP, if 0 is passed, 600s is used as the default.
 func InitExportingProcess(input ExporterInput) (*ExportingProcess, error) {
+	if input.CollectorProtocol != "tcp" && input.CollectorProtocol != "udp" {
+		return nil, fmt.Errorf("unsupported collector protocol: %s", input.CollectorProtocol)
+	}
 	var conn net.Conn
 	var err error
 	if input.TLSClientConfig != nil {
@@ -151,6 +201,15 @@ func InitExportingProcess(input ExporterInput) (*ExportingProcess, error) {
 			return nil, err
 		}
 	}
+	var isIPv6 bool
+	switch addr := conn.RemoteAddr().(type) {
+	case *net.TCPAddr:
+		isIPv6 = addr.IP.To4() == nil
+	case *net.UDPAddr:
+		isIPv6 = addr.IP.To4() == nil
+	default:
+		return nil, fmt.Errorf("unsupported net.Addr type %T", addr)
+	}
 	expProc := &ExportingProcess{
 		connToCollector: conn,
 		obsDomainID:     input.ObservationDomainID,
@@ -169,13 +228,12 @@ func InitExportingProcess(input ExporterInput) (*ExportingProcess, error) {
 			expProc.jsonBufferLen = input.JSONBufferLen
 		}
 	} else {
-		if input.MaxMsgSize == 0 {
-			expProc.maxMsgSize = entities.MaxSocketMsgSize
-		} else if input.MaxMsgSize < entities.MinSupportedMsgSize {
-			return nil, fmt.Errorf("maxMsgSize cannot be less than 512B")
-		} else {
-			expProc.maxMsgSize = input.MaxMsgSize
+		maxMsgSize, err := calculateMaxMsgSize(input.CollectorProtocol, input.MaxMsgSize, input.PathMTU, isIPv6)
+		if err != nil {
+			return nil, err
 		}
+		klog.InfoS("Calculated max IPFIX message size", "size", maxMsgSize)
+		expProc.maxMsgSize = maxMsgSize
 	}
 
 	// Start a goroutine to check whether the collector has already closed the TCP connection.

pkg/exporter/process_test.go

@@ -336,6 +336,7 @@ func TestExportingProcess_SendingDataRecordToLocalUDPServer(t *testing.T) {
 		CollectorProtocol:   conn.LocalAddr().Network(),
 		ObservationDomainID: 1,
 		TempRefTimeout:      0,
+		PathMTU:             1500,
 	}
 	exporter, err := InitExportingProcess(input)
 	if err != nil {
@@ -730,13 +731,14 @@ func TestExportingProcess_GetMsgSizeLimit(t *testing.T) {
 		CollectorProtocol:   conn.LocalAddr().Network(),
 		ObservationDomainID: 1,
 		TempRefTimeout:      1,
+		MaxMsgSize:          1000,
 	}
 	exporter, err := InitExportingProcess(input)
 	if err != nil {
 		t.Fatalf("Got error when connecting to local server %s: %v", conn.LocalAddr().String(), err)
 	}
 	t.Logf("Created exporter connecting to local server with address: %s", conn.LocalAddr().String())
-	assert.Equal(t, entities.MaxSocketMsgSize, exporter.GetMsgSizeLimit())
+	assert.Equal(t, input.MaxMsgSize, exporter.GetMsgSizeLimit())
 }
 
 func TestExportingProcess_CheckConnToCollector(t *testing.T) {
@@ -885,3 +887,84 @@ func TestSendDataRecords(t *testing.T) {
 		}
 	}
 }
+
+func TestCalculateMaxMsgSize(t *testing.T) {
+	testCases := []struct {
+		proto         string
+		requestedSize int
+		pathMTU       int
+		isIPv6        bool
+		expectedErr   string
+		expectedSize  int
+	}{
+		{
+			proto:         "tcp",
+			requestedSize: entities.MinSupportedMsgSize - 1,
+			expectedErr:   "requested message size should be between",
+		},
+		{
+			proto:         "tcp",
+			requestedSize: entities.MaxSocketMsgSize + 1,
+			expectedErr:   "requested message size should be between",
+		},
+		{
+			proto:         "tcp",
+			requestedSize: 0,
+			expectedSize:  entities.MaxSocketMsgSize,
+		},
+		{
+			proto:         "tcp",
+			requestedSize: 1000,
+			expectedSize:  1000,
+		},
+		{
+			proto:        "udp",
+			expectedSize: entities.MinSupportedMsgSize,
+		},
+		{
+			proto:         "udp",
+			requestedSize: 1000,
+			expectedSize:  1000,
+		},
+		{
+			proto:        "udp",
+			pathMTU:      1500,
+			expectedSize: 1472,
+		},
+		{
+			proto:        "udp",
+			pathMTU:      1500,
+			isIPv6:       true,
+			expectedSize: 1452,
+		},
+		{
+			proto:       "udp",
+			pathMTU:     300,
+			expectedErr: "is not large enough to accommodate min message size",
+		},
+		{
+			proto:         "udp",
+			requestedSize: 2000,
+			pathMTU:       1500,
+			expectedErr:   "exceeds max message size",
+		},
+		{
+			proto:         "udp",
+			requestedSize: 1000,
+			pathMTU:       1500,
+			expectedSize:  1000,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(fmt.Sprintf("%s-req=%d-pmtu=%d-ipv6=%t", tc.proto, tc.requestedSize, tc.pathMTU, tc.isIPv6), func(t *testing.T) {
+			size, err := calculateMaxMsgSize(tc.proto, tc.requestedSize, tc.pathMTU, tc.isIPv6)
+			if tc.expectedErr != "" {
+				assert.ErrorContains(t, err, tc.expectedErr)
+			} else {
+				require.NoError(t, err)
+				assert.Equal(t, tc.expectedSize, size)
+			}
+		})
+	}
+}