From 812f9bae92a0c0eab06e93109d8657cd815f41c5 Mon Sep 17 00:00:00 2001
From: bootjp <oh@bootjp.me>
Date: Wed, 14 Nov 2018 19:03:42 +0900
Subject: [PATCH 01/14] Add connection check at startup when out_forward.

Signed-off-by: bootjp <oh@bootjp.me>
---
 lib/fluent/plugin/out_forward.rb |  27 +++++++
 test/plugin/test_out_forward.rb  | 120 +++++++++++++++++++++++++++++++
 2 files changed, 147 insertions(+)

diff --git a/lib/fluent/plugin/out_forward.rb b/lib/fluent/plugin/out_forward.rb
index ca25092ccd..8570ecc429 100644
--- a/lib/fluent/plugin/out_forward.rb
+++ b/lib/fluent/plugin/out_forward.rb
@@ -77,6 +77,9 @@ class ConnectionClosedError < Error; end
     desc 'Ignore DNS resolution and errors at startup time.'
     config_param :ignore_network_errors_at_startup, :bool, default: false
 
+    desc 'Verify that a connection can be made with one of out_forward nodes at the time of startup.'
+    config_param :verify_connection_at_startup, :bool, default: false
+
     desc 'Compress buffered data.'
     config_param :compress, :enum, list: [:text, :gzip], default: :text
 
@@ -253,6 +256,17 @@ def start
         @sock_ack_waiting = []
         thread_create(:out_forward_receiving_ack, &method(:ack_reader))
       end
+
+      if @verify_connection_at_startup
+        @nodes.each do |node|
+          begin
+            node.verify_connection
+          rescue StandardError => e
+            log.fatal e.message
+            raise Fluent::UnrecoverableError, e.message
+          end
+        end
+      end
     end
 
     def close
@@ -568,6 +582,19 @@ def standby?
         @standby
       end
 
+      def verify_connection
+        sock = @sender.create_transfer_socket(resolved_host, port, @hostname)
+        begin
+          ri = RequestInfo.new(@sender.security ? :helo : :established)
+          if ri.state != :established
+            establish_connection(sock, ri)
+            raise if ri.state != :established
+          end
+        ensure
+          sock.close
+        end
+      end
+
       def establish_connection(sock, ri)
         while available? && ri.state != :established
           begin
diff --git a/test/plugin/test_out_forward.rb b/test/plugin/test_out_forward.rb
index 7a14a5614b..ac372abd72 100644
--- a/test/plugin/test_out_forward.rb
+++ b/test/plugin/test_out_forward.rb
@@ -217,6 +217,18 @@ def read_ack_from_sock(sock, unpacker)
     assert_equal 2, d.instance.ack_response_timeout
   end
 
+  test 'verify_connection_at_startup is disabled in default' do
+    @d = d = create_driver(CONFIG)
+    assert_equal false, d.instance.verify_connection_at_startup
+  end
+
+  test 'verify_connection_at_startup can be enabled' do
+    @d = d = create_driver(CONFIG + %[
+      verify_connection_at_startup true
+    ])
+    assert_equal true, d.instance.verify_connection_at_startup
+  end
+
   test 'send tags in str (utf-8 strings)' do
     target_input_driver = create_target_input_driver
 
