inlet/flow: make some of the socket options not fatal

If the kernel is too old for timestamping, it should not be fatal. I
prefer to not accept SO_TIMESTAMP_OLD as the size of the timestamp is
arch-dependent.

Fix #1978
This commit is contained in:
Vincent Bernat
2025-09-22 18:29:17 +02:00
parent 2c567a806f
commit 5ce0c949f2
6 changed files with 123 additions and 21 deletions

View File

@@ -10,6 +10,10 @@ identified with a specific icon:
- 🩹: bug fix - 🩹: bug fix
- 🌱: miscellaneous change - 🌱: miscellaneous change
## Unreleased
- 🩹 *inlet*: disable kernel timestamping on Linux kernel older than 5.1
## 2.0.0 - 2025-09-22 ## 2.0.0 - 2025-09-22
This release introduces a new component: the outlet. Previously, ClickHouse was This release introduces a new component: the outlet. Previously, ClickHouse was

View File

@@ -120,7 +120,8 @@ func (in *Input) Start() error {
return fmt.Errorf("unable to resolve %v: %w", in.config.Listen, err) return fmt.Errorf("unable to resolve %v: %w", in.config.Listen, err)
} }
} }
pconn, err := listenConfig.ListenPacket(in.t.Context(context.Background()), "udp", listenAddr.String()) pconn, err := listenConfig(in.r, udpSocketOptions).
ListenPacket(in.t.Context(context.Background()), "udp", listenAddr.String())
if err != nil { if err != nil {
return fmt.Errorf("unable to listen to %v: %w", listenAddr, err) return fmt.Errorf("unable to listen to %v: %w", listenAddr, err)
} }

View File

@@ -4,10 +4,13 @@
package udp package udp
import ( import (
"fmt"
"net" "net"
"syscall" "syscall"
"time" "time"
"akvorado/common/reporter"
"golang.org/x/sys/unix" "golang.org/x/sys/unix"
) )
@@ -16,19 +19,31 @@ type oobMessage struct {
Received time.Time Received time.Time
} }
// listenConfig configures a listening socket to reuse port and return overflows // socketOption describes a socket option to be applied.
var listenConfig = net.ListenConfig{ type socketOption struct {
Control: func(_, _ string, c syscall.RawConn) error { Name string
var err error Level int
c.Control(func(fd uintptr) { Option int
opts := udpSocketOptions Mandatory bool
}
// listenConfig configures a listening socket with the udpSocketOptions.
var listenConfig = func(r *reporter.Reporter, opts []socketOption) *net.ListenConfig {
return &net.ListenConfig{
Control: func(_, _ string, c syscall.RawConn) error {
var err error
for _, opt := range opts { for _, opt := range opts {
err = unix.SetsockoptInt(int(fd), unix.SOL_SOCKET, opt, 1) c.Control(func(fd uintptr) {
err = unix.SetsockoptInt(int(fd), opt.Level, opt.Option, 1)
})
if err != nil { if err != nil {
return if opt.Mandatory {
return fmt.Errorf("cannot set option %s: %w", opt.Name, err)
}
r.Warn().Err(err).Msgf("cannot set option %s", opt.Name)
} }
} }
}) return nil
return err },
}, }
} }

View File

