Add support for parameters to the script ingest processor

The script processor should support `params` to be consistent with all other script consumers.

docs/reference/ingest/ingest-node.asciidoc

@@ -1356,20 +1356,23 @@ caching see <<modules-scripting-using-caching, Script Caching>>.
 | `file`                 | no        | -       | The script file to refer to
 | `id`                   | no        | -       | The stored script id to refer to
 | `inline`               | no        | -       | An inline script to be executed
+| `params`               | no        | -       | Script Parameters
 |======
 
 You can access the current ingest document from within the script context by using the `ctx` variable.
 
-The following example sets a new field called `field_a_plus_b` to be the sum of two existing
-numeric fields `field_a` and `field_b`:
+The following example sets a new field called `field_a_plus_b_times_c` to be the sum of two existing
+numeric fields `field_a` and `field_b` multiplied by the parameter param_c:
 
 [source,js]
 --------------------------------------------------
 {
   "script": {
-    "field": "field_a_plus_b",
     "lang": "painless",
-    "inline": "return ctx.field_a + ctx.field_b"
+    "inline": "ctx.field_a_plus_b_times_c = (ctx.field_a + ctx.field_b) * params.param_c",
+    "params": {
+      "param_c": 10
+    }
   }
 }
 --------------------------------------------------

modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/ScriptProcessor.java

@@ -19,14 +19,12 @@
 
 package org.elasticsearch.ingest.common;
 
-import java.util.HashMap;
 import java.util.Map;
 
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.ingest.AbstractProcessor;
 import org.elasticsearch.ingest.IngestDocument;
 import org.elasticsearch.ingest.Processor;
-import org.elasticsearch.script.CompiledScript;
 import org.elasticsearch.script.ExecutableScript;
 import org.elasticsearch.script.Script;
 import org.elasticsearch.script.ScriptContext;
@@ -35,6 +33,7 @@ import org.elasticsearch.script.ScriptService;
 import static java.util.Collections.emptyMap;
 import static org.elasticsearch.common.Strings.hasLength;
 import static org.elasticsearch.ingest.ConfigurationUtils.newConfigurationException;
+import static org.elasticsearch.ingest.ConfigurationUtils.readOptionalMap;
 import static org.elasticsearch.ingest.ConfigurationUtils.readOptionalStringProperty;
 import static org.elasticsearch.ingest.ConfigurationUtils.readStringProperty;
 import static org.elasticsearch.script.ScriptService.ScriptType.FILE;
@@ -60,10 +59,8 @@ public final class ScriptProcessor extends AbstractProcessor {
 
     @Override
     public void execute(IngestDocument document) {
-        Map<String, Object> vars = new HashMap<>();
-        vars.put("ctx", document.getSourceAndMetadata());
-        CompiledScript compiledScript = scriptService.compile(script, ScriptContext.Standard.INGEST, emptyMap());
-        ExecutableScript executableScript = scriptService.executable(compiledScript, vars);
+        ExecutableScript executableScript = scriptService.executable(script, ScriptContext.Standard.INGEST, emptyMap());
+        executableScript.setNextVar("ctx",  document.getSourceAndMetadata());
         executableScript.run();
     }
 
@@ -87,6 +84,7 @@ public final class ScriptProcessor extends AbstractProcessor {
             String inline = readOptionalStringProperty(TYPE, processorTag, config, "inline");
             String file = readOptionalStringProperty(TYPE, processorTag, config, "file");
             String id = readOptionalStringProperty(TYPE, processorTag, config, "id");
+            Map<String, ?> params = readOptionalMap(TYPE, processorTag, config, "params");
 
             boolean containsNoScript = !hasLength(file) && !hasLength(id) && !hasLength(inline);
             if (containsNoScript) {
@@ -99,13 +97,17 @@ public final class ScriptProcessor extends AbstractProcessor {
                 throw newConfigurationException(TYPE, processorTag, null, "Only one of [file], [id], or [inline] may be configured");
             }
 
+            if(params == null) {
+                params = emptyMap();
+            }
+
             final Script script;
             if (Strings.hasLength(file)) {
-                script = new Script(file, FILE, lang, emptyMap());
+                script = new Script(file, FILE, lang, params);
             } else if (Strings.hasLength(inline)) {
-                script = new Script(inline, INLINE, lang, emptyMap());
+                script = new Script(inline, INLINE, lang, params);
             } else if (Strings.hasLength(id)) {
-                script = new Script(id, STORED, lang, emptyMap());
+                script = new Script(id, STORED, lang, params);
             } else {
                 throw newConfigurationException(TYPE, processorTag, null, "Could not initialize script");
             }

modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/ScriptProcessorTests.java

@@ -46,11 +46,9 @@ public class ScriptProcessorTests extends ESTestCase {
         int randomBytesTotal = randomBytesIn + randomBytesOut;
 
         ScriptService scriptService = mock(ScriptService.class);
-        CompiledScript compiledScript = mock(CompiledScript.class);
         Script script = new Script("_script");
-        when(scriptService.compile(any(), any(), any())).thenReturn(compiledScript);
         ExecutableScript executableScript = mock(ExecutableScript.class);
-        when(scriptService.executable(any(), any())).thenReturn(executableScript);
+        when(scriptService.executable(any(), any(), any())).thenReturn(executableScript);
 
         Map<String, Object> document = new HashMap<>();
         document.put("bytes_in", randomInt());

qa/smoke-test-ingest-with-all-dependencies/src/test/resources/rest-api-spec/test/ingest/50_script_processor_using_painless.yaml

@@ -1,5 +1,5 @@
 ---
-"Test script processor with inline script":
+"Test script processor with inline script and params":
   - do:
       ingest.put_pipeline:
         id: "my_pipeline"
@@ -10,7 +10,10 @@
               {
                 "script" : {
                   "lang" : "painless",
-                  "inline": "ctx.bytes_total = ctx.bytes_in + ctx.bytes_out"
+                  "inline": "ctx.bytes_total = (ctx.bytes_in + ctx.bytes_out) * params.factor",
+                  "params": {
+                     "factor": 10
+                  }
                 }
               }
             ]
@@ -32,7 +35,7 @@
         id: 1
   - match: { _source.bytes_in: 1234 }
   - match: { _source.bytes_out: 4321 }
-  - match: { _source.bytes_total: 5555 }
+  - match: { _source.bytes_total: 55550 }
 
 ---
 "Test script processor with file script":