Prevent garbage from coming out of the LTSV parser

Signed-off-by: Aaron Smith <[email protected]>

lib/fluent/plugin/parser_ltsv.rb

@@ -38,8 +38,10 @@ def configure(conf)
       def parse(text)
         r = {}
         text.split(@delimiter).each do |pair|
-          key, value = pair.split(@label_delimiter, 2)
-          r[key] = value
+          if pair.include? @label_delimiter
+            key, value = pair.split(@label_delimiter, 2)
+            r[key] = value
+          end
         end
         time, record = convert_values(parse_time(r), r)
         yield time, record

test/plugin/test_parser_labeled_tsv.rb

@@ -100,6 +100,21 @@ def test_parse_with_keep_time_key
     end
   end
 
+  def test_parse_and_reject_invalid_kv_pairs
+    parser = Fluent::Test::Driver::Parser.new(Fluent::Plugin::LabeledTSVParser)
+    parser.configure(
+                     'delimiter'       => ' ',
+                     'label_delimiter' => '=',
+                     )
+    text = 'A leading portion that is not LTSV    :  foo=bar baz=derp and a trailing portion'
+
+    expected = {'foo' => 'bar', 'baz' => 'derp'}
+    parser.instance.parse(text) do |time, record|
+      assert_equal expected, record
+    end
+
+  end
+
   def test_parse_with_null_value_pattern
     parser = Fluent::Test::Driver::Parser.new(Fluent::Plugin::LabeledTSVParser)
     parser.configure(