Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add option to set rcv_buffer for UDP inputs #11739

Merged
merged 18 commits into from
Apr 15, 2019
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions CHANGELOG.next.asciidoc
Original file line number Diff line number Diff line change
Expand Up @@ -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}11739[11739]

*Heartbeat*

Expand Down
4 changes: 4 additions & 0 deletions filebeat/_meta/common.reference.inputs.yml
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
6 changes: 6 additions & 0 deletions filebeat/docs/inputs/input-common-udp-options.asciidoc
Original file line number Diff line number Diff line change
Expand Up @@ -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.
4 changes: 4 additions & 0 deletions filebeat/filebeat.reference.yml
Original file line number Diff line number Diff line change
Expand Up @@ -646,6 +646,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
Expand Down
1 change: 1 addition & 0 deletions filebeat/inputsource/udp/config.go
Original file line number Diff line number Diff line change
Expand Up @@ -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"`
}
19 changes: 17 additions & 2 deletions filebeat/inputsource/udp/server.go
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,8 @@ import (
"sync"
"time"

"github.com/dustin/go-humanize"

"github.com/elastic/beats/filebeat/inputsource"
"github.com/elastic/beats/libbeat/logp"
)
Expand All @@ -40,7 +42,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{}
Expand All @@ -59,7 +61,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
}
Expand Down
8 changes: 7 additions & 1 deletion filebeat/inputsource/udp/server_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@ import (
)

const maxMessageSize = 20
const maxSocketSize = 0
const timeout = time.Second * 15

type info struct {
Expand Down Expand Up @@ -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}
}
Expand Down
4 changes: 4 additions & 0 deletions x-pack/filebeat/filebeat.reference.yml
Original file line number Diff line number Diff line change
Expand Up @@ -723,6 +723,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
Expand Down