[7.x] Configurable media type for mustache template encoding on set processor

docs/reference/ingest/processors/set.asciidoc

@@ -17,7 +17,7 @@ its value will be replaced with the provided one.
 | `copy_from` | no       | -       | The origin field which will be copied to `field`, cannot set `value` simultaneously. Supported data types are `boolean`, `number`, `array`, `object`, `string`, `date`, etc.
 | `override`  | no       | `true`  | If processor will update fields with pre-existing non-null-valued field. When set to `false`, such fields will not be touched.
 | `ignore_empty_value` | no | `false` | If `true` and `value` is a <<accessing-template-fields,template snippet>> that evaluates to `null` or the empty string, the processor quietly exits without modifying the document
-| `mime_type` | no       | `application/json` | The MIME type for encoding `value`. Applies only when `value` is a <<accessing-template-fields,template snippet>>. Must be one of `application/json`, `text/plain`, or `application/x-www-form-urlencoded`.
+| `media_type` | no       | `application/json` | The media type for encoding `value`. Applies only when `value` is a <<accessing-template-fields,template snippet>>. Must be one of `application/json`, `text/plain`, or `application/x-www-form-urlencoded`.
 include::common-options.asciidoc[]
 |======
 

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

@@ -13,6 +13,7 @@
 import org.elasticsearch.ingest.IngestDocument;
 import org.elasticsearch.ingest.Processor;
 import org.elasticsearch.ingest.ValueSource;
+import org.elasticsearch.script.Script;
 import org.elasticsearch.script.ScriptService;
 import org.elasticsearch.script.TemplateScript;
 
