INGEST: Implement Drop Processor

[original-brownbear]

• Throw DroppedDocumentException to indicate document is not to be indexed
• Catch and rethrow where necessary to prevent Exception being logged and avoid unnecessary wrapping and other operations on the exception
• In simulate API: Don't add dropped documents to returned docs array
• In index request: return "noop" type index response for dropped documents
• closes Drop document in ingest pipeline #23726

[elasticmachine]

Pinging @elastic/es-core-infra

[original-brownbear]

@rjernst can you take al look here and let me know if the following behaviour looks ok to you?:

Pipeline:

PUT _ingest/pipeline/my_index
{
    "description": "drops documents",
    "processors": [
      {
        "script": {
          "source": """
            if (ctx.message == 'text') {
              Processors.drop()
            }
          """
        }
      }
    ]
}

Standard index request:

POST foo/1/?pipeline=my_index
{
  "message": "text"
}

returns:

{
  "_index": "foo",
  "_type": "1",
  "_id": null,
  "_version": -3,
  "result": "noop",
  "_shards": {
    "total": 0,
    "successful": 0,
    "failed": 0
  }
}

Simulate API:

Request:

POST _ingest/pipeline/_simulate
{
  "pipeline" :
  {
    "description": "_description",
    "processors": [
        {
        "script": {
          "source": """
            if (ctx.message == 'text') {
              Processors.drop()
            }
          """
        }
      }
    ]
  },
  "docs": [
    {
      "_index": "index",
      "_type": "_doc",
      "_id": "id",
      "_source": {
        "message": "text"
      }
    },
    {
      "_index": "index",
      "_type": "_doc",
      "_id": "id",
      "_source": {
        "message": "text1"
      }
    }
  ]
}

returns:

{
  "docs": [
    {
      "doc": {
        "_index": "index",
        "_type": "_doc",
        "_id": "id",
        "_source": {
          "message": "text1"
        },
        "_ingest": {
          "timestamp": "2018-07-23T10:32:47.461733Z"
        }
      }
    }
  ]
}

Then I'd add tests here and we should be good :) (provided you're ok with the implementation)

[rjernst]

Sorry for the long delay in reviewing @original-brownbear. Overall this looks good, but we need to not add ingest specific code to painless.

[rjernst]

Painless shouldn't know about ingest. I think we need to discuss with @jdconrad how we can change painless to allow some exception types to be propagated verbatim instead of wrapping in a ScriptException. Perhaps we could allow execute on script classes to have checked exceptions, and those exceptions would be rethrown instead of wrapped?

[original-brownbear]

@rjernst yea this was really dirty, hence the WIP label.

@jdconrad @rjernst I like the idea of a checked Exception for this. Maybe something like ScriptExitException? We could even add an exit code field to it and maybe reuse it for other scenarios as well (thinking about ingest processors called from Painless directly failing here mostly ... those currently generate a weirdly complicated wrapped exception when failing, when in actuality you probably just want the failure in the processor to propagate up cleanly)?

Then we could do:

} catch (ScriptExitException e) {
   if (e.exitCode() == 0) {
      // Drop document quietly
   } else {
     // Log error and do other error things
   }
}

WDYT?

[jdconrad]

You can't use exceptions in this form because a user could potentially catch them in the script. We decided a while back that if we want to get around this we can use Error classes and we will never whitelist them. These bypass the users ability to catch them in script.

To add to that we currently have a PainlessError class that maybe could be used as an example for some of what you'd like to accomplish here? This gets triggered right now when a loop counter goes too high.

[rjernst]

I don't think an error is right here. This is essentially an exception the user is throwing (through calling the drop() method). Sure, they could catch it themselves. Another option (suggested by @jdconrad privately) is to have a flag on the IngestDocument, and set this, kind of like we have noop for update scripts. This might be an ok solution; we would then make the drop method take the ingest document and set the flag.

To be clear, though, my proposal was to change the execute method to something like this:

void execute(Map ctx) throws DroppedDocmentException

Painless would determine any checked exceptions when reflecting the execute method, and add a catch/rethrow into the existing try/catch within the main execute implementation. So the exception type would not be generic, but still specific to ingest, and painless would learn about it through reflection (as it does the other argument types as well).

