Implements safe evaluation of conditional expressions, logging the error without killing the pipeline

[andsel]

Release notes

Fix if statement expression evaluation against runtime errors that could crash the pipeline, just logging the offending event and continuing with the next in the batch.

What does this PR do?

Translates the org.jruby.exceptions.TypeError, IllegalArgumentException, org.jruby.exceptions.ArgumentError that could happen during EventCodition evaluation into a custom ConditionalEvaluationError which bubbles up on AST tree nodes. It's catched in the SplitDataset node.
Updates the generation of the SplitDataset so that the execution of filterEvents method inside the compute body is try-catch guarded and defer the execution to an instance of AbstractPipelineExt.ConditionalEvaluationListener to handle such error. In this particular case the error management consist in just logging the offending Event.

Why is it important/What is the impact to the user?

This PR protects the if statements against expression evaluation errors, cancel the event under processing and log it.
This avoids to crash the pipeline which encounter a runtime error during event condition evaluation, permitting to debug the root cause reporting the offending event and removing from the current processing batch.

Checklist

• My code follows the style guidelines of this project
• I have commented my code, particularly in hard-to-understand areas
• I have made corresponding changes to the documentation
• [ ] I have made corresponding change to the default configuration files (and/or docker env variables)
• I have added tests that prove my fix is effective or that my feature works

Author's Checklist

• add test case for invalid UTF-8 sequences "crashes" the logstash  #15091

How to test this PR locally

1. start with small pipeline like

input {
    stdin { codec => json_lines }
}
filter {
  if [path][to][value] > 100 {
    mutate { add_tag => "hit" }
  }
}
output {
  stdout { 
    codec => rubydebug
  }
}

and use the data samples from #16007
2. tinker with ifs also in the output, something like:

input {
    stdin { codec => json_lines }
}

filter {
  if [path][to][value] > 100 {
    mutate { add_tag => "hit" }
  } else {
    mutate { add_tag => "miss" }
  }
  mutate { add_tag => "after"}
}

output {
   if [path][to][value] > 100 {
        stdout {
            codec => rubydebug
        }
   }
}

3. then verify with batches containing ok and ko data. Add the following input for example:

file {
    path => "/tmp/pipeline_conditional_test_fixture.json"
    sincedb_path => "/tmp/logstash_andsel/sincedb"
    mode => "read"
    file_completed_action => "log"
    file_completed_log_path => "/tmp/processed.log"

    codec => json
}

Using a json file like:

{"path":{"to":{"value":101}}}
{"path":{"to":{"value":102}}}
{"path":{"to":{"value":"101"}}}
{"path":{"to":{"value":103}}}

4. whatever other test comes to your mind :-)

Related issues

• Relates Comparison conditionals crash pipelines opaquely when field on event is not correct value type #16007
• Closes invalid UTF-8 sequences "crashes" the logstash  #15091
• Closes Better conditional error messages #5593

Use cases

Screenshots

Logs

[andsel]

Note for reviewer

This PR solves the problem to protect against if failures inserting a try-catch statement in the dataset that invokes the split dataset that evaluates the EventCondition.

without this PR the following statement:

if [a] {
    noop "hit"
}

is translated to the following Dataset graph:

graph TD;
DatasetFC[Dataset filter call]-->Split[Split Dataset];
    DatasetFC-->FiltDlg[Filter Delegate.multiFilter 'hit'];
    Split-->EC[Event condition];
    Split-->CMPL[Complement];

Loading

and the following filters call-chain reuse the same Dataset filter call. So

  mutate "hit"
  mutate "hit 2"

is translated to the following AST

graph TD;
DatasetFC[Dataset filter call]-->DatasetFC2[Dataset filter call];
    DatasetFC-->FiltDlg[Filter Delegate.multiFilter 'hit2'];
    DatasetFC2-->CMPL[Next...];
    DatasetFC2-->FiltDlg2[Filter Delegate.multiFilter 'hit'];

Loading

while with new change it becomes:

