diff --git a/lib/fluent/plugin/in_syslog.rb b/lib/fluent/plugin/in_syslog.rb
index af08e4044f..d6566cfcac 100644
--- a/lib/fluent/plugin/in_syslog.rb
+++ b/lib/fluent/plugin/in_syslog.rb
@@ -93,6 +93,10 @@ def initialize
     config_param :source_host_key, :string, default: 'source_host'.freeze, deprecated: "use source_hostname_key instead"
     desc "The field name of the client's hostname."
     config_param :source_hostname_key, :string, default: nil
+    desc "The field name of the client's source address."
+    config_param :source_address_key, :string, default: nil
+    desc 'Try to resolve hostname from IP addresses or not.'
+    config_param :resolve_hostname, :bool, default: nil
     desc 'The field name of the priority.'
     config_param :priority_key, :string, default: nil
     desc 'The field name of the facility.'
@@ -131,6 +135,13 @@ def configure(conf)
       if @source_hostname_key.nil? && @include_source_host
         @source_hostname_key = @source_host_key
       end
+      if @source_hostname_key
+        if @resolve_hostname.nil?
+          @resolve_hostname = true
+        elsif !@resolve_hostname # user specifies "false" in configure
+          raise Fluent::ConfigError, "resolve_hostname must be true with source_hostname_key"
+        end
+      end
     end
 
     def start
@@ -223,6 +234,7 @@ def parse_text(text, addr, pri = nil)
         record[@priority_key] = priority if @priority_key
         record[@facility_key] = facility if @facility_key
         record[@source_hostname_key] = addr[2] if @source_hostname_key
+        record[@source_address_key] = addr[3] if @source_address_key
 
         tag = "#{@tag}.#{facility}.#{priority}"
         emit(tag, time, record)
@@ -236,10 +248,10 @@ def listen(callback)
       if @protocol_type == :udp
         @usock = SocketUtil.create_udp_socket(@bind)
         @usock.bind(@bind, @port)
-        SocketUtil::UdpHandler.new(@usock, log, @message_length_limit, callback, !!@source_hostname_key)
+        SocketUtil::UdpHandler.new(@usock, log, @message_length_limit, callback, @resolve_hostname)
       else
         # syslog family add "\n" to each message and this seems only way to split messages in tcp stream
-        Coolio::TCPServer.new(@bind, @port, SocketUtil::TcpHandler, log, "\n", callback, !!@source_hostname_key)
+        Coolio::TCPServer.new(@bind, @port, SocketUtil::TcpHandler, log, "\n", callback, @resolve_hostname)
       end
     end
 
diff --git a/test/plugin/test_in_syslog.rb b/test/plugin/test_in_syslog.rb
index 6658f69c49..3818fb1819 100755
--- a/test/plugin/test_in_syslog.rb
+++ b/test/plugin/test_in_syslog.rb
@@ -35,6 +35,74 @@ def test_configure
     }
   end
 
+  sub_test_case 'source_hostname_key and source_address_key features' do
+    test 'resolve_hostname must be true with source_hostname_key' do
+      assert_raise(Fluent::ConfigError) {
+        create_driver(CONFIG + <<EOS)
+resolve_hostname false
+source_hostname_key hostname
+EOS
+      }
+    end
+
+    LOCALHOST_HOSTNAME_GETTER = ->(){sock = UDPSocket.new(::Socket::AF_INET); sock.do_not_reverse_lookup = false; sock.connect("127.0.0.1", 2048); sock.peeraddr[2] }
+    LOCALHOST_HOSTNAME = LOCALHOST_HOSTNAME_GETTER.call
+    DUMMY_SOCK = Struct.new(:remote_host, :remote_addr, :remote_port).new(LOCALHOST_HOSTNAME, "127.0.0.1", 0)
+    data(
+      both: [:hostname, :address],
+      hostname: [:hostname],
+      address: [:address],
+    )
+    test 'source_hostname_key and source_address_key parameter feature should add record(s)' do |keys|
+      conf = CONFIG.dup
+      if keys.include?(:hostname)
+        conf << <<EOL
+source_hostname_key source_hostname
+EOL
+        end
+      if keys.include?(:address)
+        conf << <<EOL
+source_address_key source_address
+EOL
+      end
+      tests = create_test_case
+      d = create_driver(conf)
+
+      d.run do
+        u = UDPSocket.new
+        u.connect('127.0.0.1', PORT)
+        tests.each {|test|
+          u.send(test['msg'], 0)
+        }
+        sleep 1
+      end
+
+      d.emits.each { |tag, _time, record|
+        if keys.include?(:hostname)
+          assert_true record.has_key?('source_hostname')
+          assert_equal DUMMY_SOCK.remote_host, record['source_hostname']
+          unless keys.include?(:address)
+            assert_false record.has_key?('source_address')
+          end
+        end
+        if keys.include?(:address)
+          assert_true record.has_key?('source_address')
+          assert_equal DUMMY_SOCK.remote_addr, record['source_address']
+          unless keys.include?(:hostname)
+            assert_false record.has_key?('source_hostname')
+          end
+        end
+      }
+    end
+
+    data('resolve_hostname' => 'resolve_hostname true',
+         'source_hostname_key' => 'source_hostname_key source_host')
+    def test_configure_reslove_hostname(param)
+      d = create_driver([CONFIG, param].join("\n"))
+      assert_true d.instance.resolve_hostname
+    end
+  end
+
   def test_time_format
     configs = {'127.0.0.1' => CONFIG}
     configs.merge!('::1' => IPv6_CONFIG) if ipv6_enabled?