From d3c2a61a90016a2f66cd29b9ee99a95271f6d9c1 Mon Sep 17 00:00:00 2001
From: Tyler Gregory <td.g@icloud.com>
Date: Wed, 27 Mar 2019 16:12:43 -0500
Subject: [PATCH] Add config parameters for setting TCP and UDP socket buffer
 sizes

---
 docs/index.asciidoc           | 27 +++++++++++++++++++++++++++
 lib/logstash/inputs/syslog.rb | 28 +++++++++++++++++++++++++++-
 2 files changed, 54 insertions(+), 1 deletion(-)

diff --git a/docs/index.asciidoc b/docs/index.asciidoc
index 9c15972..babb85f 100644
--- a/docs/index.asciidoc
+++ b/docs/index.asciidoc
@@ -52,6 +52,8 @@ This plugin supports the following configuration options plus the <<plugins-{typ
 | <<plugins-{type}s-{plugin}-host>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-locale>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-port>> |<<number,number>>|No
+| <<plugins-{type}s-{plugin}-tcp_receive_buffer_bytes>> |<<number,number>>|No
+| <<plugins-{type}s-{plugin}-udp_receive_buffer_bytes>> |<<number,number>>|No
 | <<plugins-{type}s-{plugin}-proxy_protocol>> |<<boolean,boolean>>|No
 | <<plugins-{type}s-{plugin}-severity_labels>> |<<array,array>>|No
 | <<plugins-{type}s-{plugin}-syslog_field>> |<<string,string>>|No
@@ -122,6 +124,31 @@ The port to listen on. Remember that ports less than 1024 (privileged
 ports) may require root to use.
 
 [id="plugins-{type}s-{plugin}-proxy_protocol"]
+
+===== `tcp_receive_buffer_bytes`
+
+  * Value type is <<number,number>>
+  * There is no default value for this setting
+
+The TCP socket receive buffer size in bytes.
+If option is not set, the operating system default is used.
+The operating system will use the max allowed value if `tcp_receive_buffer_bytes` is larger than allowed.
+Consult your operating system documentation if you need to increase this max allowed value.
+
+[id="plugins-{type}s-{plugin}-tcp_receive_buffer_bytes"]
+
+===== `udp_receive_buffer_bytes`
+
+  * Value type is <<number,number>>
+  * There is no default value for this setting
+
+The UDP socket receive buffer size in bytes.
+If option is not set, the operating system default is used.
+The operating system will use the max allowed value if `udp_receive_buffer_bytes` is larger than allowed.
+Consult your operating system documentation if you need to increase this max allowed value.
+
+[id="plugins-{type}s-{plugin}-udp_receive_buffer_bytes"]
+
 ===== `proxy_protocol`
 
   * Value type is <<boolean,boolean>>
diff --git a/lib/logstash/inputs/syslog.rb b/lib/logstash/inputs/syslog.rb
index ced9b3a..0fc493a 100644
--- a/lib/logstash/inputs/syslog.rb
+++ b/lib/logstash/inputs/syslog.rb
@@ -75,6 +75,18 @@ class LogStash::Inputs::Syslog < LogStash::Inputs::Base
   #
   config :locale, :validate => :string
 
+  # The UDP socket receive buffer size in bytes.
+  # If option is not set, the operating system default is used.
+  # The operating system will use the max allowed value if receive_buffer_bytes is larger than allowed.
+  # Consult your operating system documentation if you need to increase this max allowed value.
+  config :udp_receive_buffer_bytes, :validate => :number
+
+  # The TCP socket receive buffer size in bytes.
+  # If option is not set, the operating system default is used.
+  # The operating system will use the max allowed value if receive_buffer_bytes is larger than allowed.
+  # Consult your operating system documentation if you need to increase this max allowed value.
+  config :tcp_receive_buffer_bytes, :validate => :number
+
   public
   def initialize(params)
     super
@@ -146,6 +158,13 @@ def udp_listener(output_queue)
 
     @udp.close if @udp
     @udp = UDPSocket.new(Socket::AF_INET)
+    if @udp_receive_buffer_bytes
+      @udp.setsockopt(Socket::SOL_SOCKET, Socket::SO_RCVBUF, @udp_receive_buffer_bytes)
+      rcvbuf = @udp.getsockopt(Socket::SOL_SOCKET, Socket::SO_RCVBUF).unpack("i")[0]
+      if rcvbuf != @udp_receive_buffer_bytes
+        @logger.warn("Unable to set udp_receive_buffer_bytes to desired size. Requested #{@udp_receive_buffer_bytes} but obtained #{rcvbuf} bytes.")
+      end
+    end
     @udp.bind(@host, @port)
 
     while !stop?
@@ -168,6 +187,13 @@ def tcp_listener(output_queue)
 
     while !stop?
       socket = @tcp.accept
+      if @tcp_receive_buffer_bytes
+        socket.setsockopt(Socket::SOL_SOCKET, Socket::SO_RCVBUF, @tcp_receive_buffer_bytes)
+        rcvbuf = socket.getsockopt(Socket::SOL_SOCKET, Socket::SO_RCVBUF).unpack("i")[0]
+        if rcvbuf != @tcp_receive_buffer_bytes
+          @logger.warn("Unable to set tcp_receive_buffer_bytes to desired size. Requested #{@tcp_receive_buffer_bytes} but obtained #{rcvbuf} bytes.")
+        end
+      end
       @tcp_sockets << socket
       metric.increment(:connections)
 
@@ -179,7 +205,7 @@ def tcp_listener(output_queue)
     close_tcp
   end # def tcp_listener
 
-  # tcp_receiver is executed in a thread, any uncatched exception will be bubbled up to the
+  # tcp_receiver is executed in a thread, any uncaught exception will be bubbled up to the
   # tcp server thread and all tcp connections will be closed and the listener restarted.
   def tcp_receiver(output_queue, socket)
     ip, port = socket.peeraddr[3], socket.peeraddr[1]