Add input plugin

Author: Watson1978

README.md

@@ -30,27 +30,49 @@ $ bundle
 
 ## Configuration
 
-### plugins_json (string) (optional)
+### Filter plugin
+
+#### plugins_json (string) (optional)
 
 Path to `plugins.json`.
 
 Default value: `https://raw.githubusercontent.com/fluent/fluentd-website/master/scripts/plugins.json`.
 
-
-### Deprecated: obsolete_plugins_yml (string) (optional)
+#### Deprecated: obsolete_plugins_yml (string) (optional)
 
 Path to `obsolete-plugins.yml`. This parameter is deprecated. Please use `plugins_json` parameter instead.
 
 Default value: `nil`
 
-### timeout (integer) (optional)
+#### timeout (integer) (optional)
+
+Timeout to read data of obsolete plugins.
+If it occurs timeout, it just skips to detect obsolete plugins.
+
+Default value: `5`
+
+#### raise_error (bool) (optional)
+
+Raise error if obsolete plugins are detected
+
+Default value: `false`.
+
+### Input plugin
+
+#### plugins_json (string) (optional)
+
+Path to `plugins.json`.
+
+Default value: `https://raw.githubusercontent.com/fluent/fluentd-website/master/scripts/plugins.json`.
+
+#### timeout (integer) (optional)
 
 Timeout to read data of obsolete plugins.
 If it occurs timeout, it just skips to detect obsolete plugins.
 
 Default value: `5`
 
-### raise_error (bool) (optional)
+#### raise_error (bool) (optional)
 
 Raise error if obsolete plugins are detected
 

lib/fluent/plugin/in_obsolete_plugins.rb

@@ -0,0 +1,30 @@
+require "fluent/plugin/input"
+require "fluent/plugin/obsolete_plugins_utils"
+
+module Fluent
+  module Plugin
+    class ObsoletePluginsInput < Fluent::Plugin::Input
+      Fluent::Plugin.register_input("obsolete_plugins", self)
+
+      PLUGINS_JSON_URL = "https://raw.githubusercontent.com/fluent/fluentd-website/master/scripts/plugins.json"
+
+      desc "Path to plugins.json"
+      config_param :plugins_json, :string, default: PLUGINS_JSON_URL
+      desc "Timeout value to read data of obsolete plugins"
+      config_param :timeout, :integer, default: 5
+      desc "Raise error if obsolete plugins are detected"
+      config_param :raise_error, :bool, default: false
+
+      def configure(conf)
+        super
+
+        obsolete_plugins = ObsoletePluginsUtils.obsolete_plugins_from_json(@plugins_json, timeout: @timeout)
+        ObsoletePluginsUtils.notify(log, obsolete_plugins, raise_error: @raise_error)
+      rescue Fluent::ConfigError
+        raise
+      rescue => e
+        log.info("Failed to notify obsolete plugins", error: e)
+      end
+    end
+  end
+end

test/helper.rb

@@ -3,7 +3,6 @@
 require "test/unit/rr"
 require "timecop"
 require "fluent/test"
-require "fluent/test/driver/filter"
 require "fluent/test/helpers"
 
 Test::Unit::TestCase.include(Fluent::Test::Helpers)

test/plugin/test_filter_obsolete_plugins.rb

@@ -1,4 +1,5 @@
 require "helper"
+require "fluent/test/driver/filter"
 require "fluent/plugin/filter_obsolete_plugins"
 
 class ObsoletePluginsFilterTest < Test::Unit::TestCase

test/plugin/test_in_obsolete_plugins.rb

@@ -0,0 +1,116 @@
+require "helper"
+require "fluent/test/driver/input"
+require "fluent/plugin/in_obsolete_plugins"
+require "fluent/plugin/obsolete_plugins_utils"
+
+class ObsoletePluginsInputTest < Test::Unit::TestCase
+
+  setup do
+    Fluent::Test.setup
+    $log = Fluent::Test::TestLogger.new
+    @time = Time.now
+    Timecop.freeze(@time)
+  end
+
+  teardown do
+    Timecop.return
+  end
+
+  sub_test_case "plugins_json" do
+    CONFIG_JSON = %[
+      plugins_json #{fixture_path("plugins.json")}
+    ]
+
+    test "no obsolete plugins" do
+      d = create_driver(CONFIG_JSON)
+      d.run
+      assert_equal([], d.events)
+      assert_equal([], d.logs)
+    end
+
+    test "obsolete plugins" do
+      stub(Gem).loaded_specs do
+        {
+          "fluent-plugin-tail-multiline" => nil,
+          "fluent-plugin-hostname" => nil
+        }
+      end
+      d = create_driver(CONFIG_JSON)
+      d.run
+      assert_equal([], d.events)
+      expected_logs = [
+        "#{@time} [warn]: fluent-plugin-tail-multiline is obsolete: Merged in in_tail in Fluentd v0.10.45. [fluent/fluentd#269](https://github.com/fluent/fluentd/issues/269)\n",
+        "#{@time} [warn]: fluent-plugin-hostname is obsolete: Use [filter\\_record\\_transformer](http://docs.fluentd.org/v0.12/articles/filter_record_transformer) instead.\n"
+      ]
+      assert_equal(expected_logs, d.logs)
+    end
+
+    test "raise error when detect obsolete plugins" do
+      stub(Gem).loaded_specs do
+        {
+          "fluent-plugin-tail-multiline" => nil,
+          "fluent-plugin-hostname" => nil
+        }
+      end
+
+      ex = assert_raise(Fluent::ConfigError) do
+        create_driver(CONFIG_JSON + "raise_error yes")
+      end
+      assert_equal("Detected obsolete plugins", ex.message)
+    end
+  end
+
+  sub_test_case "error handling" do
+    test "invalid json" do
+      d = create_driver("plugins_json #{fixture_path('invalid.json')}")
+
+      expected_logs = [
+        "#{@time} [info]: Failed to notify obsolete plugins error_class=JSON::ParserError error=\"expected ',' or '}' after object value, got: EOF at line 11 column 1\"\n",
+      ]
+
+      assert_equal(expected_logs, d.logs)
+    end
+
+    test "timeout with slow server" do
+      server = create_slow_webserver(port: 12345)
+
+      mock(Fluent::Plugin::ObsoletePluginsUtils).notify.never
+
+      d = create_driver(%[
+        plugins_json http://localhost:12345/plugins.json
+        timeout 1
+      ])
+
+      sleep 2
+
+      expected_logs = [
+        "#{@time} [info]: Failed to notify obsolete plugins error_class=Timeout::Error error=\"execution expired\"\n",
+      ]
+
+      assert_equal(expected_logs, d.logs)
+    ensure
+      server.shutdown
+    end
+
+  end
+
+  private
+
+  def create_driver(conf)
+    Fluent::Test::Driver::Input.new(Fluent::Plugin::ObsoletePluginsInput).configure(conf)
+  end
+
+  def create_slow_webserver(port: 12345)
+    require "webrick"
+
+    server = WEBrick::HTTPServer.new(Port: port)
+    server.mount_proc '/' do |req, res|
+      sleep 60
+
+      res['Content-Type'] = 'application/json'
+      res.body = File.read(fixture_path("plugins.json"))
+    end
+
+    server
+  end
+end