Make UDP socket buffer size configurable

cmd/agent/app/builder.go

@@ -87,9 +87,10 @@ type ProcessorConfiguration struct {
 
 // ServerConfiguration holds config for a server that receives spans from the network
 type ServerConfiguration struct {
-	QueueSize     int    `yaml:"queueSize"`
-	MaxPacketSize int    `yaml:"maxPacketSize"`
-	HostPort      string `yaml:"hostPort" validate:"nonzero"`
+	QueueSize        int    `yaml:"queueSize"`
+	MaxPacketSize    int    `yaml:"maxPacketSize"`
+	SocketBufferSize int    `yaml:"socketBufferSize"`
+	HostPort         string `yaml:"hostPort" validate:"nonzero"`
 }
 
 // HTTPServerConfiguration holds config for a server providing sampling strategies and baggage restrictions to clients
@@ -185,6 +186,7 @@ func (c *ProcessorConfiguration) applyDefaults() {
 func (c *ServerConfiguration) applyDefaults() {
 	c.QueueSize = defaultInt(c.QueueSize, defaultQueueSize)
 	c.MaxPacketSize = defaultInt(c.MaxPacketSize, defaultMaxPacketSize)
+	c.SocketBufferSize = defaultInt(c.SocketBufferSize, 0)
 }
 
 // getUDPServer gets a TBufferedServer backed server using the server configuration
@@ -198,6 +200,11 @@ func (c *ServerConfiguration) getUDPServer(mFactory metrics.Factory) (servers.Se
 	if err != nil {
 		return nil, fmt.Errorf("cannot create UDPServerTransport: %w", err)
 	}
+	if c.SocketBufferSize != 0 {
+		if err := transport.SetSocketBufferSize(c.SocketBufferSize); err != nil {
+			return nil, fmt.Errorf("cannot set UDP socket buffer size: %w", err)
+		}
+	}
 
 	return servers.NewTBufferedServer(transport, c.QueueSize, c.MaxPacketSize, mFactory)
 }

cmd/agent/app/builder_test.go

@@ -125,7 +125,7 @@ func TestBuilderWithProcessorErrors(t *testing.T) {
 	}{
 		{protocol: Protocol("bad"), err: "cannot find protocol factory for protocol bad"},
 		{protocol: compactProtocol, model: Model("bad"), err: "cannot find agent processor for data model bad"},
-		{protocol: compactProtocol, model: jaegerModel, err: "no host:port provided for udp server: {QueueSize:1000 MaxPacketSize:65000 HostPort:}"},
+		{protocol: compactProtocol, model: jaegerModel, err: "no host:port provided for udp server: {QueueSize:1000 MaxPacketSize:65000 SocketBufferSize:0 HostPort:}"},
 		{protocol: compactProtocol, model: zipkinModel, hostPort: "bad-host-port", errContains: "bad-host-port"},
 	}
 	for _, tc := range testCases {

cmd/agent/app/flags.go

@@ -26,10 +26,11 @@ import (
 )
 
 const (
-	suffixWorkers             = "workers"
-	suffixServerQueueSize     = "server-queue-size"
-	suffixServerMaxPacketSize = "server-max-packet-size"
-	suffixServerHostPort      = "server-host-port"
+	suffixWorkers                = "workers"
+	suffixServerQueueSize        = "server-queue-size"
+	suffixServerMaxPacketSize    = "server-max-packet-size"
+	suffixServerSocketBufferSize = "server-socket-buffer-size"
+	suffixServerHostPort         = "server-host-port"
 	// HTTPServerHostPort is the flag for HTTP endpoint
 	HTTPServerHostPort = "http-server.host-port"
 )
@@ -51,6 +52,7 @@ func AddFlags(flags *flag.FlagSet) {
 		flags.Int(prefix+suffixWorkers, defaultServerWorkers, "how many workers the processor should run")
 		flags.Int(prefix+suffixServerQueueSize, defaultQueueSize, "length of the queue for the UDP server")
 		flags.Int(prefix+suffixServerMaxPacketSize, defaultMaxPacketSize, "max packet size for the UDP server")
+		flags.Int(prefix+suffixServerSocketBufferSize, 0, "socket buffer size for UDP packets in bytes")
 		flags.String(prefix+suffixServerHostPort, ":"+strconv.Itoa(p.port), "host:port for the UDP server")
 	}
 	AddOTELFlags(flags)
@@ -72,6 +74,7 @@ func (b *Builder) InitFromViper(v *viper.Viper) *Builder {
 		p.Workers = v.GetInt(prefix + suffixWorkers)
 		p.Server.QueueSize = v.GetInt(prefix + suffixServerQueueSize)
 		p.Server.MaxPacketSize = v.GetInt(prefix + suffixServerMaxPacketSize)
+		p.Server.SocketBufferSize = v.GetInt(prefix + suffixServerSocketBufferSize)
 		p.Server.HostPort = portNumToHostPort(v.GetString(prefix + suffixServerHostPort))
 		b.Processors = append(b.Processors, *p)
 	}

cmd/agent/app/servers/thriftudp/socket_buffer.go

@@ -0,0 +1,31 @@
+// Copyright (c) 2020 The Jaeger Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// +build !windows
+
+package thriftudp
+
+import (
+	"net"
+	"syscall"
+)
+
+func setSocketBuffer(conn *net.UDPConn, bufferSize int) error {
+	file, err := conn.File()
+	if err != nil {
+		return err
+	}
+
+	return syscall.SetsockoptInt(int(file.Fd()), syscall.SOL_SOCKET, syscall.SO_RCVBUF, bufferSize)
+}

cmd/agent/app/servers/thriftudp/socket_buffer_windows.go

@@ -0,0 +1,24 @@
+// Copyright (c) 2020 The Jaeger Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package thriftudp
+
+import (
+	"net"
+)
+
+// Not supported on windows, so windows version just returns nil
+func setSocketBuffer(_ *net.UDPConn, _ int) error {
+	return nil
+}

cmd/agent/app/servers/thriftudp/transport.go

@@ -26,7 +26,9 @@ import (
 )
 
 //MaxLength of UDP packet
-const MaxLength = 65000
+const (
+	MaxLength = 65000
+)
 
 var errConnAlreadyClosed = errors.New("connection already closed")
 
@@ -83,6 +85,7 @@ func NewTUDPServerTransport(hostPort string) (*TUDPTransport, error) {
 	if err != nil {
 		return nil, thrift.NewTTransportException(thrift.NOT_OPEN, err.Error())
 	}
+
 	return &TUDPTransport{addr: conn.LocalAddr(), conn: conn}, nil
 }
 
@@ -153,3 +156,8 @@ func (p *TUDPTransport) Flush(_ context.Context) error {
 	p.writeBuf.Reset() // always reset the buffer, even in case of an error
 	return err
 }
+
+// SetSocketBufferSize sets udp buffer size
+func (p *TUDPTransport) SetSocketBufferSize(bufferSize int) error {
+	return setSocketBuffer(p.Conn(), bufferSize)
+}