Configurable MIME type for mustache template encoding on set processor (#65314)

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

@@ -24,6 +24,7 @@ import org.elasticsearch.ingest.ConfigurationUtils;
 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;
 
@@ -110,10 +111,11 @@ public final class SetProcessor extends AbstractProcessor {
                                    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 mimeType = ConfigurationUtils.readMimeTypeProperty(TYPE, processorTag, config, "mime_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, Map.of(Script.CONTENT_TYPE_OPTION, mimeType));
             } else {
                 Object value = config.remove("value");
                 if (value != null) {
@@ -123,8 +125,7 @@ public final class SetProcessor extends AbstractProcessor {
             }
 
             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

@@ -21,15 +21,18 @@ package org.elasticsearch.ingest.common;
 
 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 {
 
@@ -133,4 +136,26 @@ public class SetProcessorFactoryTests extends ESTestCase {
             () -> 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 testMimeType() throws Exception {
+        // valid mime type
+        String expectedMimeType = randomFrom(ConfigurationUtils.VALID_MIME_TYPES);
+        Map<String, Object> config = new HashMap<>();
+        config.put("field", "field1");
+        config.put("value", "value1");
+        config.put("mime_type", expectedMimeType);
+        String processorTag = randomAlphaOfLength(10);
+        SetProcessor setProcessor = factory.create(null, processorTag, null, config);
+        assertThat(setProcessor.getTag(), equalTo(processorTag));
+
+        // invalid mime type
+        expectedMimeType = randomValueOtherThanMany(m -> Arrays.asList(ConfigurationUtils.VALID_MIME_TYPES).contains(m),
+            () -> randomAlphaOfLengthBetween(5, 9));
+        final Map<String, Object> config2 = new HashMap<>();
+        config2.put("field", "field1");
+        config2.put("value", "value1");
+        config2.put("mime_type", expectedMimeType);
+        ElasticsearchException e = expectThrows(ElasticsearchException.class, () -> factory.create(null, processorTag, null, config2));
+        assertThat(e.getMessage(), containsString("property does not contain a supported MIME type [" + expectedMimeType + "]"));
+    }
 }

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

@@ -19,6 +19,8 @@
 
 package org.elasticsearch.ingest;
 
+import org.elasticsearch.script.Script;
+
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
@@ -72,4 +74,27 @@ public class ValueSourceMustacheIT extends AbstractScriptTestCase {
         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
+        var scriptOptions = 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 = 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

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

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

@@ -26,7 +26,6 @@ import org.elasticsearch.script.TemplateScript;
 
 import java.util.ArrayList;
 import java.util.Arrays;
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -50,13 +49,17 @@ public interface ValueSource {
     Object copyAndResolve(Map<String, Object> model);
 
     static ValueSource wrap(Object value, ScriptService scriptService) {
+        return wrap(value, scriptService, 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) {
@@ -64,7 +67,7 @@ public interface ValueSource {
             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) {
@@ -76,7 +79,7 @@ public interface ValueSource {
             // 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, Map.of());
                 return new TemplatedValue(scriptService.compile(script, TemplateScript.CONTEXT));
             } else {
                 return new ObjectValue(value);

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

@@ -19,6 +19,7 @@
 
 package org.elasticsearch.ingest;
 
+import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.ElasticsearchParseException;
 import org.elasticsearch.script.ScriptService;
 import org.elasticsearch.script.TemplateScript;
@@ -32,6 +33,7 @@ import java.util.HashMap;
 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;
@@ -114,6 +116,36 @@ public class ConfigurationUtilsTests extends ESTestCase {
         }
     }
 
+    public void testReadMimeProperty() {
+        // valid mime type
+        String expectedMimeType = randomFrom(ConfigurationUtils.VALID_MIME_TYPES);
+        config.put("mime_type", expectedMimeType);
+        String readMimeType = ConfigurationUtils.readMimeTypeProperty(null, null, config, "mime_type", "");
+        assertThat(readMimeType, equalTo(expectedMimeType));
+
+        // missing mime type with valid default
+        expectedMimeType = randomFrom(ConfigurationUtils.VALID_MIME_TYPES);
+        config.remove("mime_type");
+        readMimeType = ConfigurationUtils.readMimeTypeProperty(null, null, config, "mime_type", expectedMimeType);
+        assertThat(readMimeType, equalTo(expectedMimeType));
+
+        // invalid mime type
+        expectedMimeType = randomValueOtherThanMany(m -> Arrays.asList(ConfigurationUtils.VALID_MIME_TYPES).contains(m),
+            () -> randomAlphaOfLengthBetween(5, 9));
+        config.put("mime_type", expectedMimeType);
+        ElasticsearchException e = expectThrows(ElasticsearchException.class,
+            () -> ConfigurationUtils.readMimeTypeProperty(null, null, config, "mime_type", ""));
+        assertThat(e.getMessage(), containsString("property does not contain a supported MIME type [" + expectedMimeType + "]"));
+
+        // missing mime type with invalid default
+        final String invalidDefaultMimeType = randomValueOtherThanMany(m -> Arrays.asList(ConfigurationUtils.VALID_MIME_TYPES).contains(m),
+            () -> randomAlphaOfLengthBetween(5, 9));
+        config.remove("mime_type");
+        e = expectThrows(ElasticsearchException.class,
+            () -> ConfigurationUtils.readMimeTypeProperty(null, null, config, "mime_type", invalidDefaultMimeType));
+        assertThat(e.getMessage(), containsString("property does not contain a supported MIME type [" + invalidDefaultMimeType + "]"));
+    }
+
     public void testReadProcessors() throws Exception {
         Processor processor = mock(Processor.class);
         Map<String, Processor.Factory> registry =