Merge branch 'dev-fix-restart-on-parse-error'

Author: eladamitpxi

lib/logstash/outputs/cassandra/event_parser.rb

@@ -17,18 +17,22 @@ def initialize(options)
 
     def parse(event)
       action = {}
-      action['table'] = event.sprintf(@table)
-      filter_transform = get_filter_transform(event)
-      if filter_transform
-        action['data'] = {}
-        filter_transform.each { |filter|
-          add_event_value_from_filter_to_action(event, filter, action)
-        }
-      else
-        add_event_data_using_configured_hints(event, action)
+      begin
+        action['table'] = event.sprintf(@table)
+        filter_transform = get_filter_transform(event)
+        if filter_transform
+          action['data'] = {}
+          filter_transform.each { |filter|
+            add_event_value_from_filter_to_action(event, filter, action)
+          }
+        else
+          add_event_data_using_configured_hints(event, action)
+        end
+        @logger.debug('event parsed to action', :action => action)
+      rescue Exception => e
+        @logger.error('failed parsing event', :event => event, :error => e)
+        action = nil
       end
-
-      @logger.debug('event parsed to action', :action => action)
       action
     end
 

lib/logstash/outputs/cassandra/safe_submitter.rb

@@ -57,8 +57,10 @@ def prepare_queries(actions)
       remaining_queries = Queue.new
       actions.each do |action|
         begin
-          query = get_query(action)
-          remaining_queries << { :query => query, :arguments => action['data'].values }
+          if action
+            query = get_query(action)
+            remaining_queries << { :query => query, :arguments => action['data'].values }
+          end
         rescue Exception => e
           @logger.error('Failed to prepare query', :action => action, :exception => e, :backtrace => e.backtrace)
         end

logstash-output-cassandra.gemspec

@@ -1,7 +1,7 @@
 Gem::Specification.new do |s|
 
   s.name            = 'logstash-output-cassandra'
-  s.version         = '0.9.0'
+  s.version         = '0.9.1'
   s.licenses        = [ 'Apache License (2.0)' ]
   s.summary         = 'Store events into Cassandra'
   s.description     = 'This gem is a logstash plugin required to be installed on top of the Logstash core pipeline using $LS_HOME/bin/plugin install gemname. This gem is not a stand-alone program'

spec/unit/outputs/event_parser_spec.rb

@@ -170,9 +170,10 @@
           options = default_opts.update({ 'filter_transform' => [{ 'event_key' => 'a_field', 'column_name' => 'a_column', 'cassandra_type' => 'what?!' }] })
           sut_instance = sut.new(options)
           sample_event['a_field'] = 'a_value'
-          expect(options['logger']).to(receive(:error))
+          expect(options['logger']).to(receive(:error)).at_least(:once)
 
-          expect { sut_instance.parse(sample_event) }.to raise_error(/Cannot convert/)
+          result = sut_instance.parse(sample_event)
+          expect(result).to be_nil
         end
       end
     end
@@ -238,21 +239,23 @@
     it 'fails for unknown hint types' do
       options = default_opts.update({ 'hints' => { 'a_field' => 'not_a_real_type' } })
       sut_instance = sut.new(options)
-      expect(options['logger']).to(receive(:error))
-
+      expect(options['logger']).to(receive(:error)).at_least(:once)
       sample_event['a_field'] = 'a value'
 
-      expect { sut_instance.parse(sample_event) }.to raise_error(/Cannot convert/)
+      result = sut_instance.parse(sample_event)
+
+      expect(result).to be_nil
     end
 
     it 'fails for unsuccessful hint conversion' do
       options = default_opts.update({ 'hints' => { 'a_field' => 'int' } })
-      expect(options['logger']).to(receive(:error))
-
+      expect(options['logger']).to(receive(:error)).at_least(:once)
       sut_instance = sut.new(options)
-
       sample_event['a_field'] = 'i am not an int!!!'
-      expect { sut_instance.parse(sample_event) }.to raise_error(/Cannot convert/)
+
+      result = sut_instance.parse(sample_event)
+
+      expect(result).to be_nil
     end
   end
 
@@ -297,8 +300,11 @@
       options = default_opts.update({ 'ignore_bad_values' => true, 'hints' => { 'a_field' => 'map<float>' } })
       sut_instance = sut.new(options)
       sample_event['a_field'] = 'i am not a set'
+      expect(options['logger']).to(receive(:error))
+
+      result = sut_instance.parse(sample_event)
 
-      expect { sut_instance.parse(sample_event) }.to raise_error ArgumentError
+      expect(result).to be_nil
     end
   end
 end

spec/unit/outputs/safe_submitter_spec.rb

@@ -109,6 +109,15 @@ def generate_future_double
       return future_double
     end
 
+    it 'does nothing in case it got a nil action' do
+      doubles = setup_session_double(default_options)
+      expect(doubles[:session_double]).to_not(receive(:prepare))
+      expect(doubles[:session_double]).to_not(receive(:execute_async))
+      sut_instance = sut.new(default_options)
+
+      expect { sut_instance.submit([nil]) }.to_not raise_error
+    end
+
     it 'prepares and executes the query' do
       doubles = setup_session_double(default_options)
       expect(doubles[:session_double]).to(receive(:prepare).with(expected_query_for_one_action)).and_return('eureka')