[original-brownbear]

@rjernst @jdconrad

Painless would determine any checked exceptions when reflecting the execute method, and add a catch/rethrow into the existing try/catch within the main execute implementation. So the exception type would not be generic, but still specific to ingest, and painless would learn about it through reflection (as it does the other argument types as well).

👍 This makes sense to me, maybe we could start moving on that?

[original-brownbear]

@rjernst ping :)

[original-brownbear]

Discussed this with @rjernst and we came up with an approach to this that
doesn't force the use of Exceptions and catching them on us.

As discussed earlier, the long-term goal is moving to representing the pipeline
as a script instead of as JSON.
If we were to represent a pipeline as a script like e.g.:

if(doc.foo == 'bar') {
   return 'null'
}

// legacy script processor called from new-style pipeline
scriptProcessor({
 "script": "script source",
 "params": {
   "param1": "val1" 
 }
}).execute(doc)

return doc

then handling dropping a document as simply returning null seems like a natural approach.
The practical issue we have in getting there with the current scripts is that they have no
return value, so supporting drop in legacy scripts is not possible in this way.

Now that we do have conditionals for any processor by setting the if field having drop
as part of the current scripts is less valuable though. We could simply implement a Drop
processor in the normal JSON pipeline and have its only configuration setting be the if condition.
This still leaves the question of how to actually make the execution of the pipeline break out
as a result of the drop processor.

In order to be able to break out of the execution and make the pipeline more similar to
what the above script based pipeline configuration looks like we were thinking of:

• Change the current org.elasticsearch.ingest.Processor#execute method from void to
  returning IngestDocument.
  • Make all existing processor simply return the input document for now
  • Make the drop processor return null.
  • Adjust foreach, pipeline processor and composite processor accordingly so that they
    simply break out if a pipeline/processor returns null and return null themselves
  • Do not index document if pipeline returns null

(correct me if I misrepresented anything @rjernst :) )

[original-brownbear]

@rjernst Implemented the above approach here (needs tests still and I might have to clean up a few left over formatting changes from the earlier impl. but functionally it seems fine), so in practice this works now:

Define drop processor with condition:

PUT _ingest/pipeline/my_index
{
    "description": "drops documents",
    "processors": [
      {
        "drop": {
          "if": "ctx.message == 'text'"
        }
      }
    ]
}

Try to index document that should be dropped:

POST foo/1/?pipeline=my_index
{
  "message": "text"
}

Returns:

{
  "_index": "foo",
  "_type": "1",
  "_id": null,
  "_version": -3,
  "result": "noop",
  "_shards": {
    "total": 0,
    "successful": 0,
    "failed": 0
  }
}

Think this is what we want functionally (then I'd clean it up real quick and add some tests :) and we should have this ready shortly).

[rjernst]

Thanks! LGTM

[rjernst]

technically shouldn't the ingestDocument be chained for each successive execute call?

[original-brownbear]

Jup, that would be cleaner now => Will adjust that tomorrow :)

[original-brownbear]

@rjernst thanks!

[original-brownbear]

urgh ... I just realized I completely forgot about adding tests here and just instinctively merged on the LGTM ✔️ => sorry about that PR for tests incoming shortly

[original-brownbear]

Tests -> #33430

[jakelandis]

@original-brownbear - is this still pending back port to 6.5 ? Any concerns about the change to the execute signature for the 6.x branch ?

https://github.com/elastic/elasticsearch/tree/6.x/modules/ingest-common/src/main/java/org/elasticsearch/ingest/common

[original-brownbear]

@jakelandis no not really but just to be safe I was waiting for #33430 to be reviewed/merged so we have a bit of coverage :)

[jakelandis]

@original-brownbear - now that #33430 is merged, should be backport this for 6.5 ?

[guruxu]

Folks, it would really be helpful to us plugin developers to plan ahead of time, if you could 1) add a deprecation annotation in advance and 2) release note the change. Re: https://github.com/elastic/elasticsearch/pull/32278/files#diff-41059c22f314a0bd73e614cc788a483bR60

[jakelandis]

@guruxu - apologies for the troubles this has caused you. We will try to be more careful with these types of changes.