Fix IPv6 support.

In Logstash 5.x and below, JRuby 1.7.x is used, and
UDPSocket.new(Socket::AF_INET) somehow accidentally works with IPv6.

Logstash 6.x uses JRuby 9.x where this now behaves correctly (AF_INET is
ipv4 only), but triggers a bug where ipv6 was accidentally disabled in
the udp input. Oops!

This change is small and adds tests to cover the
default scenario that 127.0.0.1 and ::1 both work for the default.

lib/logstash/inputs/udp.rb

@@ -82,7 +82,11 @@ def udp_listener(output_queue)
       @udp.close
     end
 
-    @udp = UDPSocket.new(Socket::AF_INET)
+    if IPAddr.new(@host).ipv6?
+      @udp = UDPSocket.new(Socket::AF_INET6)
+    elsif IPAddr.new(@host).ipv4?
+      @udp = UDPSocket.new(Socket::AF_INET)
+    end
     # set socket receive buffer size if configured
     if @receive_buffer_bytes
       @udp.setsockopt(Socket::SOL_SOCKET, Socket::SO_RCVBUF, @receive_buffer_bytes)

spec/inputs/udp_spec.rb

@@ -11,7 +11,8 @@
   let!(:helper) { UdpHelpers.new }
   let(:client) { LogStash::Inputs::Test::UDPClient.new(port) }
   let(:port) { rand(1024..65535) }
-  let(:config) { { "port" => port } }
+  let(:host) { "0.0.0.0" }
+  let(:config) { { "port" => port, "host" => host } }
   subject { LogStash::Plugin.lookup("input","udp").new(config) }
 
   after :each do
@@ -25,26 +26,38 @@
   end
 
   describe "receive" do
-    let(:nevents) { 10 }
+  	shared_examples "receiving" do
+      before(:each) do
+        subject.register
+      end
 
-    let(:events) do
-      helper.input(subject, nevents) do
-        nevents.times do |i|
-          client.send("msg #{i}")
+      let(:nevents) { 10 }
+
+      let(:events) do
+        helper.input(subject, nevents) do
+          nevents.times do |i|
+            client.send("msg #{i}")
+          end
         end
       end
-    end
 
-    before(:each) do
-      subject.register
+      it "should receive events been generated" do
+        expect(events.size).to be(nevents)
+        messages = events.map { |event| event.get("message")}
+        messages.each do |message|
+          expect(message).to match(/msg \d+/)
+        end
+      end
     end
 
-    it "should receive events been generated" do
-      expect(events.size).to be(nevents)
-      messages = events.map { |event| event.get("message")}
-      messages.each do |message|
-        expect(message).to match(/msg \d+/)
-      end
+    context "ipv4" do
+      let(:client) { LogStash::Inputs::Test::UDPClient.new(port, "127.0.0.1") }
+      include_examples "receiving"
+    end
+    context "ipv6" do
+      let(:host) { "::1" }
+      let(:client) { LogStash::Inputs::Test::UDPClient.new(port, "::1") }
+      include_examples "receiving"
     end
   end
 
@@ -74,4 +87,4 @@
   it_behaves_like "an interruptible input plugin" do
     # see https://github.com/elastic/logstash-devutils/blob/9c4a1fbf2b0c4547e428c5a40ed84f60aad17f97/lib/logstash/devutils/rspec/shared_examples.rb
   end
-end
+end

spec/support/client.rb

@@ -1,21 +1,25 @@
 # encoding: utf-8
 require "socket"
+require "ipaddr"
 
 module LogStash::Inputs::Test
 
   class UDPClient
     attr_reader :host, :port, :socket
 
-    def initialize(port)
+    def initialize(port, host = "0.0.0.0")
+      @host = host
       @port = port
-      @host = "0.0.0.0"
-      @socket = UDPSocket.new
-      socket.connect(host, port)
+      if IPAddr.new(@host).ipv6?
+        @socket = UDPSocket.new(Socket::AF_INET6)
+      elsif IPAddr.new(@host).ipv4?
+        @socket = UDPSocket.new(Socket::AF_INET)
+      end
+      @socket.connect(host, port)
     end
 
     def send(msg)
       begin
-        socket.connect(host, port) if socket.closed?
         socket.send(msg, 0)
       rescue  => e
         puts("send exception, retrying", e.inspect)