@@ -99,10 +100,13 @@ public SetProcessor create(Map<String, Processor.Factory> registry, String proce
                                    String description, Map<String, Object> config) throws Exception {
             String field = ConfigurationUtils.readStringProperty(TYPE, processorTag, config, "field");
             String copyFrom = ConfigurationUtils.readOptionalStringProperty(TYPE, processorTag, config, "copy_from");
+            String mediaType = ConfigurationUtils.readMediaTypeProperty(TYPE, processorTag, config, "media_type", "application/json");
             ValueSource valueSource = null;
             if (copyFrom == null) {
                 Object value = ConfigurationUtils.readObject(TYPE, processorTag, config, "value");
-                valueSource = ValueSource.wrap(value, scriptService);
+                valueSource = ValueSource.wrap(
+                    value, scriptService, org.elasticsearch.common.collect.Map.of(Script.CONTENT_TYPE_OPTION, mediaType)
+                );
             } else {
                 Object value = config.remove("value");
                 if (value != null) {
@@ -112,8 +116,7 @@ public SetProcessor create(Map<String, Processor.Factory> registry, String proce
             }
 
             boolean overrideEnabled = ConfigurationUtils.readBooleanProperty(TYPE, processorTag, config, "override", true);
-            TemplateScript.Factory compiledTemplate = ConfigurationUtils.compileTemplate(TYPE, processorTag,
-                "field", field, scriptService);
+            TemplateScript.Factory compiledTemplate = ConfigurationUtils.compileTemplate(TYPE, processorTag, "field", field, scriptService);
             boolean ignoreEmptyValue = ConfigurationUtils.readBooleanProperty(TYPE, processorTag, config, "ignore_empty_value", false);
 
             return new SetProcessor(

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

@@ -10,15 +10,18 @@
 
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.ElasticsearchParseException;
+import org.elasticsearch.ingest.ConfigurationUtils;
 import org.elasticsearch.ingest.TestTemplateService;
 import org.elasticsearch.test.ESTestCase;
 import org.junit.Before;
 
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
 
 import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.Matchers.containsString;
 
 public class SetProcessorFactoryTests extends ESTestCase {
 
@@ -122,4 +125,26 @@ public void testCreateWithCopyFromAndValue() throws Exception {
             () -> factory.create(null, processorTag, null, config));
         assertThat(exception.getMessage(), equalTo("[copy_from] cannot set both `copy_from` and `value` in the same processor"));
     }
+
+    public void testMediaType() throws Exception {
+        // valid media type
+        String expectedMediaType = randomFrom(ConfigurationUtils.VALID_MEDIA_TYPES);
+        Map<String, Object> config = new HashMap<>();
+        config.put("field", "field1");
+        config.put("value", "value1");
+        config.put("media_type", expectedMediaType);
+        String processorTag = randomAlphaOfLength(10);
+        SetProcessor setProcessor = factory.create(null, processorTag, null, config);
+        assertThat(setProcessor.getTag(), equalTo(processorTag));
+
+        // invalid media type
+        expectedMediaType = randomValueOtherThanMany(m -> Arrays.asList(ConfigurationUtils.VALID_MEDIA_TYPES).contains(m),
+            () -> randomAlphaOfLengthBetween(5, 9));
+        final Map<String, Object> config2 = new HashMap<>();
+        config2.put("field", "field1");
+        config2.put("value", "value1");
+        config2.put("media_type", expectedMediaType);
+        ElasticsearchException e = expectThrows(ElasticsearchException.class, () -> factory.create(null, processorTag, null, config2));
+        assertThat(e.getMessage(), containsString("property does not contain a supported media type [" + expectedMediaType + "]"));
+    }
 }

qa/smoke-test-ingest-with-all-dependencies/src/test/java/org/elasticsearch/ingest/ValueSourceMustacheIT.java

@@ -8,6 +8,8 @@
 
 package org.elasticsearch.ingest;
 
+import org.elasticsearch.script.Script;
+
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
@@ -61,4 +63,27 @@ public void testAccessSourceViaTemplate() {
         assertThat(ingestDocument.hasField("index"), is(false));
     }
 
+    public void testWithConfigurableEncoders() {
+        Map<String, Object> model = new HashMap<>();
+        model.put("log_line", "10.10.1.1 - - [17/Nov/2020:04:59:43 +0000] \"GET /info HTTP/1.1\" 200 6229 \"-\" \"-\"  2");
+
+        // default encoder should be application/json
+        ValueSource valueSource = ValueSource.wrap("{{log_line}}", scriptService);
+        Object result = valueSource.copyAndResolve(model);
+        assertThat(result,
+            equalTo("10.10.1.1 - - [17/Nov/2020:04:59:43 +0000] \\\"GET /info HTTP/1.1\\\" 200 6229 \\\"-\\\" \\\"-\\\"  2"));
+
+        // text/plain encoder
+        Map<String, String> scriptOptions = org.elasticsearch.common.collect.Map.of(Script.CONTENT_TYPE_OPTION, "text/plain");
+        valueSource = ValueSource.wrap("{{log_line}}", scriptService, scriptOptions);
+        result = valueSource.copyAndResolve(model);
+        assertThat(result, equalTo("10.10.1.1 - - [17/Nov/2020:04:59:43 +0000] \"GET /info HTTP/1.1\" 200 6229 \"-\" \"-\"  2"));
+
+        // application/x-www-form-urlencoded encoder
+        scriptOptions = org.elasticsearch.common.collect.Map.of(Script.CONTENT_TYPE_OPTION, "application/x-www-form-urlencoded");
+        valueSource = ValueSource.wrap("{{log_line}}", scriptService, scriptOptions);
+        result = valueSource.copyAndResolve(model);
+        assertThat(result, equalTo("10.10.1.1+-+-+%5B17%2FNov%2F2020%3A04%3A59%3A43+%2B0000%5D+%22GET+%2Finfo+HTTP%2F1.1%22+200" +
+            "+6229+%22-%22+%22-%22++2"));
+    }
 }

server/src/main/java/org/elasticsearch/ingest/ConfigurationUtils.java

@@ -38,6 +38,7 @@ public final class ConfigurationUtils {
 
     public static final String TAG_KEY = "tag";
     public static final String DESCRIPTION_KEY = "description";
+    public static final String[] VALID_MEDIA_TYPES = {"application/json", "text/plain", "application/x-www-form-urlencoded"};
 
     private ConfigurationUtils() {
     }
@@ -294,6 +295,18 @@ public static Object readObject(String processorType, String processorTag, Map<S
         return value;
     }
 
+    public static String readMediaTypeProperty(String processorType, String processorTag, Map<String, Object> configuration,
+        String propertyName, String defaultValue) {
+        String mediaType = readStringProperty(processorType, processorTag, configuration, propertyName, defaultValue);
+
+        if (Arrays.asList(VALID_MEDIA_TYPES).contains(mediaType) == false) {
+            throw newConfigurationException(processorType, processorTag, propertyName,
+                "property does not contain a supported media type [" + mediaType + "]");
+        }
+
+        return mediaType;
+    }
+
     public static ElasticsearchException newConfigurationException(String processorType, String processorTag,
                                                                         String propertyName, String reason) {
         String msg;

server/src/main/java/org/elasticsearch/ingest/ValueSource.java

@@ -15,7 +15,6 @@
 
 import java.util.ArrayList;
 import java.util.Arrays;
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -39,21 +38,25 @@ public interface ValueSource {
     Object copyAndResolve(Map<String, Object> model);
 
     static ValueSource wrap(Object value, ScriptService scriptService) {
+        return wrap(value, scriptService, org.elasticsearch.common.collect.Map.of());
+    }
+
+    static ValueSource wrap(Object value, ScriptService scriptService, Map<String, String> scriptOptions) {
 
         if (value instanceof Map) {
             @SuppressWarnings("unchecked")
             Map<Object, Object> mapValue = (Map) value;
             Map<ValueSource, ValueSource> valueTypeMap = new HashMap<>(mapValue.size());
             for (Map.Entry<Object, Object> entry : mapValue.entrySet()) {
-                valueTypeMap.put(wrap(entry.getKey(), scriptService), wrap(entry.getValue(), scriptService));
+                valueTypeMap.put(wrap(entry.getKey(), scriptService, scriptOptions), wrap(entry.getValue(), scriptService, scriptOptions));
             }
             return new MapValue(valueTypeMap);
         } else if (value instanceof List) {
             @SuppressWarnings("unchecked")
             List<Object> listValue = (List) value;
             List<ValueSource> valueSourceList = new ArrayList<>(listValue.size());
             for (Object item : listValue) {
-                valueSourceList.add(wrap(item, scriptService));
+                valueSourceList.add(wrap(item, scriptService, scriptOptions));
             }
             return new ListValue(valueSourceList);
         } else if (value == null || value instanceof Number || value instanceof Boolean) {
@@ -65,7 +68,9 @@ static ValueSource wrap(Object value, ScriptService scriptService) {
             // installed for use by REST tests. `value` will not be
             // modified if templating is not available
             if (scriptService.isLangSupported(DEFAULT_TEMPLATE_LANG) && ((String) value).contains("{{")) {
-                Script script = new Script(ScriptType.INLINE, DEFAULT_TEMPLATE_LANG, (String) value, Collections.emptyMap());
+                Script script = new Script(
+                    ScriptType.INLINE, DEFAULT_TEMPLATE_LANG, (String) value, scriptOptions, org.elasticsearch.common.collect.Map.of()
+                );
                 return new TemplatedValue(scriptService.compile(script, TemplateScript.CONTEXT));
             } else {
                 return new ObjectValue(value);

server/src/test/java/org/elasticsearch/ingest/ConfigurationUtilsTests.java

@@ -8,6 +8,7 @@
 
 package org.elasticsearch.ingest;
 
+import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.ElasticsearchParseException;
 import org.elasticsearch.script.ScriptService;
 import org.elasticsearch.script.TemplateScript;
@@ -21,6 +22,7 @@
 import java.util.List;
 import java.util.Map;
 
+import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.is;
@@ -103,6 +105,38 @@ public void testReadStringOrIntPropertyInvalidType() {
         }
     }
 
+    public void testReadMediaProperty() {
+        // valid media type
+        String expectedMediaType = randomFrom(ConfigurationUtils.VALID_MEDIA_TYPES);
+        config.put("media_type", expectedMediaType);
+        String readMediaType = ConfigurationUtils.readMediaTypeProperty(null, null, config, "media_type", "");
+        assertThat(readMediaType, equalTo(expectedMediaType));
+
+        // missing media type with valid default
+        expectedMediaType = randomFrom(ConfigurationUtils.VALID_MEDIA_TYPES);
+        config.remove("media_type");
+        readMediaType = ConfigurationUtils.readMediaTypeProperty(null, null, config, "media_type", expectedMediaType);
+        assertThat(readMediaType, equalTo(expectedMediaType));
+
+        // invalid media type
+        expectedMediaType = randomValueOtherThanMany(m -> Arrays.asList(ConfigurationUtils.VALID_MEDIA_TYPES).contains(m),
+            () -> randomAlphaOfLengthBetween(5, 9));
+        config.put("media_type", expectedMediaType);
+        ElasticsearchException e = expectThrows(ElasticsearchException.class,
+            () -> ConfigurationUtils.readMediaTypeProperty(null, null, config, "media_type", ""));
+        assertThat(e.getMessage(), containsString("property does not contain a supported media type [" + expectedMediaType + "]"));
+
+        // missing media type with invalid default
+        final String invalidDefaultMediaType = randomValueOtherThanMany(
+            m -> Arrays.asList(ConfigurationUtils.VALID_MEDIA_TYPES).contains(m),
+            () -> randomAlphaOfLengthBetween(5, 9)
+        );
+        config.remove("media_type");
+        e = expectThrows(ElasticsearchException.class,
+            () -> ConfigurationUtils.readMediaTypeProperty(null, null, config, "media_type", invalidDefaultMediaType));
+        assertThat(e.getMessage(), containsString("property does not contain a supported media type [" + invalidDefaultMediaType + "]"));
+    }
+
     public void testReadProcessors() throws Exception {
         Processor processor = mock(Processor.class);
         Map<String, Processor.Factory> registry =