Logstash: Drop filter does not interrupt event filtering

[hpello]

• Version: 6.4.1
• Operating System: Docker: docker.elastic.co/logstash/logstash-oss:6.4.1

• Config File:
  logstash.conf:

  input {
  stdin {
    codec => "json"
  }
  }
  
  filter {
  if ![time] {
    ruby { code => "logger.warn('Drop invalid event:', event.to_hash)" }
    drop {}
  }
  
  # uncomment next line and problem disappears!
  # ruby { code => "" }
  
  if [time] < "2018" {
    drop {}
  }
  }
  
  output {
  stdout { codec => rubydebug { metadata => true } }
  }
  

• Steps to Reproduce:
• Launch logstash:
  docker run --rm -it \ -v $PWD/logstash.conf:/usr/share/logstash/pipeline/logstash.conf:ro \ docker.elastic.co/logstash/logstash-oss:6.4.1
• When input is ready, send an empty json object to stdin: {}
• Watch debug message being displayed and event supposedly being dropped, but pipeline crashing afterwards because it tries to compare a nil value with a string.
• Interestingly, if you insert a line of ruby code after "drop" filter but before "if" statement, the problem disappears, i.e. "if" statement is not executed.
• Logs:

Sending Logstash logs to /usr/share/logstash/logs which is now configured via log4j2.properties
[2018-10-03T13:28:00,933][INFO ][logstash.setting.writabledirectory] Creating directory {:setting=>"path.queue", :path=>"/usr/share/logstash/data/queue"}
[2018-10-03T13:28:00,945][INFO ][logstash.setting.writabledirectory] Creating directory {:setting=>"path.dead_letter_queue", :path=>"/usr/share/logstash/data/dead_letter_queue"}
[2018-10-03T13:28:01,412][INFO ][logstash.agent           ] No persistent UUID file found. Generating new UUID {:uuid=>"6f3d9f64-5407-42f1-8311-f9487bdcd8eb", :path=>"/usr/share/logstash/data/uuid"}
[2018-10-03T13:28:02,094][INFO ][logstash.runner          ] Starting Logstash {"logstash.version"=>"6.4.1"}
[2018-10-03T13:28:06,210][INFO ][logstash.pipeline        ] Starting pipeline {:pipeline_id=>"main", "pipeline.workers"=>2, "pipeline.batch.size"=>125, "pipeline.batch.delay"=>50}
[2018-10-03T13:28:06,378][INFO ][logstash.inputs.stdin    ] Automatically switching from json to json_lines codec {:plugin=>"stdin"}
[2018-10-03T13:28:06,474][INFO ][logstash.pipeline        ] Pipeline started successfully {:pipeline_id=>"main", :thread=>"#<Thread:0x71828c37 run>"}
The stdin plugin is now waiting for input:
[2018-10-03T13:28:06,569][INFO ][logstash.agent           ] Pipelines running {:count=>1, :running_pipelines=>[:main], :non_running_pipelines=>[]}
[2018-10-03T13:28:07,034][INFO ][logstash.agent           ] Successfully started Logstash API endpoint {:port=>9600}
{}
[2018-10-03T13:28:17,492][WARN ][logstash.filters.ruby    ] Drop invalid event: {"@version"=>"1", "@timestamp"=>2018-10-03T13:28:17.323Z, "host"=>"b64f1ca17e28"}
[2018-10-03T13:28:17,631][ERROR][logstash.pipeline        ] Exception in pipelineworker, the pipeline stopped processing new events, please check your filter configuration and restart Logstash. {:pipeline_id=>"main", "exception"=>"undefined method `<' for nil:NilClass", "backtrace"=>["(eval):153:in `block in initialize'", "org/jruby/RubyArray.java:1734:in `each'", "(eval):151:in `block in initialize'", "(eval):95:in `block in filter_func'", "/usr/share/logstash/logstash-core/lib/logstash/pipeline.rb:341:in `filter_batch'", "/usr/share/logstash/logstash-core/lib/logstash/pipeline.rb:320:in `worker_loop'", "/usr/share/logstash/logstash-core/lib/logstash/pipeline.rb:286:in `block in start_workers'"], :thread=>"#<Thread:0x71828c37 sleep>"}
[2018-10-03T13:28:17,801][FATAL][logstash.runner          ] An unexpected error occurred! {:error=>#<NoMethodError: undefined method `<' for nil:NilClass>, :backtrace=>["(eval):153:in `block in initialize'", "org/jruby/RubyArray.java:1734:in `each'", "(eval):151:in `block in initialize'", "(eval):95:in `block in filter_func'", "/usr/share/logstash/logstash-core/lib/logstash/pipeline.rb:341:in `filter_batch'", "/usr/share/logstash/logstash-core/lib/logstash/pipeline.rb:320:in `worker_loop'", "/usr/share/logstash/logstash-core/lib/logstash/pipeline.rb:286:in `block in start_workers'"]}
[2018-10-03T13:28:17,888][ERROR][org.logstash.Logstash    ] java.lang.IllegalStateException: Logstash stopped processing because of an error: (SystemExit) exit

[jsvd]

I can confirm this. A simpler reproduction pipeline is:

input { generator { count => 1 } }
filter {
  drop {}
  if [notafield] < "crash" { }
}
output {}

This seems related to an issue discussed a year and a half ago: https://github.com/elastic/logstash/issues/6744#issuecomment-295156821

The problem seems to be that we rely on the batch iteration of the next filter element to exclude the cancelled events of the previous one. If the next element is a filter plugin, the base class takes care of filtering out these for you (unless you use multi_filter). But if the next element is a conditional we aren't filtering those elements out, leading to the event getting evaluated and causing the crash.

[colinsurprenant]

@jsvd do we have the same behaviour with both the Ruby and Java pipelines?

[jsvd]

the experimental java pipeline does not exhibit this behaviour

[guyboertje]

hmmm.

[guyboertje]

On line 64 of WorkerLoop.java the dequeued QueueBatch object has its to_a method called (skips cancelled events straight out of the queue), and hands the RubyArray into the execution graph.

The cancelled state of an event is not checked inside the execution graph.

Deeper in the Java execution though, the comparison is not done via method calls on the LHS so RubyNil is not less than a RubyString so not error - but the cancelled event is still processed, only with no (buggy)side effects.