Merge pull request #9 from HatsuneMiku3939/feature/proxy-auth-support

add proxy, basic_auth support

Author: HatsuneMiku3939

README.md

@@ -139,6 +139,18 @@ If status_only is true, body is not parsed.
 
 The timeout of each request.
 
+### proxy (string) (optional, default: nil)
+
+The HTTP proxy URL to use for each requests
+
+### user (string) (optional, default: nil)
+
+The user for basic auth
+
+### password (string) (optional, default: nil)
+
+The password for basic auth
+
 ## In case of remote error
 
 ### Can receive response from remote

Rakefile

@@ -10,6 +10,11 @@ Rake::TestTask.new(:test) do |t|
     FileUtils.rm "stub_server.log"
   end
 
+  if File.exists? "stub_proxy.log"
+    puts "clear stub_proxy.log"
+    FileUtils.rm "stub_proxy.log"
+  end
+
   t.libs.push("lib", "test")
   t.test_files = FileList["test/**/test_*.rb"]
   t.verbose = true

lib/fluent/plugin/in_http_pull.rb

@@ -35,8 +35,15 @@ def initialize
       config_param :interval, :time
       desc 'status_only'
       config_param :status_only, :bool, default: false
-      desc 'timeout second of each request'
+      desc 'The timeout stime of each request'
       config_param :timeout, :time, default: 10
+      desc 'The HTTP proxy URL to use for each requests'
+      config_param :proxy, :string, default: nil
+
+      desc 'user of basic auth'
+      config_param :user, :string, default: nil
+      desc 'password of basic auth'
+      config_param :password, :string, default: nil
 
       def configure(conf)
         compat_parameters_convert(conf, :parser)
@@ -55,9 +62,13 @@ def on_timer
         record = { "url" => @url }
 
         begin
-          res = RestClient::Request.execute(method: :get,
-                                            url: @url,
-                                            timeout: @timeout)
+          request_options = { method: :get, url: @url, timeout: @timeout }
+
+          request_options[:proxy] = @proxy if @proxy
+          request_options[:user] = @user if @user
+          request_options[:password] = @password if @password
+
+          res = RestClient::Request.execute request_options
           record["status"] = res.code
           record["body"] = res.body
         rescue StandardError => err

test/helper.rb

@@ -23,6 +23,7 @@
 
 # require stub_server
 require "test/helper/stub_server"
+require "test/helper/stub_proxy"
 
 Test::Unit::TestCase.include(Fluent::Test::Helpers)
 Test::Unit::TestCase.extend(Fluent::Test::Helpers)

test/helper/stub_proxy.rb

@@ -0,0 +1,39 @@
+require 'webrick'
+require 'webrick/httpproxy'
+
+class StubProxy
+  def initialize
+    create_proxy
+  end
+
+  def start
+    @thread = Thread.new { @proxy.start }
+  end
+
+  def shutdown
+    @proxy.shutdown
+
+    # wait until webrick was shutting down
+    while true
+      break if @thread.status == false
+
+      # issue webrick shutdown once more
+      @proxy.shutdown
+      sleep 1
+    end
+
+    # then exit thread
+    @thread.exit
+  end
+
+  private
+  def create_proxy
+    @log_file = File.open("stub_proxy.log", "a+")
+    @log = WEBrick::Log.new @log_file
+    @access_log = [
+      [@log_file, WEBrick::AccessLog::COMBINED_LOG_FORMAT],
+    ]
+
+    @proxy = WEBrick::HTTPProxyServer.new :Port => 4040, :Logger => @log, :AccessLog => @access_log
+  end
+end

test/helper/stub_server.rb

@@ -10,6 +10,7 @@ def initialize
     @server.mount_proc '/timeout', &method(:timeout)
     @server.mount_proc '/internal_error', &method(:internal_error)
     @server.mount_proc '/redirect', &method(:redirect)
+    @server.mount_proc '/protected', &method(:protected)
   end
 
   def start
@@ -70,4 +71,14 @@ def internal_error(req, res)
   def redirect(req, res)
     res.set_redirect WEBrick::HTTPStatus::TemporaryRedirect, "http://127.0.0.1:3939/"
   end
+
+  def protected(req, res)
+    WEBrick::HTTPAuth.basic_auth(req, res, 'protected') do |user, password|
+      user == 'HatsuneMiku' && password == '3939'
+    end
+
+    res.status = 200
+    res['Content-Type'] = 'application/json'
+    res.body = '{ "status": "OK" }'
+  end
 end

test/plugin/test_in_http_pull.rb

@@ -7,8 +7,6 @@ class HttpPullInputTest < Test::Unit::TestCase
   @stub_server = nil
 
   setup do
-    Fluent::Test.setup
-
     @stub_server = StubServer.new
     @stub_server.start
   end
@@ -39,6 +37,27 @@ class HttpPullInputTest < Test::Unit::TestCase
 
       assert_equal(10, d.instance.timeout)
     end
