Add timeout parameter and error handling

Author: Watson1978

README.md

@@ -43,6 +43,13 @@ Path to `obsolete-plugins.yml`. This parameter is deprecated. Please use `plugin
 
 Default value: `nil`
 
+### 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

lib/fluent/plugin/filter_obsolete_plugins.rb

@@ -27,6 +27,8 @@ class ObsoletePluginsFilter < Fluent::Plugin::Filter
       config_param :obsolete_plugins_yml, :string, default: nil, deprecated: "use plugins_json parameter instead"
       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
 
@@ -35,12 +37,16 @@ def configure(conf)
 
         obsolete_plugins =
           if @obsolete_plugins_yml
-            ObsoletePluginsUtils.obsolete_plugins_from_yaml(@obsolete_plugins_yml)
+            ObsoletePluginsUtils.obsolete_plugins_from_yaml(@obsolete_plugins_yml, timeout: @timeout)
           else
-            ObsoletePluginsUtils.obsolete_plugins_from_json(@plugins_json)
+            ObsoletePluginsUtils.obsolete_plugins_from_json(@plugins_json, timeout: @timeout)
           end
 
         ObsoletePluginsUtils.notify(log, obsolete_plugins, raise_error: @raise_error)
+      rescue Fluent::ConfigError
+        raise
+      rescue
+        # ignore other exception
       end
 
       def filter(tag, time, record)

lib/fluent/plugin/obsolete_plugins_utils.rb

@@ -2,20 +2,25 @@
 require "open-uri"
 require "yaml"
 require "json"
+require "timeout"
 
 class Fluent::Plugin::ObsoletePluginsUtils
-  def self.obsolete_plugins_from_yaml(url)
-    URI.open(url) do |io|
-      YAML.safe_load(io.read)
+  def self.obsolete_plugins_from_yaml(url, timeout: 5)
+    Timeout.timeout(timeout) do
+      URI.open(url) do |io|
+        YAML.safe_load(io.read)
+      end
     end
   end
 
-  def self.obsolete_plugins_from_json(url)
-    plugins = URI.open(url) do |io|
-      # io.read causes Encoding::UndefinedConversionError with UTF-8 data when Ruby is started with "-Eascii-8bit:ascii-8bit".
-      # It set the proper encoding to avoid the error.
-      io.set_encoding("UTF-8", "UTF-8")
-      JSON.parse(io.read)
+  def self.obsolete_plugins_from_json(url, timeout: 5)
+    plugins = Timeout.timeout(timeout) do
+      URI.open(url) do |io|
+        # io.read causes Encoding::UndefinedConversionError with UTF-8 data when Ruby is started with "-Eascii-8bit:ascii-8bit".
+        # It set the proper encoding to avoid the error.
+        io.set_encoding("UTF-8", "UTF-8")
+        JSON.parse(io.read)
+      end
     end
     plugins.select { |plugin| plugin["obsolete"] }.reduce({}) do |result, plugin|
       result[plugin["name"]] = plugin["note"]

test/fixtures/invalid.json

@@ -0,0 +1,11 @@
+[
+  {
+    "obsolete": null,
+    "note": null,
+    "name": "fluent-plugin-s3",
+    "info": "Amazon S3 output plugin for Fluentd event collector",
+    "authors": "Sadayuki Furuhashi, Masahiro Nakagawa",
+    "version": "1.8.3",
+    "downloads": 134522742,
+    "homepage_uri": "https://github.com/fluent/fluent-plugin-s3",
+    "source_code_uri": null

test/plugin/test_filter_obsolete_plugins.rb

@@ -1,5 +1,5 @@
 require "helper"
-require "fluent/plugin/filter_obsolete_plugins.rb"
+require "fluent/plugin/filter_obsolete_plugins"
 
 class ObsoletePluginsFilterTest < Test::Unit::TestCase
 
@@ -114,9 +114,54 @@ class ObsoletePluginsFilterTest < Test::Unit::TestCase
     end
   end
 
+  sub_test_case "error handling" do
+    test "ignore error with invalid json" do
+      assert_nothing_raised {
+        create_driver("plugins_json #{fixture_path('invalid.json')}")
+      }
+    end
+
+    test "timeout with slow server and skip detecting obsolete plugins" 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
+      ])
+
+      d.run(default_tag: "test") do
+        d.feed({ message: "This is test message." })
+      end
+      assert_equal([{ message: "This is test message." }], d.filtered_records)
+
+      sleep 2
+
+      assert_equal([], d.logs)
+    ensure
+      server.shutdown
+    end
+
+  end
+
   private
 
   def create_driver(conf)
     Fluent::Test::Driver::Filter.new(Fluent::Plugin::ObsoletePluginsFilter).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