graph TD;
DatasetFC[Dataset filter call]-->Split[Split Dataset];
    DatasetFC-->FiltDlg[Filter Delegate.multiFilter 'hit'];
    Split-->EC[Event condition];
    DatasetFC-->ErrList[error handler]
    Split-->CMPL[Complement];

Loading

So while in without this PR the same Dataset filter was used both inside an if statement and to chain filter calls. With this PR the dataset that wraps a filter after an if statement are different and so the reason why to create this "noop chain" which is used also in the pipeline_reuse_test.conf, so if we don't chain here we don't have the dataset that wraps the filter without the try statement.

[andsel]

We the else branch because a datasource that wraps an output delegator and evaluates the complement of the condition is already used in the pipeline_reuse_test.conf output section.

[andsel]

We need to remove this output invocation else the datasource that invokes all the outputs has 5 children datasources (1 for the if, 1 for the complement and 3 for the stdout filters, out for each invocation). This would generate a Datasource with 5 children, while in the pipeline_reuse_test.conf output section the children are 4, (1 for the first ifstatement, 1 for the complement of the first if, 1 for the second ifstatement, 1 for the complement of the second if).

[andsel]

This is the complement branch (the right one, the else of the empty branch of an if). The dataset execute positive path on if and the complement. If an evaluation error occur in EventCondition it will happen also in the complementary evaluation of the same event condition. So silence the error, the data remains empty so the execution flow interrupts here for the batch.

[jsvd]

This PR should have a much more comprehensive test suite given the kind of problem it aims to address.

From my tests, out of 3 pipelines only 1 didn't crash:

input {
    generator {
      message => '{"path":{"to":{"value":"101"}}}'
      codec => json
      count => 1
    }
}
filter {
  if [path][to][value] > 100 { mutate { add_tag => "hit" } }
}
output { stdout { codec => rubydebug } }

which gave:

[2024-07-22T15:39:11,572][WARN ][org.logstash.execution.AbstractPipelineExt][main] Error in condition evaluation with event {"path":{"to":{"value":"101"}}}

But for example:

input {
    generator {
      message => '{}'
      codec => json
      count => 1
    }
}
filter {
  if [path][to][value] > 100 { mutate { add_tag => "hit" } }
}
output { stdout { codec => rubydebug } }

crashed with:

[2024-07-22T15:42:06,915][ERROR][logstash.javapipeline    ][main] Pipeline worker error, the pipeline will be stopped {:pipeline_id=>"main", :error=>"Cannot invoke \"Object.toString()\" because \"unexpected\" is null", :exception=>Java::JavaLang::NullPointerException, :backtrace=>["org.logstash.config.ir.compiler.EventCondition$Compiler$UnexpectedTypeException.getUnexpectedTypeDetails(EventCondition.java:712)", "org.logstash.config.ir.compiler.EventCondition$Compiler$UnexpectedTypeException.<init>(EventCondition.java:700)", 

Similarly with:

input {
    generator {
      message => '{"path":{"to":{"value":[101, 102]}}}'
      codec => json
      count => 1
    }
}
filter {
  if [path][to][value] > 100 { mutate { add_tag => "hit" } }
}
output { stdout { codec => rubydebug } }

[jsvd]

the issue with doing LOGGER.warn("string {}", event, err) is that it prints the exception on a new line:

[2024-07-24T12:42:34,289][WARN ][org.logstash.execution.AbstractPipelineExt][main] Error in condition evaluation with event null
org.logstash.config.ir.compiler.ConditionalEvaluationError: org.logstash.config.ir.compiler.EventCondition$Compiler$UnexpectedTypeException: Unexpected input type combination left <no-class>:<null-value>, right class org.jruby.RubyFixnum:100
	at org.logstash.config.ir.compiler.Utils.filterEvents(Utils.java:56) ~[logstash-core.jar:?]
	at org.logstash.generated.CompiledDataset1.compute(Unknown Source) ~[?:?]
	at org.logstash.generated.CompiledDataset2.compute(Unknown Source) ~[?:?]
	at org.logstash.generated.CompiledDataset3.compute(Unknown Source) ~[?:?]
	at org.logstash.config.ir.CompiledPipeline$CompiledUnorderedExecution.compute(CompiledPipeline.java:364) ~[logstash-core.jar:?]
	at org.logstash.config.ir.CompiledPipeline$CompiledUnorderedExecution.compute(CompiledPipeline.java:358) ~[logstash-core.jar:?]
	at org.logstash.execution.ObservedExecution.lambda$compute$0(ObservedExecution.java:17) ~[logstash-core.jar:?]
	at org.logstash.execution.WorkerObserver.lambda$observeExecutionComputation$0(WorkerObserver.java:39) ~[logstash-core.jar:?]
	at org.logstash.instrument.metrics.timer.ConcurrentLiveTimerMetric.time(ConcurrentLiveTimerMetric.java:47) ~[logstash-core.jar:?]
	at org.logstash.execution.WorkerObserver.lambda$executeWithTimers$1(WorkerObserver.java:50) ~[logstash-core.jar:?]
	at org.logstash.instrument.metrics.timer.ConcurrentLiveTimerMetric.time(ConcurrentLiveTimerMetric.java:47) ~[logstash-core.jar:?]
	at org.logstash.execution.WorkerObserver.executeWithTimers(WorkerObserver.java:50) ~[logstash-core.jar:?]
	at org.logstash.execution.WorkerObserver.observeExecutionComputation(WorkerObserver.java:38) ~[logstash-core.jar:?]
	at org.logstash.execution.ObservedExecution.compute(ObservedExecution.java:17) ~[logstash-core.jar:?]
	at org.logstash.execution.WorkerLoop.abortableCompute(WorkerLoop.java:113) ~[logstash-core.jar:?]
	at org.logstash.execution.WorkerLoop.run(WorkerLoop.java:86) ~[logstash-core.jar:?]
	at jdk.internal.reflect.DirectMethodHandleAccessor.invoke(DirectMethodHandleAccessor.java:103) ~[?:?]
	at java.lang.reflect.Method.invoke(Method.java:580) ~[?:?]
	at org.jruby.javasupport.JavaMethod.invokeDirectWithExceptionHandling(JavaMethod.java:300) ~[jruby.jar:?]
	at org.jruby.javasupport.JavaMethod.invokeDirect(JavaMethod.java:164) ~[jruby.jar:?]
	at org.jruby.java.invokers.InstanceMethodInvoker.call(InstanceMethodInvoker.java:32) ~[jruby.jar:?]
	at org.jruby.runtime.callsite.CachingCallSite.call(CachingCallSite.java:193) ~[jruby.jar:?]
	at org.jruby.ir.interpreter.InterpreterEngine.processCall(InterpreterEngine.java:346) ~[jruby.jar:?]
	at org.jruby.ir.interpreter.StartupInterpreterEngine.interpret(StartupInterpreterEngine.java:66) ~[jruby.jar:?]
	at org.jruby.ir.interpreter.Interpreter.INTERPRET_BLOCK(Interpreter.java:118) ~[jruby.jar:?]
	at org.jruby.runtime.MixedModeIRBlockBody.commonYieldPath(MixedModeIRBlockBody.java:136) ~[jruby.jar:?]
	at org.jruby.runtime.IRBlockBody.call(IRBlockBody.java:66) ~[jruby.jar:?]
	at org.jruby.runtime.IRBlockBody.call(IRBlockBody.java:58) ~[jruby.jar:?]
	at org.jruby.runtime.Block.call(Block.java:144) ~[jruby.jar:?]
	at org.jruby.RubyProc.call(RubyProc.java:354) ~[jruby.jar:?]
	at org.jruby.internal.runtime.RubyRunnable.run(RubyRunnable.java:111) ~[jruby.jar:?]
	at java.lang.Thread.run(Thread.java:1583) ~[?:?]
Caused by: org.logstash.config.ir.compiler.EventCondition$Compiler$UnexpectedTypeException: Unexpected input type combination left <no-class>:<null-value>, right class org.jruby.RubyFixnum:100
	at org.logstash.config.ir.compiler.EventCondition$Compiler.compare(EventCondition.java:458) ~[logstash-core.jar:?]
	at org.logstash.config.ir.compiler.EventCondition$Compiler.lambda$compareFieldToConstant$11(EventCondition.java:449) ~[logstash-core.jar:?]
	at org.logstash.config.ir.compiler.Utils.filterEvents(Utils.java:52) ~[logstash-core.jar:?]
	... 31 more

But when using log.format=json the exception doesn't show up:

{"level":"WARN","loggerName":"org.logstash.execution.AbstractPipelineExt","timeMillis":1721821446156,"thread":"[main]>worker8","logEvent":{"message":"Error in condition evaluation with event null"}}

Maybe we can log only the lastErrorEvaluationReceived at warn level, but add the rest to a debug log level entry.

[andsel]

So we have this problem of loosing the exceptions for each exception logged if the format is JSON.
I think we could do something like:

LOGGER.warn("Error in condition evaluation with error {} on event {}", lastErrorEvaluationReceived, err.failedEvent().getField("[event][original]"));
try (StringWriter sw = new StringWriter(); PrintWriter pw = new PrintWriter(sw)) {
    err.printStackTrace(pw);
    LOGGER.debug("{}", sw);
}

Which would generate the following JSON log:

{"level":"DEBUG",
"loggerName":"org.logstash.execution.AbstractPipelineExt",
"timeMillis":1721835389378,
"thread":"[main]>worker1",
"logEvent":{
    "message":"org.logstash.config.ir.compiler.ConditionalEvaluationError: org.logstash.config.ir.compiler.EventCondition$Compiler$UnexpectedTypeException: Unexpected input type combination left class org.logstash.ConvertedList:ConvertedList{delegate=[101, 102]}, right class org.jruby.RubyFixnum:100\n\tat org.logstash.config.ir.compiler.Utils.filterEvents(Utils.java:56)\n\tat org.logstash.generated.CompiledDataset1.compute(Unknown Source)\n\tat org.logstash.generated.CompiledDataset2.compute(Unknown Source)\n\tat org.logstash.generated.CompiledDataset3.compute(Unknown Source)\n\tat org.logstash.config.ir.CompiledPipeline$CompiledUnorderedExecution.compute(CompiledPipeline.java:364)\n\tat org.logstash.config.ir.CompiledPipeline$CompiledUnorderedExecution.compute(CompiledPipeline.java:358)\n\tat org.logstash.execution.ObservedExecution.lambda$compute$0(ObservedExecution.java:17)\n\tat org.logstash.execution.WorkerObserver.lambda$observeExecutionComputation$0(WorkerObserver.java:39)\n\tat org.logstash.instrument.metrics.timer.ConcurrentLiveTimerMetric.time(ConcurrentLiveTimerMetric.java:47)\n\tat org.logstash.execution.WorkerObserver.lambda$executeWithTimers$1(WorkerObserver.java:50)\n\tat org.logstash.instrument.metrics.timer.ConcurrentLiveTimerMetric.time(ConcurrentLiveTimerMetric.java:47)\n\tat org.logstash.execution.WorkerObserver.executeWithTimers(WorkerObserver.java:50)\n\tat org.logstash.execution.WorkerObserver.observeExecutionComputation(WorkerObserver.java:38)\n\tat org.logstash.execution.ObservedExecution.compute(ObservedExecution.java:17)\n\tat org.logstash.execution.WorkerLoop.abortableCompute(WorkerLoop.java:113)\n\tat org.logstash.execution.WorkerLoop.run(WorkerLoop.java:86)\n\tat java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)\n\tat java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:77)\n\tat java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\n\tat java.base/java.lang.reflect.Method.invoke(Method.java:568)\n\tat org.jruby.javasupport.JavaMethod.invokeDirectWithExceptionHandling(JavaMethod.java:300)\n\tat org.jruby.javasupport.JavaMethod.invokeDirect(JavaMethod.java:164)\n\tat org.jruby.java.invokers.InstanceMethodInvoker.call(InstanceMethodInvoker.java:32)\n\tat org.jruby.runtime.callsite.CachingCallSite.cacheAndCall(CachingCallSite.java:456)\n\tat org.jruby.runtime.callsite.CachingCallSite.call(CachingCallSite.java:195)\n\tat org.jruby.ir.interpreter.InterpreterEngine.processCall(InterpreterEngine.java:346)\n\tat org.jruby.ir.interpreter.StartupInterpreterEngine.interpret(StartupInterpreterEngine.java:66)\n\tat org.jruby.ir.interpreter.Interpreter.INTERPRET_BLOCK(Interpreter.java:118)\n\tat org.jruby.runtime.MixedModeIRBlockBody.commonYieldPath(MixedModeIRBlockBody.java:136)\n\tat org.jruby.runtime.IRBlockBody.call(IRBlockBody.java:66)\n\tat org.jruby.runtime.IRBlockBody.call(IRBlockBody.java:58)\n\tat org.jruby.runtime.Block.call(Block.java:144)\n\tat org.jruby.RubyProc.call(RubyProc.java:354)\n\tat org.jruby.internal.runtime.RubyRunnable.run(RubyRunnable.java:111)\n\tat java.base/java.lang.Thread.run(Thread.java:840)\nCaused by: org.logstash.config.ir.compiler.EventCondition$Compiler$UnexpectedTypeException: Unexpected input type combination left class org.logstash.ConvertedList:ConvertedList{delegate=[101, 102]}, right class org.jruby.RubyFixnum:100\n\tat org.logstash.config.ir.compiler.EventCondition$Compiler.compare(EventCondition.java:458)\n\tat org.logstash.config.ir.compiler.EventCondition$Compiler.lambda$compareFieldToConstant$11(EventCondition.java:449)\n\tat org.logstash.config.ir.compiler.Utils.filterEvents(Utils.java:52)\n\t... 34 more\n"
  }
}

[andsel]

Fixed by ee1837a

[jsvd]

The warn-level message won't be useful if the document doesn't have event.original. I suggest we keep the warn level message more generic and log the entire event in DEBUG level.

Also, this PR is missing documentation about the consequences of an invalid conditional, here https://www.elastic.co/guide/en/logstash/current/event-dependent-configuration.html#conditionals

[jsvd]

There's no guarantee the event will contain an [event][original] field, resulting in a confusing message:

[2024-08-13T12:15:51,184][WARN ][org.logstash.execution.AbstractPipelineExt][main] Error in condition evaluation with error org.logstash.config.ir.compiler.EventCondition$Compiler$UnexpectedTypeException: Unexpected input type combination left <no-class>:<null-value>, right class org.jruby.RubyFixnum:100 on event null

I suggest dumping the entire document but only on DEBUG level.

[andsel]

Hi @jsvd, I've split the log in warn with generic message and debug with the toMap representation of the event.
Added also org.jruby.exceptions.ArgumentError to the list of errors that can be catched during conditional expression evaluation, plus improved the guide with a short comment on what happens when an expression evaluation fails during conditional execution.

[elastic-sonarqube]

Quality Gate passed

Issues
0 New issues
0 Fixed issues
0 Accepted issues

Measures
0 Security Hotspots
No data about Coverage
0.0% Duplication on New Code

See analysis details on SonarQube

[elasticmachine]

💚 Build Succeeded

• Buildkite Build
• Commit: 41e5661

History

• 💚 Build #1406 succeeded f60892c
• 💚 Build #1398 succeeded e2f3197
• 💔 Build #1397 failed ea5d094
• 💚 Build #1395 succeeded 135ae08
• 💚 Build #1392 succeeded 01fcab7
• 💚 Build #1391 succeeded beb47ad

cc @andsel

[paulparas]

Hi @andsel

In which logstash version would this fix be released?

Thanks
Paras

[andsel]

Hi @paulparas thank's for pointing it out, it's going to be released in 8.16.0