Merge pull request #147 from richm/issue-146

[master] test for eventstream has empty method and if class is EventStream

Author: jcantrill

lib/fluent/plugin/filter_kubernetes_metadata.rb

@@ -299,7 +299,7 @@ def filter_stream(tag, es)
     end
 
     def filter_stream_from_files(tag, es)
-      return es if es.nil? || es.empty?
+      return es if (es.respond_to?(:empty?) && es.empty?) || !es.is_a?(Fluent::EventStream)
       new_es = Fluent::MultiEventStream.new
 
       match_data = tag.match(@tag_to_kubernetes_name_regexp_compiled)
@@ -325,7 +325,7 @@ def filter_stream_from_files(tag, es)
     end
 
     def filter_stream_from_journal(tag, es)
-      return es if es.nil? || es.empty?
+      return es if (es.respond_to?(:empty?) && es.empty?) || !es.is_a?(Fluent::EventStream)
       new_es = Fluent::MultiEventStream.new
       batch_miss_cache = {}
       es.each do |time, record|

test/plugin/test_filter_kubernetes_metadata.rb

@@ -160,20 +160,40 @@ def emit_with_tag(tag, msg={}, config='
 
     test 'nil event stream from journal' do
      #not certain how this is possible but adding test to properly
-     #guard against this condition we have seen
-
+     #guard against this condition we have seen - test for nil,
+     #empty, no empty method, not an event stream
      plugin = create_driver.instance
-     plugin.filter_stream_from_journal('tag', nil)
-     plugin.filter_stream_from_journal('tag', Fluent::MultiEventStream.new)
+     [nil, Fluent::MultiEventStream.new, 1, [1]].each do |es|
+       assert_equal es, plugin.filter_stream_from_journal('tag', es)
+     end
+     # and make sure OneEventStream works
+     ts = Time.now()
+     rec = {"message"=>"hello"}
+     es = Fluent::OneEventStream.new(ts, rec)
+     newes = plugin.filter_stream_from_journal('tag', es)
+     newes.each do |newts, newrec|
+      assert_equal ts, newts
+      assert_equal rec, newrec
+     end
     end
 
     test 'nil event stream from files' do
      #not certain how this is possible but adding test to properly
      #guard against this condition we have seen
 
      plugin = create_driver.instance
-     plugin.filter_stream_from_files('tag', nil)
-     plugin.filter_stream_from_files('tag', Fluent::MultiEventStream.new)
+     [nil, Fluent::MultiEventStream.new, 1, [1]].each do |es|
+       assert_equal es, plugin.filter_stream_from_files('tag', es)
+     end
+     # and make sure OneEventStream works
+     ts = Time.now()
+     rec = {"message"=>"hello"}
+     es = Fluent::OneEventStream.new(ts, rec)
+     newes = plugin.filter_stream_from_journal('tag', es)
+     newes.each do |newts, newrec|
+      assert_equal ts, newts
+      assert_equal rec, newrec
+     end
     end
 
     test 'inability to connect to the api server handles exception and doensnt block pipeline' do