in_systemd.rb in fluent-plugin-systemd-0.3.1

- old
+ new

@@ -4,11 +4,11 @@
 require "fluent/plugin/systemd/pos_writer"
 require "fluent/plugin/systemd/entry_mutator"
 
 module Fluent
   module Plugin
-    class SystemdInput < Input
+    class SystemdInput < Input # rubocop:disable Metrics/ClassLength
       Fluent::Plugin.register_input("systemd", self)
 
       helpers :timer, :storage
 
       DEFAULT_STORAGE_TYPE = "local"
@@ -59,10 +59,12 @@
       end
 
       private
 
       def init_journal
+        # TODO: ruby 2.3
+        @journal.close if @journal # rubocop:disable Style/SafeNavigation
         @journal = Systemd::Journal.new(path: @path)
         # make sure initial call to wait doesn't return :invalidate
         # see https://github.com/ledbettj/systemd-journal/issues/70
         @journal.wait(0)
         @journal.filter(*@filters)
@@ -126,10 +128,15 @@
         @mutator.run(entry)
       end
 
       def watch
         while @journal.move_next
-          yield @journal.current_entry
+          begin
+            yield @journal.current_entry
+          rescue Systemd::JournalError => e
+            log.warn("Error Parsing Journal: #{e.class}: #{e.message}")
+            next
+          end
           @pos_storage.put(:journal, @journal.cursor)
         end
       end
     end
   end