Merge pull request #49 from chancez/0.8_handle_corrupted_logs

Gracefully handle corrupted journal log entries

.travis.yml

@@ -2,3 +2,5 @@ language: ruby
 sudo: required
 services:
   - docker
+rvm:
+  - 2.3

Rakefile

@@ -14,7 +14,7 @@ task build: "docker:test"
 task default: :reevoocop
 
 namespace :docker do
-  distros = [:ubuntu, :"tdagent-ubuntu", :"tdagent-centos"]
+  distros = %i[ubuntu tdagent-ubuntu tdagent-centos]
   task test: distros
 
   distros.each do |distro|

lib/fluent/plugin/in_systemd.rb

@@ -85,7 +85,12 @@ def watch
       while @running
         init_journal if @journal.wait(0) == :invalidate
         while @journal.move_next && @running
-          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_writer.update(@journal.cursor)
         end
         # prevent a loop of death

test/plugin/test_in_systemd.rb

@@ -12,6 +12,12 @@ def setup
       path test/fixture
     )
 
+    @badmsg_config = %(
+      tag test
+      path test/fixture/corrupt
+      read_from_head true
+    )
+
     @strip_config = base_config + %(
       strip_underscores true
     )
@@ -37,7 +43,8 @@ def setup
     )
   end
 
-  attr_reader :journal, :base_config, :pos_path, :pos_config, :head_config, :filter_config, :strip_config, :tail_config
+  attr_reader :journal, :base_config, :pos_path, :pos_config, :head_config,
+    :filter_config, :strip_config, :tail_config, :badmsg_config
 
   def create_driver(config)
     Fluent::Test::InputTestDriver.new(Fluent::SystemdInput).configure(config)
@@ -112,7 +119,6 @@ def test_reading_from_the_journal_tail_with_strip_underscores
     d.run
   end
 
-
   def test_pos_file_is_written
     d = create_driver(pos_config)
     d.run
@@ -140,7 +146,7 @@ def test_reading_from_a_pos
     assert_equal 143, d.emits.size
   end
 
-  def test_reading_from_an_invalid_pos # rubocop:disable Metrics/AbcSize
+  def test_reading_from_an_invalid_pos
     file = File.open(pos_path, "w+")
     file.print "thisisinvalid"
     file.close
@@ -180,4 +186,10 @@ def test_reading_from_the_journal_tail_explicit_setting
     d.run
   end
 
+  def test_continue_on_bad_message
+    d = create_driver(badmsg_config)
+    d.run
+    assert_equal 460, d.events.size
+  end
+
 end