Add config parameters for setting TCP and UDP socket buffer sizes

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
@@ -129,6 +131,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>>

lib/logstash/inputs/syslog.rb

@@ -74,6 +74,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 register
     @metric_errors = metric.namespace(:errors)
@@ -140,6 +152,13 @@ def udp_listener(output_queue)
     @udp.close if @udp
     @udp = UDPSocket.new(Socket::AF_INET)
     @udp.do_not_reverse_lookup = true
+    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?
@@ -163,6 +182,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)
 
@@ -174,7 +200,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]