Validate the action before sending the bulk request to ES.

[mashhurs]

What this PR does?

Plugin receives 400 Validation Failed: 1: no requests added exception when requesting unsupported bulk action.
This PR introduces a validation check on action before sending bulk request to ES.

Closes #1079

Test

Manual testing

• Used configuration to read from JSON file and send to ES. File contains an action field which is not in [index, create, update, delete]

   input {
      file {
          path => "/path/to/files/*.json"
          start_position => "beginning"
          codec => "json"
      }
   }
   output {
     elasticsearch {
       ....
       action => "%{[@metadata][action]}"
     }
   }

• Result

[2022-09-09T11:59:44,810][WARN ][logstash.outputs.elasticsearch][main][ca10f590b6ab3366bc38f754293adbd2499c9d97aedd7933d11542cd26cd056d] Could not index event to Elasticsearch because its action is not supported. Action: ["partialupdate", {:_id=>"5412e1fe-7822-4440-86b6-5b508ab99229", :_index=>"logstash-test-output", :routing=>nil}, {"eventData"=>....}]

Unit testing

• Run the unit test
  rspec spec/unit/outputs/elasticsearch_spec.rb

Logs

[2022-09-21T09:55:33,170][WARN ][logstash.outputs.elasticsearch][main][02e543d9daaf1e2af9190a6aef2cb91aa012fabaf7756093f487d85b2b9224de] Badly formatted index, after interpolation still contains placeholder: [%{[index_name]}], {"action"=>"index", "host"=>{"name"=>"dhcp-128-189-70-158"}, "name"=>"TEST", "source"=>"template", "@timestamp"=>2022-09-21T16:55:33.045630Z, "type"=>"TEST", "version"=>1, "@version"=>"1", "log"=>{"file"=>{"path"=>"/Users/mashhur/Dev/elastic/SDH/test2.json"}}, "event"=>{"original"=>"{\"name\":\"TEST\",\"type\":\"TEST\",\"action\":\"index\",\"source\":\"template\",\"version\":1}"}}

[2022-09-21T09:55:33,171][WARN ][logstash.outputs.elasticsearch][main][02e543d9daaf1e2af9190a6aef2cb91aa012fabaf7756093f487d85b2b9224de] Elasticsearch doesn't support [partialupdate] action, {"action"=>"partialupdate", "host"=>{"name"=>"dhcp-128-189-70-158"}, "source"=>"template", "type"=>"TEST", "index_name"=>"logstash-output", "@version"=>"1", "log"=>{"file"=>{"path"=>"/Users/mashhur/Dev/elastic/SDH/test2.json"}}, "event"=>{"original"=>"{\"name\":\"TEST\",\"type\":\"TEST\",\"action\":\"partialupdate\",\"source\":\"template\",\"version\":1,\"index_name\":\"logstash-output\"}"}, "name"=>"TEST", "@timestamp"=>2022-09-21T16:55:33.052311Z, "version"=>1}


[2022-09-21T09:55:33,171][WARN ][logstash.outputs.elasticsearch][main][02e543d9daaf1e2af9190a6aef2cb91aa012fabaf7756093f487d85b2b9224de] Can't map the event, needs to configure the pipeline settings according to the input. The event sent to DLQ: Badly formatted index, after interpolation still contains placeholder: [%{[index_name]}], {"action"=>"index", "host"=>{"name"=>"dhcp-128-189-70-158."}, "name"=>"TEST", "source"=>"template", "@timestamp"=>2022-09-21T16:55:33.045630Z, "type"=>"TEST", "version"=>1, "@version"=>"1", "log"=>{"file"=>{"path"=>"/Users/mashhur/Dev/elastic/SDH/test2.json"}}, "event"=>{"original"=>"{\"name\":\"TEST\",\"type\":\"TEST\",\"action\":\"index\",\"source\":\"template\",\"version\":1}"}}


[2022-09-21T09:55:33,171][WARN ][logstash.outputs.elasticsearch][main][02e543d9daaf1e2af9190a6aef2cb91aa012fabaf7756093f487d85b2b9224de] Can't map the event, needs to configure the pipeline settings according to the input. The event sent to DLQ: Elasticsearch doesn't support [partialupdate] action, {"action"=>"partialupdate", "host"=>{"name"=>"dhcp-128-189-70-158"}, "source"=>"template", "type"=>"TEST", "index_name"=>"logstash-output", "@version"=>"1", "log"=>{"file"=>{"path"=>"/Users/mashhur/Dev/elastic/SDH/test2.json"}}, "event"=>{"original"=>"{\"name\":\"TEST\",\"type\":\"TEST\",\"action\":\"partialupdate\",\"source\":\"template\",\"version\":1,\"index_name\":\"logstash-output\"}"}, "name"=>"TEST", "@timestamp"=>2022-09-21T16:55:33.052311Z, "version"=>1}

[yaauie]

We have an ordering problem, since the per-event responses must be emitted in the same order as the actions we received.

[andsel]

Sorry to chime in, I saw this while working on #1084, and had similar problem in pre-validate a condition (in my case the sprintf of the index is fully resolved) and in case send to DLQ.

[andsel]

The changes you proposed seems good to me.
Left a couple of notes on test's expectations, if should be re-enabled plus a concern about the eventuality to log something in case a mapping error happens.

[andsel]

Should be re-enabled to check also the failed ones?

[mashhurs]

Yup, added in new commits.

[andsel]

If dlq is enabled then no log lines are logged. I think that any mapping error, that route events to DLQ, should be clearly highlighted into the logs, so a warn log line should be present to eventually understand what's happening.

[yaauie]

To avoid flooding I would like to log a single INFO for the batch if the DLQ is enabled (including the quantity of events from the batch), or to defer to the per-event WARN if there is no DLQ and we need to drop the event. Users who have the DLQ enabled should already be in a position to monitor its growth.

[yaauie]

To avoid flooding I would like to log a single INFO for the batch if the DLQ is enabled (including the quantity of events from the batch), or to defer to the per-event WARN if there is no DLQ and we need to drop the event. Users who have the DLQ enabled should already be in a position to monitor its growth.

[yaauie]

I would prefer a rephrase of this, so that event_mapping_errors was either an array of event/message tuples or an array of objects that have event/message properties, since action in this code is somewhat ambiguous and can mean either a String action or an action-tuple as prepared for a bulk request.

When we are building this array, we already have access to the event and an object that has the message (the exception), and when we are raising the exception we have access to the event. We could either create a specialized struct, or can embed the event into the exception that we are already handling.

Additionally, we have everything we need to produce our detail_message when populating the array (or even when throwing the relevant exception that leads to an entry in the array), so I would rather pull that up.

[mashhurs]

@yaauie , I really liked the idea of having rspec-collection_matchers when unit tests check agains collections sizes. I understood more about use cases after taking a look at this thread.
However, in order to use collection_matchers, we need to onboard it with LS-core (add gem "rspec-collection_matchers", :group => :development in Gemfile.template file).

Let me know your thoughts if I can add it to plugin level dependency.

[yaauie]

I could go either way -- the failure scenario reads "clear enough" for these specs.

The normal way of adding the dependency would be to declare a development dependency in this project's gemspec, not in Logstash core.

[yaauie]

Left one follow-up on the collections expectations. Feel free to either chase it down or merge as-is.

[yaauie]

LGTM 🎉