@@ -15,13 +15,30 @@ import (
var ( var (
oobLength = syscall.CmsgLen(4) + syscall.CmsgLen(16) // uint32 + 2*int64 oobLength = syscall.CmsgLen(4) + syscall.CmsgLen(16) // uint32 + 2*int64
udpSocketOptions = []int{ udpSocketOptions = []socketOption{
// Allow multiple listeners to bind to the same IP/port {
unix.SO_REUSEADDR, unix.SO_REUSEPORT, // Allow multiple listeners to bind to the same IP
// Get the number of dropped packets Name: "SO_REUSEADDR",
unix.SO_RXQ_OVFL, Level: unix.SOL_SOCKET,
// Ask the kernel to timestamp incoming packets Option: unix.SO_REUSEADDR,
unix.SO_TIMESTAMP_NEW | unix.SOF_TIMESTAMPING_RX_SOFTWARE, Mandatory: true,
}, {
// Allow multiple listeners to bind to the same port
Name: "SO_REUSEPORT",
Level: unix.SOL_SOCKET,
Option: unix.SO_REUSEPORT,
Mandatory: true,
}, {
// Get the number of dropped packets
Name: "SO_RXQ_OVFL",
Level: unix.SOL_SOCKET,
Option: unix.SO_RXQ_OVFL,
}, {
// Ask the kernel to timestamp incoming packets
Name: "SO_TIMESTAMP_NEW",
Level: unix.SOL_SOCKET,
Option: unix.SO_TIMESTAMP_NEW | unix.SOF_TIMESTAMPING_RX_SOFTWARE,
},
} }
) )

View File

@@ -9,7 +9,21 @@ import "golang.org/x/sys/unix"
var ( var (
oobLength = 0 oobLength = 0
udpSocketOptions = []int{unix.SO_REUSEADDR, unix.SO_REUSEPORT} udpSocketOptions = []socketOption{
{
// Allow multiple listeners to bind to the same IP
Name: "SO_REUSEADDR",
Level: unix.SOL_SOCKET,
Option: unix.SO_REUSEADDR,
Mandatory: true,
}, {
// Allow multiple listeners to bind to the same port
Name: "SO_REUSEPORT",
Level: unix.SOL_SOCKET,
Option: unix.SO_REUSEPORT,
Mandatory: true,
},
}
) )
// parseSocketControlMessage always returns 0. // parseSocketControlMessage always returns 0.

View File

@@ -11,13 +11,19 @@ import (
"runtime" "runtime"
"testing" "testing"
"time" "time"
"akvorado/common/reporter"
"golang.org/x/sys/unix"
) )
func TestParseSocketControlMessage(t *testing.T) { func TestParseSocketControlMessage(t *testing.T) {
if runtime.GOOS != "linux" { if runtime.GOOS != "linux" {
t.Skip("Skip Linux-only test") t.Skip("Skip Linux-only test")
} }
server, err := listenConfig.ListenPacket(context.Background(), "udp", "127.0.0.1:0") r := reporter.NewMock(t)
server, err := listenConfig(r, udpSocketOptions).
ListenPacket(context.Background(), "udp", "127.0.0.1:0")
if err != nil { if err != nil {
t.Fatalf("ListenPacket() error:\n%+v", err) t.Fatalf("ListenPacket() error:\n%+v", err)
} }
@@ -78,3 +84,48 @@ outer:
t.Fatal("too many drops detected") t.Fatal("too many drops detected")
} }
} }
func TestListenConfig(t *testing.T) {
r := reporter.NewMock(t)
t.Run("one mandatory option", func(t *testing.T) {
_, err := listenConfig(r, []socketOption{
{
Name: "SO_REUSEADDR",
Level: unix.SOL_SOCKET,
Option: unix.SO_REUSEADDR,
Mandatory: true,
},
}).ListenPacket(t.Context(), "udp", "127.0.0.1:0")
if err != nil {
t.Fatalf("ListenPacket() error:\n%+v", err)
}
})
t.Run("one mandatory not implemented option", func(t *testing.T) {
_, err := listenConfig(r, []socketOption{
{
Name: "SO_UNKNOWN",
Level: unix.SOL_SOCKET,
Option: 9999,
Mandatory: true,
},
}).ListenPacket(t.Context(), "udp", "127.0.0.1:0")
if err == nil {
t.Fatal("ListenPacket() did not error error")
}
})
t.Run("one optional not implemented option", func(t *testing.T) {
_, err := listenConfig(r, []socketOption{
{
Name: "SO_UNKNOWN",
Level: unix.SOL_SOCKET,
Option: 9999,
},
}).ListenPacket(t.Context(), "udp", "127.0.0.1:0")
if err != nil {
t.Fatalf("ListenPacket() error:\n%+v", err)
}
})
}