Merge pull request #4 from HatsuneMiku3939/fetaure/add-remote-timeout

add initial implementation of remote timeout

Author: HatsuneMiku3939

README.md

@@ -39,7 +39,7 @@ $ bundle
 
   tag test
   url http://www.google.com
-  interval 1
+  interval 1s
 
   status_only true
 </source>
@@ -67,7 +67,7 @@ $ bundle
 
   tag fluentd.status
   url http://localhost:24220/api/plugins.json
-  interval 1
+  interval 1s
 </source>
 
 <match fluentd.status>
@@ -88,7 +88,7 @@ $ bundle
 
   tag es.cluster.health
   url http://localhost:9200/_cluster/health
-  interval 1
+  interval 1s
 </source>
 
 <match es.cluster.health>
@@ -111,14 +111,18 @@ The tag of the event.
 
 The url of remote server.
 
-### interval (integer) (required)
+### interval (time) (required)
 
-The second interval time between periodic request.
+The interval time between periodic request.
 
 ### status_only (bool) (optional, default: false)
 
 If atatus_only is true, body is not parsed.
 
+### timeout (integer) (optional, default: 10)
+
+Timeout second of each request.
+
 ## In case of remote error
 
 ### Can receive response from remote

lib/fluent/plugin/in_http_pull.rb

@@ -35,6 +35,8 @@ def initialize
       config_param :interval, :time
       desc 'status_only'
       config_param :status_only, :bool, default: false
+      desc 'timeout second of each request'
+      config_param :timeout, :integer, default: 10
 
       def configure(conf)
         super
@@ -50,14 +52,18 @@ def on_timer
         log = { "url" => @url }
 
         begin
-          res = RestClient.get(@url)
+          res = RestClient::Request.execute(method: :get,
+                                            url: @url,
+                                            timeout: @timeout)
           log["status"] = res.code
           log["body"] = res.body
-        rescue RestClient::ExceptionWithResponse => err
-          log["status"] = err.code
-          log["error"] = err.message
-        rescue Exception => err
-          log["status"] = 0
+        rescue StandardError => err
+          if err.respond_to? :code
+            log["status"] = err.code
+          else
+            log["status"] = 0
+          end
+
           log["error"] = err.message
         end
 

test/plugin/test_in_http_pull.rb

@@ -4,59 +4,91 @@
 require 'ostruct'
 
 class HttpPullInputTest < Test::Unit::TestCase
-  TEST_INTERVAL_3_CONFIG = %[
-    tag test
-    url http://127.0.0.1
+  setup do
+    Fluent::Test.setup
+  end
+
+  sub_test_case "default value of each options" do
+    TEST_DEFAULT_VALUE_CONFIG = %[
+      tag test
+      url http://127.0.0.1
+      interval 3s
+    ]
 
-    interval 3
-    status_only true
-  ]
+    test 'status_only' do
+      d = create_driver TEST_DEFAULT_VALUE_CONFIG
+      assert_equal("test", d.instance.tag)
 
-  TEST_INTERVAL_5_CONFIG = %[
-    tag test
-    url http://127.0.0.1
+      assert_equal(false, d.instance.status_only)
+    end
 
-    interval 5
-  ]
+    test 'timeout' do
+      d = create_driver TEST_DEFAULT_VALUE_CONFIG
+      assert_equal("test", d.instance.tag)
 
-  setup do
-    Fluent::Test.setup
+      assert_equal(10, d.instance.timeout)
+    end
   end
 
   sub_test_case "success case with status only" do
+    TEST_INTERVAL_3_CONFIG = %[
+      tag test
+      url http://127.0.0.1
+      timeout 10
+
+      interval 3s
+      status_only true
+    ]
+
+    TEST_INTERVAL_5_CONFIG = %[
+      tag test
+      url http://127.0.0.1
+      timeout 10
+
+      interval 5s
+    ]
+
     setup do
-      mock(RestClient).get("http://127.0.0.1").times(2) do
-        OpenStruct.new({code: 200, body: '{"status": "OK"}'})
-      end
+      mock(RestClient::Request).
+        execute(method: :get,
+                url: "http://127.0.0.1",
+                timeout: 10).
+        times(2) do
+          OpenStruct.new({code: 200, body: '{"status": "OK"}'})
+        end
     end
 
     test 'interval 3 with status_only' do
       d = create_driver TEST_INTERVAL_3_CONFIG
-      assert_equal "test", d.instance.tag
+      assert_equal("test", d.instance.tag)
+      assert_equal(3, d.instance.interval)
 
       d.run(timeout: 5) do
         sleep 7
       end
-      assert_equal 2, d.events.size
+      assert_equal(2, d.events.size)
 
       d.events.each do |tag, time, record|
         assert_equal("test", tag)
+
         assert_equal({"url"=>"http://127.0.0.1","status"=>200}, record)
         assert(time.is_a?(Fluent::EventTime))
       end
     end
 
     test 'interval 5' do
       d = create_driver TEST_INTERVAL_5_CONFIG
-      assert_equal "test", d.instance.tag
+      assert_equal("test", d.instance.tag)
+      assert_equal(5, d.instance.interval)
 
       d.run(timeout: 7) do
         sleep 11
       end
-      assert_equal 2, d.events.size
+      assert_equal(2, d.events.size)
 
       d.events.each do |tag, time, record|
         assert_equal("test", tag)
+
         assert_equal({"url"=>"http://127.0.0.1","status"=>200, "message"=>{"status"=>"OK"}}, record)
         assert(time.is_a?(Fluent::EventTime))
       end
@@ -71,14 +103,62 @@ class HttpPullInputTest < Test::Unit::TestCase
     ]
     test "connection refused by remote" do
       d = create_driver TEST_REFUSED_CONFIG
-      d.run(timeout: 2) { sleep 3 }
+      assert_equal("test", d.instance.tag)
 
-      assert_equal 3, d.events.size
+      d.run(timeout: 2) do
+        sleep 3
+      end
+
+      assert_equal(3, d.events.size)
       d.events.each do |tag, time, record|
         assert_equal("test", tag)
+
         assert_equal("http://127.0.0.1:5927", record["url"])
+        assert(time.is_a?(Fluent::EventTime))
+
         assert_equal(0, record["status"])
+        assert_not_nil(record["error"])
+      end
+    end
+  end
+
+  sub_test_case "fail when remote timeout" do
+    TEST_TIMEOUT_FAIL_CONFIG = %[
+      tag test
+      url http://127.0.0.1
+      timeout 2
+
+      interval 3
+    ]
+
+    setup do
+      mock(RestClient::Request).
+        execute(method: :get,
+                url: "http://127.0.0.1",
+                timeout: 2).
+        times(2) do
+          sleep 2
+          raise RestClient::Exceptions::Timeout.new
+        end
+    end
+
+    test "timeout" do
+      d = create_driver TEST_TIMEOUT_FAIL_CONFIG
+      assert_equal("test", d.instance.tag)
+
+      d.run(timeout: 5) do
+        sleep 7
+      end
+      assert_equal(2, d.events.size)
+
+      d.events.each do |tag, time, record|
+        assert_equal("test", tag)
+
+        assert_equal("http://127.0.0.1", record["url"])
         assert(time.is_a?(Fluent::EventTime))
+
+        assert_equal(0, record["status"])
+        assert_not_nil(record["error"])
       end
     end
   end