+
+    test 'user' do
+      d = create_driver TEST_DEFAULT_VALUE_CONFIG
+      assert_equal("test", d.instance.tag)
+
+      assert_equal(nil, d.instance.user)
+    end
+
+    test 'password' do
+      d = create_driver TEST_DEFAULT_VALUE_CONFIG
+      assert_equal("test", d.instance.tag)
+
+      assert_equal(nil, d.instance.password)
+    end
+
+    test 'proxy' do
+      d = create_driver TEST_DEFAULT_VALUE_CONFIG
+      assert_equal("test", d.instance.tag)
+
+      assert_equal(nil, d.instance.proxy)
+    end
   end
 
   sub_test_case "success case" do
@@ -245,6 +264,195 @@ class HttpPullInputTest < Test::Unit::TestCase
     end
   end
 
+  sub_test_case "remote is prtected by basic auth" do
+    TEST_AUTH_SUCCESS_CONFIG = %[
+      tag test
+      url http://127.0.0.1:3939/protected
+      timeout 2s
+      user HatsuneMiku
+      password 3939
+
+      interval 3s
+      format json
+    ]
+
+    TEST_AUTH_FAIL_CONFIG = %[
+      tag test
+      url http://127.0.0.1:3939/protected
+      timeout 2s
+      user HatsuneMiku
+      password wrong_password
+
+      interval 3s
+      format json
+    ]
+
+    TEST_AUTH_FAIL_NOT_GIVEN_CONFIG = %[
+      tag test
+      url http://127.0.0.1:3939/protected
+      timeout 2s
+
+      interval 3s
+      format json
+    ]
+
+    test 'interval 3 with corrent password' do
+      d = create_driver TEST_AUTH_SUCCESS_CONFIG
+      assert_equal("test", d.instance.tag)
+      assert_equal(3, d.instance.interval)
+
+      d.run(timeout: 8) do
+        sleep 7
+      end
+      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:3939/protected","status"=>200, "message"=>{"status"=>"OK"}}, record)
+        assert(time.is_a?(Fluent::EventTime))
+      end
+    end
+
+    test 'interval 3 with wrong password' do
+      d = create_driver TEST_AUTH_FAIL_CONFIG
+      assert_equal("test", d.instance.tag)
+      assert_equal(3, d.instance.interval)
+
+      d.run(timeout: 8) 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:3939/protected", record["url"])
+        assert(time.is_a?(Fluent::EventTime))
+
+        assert_equal(401, record["status"])
+        assert_not_nil(record["error"])
+      end
+    end
+
+    test 'interval 3 without auth info' do
+      d = create_driver TEST_AUTH_FAIL_CONFIG
+      assert_equal("test", d.instance.tag)
+      assert_equal(3, d.instance.interval)
+
+      d.run(timeout: 8) 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:3939/protected", record["url"])
+        assert(time.is_a?(Fluent::EventTime))
+
+        assert_equal(401, record["status"])
+        assert_not_nil(record["error"])
+      end
+    end
+  end
+
+  sub_test_case "success case behind proxy" do
+    TEST_INTERVAL_3_PROXY_CONFIG = %[
+      tag test
+      url http://127.0.0.1:3939
+      proxy http://127.0.0.1:4040
+
+      interval 3s
+      format none
+      status_only true
+    ]
+
+    TEST_INTERVAL_3_REDIRECT_PROXY_CONFIG = %[
+      tag test
+      url http://127.0.0.1:3939/redirect
+      proxy http://127.0.0.1:4040
+
+      interval 3s
+      format json
+    ]
+
+    TEST_AUTH_SUCCESS_PROXY_CONFIG = %[
+      tag test
+      url http://127.0.0.1:3939/protected
+      proxy http://127.0.0.1:4040
+      timeout 2s
+      user HatsuneMiku
+      password 3939
+
+      interval 3s
+      format json
+    ]
+
+    setup do
+      @proxy_server = StubProxy.new
+      @proxy_server.start
+    end
+
+    teardown do
+      @proxy_server.shutdown
+    end
+
+    test 'interval 3 with status_only' do
+      d = create_driver TEST_INTERVAL_3_PROXY_CONFIG
+      assert_equal("test", d.instance.tag)
+      assert_equal(3, d.instance.interval)
+
+      d.run(timeout: 8) do
+        sleep 7
+      end
+      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:3939","status"=>200}, record)
+        assert(time.is_a?(Fluent::EventTime))
+      end
+    end
+
+    test 'interval 3 with redirect' do
+      d = create_driver TEST_INTERVAL_3_REDIRECT_PROXY_CONFIG
+      assert_equal("test", d.instance.tag)
+      assert_equal(3, d.instance.interval)
+
+      d.run(timeout: 8) do
+        sleep 7
+      end
+      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:3939/redirect","status"=>200, "message"=>{"status"=>"OK"}}, record)
+        assert(time.is_a?(Fluent::EventTime))
+      end
+    end
+
+    test 'interval 3 with corrent password' do
+      d = create_driver TEST_AUTH_SUCCESS_PROXY_CONFIG
+      assert_equal("test", d.instance.tag)
+      assert_equal(3, d.instance.interval)
+
+      d.run(timeout: 8) do
+        sleep 7
+      end
+      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:3939/protected","status"=>200, "message"=>{"status"=>"OK"}}, record)
+        assert(time.is_a?(Fluent::EventTime))
+      end
+    end
+  end
+
   private
 
   def create_driver(conf)