@@ -774,4 +786,112 @@ def create_target_input_driver(response_stub: nil, disconnect: false, conf: TARG
       i.configure(conf)
     end
   end
+
+  test 'verify_connection_at_startup test' do
+      @d = d = create_driver(CONFIG + %[
+        verify_connection_at_startup true
+        <buffer tag>
+          flush_mode immediate
+          retry_type periodic
+          retry_wait 30s
+          flush_at_shutdown false # suppress errors in d.instance_shutdown
+        </buffer>
+      ])
+    assert_raise Fluent::UnrecoverableError do
+      d.instance_start
+    end
+    d.instance_shutdown
+  end
+
+  test 'authentication_with_shared_key_miss_match' do
+    input_conf = TARGET_CONFIG + %[
+                   <security>
+                     self_hostname in.localhost
+                     shared_key fluentd-sharedkey
+                   </security>
+                 ]
+    target_input_driver = create_target_input_driver(conf: input_conf)
+
+    output_conf = %[
+      send_timeout 30
+      heartbeat_type transport
+      transport tls
+      tls_verify_hostname false
+      verify_connection_at_startup true
+      require_ack_response true
+      ack_response_timeout 5s
+      <security>
+        self_hostname localhost
+        shared_key key_miss_match
+      </security>
+      <buffer tag>
+        flush_mode immediate
+        retry_type periodic
+        retry_wait 30s
+        flush_at_shutdown false # suppress errors in d.instance_shutdown
+        flush_thread_interval 31s
+      </buffer>
+
+      <server>
+        host #{TARGET_HOST}
+        port #{TARGET_PORT}
+      </server>
+    ]
+    @d = d = create_driver(output_conf)
+
+    target_input_driver.run(expect_records: 1, timeout: 15) do
+      assert_raise Fluent::UnrecoverableError do
+        d.instance_start
+      end
+      d.instance_shutdown
+    end
+  end
+
+  test 'authentication_with_shared_key_match' do
+      input_conf = TARGET_CONFIG + %[
+                       <security>
+                         self_hostname in.localhost
+                         shared_key fluentd-sharedkey
+                         <client>
+                           host 127.0.0.1
+                         </client>
+                       </security>
+                     ]
+      target_input_driver = create_target_input_driver(conf: input_conf)
+
+      output_conf = %[
+          send_timeout 51
+          verify_connection_at_startup true
+          <security>
+            self_hostname localhost
+            shared_key fluentd-sharedkey
+          </security>
+          <server>
+            name test
+            host #{TARGET_HOST}
+            port #{TARGET_PORT}
+            shared_key fluentd-sharedkey
+          </server>
+        ]
+      @d = d = create_driver(output_conf)
+
+      time = event_time("2011-01-02 13:14:15 UTC")
+      records = [
+          {"a" => 1},
+          {"a" => 2}
+      ]
+
+      target_input_driver.run(expect_records: 2, timeout: 15) do
+        d.run(default_tag: 'test') do
+          records.each do |record|
+            d.feed(time, record)
+          end
+        end
+      end
+
+      events = target_input_driver.events
+      assert{ events != [] }
+      assert_equal(['test', time, records[0]], events[0])
+      assert_equal(['test', time, records[1]], events[1])
+    end
 end

From ef2e4b796a462252e9d0e8606dcf2776b9ec2a79 Mon Sep 17 00:00:00 2001
From: bootjp <oh@bootjp.me>
Date: Fri, 16 Nov 2018 10:24:09 +0900
Subject: [PATCH 02/14] Fix test description.

Signed-off-by: bootjp <oh@bootjp.me>
---
 test/plugin/test_out_forward.rb | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/test/plugin/test_out_forward.rb b/test/plugin/test_out_forward.rb
index ac372abd72..9adb61b06d 100644
--- a/test/plugin/test_out_forward.rb
+++ b/test/plugin/test_out_forward.rb
@@ -787,7 +787,7 @@ def create_target_input_driver(response_stub: nil, disconnect: false, conf: TARG
     end
   end
 
-  test 'verify_connection_at_startup test' do
+  test 'verify_connection_at_startup_nodes_are_not_available' do
       @d = d = create_driver(CONFIG + %[
         verify_connection_at_startup true
         <buffer tag>
@@ -803,7 +803,7 @@ def create_target_input_driver(response_stub: nil, disconnect: false, conf: TARG
     d.instance_shutdown
   end
 
-  test 'authentication_with_shared_key_miss_match' do
+  test 'verify_connection_at_startup_nodes_shared_key_miss_match' do
     input_conf = TARGET_CONFIG + %[
                    <security>
                      self_hostname in.localhost
@@ -847,7 +847,7 @@ def create_target_input_driver(response_stub: nil, disconnect: false, conf: TARG
     end
   end
 
-  test 'authentication_with_shared_key_match' do
+  test 'verify_connection_at_startup_nodes_shared_key_match' do
       input_conf = TARGET_CONFIG + %[
                        <security>
                          self_hostname in.localhost

From 77d8943eea0043e3302eb126f948a0556d42d8ab Mon Sep 17 00:00:00 2001
From: bootjp <oh@bootjp.me>
Date: Thu, 22 Nov 2018 12:57:18 +0900
Subject: [PATCH 03/14] Fix error message.

Signed-off-by: bootjp <oh@bootjp.me>
---
 lib/fluent/plugin/out_forward.rb | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/lib/fluent/plugin/out_forward.rb b/lib/fluent/plugin/out_forward.rb
index 8570ecc429..780545d5a9 100644
--- a/lib/fluent/plugin/out_forward.rb
+++ b/lib/fluent/plugin/out_forward.rb
@@ -262,7 +262,7 @@ def start
           begin
             node.verify_connection
           rescue StandardError => e
-            log.fatal e.message
+            log.fatal "forward's connection setting error: #{e.message}"
             raise Fluent::UnrecoverableError, e.message
           end
         end

From 8a7022dcc2d6f95cdc2b99f4637465823efe7dd5 Mon Sep 17 00:00:00 2001
From: bootjp <oh@bootjp.me>
Date: Thu, 22 Nov 2018 13:01:26 +0900
Subject: [PATCH 04/14] Fix style of test case.

Signed-off-by: bootjp <oh@bootjp.me>
---
 test/plugin/test_out_forward.rb | 87 +++++++++++++++++----------------
 1 file changed, 44 insertions(+), 43 deletions(-)

diff --git a/test/plugin/test_out_forward.rb b/test/plugin/test_out_forward.rb
index 9adb61b06d..946f303fb2 100644
--- a/test/plugin/test_out_forward.rb
+++ b/test/plugin/test_out_forward.rb
@@ -219,14 +219,14 @@ def read_ack_from_sock(sock, unpacker)
 
   test 'verify_connection_at_startup is disabled in default' do
     @d = d = create_driver(CONFIG)
-    assert_equal false, d.instance.verify_connection_at_startup
+    assert_false d.instance.verify_connection_at_startup
   end
 
   test 'verify_connection_at_startup can be enabled' do
     @d = d = create_driver(CONFIG + %[
       verify_connection_at_startup true
     ])
-    assert_equal true, d.instance.verify_connection_at_startup
+    assert_true d.instance.verify_connection_at_startup
   end
 
   test 'send tags in str (utf-8 strings)' do
@@ -787,7 +787,8 @@ def create_target_input_driver(response_stub: nil, disconnect: false, conf: TARG
     end
   end
 
-  test 'verify_connection_at_startup_nodes_are_not_available' do
+  sub_test_case 'verify_connection_at_startup' do
+    test 'nodes are not available' do
       @d = d = create_driver(CONFIG + %[
         verify_connection_at_startup true
         <buffer tag>
@@ -797,57 +798,56 @@ def create_target_input_driver(response_stub: nil, disconnect: false, conf: TARG
           flush_at_shutdown false # suppress errors in d.instance_shutdown
         </buffer>
       ])
-    assert_raise Fluent::UnrecoverableError do
-      d.instance_start
+      assert_raise Fluent::UnrecoverableError do
+        d.instance_start
+      end
+      d.instance_shutdown
     end
-    d.instance_shutdown
-  end
 
-  test 'verify_connection_at_startup_nodes_shared_key_miss_match' do
-    input_conf = TARGET_CONFIG + %[
+    test 'nodes_shared_key_miss_match' do
+      input_conf = TARGET_CONFIG + %[
                    <security>
                      self_hostname in.localhost
                      shared_key fluentd-sharedkey
                    </security>
                  ]
-    target_input_driver = create_target_input_driver(conf: input_conf)
-
-    output_conf = %[
-      send_timeout 30
-      heartbeat_type transport
-      transport tls
-      tls_verify_hostname false
-      verify_connection_at_startup true
-      require_ack_response true
-      ack_response_timeout 5s
-      <security>
-        self_hostname localhost
-        shared_key key_miss_match
-      </security>
-      <buffer tag>
-        flush_mode immediate
-        retry_type periodic
-        retry_wait 30s
-        flush_at_shutdown false # suppress errors in d.instance_shutdown
-        flush_thread_interval 31s
-      </buffer>
+      target_input_driver = create_target_input_driver(conf: input_conf)
+      output_conf = %[
+        send_timeout 30
+        heartbeat_type transport
+        transport tls
+        tls_verify_hostname false
+        verify_connection_at_startup true
+        require_ack_response true
+        ack_response_timeout 5s
+        <security>
+          self_hostname localhost
+          shared_key key_miss_match
+        </security>
+        <buffer tag>
+          flush_mode immediate
+          retry_type periodic
+          retry_wait 30s
+          flush_at_shutdown false # suppress errors in d.instance_shutdown
+          flush_thread_interval 31s
+        </buffer>
 
-      <server>
-        host #{TARGET_HOST}
-        port #{TARGET_PORT}
-      </server>
-    ]
-    @d = d = create_driver(output_conf)
+        <server>
+          host #{TARGET_HOST}
+          port #{TARGET_PORT}
+        </server>
+      ]
+      @d = d = create_driver(output_conf)
 
-    target_input_driver.run(expect_records: 1, timeout: 15) do
-      assert_raise Fluent::UnrecoverableError do
-        d.instance_start
+      target_input_driver.run(expect_records: 1, timeout: 15) do
+        assert_raise Fluent::UnrecoverableError do
+          d.instance_start
+        end
+        d.instance_shutdown
       end
-      d.instance_shutdown
     end
-  end
 
-  test 'verify_connection_at_startup_nodes_shared_key_match' do
+    test 'nodes_shared_key_match' do
       input_conf = TARGET_CONFIG + %[
                        <security>
                          self_hostname in.localhost
@@ -872,7 +872,7 @@ def create_target_input_driver(response_stub: nil, disconnect: false, conf: TARG
             port #{TARGET_PORT}
             shared_key fluentd-sharedkey
           </server>
-        ]
+      ]
       @d = d = create_driver(output_conf)
 
       time = event_time("2011-01-02 13:14:15 UTC")
@@ -894,4 +894,5 @@ def create_target_input_driver(response_stub: nil, disconnect: false, conf: TARG
       assert_equal(['test', time, records[0]], events[0])
       assert_equal(['test', time, records[1]], events[1])
     end
+  end
 end

From c4d0a88b7f6b2772be66f84d5b6590cd03ef5310 Mon Sep 17 00:00:00 2001
From: Tatsuki Sugiura <sugi@nemui.org>
Date: Thu, 22 Nov 2018 13:14:04 +0900
Subject: [PATCH 05/14] Keep `class` method to avoid error on Sigdump.dump.

When stopping fluentd, sometime I get follwoing error message on stderr;

```
Unexpected error undefined method `class' for #<Fluent::Plugin::RecordTransformerFilter::RubyPlaceholderExpander::CleanroomExpander:0x00007f7f1aca50d0>
  /opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/sigdump-0.2.4/lib/sigdump.rb:74:in `block in dump_object_count'
  /opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/sigdump-0.2.4/lib/sigdump.rb:73:in `each_object'
  /opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/sigdump-0.2.4/lib/sigdump.rb:73:in `dump_object_count'
  /opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/sigdump-0.2.4/lib/sigdump.rb:18:in `block in dump'
  /opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/sigdump-0.2.4/lib/sigdump.rb:136:in `open'
  /opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/sigdump-0.2.4/lib/sigdump.rb:136:in `_open_dump_path'
  /opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/sigdump-0.2.4/lib/sigdump.rb:14:in `dump'
  /opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/serverengine-2.0.7/lib/serverengine/server.rb:74:in `dump'
  /opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/serverengine-2.0.7/lib/serverengine/server.rb:112:in `block (2 levels) in install_signal_handlers'
  /opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/serverengine-2.0.7/lib/serverengine/signal_thread.rb:96:in `main'
```

This cause Sigdump.dump fails with CleanroomExpander because it will
call `class` method for all objects to gathering stats.

This patch just keep `class` method to avoid error.

Signed-off-by: Tatsuki Sugiura <sugi@nemui.org>
---
 lib/fluent/plugin/filter_record_transformer.rb | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/lib/fluent/plugin/filter_record_transformer.rb b/lib/fluent/plugin/filter_record_transformer.rb
index c6a74b2fd4..02ba686a39 100644
--- a/lib/fluent/plugin/filter_record_transformer.rb
+++ b/lib/fluent/plugin/filter_record_transformer.rb
@@ -316,7 +316,7 @@ def expand(__str_to_eval__, tag, time, record, tag_parts, tag_prefix, tag_suffix
         end
 
         (Object.instance_methods).each do |m|
-          undef_method m unless m.to_s =~ /^__|respond_to_missing\?|object_id|public_methods|instance_eval|method_missing|define_singleton_method|respond_to\?|new_ostruct_member/
+          undef_method m unless m.to_s =~ /^__|respond_to_missing\?|object_id|public_methods|instance_eval|method_missing|define_singleton_method|respond_to\?|new_ostruct_member|^class$/
         end
       end
     end

From 32483a7f2617742efc466bae3f03fb6afbf0190a Mon Sep 17 00:00:00 2001
From: Konstantin Grachev <me@grachevko.ru>
Date: Sun, 25 Nov 2018 16:24:28 +0300
Subject: [PATCH 06/14] [Bugfix] Nginx Parser: http_x_forwarded_for can contain
 multiple IP divided by comma

Signed-off-by: Konstantin Grachev <me@grachevko.ru>
---
 lib/fluent/plugin/parser_nginx.rb |  2 +-
 test/plugin/test_parser_nginx.rb  | 20 ++++++++++++++++++++
 2 files changed, 21 insertions(+), 1 deletion(-)

diff --git a/lib/fluent/plugin/parser_nginx.rb b/lib/fluent/plugin/parser_nginx.rb
index 9717e1bcbe..e9090ef06d 100644
--- a/lib/fluent/plugin/parser_nginx.rb
+++ b/lib/fluent/plugin/parser_nginx.rb
@@ -21,7 +21,7 @@ module Plugin
     class NginxParser < RegexpParser
       Plugin.register_parser("nginx", self)
 
-      config_set_default :expression, /^(?<remote>[^ ]*) (?<host>[^ ]*) (?<user>[^ ]*) \[(?<time>[^\]]*)\] "(?<method>\S+)(?: +(?<path>[^\"]*?)(?: +\S*)?)?" (?<code>[^ ]*) (?<size>[^ ]*)(?: "(?<referer>[^\"]*)" "(?<agent>[^\"]*)"(?:\s+(?<http_x_forwarded_for>[^ ]+))?)?$/
+      config_set_default :expression, /^(?<remote>[^ ]*) (?<host>[^ ]*) (?<user>[^ ]*) \[(?<time>[^\]]*)\] "(?<method>\S+)(?: +(?<path>[^\"]*?)(?: +\S*)?)?" (?<code>[^ ]*) (?<size>[^ ]*)(?: "(?<referer>[^\"]*)" "(?<agent>[^\"]*)"(?:\s+\"?(?<http_x_forwarded_for>[^\"]*)\"?)?)?$/
       config_set_default :time_format, "%d/%b/%Y:%H:%M:%S %z"
     end
   end
diff --git a/test/plugin/test_parser_nginx.rb b/test/plugin/test_parser_nginx.rb
index fc06485da4..7a6c3cd8eb 100644
--- a/test/plugin/test_parser_nginx.rb
+++ b/test/plugin/test_parser_nginx.rb
@@ -28,6 +28,18 @@ def setup
       'agent'                => 'Opera/12.0',
       'http_x_forwarded_for' => '-'
     }
+    @expected_extended_multiple_ip = {
+      'remote'               => '127.0.0.1',
+      'host'                 => '192.168.0.1',
+      'user'                 => '-',
+      'method'               => 'GET',
+      'path'                 => '/',
+      'code'                 => '200',
+      'size'                 => '777',
+      'referer'              => '-',
+      'agent'                => 'Opera/12.0',
+      'http_x_forwarded_for' => '127.0.0.1, 192.168.0.1'
+    }
   end
 
   def create_driver
@@ -65,4 +77,12 @@ def test_parse_with_http_x_forwarded_for
       assert_equal(@expected_extended, record)
     }
   end
+
+  def test_parse_with_http_x_forwarded_for_multiple_ip
+    d = create_driver
+    d.instance.parse('127.0.0.1 192.168.0.1 - [28/Feb/2013:12:00:00 +0900] "GET / HTTP/1.1" 200 777 "-" "Opera/12.0" "127.0.0.1, 192.168.0.1"') { |time, record|
+      assert_equal(event_time('28/Feb/2013:12:00:00 +0900', format: '%d/%b/%Y:%H:%M:%S %z'), time)
+      assert_equal(@expected_extended_multiple_ip, record)
+    }
+  end
 end

From ba3471630e6d3e90bc3e0554816579eeb0194c8a Mon Sep 17 00:00:00 2001
From: Masahiro Nakagawa <repeatedly@gmail.com>
Date: Tue, 27 Nov 2018 07:34:59 +0900
Subject: [PATCH 07/14] out_forward: Separate parameter names for certificates

Signed-off-by: Masahiro Nakagawa <repeatedly@gmail.com>
---
 lib/fluent/plugin/out_forward.rb | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/lib/fluent/plugin/out_forward.rb b/lib/fluent/plugin/out_forward.rb
index c93706732a..56a165be39 100644
--- a/lib/fluent/plugin/out_forward.rb
+++ b/lib/fluent/plugin/out_forward.rb
@@ -92,7 +92,8 @@ class ConnectionClosedError < Error; end
     config_param :tls_verify_hostname, :bool, default: true
     desc 'The additional CA certificate path for TLS.'
     config_param :tls_ca_cert_path, :array, value_type: :string, default: nil
-    config_param :tls_cert_path, :array, value_type: :string, default: nil, deprecated: "Use tls_ca_cert_path instead"
+    desc 'The additional certificate path for TLS.'
+    config_param :tls_cert_path, :array, value_type: :string, default: nil
 
     config_section :security, required: false, multi: false do
       desc 'The hostname'
@@ -167,7 +168,7 @@ def configure(conf)
       end
 
       if @transport == :tls
-        # for backward compatibility
+        # socket helper adds CA cert or signed certificate to same cert store internally so unify it in this place.
         if @tls_cert_path && !@tls_cert_path.empty?
           @tls_ca_cert_path = @tls_cert_path
         end

From cdfc43885d9c9eed4f7355fb403b3d88a6d91fea Mon Sep 17 00:00:00 2001
From: Masahiro Nakagawa <repeatedly@gmail.com>
Date: Tue, 27 Nov 2018 09:34:09 +0900
Subject: [PATCH 08/14] out_forward: Apply change to test

Signed-off-by: Masahiro Nakagawa <repeatedly@gmail.com>
---
 test/plugin/test_out_forward.rb | 11 +++++------
 1 file changed, 5 insertions(+), 6 deletions(-)

diff --git a/test/plugin/test_out_forward.rb b/test/plugin/test_out_forward.rb
index 1b9be7a123..aaa11fbb68 100644
--- a/test/plugin/test_out_forward.rb
+++ b/test/plugin/test_out_forward.rb
@@ -157,14 +157,16 @@ def read_ack_from_sock(sock, unpacker)
     assert{ logs.any?{|log| log.include?(expected_log) && log.include?(expected_detail) } }
   end
 
-  test 'configure tls_cert_path is deprecated' do
+  data('CA cert'     => 'tls_ca_cert_path',
+       'non CA cert' => 'tls_cert_path')
+  test 'configure tls_cert_path/tls_ca_cert_path' do |param|
     dummy_cert_path = File.join(TMP_DIR, "dummy_cert.pem")
     FileUtils.touch(dummy_cert_path)
     conf = %[
       send_timeout 5
       transport tls
       tls_insecure_mode true
-      tls_cert_path #{dummy_cert_path}
+      #{param} #{dummy_cert_path}
       <server>
         host #{TARGET_HOST}
         port #{TARGET_PORT}
@@ -172,10 +174,7 @@ def read_ack_from_sock(sock, unpacker)
     ]
 
     @d = d = create_driver(conf)
-    expected_log = "'tls_cert_path' parameter is deprecated: Use tls_ca_cert_path instead"
-    logs = d.logs
-    assert{ logs.any?{|log| log.include?(expected_log) } }
-    assert_equal([dummy_cert_path], d.instance.tls_cert_path)
+    # In the plugin, tls_ca_cert_path is used for both cases
     assert_equal([dummy_cert_path], d.instance.tls_ca_cert_path)
   end
 

From f70e743a94d12f04832e9a1cfda9a0f0b965d5a9 Mon Sep 17 00:00:00 2001
From: Kenji Okimoto <okimoto@clear-code.com>
Date: Tue, 27 Nov 2018 15:56:45 +0900
Subject: [PATCH 09/14] Delete top level using bracket style properly

In previous version, following configuration does not work properly:

```
<source>
  @type dummy
  dummy [
    {"foo.bar": "test1234", "message": "Hello"}
  ]
  tag dummy
</source>

<filter dummy>
  @type record_transformer
  remove_keys "$['foo.bar']"
</filter>

<match dummy>
  @type stdout
</match>
```

This shows like following:

```
2018-11-27 15:19:18 +0900 [info]: #0 fluentd worker is now running worker=0
2018-11-27 15:19:19.008586045 +0900 dummy: {"foo.bar":"test1234","message":"Hello"}
2018-11-27 15:19:20.009721132 +0900 dummy: {"foo.bar":"test1234","message":"Hello"}
2018-11-27 15:19:21.010784035 +0900 dummy: {"foo.bar":"test1234","message":"Hello"}
```

In this version, it works well.

See also #2109

Signed-off-by: Kenji Okimoto <okimoto@clear-code.com>
---
 lib/fluent/plugin_helper/record_accessor.rb | 10 ++++++++--
 test/plugin_helper/test_record_accessor.rb  | 13 +++++++++++++
 2 files changed, 21 insertions(+), 2 deletions(-)

diff --git a/lib/fluent/plugin_helper/record_accessor.rb b/lib/fluent/plugin_helper/record_accessor.rb
index 59062379bd..b34da5e3f8 100644
--- a/lib/fluent/plugin_helper/record_accessor.rb
+++ b/lib/fluent/plugin_helper/record_accessor.rb
@@ -40,8 +40,14 @@ def initialize(param)
           if @keys.is_a?(Array)
             @last_key = @keys.last
             @dig_keys = @keys[0..-2]
-            mcall = method(:call_dig)
-            mdelete = method(:delete_nest)
+            if @dig_keys.empty?
+              @keys = @keys.first
+              mcall = method(:call_index)
+              mdelete = method(:delete_top)
+            else
+              mcall = method(:call_dig)
+              mdelete = method(:delete_nest)
+            end
           else
             # Call [] for single key to reduce dig overhead
             mcall = method(:call_index)
diff --git a/test/plugin_helper/test_record_accessor.rb b/test/plugin_helper/test_record_accessor.rb
index f3e052bbcf..5d942ed77f 100644
--- a/test/plugin_helper/test_record_accessor.rb
+++ b/test/plugin_helper/test_record_accessor.rb
@@ -110,6 +110,12 @@ class Dummy < Fluent::Plugin::TestBase
       assert_equal r[param], accessor.call(r)
     end
 
+    test "access single dot key using bracket style" do
+      r = {'key1' => 'v1', 'ke y2' => 'v2', 'this.is.key3' => 'v3'}
+      accessor = @d.record_accessor_create('$["this.is.key3"]')
+      assert_equal 'v3', accessor.call(r)
+    end
+
     test "nested bracket keys with dot" do
       r = {'key1' => {'this.is.key3' => 'value'}}
       accessor = @d.record_accessor_create("$['key1']['this.is.key3']")
@@ -164,6 +170,13 @@ class Dummy < Fluent::Plugin::TestBase
       assert_not_include(r, param)
     end
 
+    test "delete top key using bracket style" do
+      r = {'key1' => 'v1', 'ke y2' => 'v2', 'this.is.key3' => 'v3'}
+      accessor = @d.record_accessor_create('$["this.is.key3"]')
+      accessor.delete(r)
+      assert_not_include(r, 'this.is.key3')
+    end
+
     data('bracket' => "$['key1'][0]['ke y2']",
          'bracket w/ double quotes' => '$["key1"][0]["ke y2"]')
     test "delete nested keys ['key1', 0, 'ke y2']" do |param|

From a82e9c619052299646265cbb8a1b9c87270a0d24 Mon Sep 17 00:00:00 2001
From: Kenji Okimoto <okimoto@clear-code.com>
Date: Tue, 27 Nov 2018 17:36:39 +0900
Subject: [PATCH 10/14] fluent-cat: Implement retry feature

Signed-off-by: Kenji Okimoto <okimoto@clear-code.com>
---
 lib/fluent/command/cat.rb | 28 +++++++++++++++++-----------
 1 file changed, 17 insertions(+), 11 deletions(-)

diff --git a/lib/fluent/command/cat.rb b/lib/fluent/command/cat.rb
index 935db5dfd1..b053f251f7 100644
--- a/lib/fluent/command/cat.rb
+++ b/lib/fluent/command/cat.rb
@@ -107,6 +107,8 @@
 class Writer
   include MonitorMixin
 
+  RetryLimitError = Class.new(StandardError)
+
   class TimerThread
     def initialize(writer)
       @writer = writer
@@ -236,21 +238,24 @@ def get_socket
   end
 
   def try_connect
-    now = Time.now.to_i
-
-    unless @error_history.empty?
-      # wait before re-connecting
-      wait = @retry_wait * (2 ** (@error_history.size-1))
-      if now <= @socket_time + wait
-        return false
+    begin
+      now = Time.now.to_i
+
+      unless @error_history.empty?
+        # wait before re-connecting
+        wait = 1 #@retry_wait * (2 ** (@error_history.size-1))
+        if now <= @socket_time + wait
+          sleep(wait)
+          try_connect
+        end
       end
-    end
 
-    begin
       @socket = @connector.call
       @error_history.clear
       return true
 
+    rescue RetryLimitError => ex
+      raise ex
     rescue
       $stderr.puts "connect failed: #{$!}"
       @error_history << $!
@@ -263,9 +268,10 @@ def try_connect
         }
         @pending.clear
         @error_history.clear
+        raise RetryLimitError, "exceed retry limit"
+      else
+        retry
       end
-
-      return false
     end
   end
 

From 9a6baa2f8e0c435170d48bc90177112c7413aa38 Mon Sep 17 00:00:00 2001
From: Kenji Okimoto <okimoto@clear-code.com>
Date: Tue, 27 Nov 2018 17:38:52 +0900
Subject: [PATCH 11/14] fluent-cat: Add --retry-limit to specify the number of
 retry limit

Signed-off-by: Kenji Okimoto <okimoto@clear-code.com>
---
 lib/fluent/command/cat.rb | 11 ++++++++---
 1 file changed, 8 insertions(+), 3 deletions(-)

diff --git a/lib/fluent/command/cat.rb b/lib/fluent/command/cat.rb
index b053f251f7..f218ac1776 100644
--- a/lib/fluent/command/cat.rb
+++ b/lib/fluent/command/cat.rb
@@ -34,6 +34,7 @@
 format = 'json'
 message_key = 'message'
 time_as_integer = false
+retry_limit = 5
 
 op.on('-p', '--port PORT', "fluent tcp port (default: #{port})", Integer) {|i|
   port = i
@@ -75,6 +76,10 @@
   time_as_integer = true
 }
 
+op.on('--retry-limit N', "Specify the number of retry limit (default: #{retry_limit})", Integer) {|n|
+  retry_limit = n
+}
+
 singleton_class.module_eval do
   define_method(:usage) do |msg|
     puts op.to_s
@@ -132,7 +137,7 @@ def run
     end
   end
 
-  def initialize(tag, connector, time_as_integer: false)
+  def initialize(tag, connector, time_as_integer: false, retry_limit: 5)
     @tag = tag
     @connector = connector
     @socket = false
@@ -144,7 +149,7 @@ def initialize(tag, connector, time_as_integer: false)
     @pending = []
     @pending_limit = 1024  # TODO
     @retry_wait = 1
-    @retry_limit = 5  # TODO
+    @retry_limit = retry_limit
     @time_as_integer = time_as_integer
 
     super()
@@ -291,7 +296,7 @@ def abort_message(time, record)
   }
 end
 
-w = Writer.new(tag, connector, time_as_integer: time_as_integer)
+w = Writer.new(tag, connector, time_as_integer: time_as_integer, retry_limit: retry_limit)
 w.start
 
 case format

From 3dabdc5c42918049933156db7c07437346b76773 Mon Sep 17 00:00:00 2001
From: Masahiro Nakagawa <repeatedly@gmail.com>
Date: Wed, 28 Nov 2018 06:05:14 +0900
Subject: [PATCH 12/14] v1.3.1

Signed-off-by: Masahiro Nakagawa <repeatedly@gmail.com>
---
 CHANGELOG.md          | 23 +++++++++++++++++++++++
 lib/fluent/version.rb |  2 +-
 2 files changed, 24 insertions(+), 1 deletion(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index a9ae6630d4..964ef10bb0 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,28 @@
 # v1.3
 
+## Release v1.3.1 - 2018/11/27
+
+### Enhancements
+
+* out_forward: Separate parameter names for certificate
+  https://github.com/fluent/fluentd/pull/2181
+  https://github.com/fluent/fluentd/pull/2190
+* out_forward: Add `verify_connection_at_startup` parameter to check connection setting at startup phase
+  https://github.com/fluent/fluentd/pull/2184
+* config: Check right slash position in regexp type
+  https://github.com/fluent/fluentd/pull/2176
+* parser_nginx: Support multiple IPs in `http_x_forwarded_for` field
+  https://github.com/fluent/fluentd/pull/2171
+
+### Bug fixes
+
+* fluent-cat: Fix retry limit handling
+  https://github.com/fluent/fluentd/pull/2193
+* record_accessor helper: Delete top level field with bracket style
+  https://github.com/fluent/fluentd/pull/2192
+* filter_record_transformer: Keep `class` methond to avoid undefined method error
+  https://github.com/fluent/fluentd/pull/2186
+
 ## Release v1.3.0 - 2018/11/10
 
 ### New features
diff --git a/lib/fluent/version.rb b/lib/fluent/version.rb
index 7b2f4d8707..8b4b254505 100644
--- a/lib/fluent/version.rb
+++ b/lib/fluent/version.rb
@@ -16,6 +16,6 @@
 
 module Fluent
 
-  VERSION = '1.3.0'
+  VERSION = '1.3.1'
 
 end

From e568e2345f00bf8bd6bd6dbd469922ebd62c3bca Mon Sep 17 00:00:00 2001
From: Adam Malcontenti-Wilson <amalcontenti-wilson@zendesk.com>
Date: Wed, 5 Dec 2018 11:24:52 -0800
Subject: [PATCH 13/14] Create pos file directory if it doesn't exist

Signed-off-by: Adam Malcontenti-Wilson <amalcontenti-wilson@zendesk.com>
---
 lib/fluent/plugin/in_tail.rb |  4 +++-
 test/plugin/test_in_tail.rb  | 19 +++++++++++++++++++
 2 files changed, 22 insertions(+), 1 deletion(-)

diff --git a/lib/fluent/plugin/in_tail.rb b/lib/fluent/plugin/in_tail.rb
index dbe628bea7..67bbec48de 100644
--- a/lib/fluent/plugin/in_tail.rb
+++ b/lib/fluent/plugin/in_tail.rb
@@ -185,6 +185,8 @@ def start
       super
 
       if @pos_file
+        pos_file_dir = File.dirname(@pos_file)
+        FileUtils.mkdir_p(pos_file_dir) unless Dir.exist?(pos_file_dir)
         @pf_file = File.open(@pos_file, File::RDWR|File::CREAT|File::BINARY, @file_perm)
         @pf_file.sync = true
         @pf = PositionFile.parse(@pf_file)
@@ -946,7 +948,7 @@ def initialize(file, file_mutex, seek, pos, inode)
         @file = file
         @file_mutex = file_mutex
         @seek = seek
-        @pos = pos 
+        @pos = pos
         @inode = inode
       end
 
diff --git a/test/plugin/test_in_tail.rb b/test/plugin/test_in_tail.rb
index 40ffa0d0a0..ef5b5a5a90 100644
--- a/test/plugin/test_in_tail.rb
+++ b/test/plugin/test_in_tail.rb
@@ -1011,6 +1011,25 @@ def count_timer_object
     end
   end
 
+  def test_pos_file_dir_creation
+    config = config_element("", "", {
+      "tag" => "tail",
+      "path" => "#{TMP_DIR}/*.txt",
+      "format" => "none",
+      "pos_file" => "#{TMP_DIR}/pos/tail.pos",
+      "read_from_head" => true,
+      "refresh_interval" => 1
+    })
+    d = create_driver(config, false)
+    d.run(expect_emits: 1, shutdown: false) do
+      File.open("#{TMP_DIR}/tail.txt", "ab") { |f| f.puts "test3\n" }
+    end
+    assert_path_exist("#{TMP_DIR}/pos/tail.pos")
+    cleanup_directory(TMP_DIR)
+
+    d.instance_shutdown
+  end
+
   def test_z_refresh_watchers
     plugin = create_driver(EX_CONFIG, false).instance
     sio = StringIO.new

From f8be39fefccece2275ebd0a45a1a19c7ab797ff2 Mon Sep 17 00:00:00 2001
From: Kenji Okimoto <okimoto@clear-code.com>
Date: Tue, 4 Dec 2018 12:43:57 +0900
Subject: [PATCH 14/14] Fix logs when retrying

In the previous version, the timestamp of next_retry_seconds is the same
as logs when retrying.

Fix #2201

Signed-off-by: Kenji Okimoto <okimoto@clear-code.com>
---
 lib/fluent/plugin/output.rb | 26 +++++++++++++-------------
 1 file changed, 13 insertions(+), 13 deletions(-)

diff --git a/lib/fluent/plugin/output.rb b/lib/fluent/plugin/output.rb
index 66077254c9..fad8c2fe9e 100644
--- a/lib/fluent/plugin/output.rb
+++ b/lib/fluent/plugin/output.rb
@@ -1222,29 +1222,29 @@ def update_retry_state(chunk_id, using_secondary, error = nil)
 
           # @retry exists
 
-          if error
-            if @retry.limit?
+          if @retry.limit?
+            if error
               records = @buffer.queued_records
               msg = "failed to flush the buffer, and hit limit for retries. dropping all chunks in the buffer queue."
               log.error msg, retry_times: @retry.steps, records: records, error: error
               log.error_backtrace error.backtrace
-            elsif using_secondary
-              msg = "failed to flush the buffer with secondary output."
-              log.warn msg, retry_time: @retry.steps, next_retry_seconds: @retry.next_time, chunk: chunk_id_hex, error: error
-              log.warn_backtrace error.backtrace
-            else
-              msg = "failed to flush the buffer."
-              log.warn msg, retry_time: @retry.steps, next_retry_seconds: @retry.next_time, chunk: chunk_id_hex, error: error
-              log.warn_backtrace error.backtrace
             end
-          end
-
-          if @retry.limit?
             @buffer.clear_queue!
             log.debug "buffer queue cleared"
             @retry = nil
           else
             @retry.step
+            if error
+              if using_secondary
+                msg = "failed to flush the buffer with secondary output."
+                log.warn msg, retry_time: @retry.steps, next_retry_seconds: @retry.next_time, chunk: chunk_id_hex, error: error
+                log.warn_backtrace error.backtrace
+              else
+                msg = "failed to flush the buffer."
+                log.warn msg, retry_time: @retry.steps, next_retry_seconds: @retry.next_time, chunk: chunk_id_hex, error: error
+                log.warn_backtrace error.backtrace
+              end
+            end
           end
         end
       end