Merge pull request #41 from jtwile2/continue-on-bad-msg

Continue Reading on Bad Messages

Author: errm

lib/fluent/plugin/in_systemd.rb

@@ -128,7 +128,12 @@ def formatted(entry)
 
       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

test/fixture/corrupt/test.badmsg.journal

@@ -58,6 +58,12 @@ def setup
       path test/fixture
     )
 
+    @badmsg_config = %(
+      tag test
+      path test/fixture/corrupt
+      read_from_head true
+    )
+
     # deprecated
     @strip_config = base_config + %(
       strip_underscores true
@@ -92,7 +98,8 @@ def setup
   end
 
   attr_reader :journal, :base_config, :pos_path, :pos_config, :head_config,
-    :filter_config, :strip_config, :tail_config, :not_present_config, :storage_path
+    :filter_config, :strip_config, :tail_config, :not_present_config,
+    :badmsg_config, :storage_path
 
   def create_driver(config)
     Fluent::Test::Driver::Input.new(Fluent::Plugin::SystemdInput).configure(config)
@@ -250,4 +257,11 @@ def test_journal_not_present
     d.run(timeout: 5)
     assert_match "Systemd::JournalError: No such file or directory retrying in 1s", d.logs.last
   end
+
+  def test_continue_on_bad_message
+    d = create_driver(badmsg_config)
+    d.run(expect_emits: 460)
+    assert_equal 460, d.events.size
+    assert_equal 0, d.error_events.size
+  end
 end