Add option to set rcv_buffer for UDP inputs

CHANGELOG.next.asciidoc

@@ -17,6 +17,10 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Auditd module: Normalized value of `event.category` field from `user-login` to `authentication`. {pull}11432[11432]
 
 *Filebeat*
+- Modify apache/error dataset to follow ECS. {pull}8963[8963]
+- Rename many `traefik.access.*` fields to map to ECS. {pull}9005[9005]
+- Fix parsing of GC entries in elasticsearch server log. {issue}9513[9513] {pull}9810[9810]
+- Add read_buffer configuration option. {pull}9906[9906]
 
 *Heartbeat*
 

filebeat/_meta/common.reference.inputs.yml

@@ -241,6 +241,10 @@ filebeat.inputs:
   # Maximum size of the message received over UDP
   #max_message_size: 10KiB
 
+  # Size of the UDP read buffer in bytes
+  #read_buffer: 0
+
+
 #------------------------------ TCP input --------------------------------
 # Experimental: Config options for the TCP input
 #- type: tcp

filebeat/docs/inputs/input-common-udp-options.asciidoc

@@ -15,3 +15,9 @@ The maximum size of the message received over UDP. The default is `10KiB`.
 ==== `host`
 
 The host and UDP port to listen on for event streams.
+
+[float]
+[id="{beatname_lc}-input-{type}-udp-read-buffer"]
+=== `read_buffer`
+
+The size of the read buffer on the UDP socket.

filebeat/input/udp/config.go

@@ -35,7 +35,7 @@ var defaultConfig = config{
 		// TODO: What should be default port?
 		Host: "localhost:8080",
 		// TODO: What should be the default timeout?
-		Timeout: time.Minute * 5,
+		Timeout:    time.Minute * 5,
 	},
 }
 

filebeat/inputsource/udp/config.go

@@ -28,4 +28,5 @@ type Config struct {
 	Host           string           `config:"host"`
 	MaxMessageSize cfgtype.ByteSize `config:"max_message_size" validate:"positive,nonzero"`
 	Timeout        time.Duration    `config:"timeout"`
+	ReadBuffer     cfgtype.ByteSize `config:"read_buffer" validate:"positive,nonzero"`
 }

filebeat/inputsource/udp/server.go

@@ -24,6 +24,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/dustin/go-humanize"
 	"github.com/elastic/beats/filebeat/inputsource"
 	"github.com/elastic/beats/libbeat/logp"
 )
@@ -40,7 +41,7 @@ const windowErrBuffer = "A message sent on a datagram socket was larger than the
 type Server struct {
 	config   *Config
 	callback inputsource.NetworkFunc
-	Listener net.PacketConn
+	Listener *net.UDPConn
 	log      *logp.Logger
 	wg       sync.WaitGroup
 	done     chan struct{}
@@ -59,7 +60,20 @@ func New(config *Config, callback inputsource.NetworkFunc) *Server {
 // Start starts the UDP Server and listen to incoming events.
 func (u *Server) Start() error {
 	var err error
-	u.Listener, err = net.ListenPacket("udp", u.config.Host)
+	udpAdddr, err := net.ResolveUDPAddr("udp", u.config.Host)
+	if err != nil {
+		return err
+	}
+	u.Listener, err = net.ListenUDP("udp", udpAdddr)
+	if err != nil {
+		return err
+	}
+    socketSize := int(u.config.ReadBuffer) * humanize.KiByte
+	if socketSize != 0 {
+		if err := u.Listener.SetReadBuffer(int(u.config.ReadBuffer)); err != nil {
+			return err
+		}
+	}
 	if err != nil {
 		return err
 	}

filebeat/inputsource/udp/server_test.go

@@ -29,6 +29,7 @@ import (
 )
 
 const maxMessageSize = 20
+const maxSocketSize = 0
 const timeout = time.Second * 15
 
 type info struct {
@@ -56,7 +57,12 @@ func TestReceiveEventFromUDP(t *testing.T) {
 
 	ch := make(chan info)
 	host := "localhost:0"
-	config := &Config{Host: host, MaxMessageSize: maxMessageSize, Timeout: timeout}
+	config := &Config{
+		Host:           host,
+		MaxMessageSize: maxMessageSize,
+		Timeout:        timeout,
+		ReadBuffer:     maxSocketSize,
+	}
 	fn := func(message []byte, metadata inputsource.NetworkMetadata) {
 		ch <- info{message: message, mt: metadata}
 	}