Ingest Attachment: Allow to prevent base64 conversions by using raw bytes (#16601)

CBOR is natively supported in Elasticsearch and allows for byte arrays.
This means, that by using CBOR the user can prevent base64 conversions
for the data being sent back and forth.

This PR adds support to extract data from a byte array in addition to
a string. This also required to add a ByteArrayValueSource class.

core/src/main/java/org/elasticsearch/action/ingest/WriteableIngestDocument.java

@@ -24,11 +24,9 @@ import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.xcontent.ToXContent;
 import org.elasticsearch.common.xcontent.XContentBuilder;
-import org.elasticsearch.common.xcontent.XContentBuilderString;
 import org.elasticsearch.ingest.core.IngestDocument;
 
 import java.io.IOException;
-import java.util.Collections;
 import java.util.Map;
 import java.util.Objects;
 

core/src/main/java/org/elasticsearch/ingest/core/IngestDocument.java

@@ -19,6 +19,7 @@
 
 package org.elasticsearch.ingest.core;
 
+import org.elasticsearch.common.Base64;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.index.mapper.internal.IdFieldMapper;
 import org.elasticsearch.index.mapper.internal.IndexFieldMapper;
@@ -29,9 +30,11 @@ import org.elasticsearch.index.mapper.internal.TTLFieldMapper;
 import org.elasticsearch.index.mapper.internal.TimestampFieldMapper;
 import org.elasticsearch.index.mapper.internal.TypeFieldMapper;
 
+import java.io.IOException;
 import java.text.DateFormat;
 import java.text.SimpleDateFormat;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Date;
 import java.util.HashMap;
 import java.util.List;
@@ -40,6 +43,8 @@ import java.util.Map;
 import java.util.Objects;
 import java.util.TimeZone;
 
+import static java.nio.charset.StandardCharsets.UTF_8;
+
 /**
  * Represents a single document being captured before indexing and holds the source and metadata (like id, type and index).
  */
@@ -111,6 +116,31 @@ public final class IngestDocument {
         return cast(path, context, clazz);
     }
 
+    /**
+     * Returns the value contained in the document for the provided path as a byte array.
+     * If the path value is a string, a base64 decode operation will happen.
+     * If the path value is a byte array, it is just returned
+     * @param path The path within the document in dot-notation
+     * @return the byte array for the provided path if existing
+     * @throws IllegalArgumentException if the path is null, empty, invalid, if the field doesn't exist
+     * or if the field that is found at the provided path is not of the expected type.
+     */
+    public byte[] getFieldValueAsBytes(String path) {
+        Object object = getFieldValue(path, Object.class);
+        if (object instanceof byte[]) {
+            return (byte[]) object;
+        } else if (object instanceof String) {
+            try {
+                return Base64.decode(object.toString().getBytes(UTF_8));
+            } catch (IOException e) {
+                throw new IllegalArgumentException("Could not base64 decode path [ " + path + "]", e);
+            }
+        } else {
+            throw new IllegalArgumentException("Content field [" + path + "] of unknown type [" + object.getClass().getName() +
+                "], must be string or byte array");
+        }
+    }
+
     /**
      * Checks whether the document contains a value for the provided path
      * @param path The path within the document in dot-notation
@@ -490,6 +520,9 @@ public final class IngestDocument {
                 copy.add(deepCopy(itemValue));
             }
             return copy;
+        } else if (value instanceof byte[]) {
+            byte[] bytes = (byte[]) value;
+            return Arrays.copyOf(bytes, bytes.length);
         } else if (value == null || value instanceof String || value instanceof Integer ||
             value instanceof Long || value instanceof Float ||
             value instanceof Double || value instanceof Boolean) {

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

@@ -20,6 +20,7 @@
 package org.elasticsearch.ingest.core;
 
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -59,6 +60,8 @@ public interface ValueSource {
             return new ListValue(valueSourceList);
         } else if (value == null || value instanceof Number || value instanceof Boolean) {
             return new ObjectValue(value);
+        } else if (value instanceof byte[]) {
+            return new ByteValue((byte[]) value);
         } else if (value instanceof String) {
             return new TemplatedValue(templateService.compile((String) value));
         } else {
@@ -160,6 +163,35 @@ public interface ValueSource {
         }
     }
 
+    final class ByteValue implements ValueSource {
+
+        private final byte[] value;
+
+        ByteValue(byte[] value) {
+            this.value = value;
+        }
+
+        @Override
+        public Object copyAndResolve(Map<String, Object> model) {
+            return value;
+        }
+
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) return true;
+            if (o == null || getClass() != o.getClass()) return false;
+
+            ByteValue objectValue = (ByteValue) o;
+            return Arrays.equals(value, objectValue.value);
+        }
+
+        @Override
+        public int hashCode() {
+            return Arrays.hashCode(value);
+        }
+
+    }
+
     final class TemplatedValue implements ValueSource {
 
         private final TemplateService.Template template;

core/src/test/java/org/elasticsearch/action/ingest/SimulateDocumentSimpleResultTests.java

@@ -21,12 +21,13 @@ package org.elasticsearch.action.ingest;
 
 import org.elasticsearch.common.io.stream.BytesStreamOutput;
 import org.elasticsearch.common.io.stream.StreamInput;
-import org.elasticsearch.ingest.core.IngestDocument;
 import org.elasticsearch.ingest.RandomDocumentPicks;
+import org.elasticsearch.ingest.core.IngestDocument;
 import org.elasticsearch.test.ESTestCase;
 
 import java.io.IOException;
 
+import static org.elasticsearch.ingest.core.IngestDocumentTests.assertIngestDocument;
 import static org.hamcrest.CoreMatchers.equalTo;
 import static org.hamcrest.CoreMatchers.instanceOf;
 
@@ -47,11 +48,13 @@ public class SimulateDocumentSimpleResultTests extends ESTestCase {
         StreamInput streamInput = StreamInput.wrap(out.bytes());
         SimulateDocumentBaseResult otherSimulateDocumentBaseResult = new SimulateDocumentBaseResult(streamInput);
 
-        assertThat(otherSimulateDocumentBaseResult.getIngestDocument(), equalTo(simulateDocumentBaseResult.getIngestDocument()));
         if (isFailure) {
+            assertThat(otherSimulateDocumentBaseResult.getIngestDocument(), equalTo(simulateDocumentBaseResult.getIngestDocument()));
             assertThat(otherSimulateDocumentBaseResult.getFailure(), instanceOf(IllegalArgumentException.class));
             IllegalArgumentException e = (IllegalArgumentException) otherSimulateDocumentBaseResult.getFailure();
             assertThat(e.getMessage(), equalTo("test"));
+        } else {
+            assertIngestDocument(otherSimulateDocumentBaseResult.getIngestDocument(), simulateDocumentBaseResult.getIngestDocument());
         }
     }
 }

core/src/test/java/org/elasticsearch/action/ingest/SimulateExecutionServiceTests.java

@@ -23,7 +23,6 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.ingest.RandomDocumentPicks;
 import org.elasticsearch.ingest.TestProcessor;
 import org.elasticsearch.ingest.core.CompoundProcessor;
-import org.elasticsearch.ingest.core.Processor;
 import org.elasticsearch.ingest.core.IngestDocument;
 import org.elasticsearch.ingest.core.Pipeline;
 import org.elasticsearch.test.ESTestCase;
@@ -34,6 +33,7 @@ import org.junit.Before;
 import java.util.Collections;
 import java.util.Map;
 
+import static org.elasticsearch.ingest.core.IngestDocumentTests.assertIngestDocument;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.not;
@@ -44,7 +44,6 @@ public class SimulateExecutionServiceTests extends ESTestCase {
 
     private ThreadPool threadPool;
     private SimulateExecutionService executionService;
-    private Processor processor;
     private IngestDocument ingestDocument;
 
     @Before
@@ -55,7 +54,6 @@ public class SimulateExecutionServiceTests extends ESTestCase {
                         .build()
         );
         executionService = new SimulateExecutionService(threadPool);
-        processor = new TestProcessor("id", "mock", ingestDocument -> {});
         ingestDocument = RandomDocumentPicks.randomIngestDocument(random());
     }
 
@@ -73,17 +71,19 @@ public class SimulateExecutionServiceTests extends ESTestCase {
         SimulateDocumentVerboseResult simulateDocumentVerboseResult = (SimulateDocumentVerboseResult) actualItemResponse;
         assertThat(simulateDocumentVerboseResult.getProcessorResults().size(), equalTo(2));
         assertThat(simulateDocumentVerboseResult.getProcessorResults().get(0).getProcessorTag(), equalTo("test-id"));
-        assertThat(simulateDocumentVerboseResult.getProcessorResults().get(0).getIngestDocument(), not(sameInstance(ingestDocument)));
-        assertThat(simulateDocumentVerboseResult.getProcessorResults().get(0).getIngestDocument(), equalTo(ingestDocument));
-        assertThat(simulateDocumentVerboseResult.getProcessorResults().get(0).getIngestDocument().getSourceAndMetadata(), not(sameInstance(ingestDocument.getSourceAndMetadata())));
+        IngestDocument firstProcessorIngestDocument = simulateDocumentVerboseResult.getProcessorResults().get(0).getIngestDocument();
+        assertThat(firstProcessorIngestDocument, not(sameInstance(this.ingestDocument)));
+        assertIngestDocument(firstProcessorIngestDocument, this.ingestDocument);
+        assertThat(firstProcessorIngestDocument.getSourceAndMetadata(), not(sameInstance(this.ingestDocument.getSourceAndMetadata())));
 
         assertThat(simulateDocumentVerboseResult.getProcessorResults().get(0).getFailure(), nullValue());
         assertThat(simulateDocumentVerboseResult.getProcessorResults().get(1).getProcessorTag(), equalTo("test-id"));
-        assertThat(simulateDocumentVerboseResult.getProcessorResults().get(1).getIngestDocument(), not(sameInstance(ingestDocument)));
-        assertThat(simulateDocumentVerboseResult.getProcessorResults().get(1).getIngestDocument(), equalTo(ingestDocument));
-        assertThat(simulateDocumentVerboseResult.getProcessorResults().get(1).getIngestDocument().getSourceAndMetadata(), not(sameInstance(ingestDocument.getSourceAndMetadata())));
-        assertThat(simulateDocumentVerboseResult.getProcessorResults().get(1).getIngestDocument().getSourceAndMetadata(),
-            not(sameInstance(simulateDocumentVerboseResult.getProcessorResults().get(0).getIngestDocument().getSourceAndMetadata())));
+        IngestDocument secondProcessorIngestDocument = simulateDocumentVerboseResult.getProcessorResults().get(1).getIngestDocument();
+        assertThat(secondProcessorIngestDocument, not(sameInstance(this.ingestDocument)));
+        assertIngestDocument(secondProcessorIngestDocument, this.ingestDocument);
+        assertThat(secondProcessorIngestDocument.getSourceAndMetadata(), not(sameInstance(this.ingestDocument.getSourceAndMetadata())));
+        assertThat(secondProcessorIngestDocument.getSourceAndMetadata(),
+            not(sameInstance(firstProcessorIngestDocument.getSourceAndMetadata())));
         assertThat(simulateDocumentVerboseResult.getProcessorResults().get(1).getFailure(), nullValue());
     }
 
@@ -113,7 +113,7 @@ public class SimulateExecutionServiceTests extends ESTestCase {
         assertThat(simulateDocumentVerboseResult.getProcessorResults().get(0).getProcessorTag(), equalTo("processor_0"));
         assertThat(simulateDocumentVerboseResult.getProcessorResults().get(0).getFailure(), nullValue());
         assertThat(simulateDocumentVerboseResult.getProcessorResults().get(0).getIngestDocument(), not(sameInstance(ingestDocument)));
-        assertThat(simulateDocumentVerboseResult.getProcessorResults().get(0).getIngestDocument(), equalTo(ingestDocument));
+        assertIngestDocument(simulateDocumentVerboseResult.getProcessorResults().get(0).getIngestDocument(), ingestDocument);
         assertThat(simulateDocumentVerboseResult.getProcessorResults().get(0).getIngestDocument().getSourceAndMetadata(), not(sameInstance(ingestDocument.getSourceAndMetadata())));
         assertThat(simulateDocumentVerboseResult.getProcessorResults().get(1).getProcessorTag(), equalTo("processor_1"));
         assertThat(simulateDocumentVerboseResult.getProcessorResults().get(1).getIngestDocument(), nullValue());
@@ -148,13 +148,13 @@ public class SimulateExecutionServiceTests extends ESTestCase {
         metadata.put(CompoundProcessor.ON_FAILURE_PROCESSOR_TYPE_FIELD, "mock");
         metadata.put(CompoundProcessor.ON_FAILURE_PROCESSOR_TAG_FIELD, "processor_0");
         metadata.put(CompoundProcessor.ON_FAILURE_MESSAGE_FIELD, "processor failed");
-        assertThat(simulateDocumentVerboseResult.getProcessorResults().get(1).getIngestDocument(), equalTo(ingestDocumentWithOnFailureMetadata));
+        assertIngestDocument(simulateDocumentVerboseResult.getProcessorResults().get(1).getIngestDocument(), ingestDocumentWithOnFailureMetadata);
 
         assertThat(simulateDocumentVerboseResult.getProcessorResults().get(1).getFailure(), nullValue());
 
         assertThat(simulateDocumentVerboseResult.getProcessorResults().get(2).getProcessorTag(), equalTo("processor_2"));
         assertThat(simulateDocumentVerboseResult.getProcessorResults().get(2).getIngestDocument(), not(sameInstance(ingestDocument)));
-        assertThat(simulateDocumentVerboseResult.getProcessorResults().get(2).getIngestDocument(), equalTo(ingestDocument));
+        assertIngestDocument(simulateDocumentVerboseResult.getProcessorResults().get(2).getIngestDocument(), ingestDocument);
         assertThat(simulateDocumentVerboseResult.getProcessorResults().get(2).getFailure(), nullValue());
     }
 

core/src/test/java/org/elasticsearch/action/ingest/SimulatePipelineResponseTests.java

@@ -30,6 +30,7 @@ import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
 
+import static org.elasticsearch.ingest.core.IngestDocumentTests.assertIngestDocument;
 import static org.hamcrest.CoreMatchers.equalTo;
 import static org.hamcrest.CoreMatchers.instanceOf;
 import static org.hamcrest.CoreMatchers.nullValue;
@@ -90,7 +91,9 @@ public class SimulatePipelineResponseTests extends ESTestCase {
                 for (SimulateProcessorResult simulateProcessorResult : simulateDocumentVerboseResult.getProcessorResults()) {
                     SimulateProcessorResult expectedProcessorResult = expectedProcessorResultIterator.next();
                     assertThat(simulateProcessorResult.getProcessorTag(), equalTo(expectedProcessorResult.getProcessorTag()));
-                    assertThat(simulateProcessorResult.getIngestDocument(), equalTo(expectedProcessorResult.getIngestDocument()));
+                    if (simulateProcessorResult.getIngestDocument() != null) {
+                        assertIngestDocument(simulateProcessorResult.getIngestDocument(), expectedProcessorResult.getIngestDocument());
+                    }
                     if (expectedProcessorResult.getFailure() == null) {
                         assertThat(simulateProcessorResult.getFailure(), nullValue());
                     } else {
@@ -103,7 +106,9 @@ public class SimulatePipelineResponseTests extends ESTestCase {
                 SimulateDocumentBaseResult expectedSimulateDocumentBaseResult = (SimulateDocumentBaseResult) expectedResultIterator.next();
                 assertThat(result, instanceOf(SimulateDocumentBaseResult.class));
                 SimulateDocumentBaseResult simulateDocumentBaseResult = (SimulateDocumentBaseResult) result;
-                assertThat(simulateDocumentBaseResult.getIngestDocument(), equalTo(expectedSimulateDocumentBaseResult.getIngestDocument()));
+                if (simulateDocumentBaseResult.getIngestDocument() != null) {
+                    assertIngestDocument(simulateDocumentBaseResult.getIngestDocument(), expectedSimulateDocumentBaseResult.getIngestDocument());
+                }
                 if (expectedSimulateDocumentBaseResult.getFailure() == null) {
                     assertThat(simulateDocumentBaseResult.getFailure(), nullValue());
                 } else {

core/src/test/java/org/elasticsearch/action/ingest/WriteableIngestDocumentTests.java

@@ -21,22 +21,23 @@ package org.elasticsearch.action.ingest;
 
 import org.elasticsearch.common.io.stream.BytesStreamOutput;
 import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.XContentFactory;
+import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.ingest.RandomDocumentPicks;
 import org.elasticsearch.ingest.core.IngestDocument;
 import org.elasticsearch.test.ESTestCase;
-import org.elasticsearch.test.XContentTestUtils;
 
 import java.io.IOException;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
 
-import static org.elasticsearch.test.XContentTestUtils.convertToMap;
-import static org.elasticsearch.test.XContentTestUtils.differenceBetweenMapsIgnoringArrayOrder;
+import static org.elasticsearch.common.xcontent.ToXContent.EMPTY_PARAMS;
+import static org.elasticsearch.ingest.core.IngestDocumentTests.assertIngestDocument;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.not;
-import static org.hamcrest.Matchers.nullValue;
 
 public class WriteableIngestDocumentTests extends ESTestCase {
 
@@ -107,14 +108,13 @@ public class WriteableIngestDocumentTests extends ESTestCase {
         for (int i = 0; i < numFields; i++) {
             ingestMetadata.put(randomAsciiOfLengthBetween(5, 10), randomAsciiOfLengthBetween(5, 10));
         }
-        Map<String, Object> document = RandomDocumentPicks.randomSource(random());
         WriteableIngestDocument writeableIngestDocument = new WriteableIngestDocument(new IngestDocument(sourceAndMetadata, ingestMetadata));
 
         BytesStreamOutput out = new BytesStreamOutput();
         writeableIngestDocument.writeTo(out);
         StreamInput streamInput = StreamInput.wrap(out.bytes());
         WriteableIngestDocument otherWriteableIngestDocument = new WriteableIngestDocument(streamInput);
-        assertThat(otherWriteableIngestDocument, equalTo(writeableIngestDocument));
+        assertIngestDocument(otherWriteableIngestDocument.getIngestDocument(), writeableIngestDocument.getIngestDocument());
     }
 
     @SuppressWarnings("unchecked")
@@ -122,7 +122,13 @@ public class WriteableIngestDocumentTests extends ESTestCase {
         IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random());
         WriteableIngestDocument writeableIngestDocument = new WriteableIngestDocument(new IngestDocument(ingestDocument));
 
-        Map<String, Object> toXContentMap = convertToMap(writeableIngestDocument);
+        // using a cbor builder here, so that byte arrays do not get converted, so equalTo() below works
+        XContentBuilder builder = XContentFactory.cborBuilder();
+        builder.startObject();
+        writeableIngestDocument.toXContent(builder, EMPTY_PARAMS);
+        builder.endObject();
+        Map<String, Object> toXContentMap = XContentHelper.convertToMap(builder.bytes(), false).v2();
+
         Map<String, Object> toXContentDoc = (Map<String, Object>) toXContentMap.get("doc");
         Map<String, Object> toXContentSource = (Map<String, Object>) toXContentDoc.get("_source");
         Map<String, String> toXContentIngestMetadata = (Map<String, String>) toXContentDoc.get("_ingest");
@@ -137,9 +143,7 @@ public class WriteableIngestDocumentTests extends ESTestCase {
             }
         }
 
-        String sourceDiff = differenceBetweenMapsIgnoringArrayOrder(toXContentSource, ingestDocument.getSourceAndMetadata());
-        assertThat(sourceDiff, is(nullValue()));
-
-        assertThat(toXContentIngestMetadata, equalTo(ingestDocument.getIngestMetadata()));
+        IngestDocument serializedIngestDocument = new IngestDocument(toXContentSource, toXContentIngestMetadata);
+        assertThat(serializedIngestDocument, equalTo(serializedIngestDocument));
     }
 }

core/src/test/java/org/elasticsearch/ingest/core/IngestDocumentTests.java

@@ -19,6 +19,7 @@
 
 package org.elasticsearch.ingest.core;
 
+import org.elasticsearch.action.ingest.WriteableIngestDocumentTests;
 import org.elasticsearch.ingest.RandomDocumentPicks;
 import org.elasticsearch.test.ESTestCase;
 import org.junit.Before;
@@ -35,6 +36,7 @@ import java.util.Locale;
 import java.util.Map;
 
 import static org.hamcrest.Matchers.both;
+import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.endsWith;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThanOrEqualTo;
@@ -85,7 +87,8 @@ public class IngestDocumentTests extends ESTestCase {
         assertThat(ingestDocument.getFieldValue("_index", String.class), equalTo("index"));
         assertThat(ingestDocument.getFieldValue("_type", String.class), equalTo("type"));
         assertThat(ingestDocument.getFieldValue("_id", String.class), equalTo("id"));
-        assertThat(ingestDocument.getFieldValue("_ingest.timestamp", String.class), both(notNullValue()).and(not(equalTo("bogus_timestamp"))));
+        assertThat(ingestDocument.getFieldValue("_ingest.timestamp", String.class),
+                both(notNullValue()).and(not(equalTo("bogus_timestamp"))));
         assertThat(ingestDocument.getFieldValue("_source._ingest.timestamp", String.class), equalTo("bogus_timestamp"));
     }
 
@@ -106,14 +109,14 @@ public class IngestDocumentTests extends ESTestCase {
         try {
             ingestDocument.getFieldValue("_source.", Object.class);
             fail("get field value should have failed");
-        } catch(IllegalArgumentException e) {
+        } catch (IllegalArgumentException e) {
             assertThat(e.getMessage(), equalTo("path [_source.] is not valid"));
         }
 
         try {
             ingestDocument.getFieldValue("_ingest.", Object.class);
             fail("get field value should have failed");
-        } catch(IllegalArgumentException e) {
+        } catch (IllegalArgumentException e) {
             assertThat(e.getMessage(), equalTo("path [_ingest.] is not valid"));
         }
     }
@@ -126,14 +129,14 @@ public class IngestDocumentTests extends ESTestCase {
         try {
             ingestDocument.getFieldValue("int", String.class);
             fail("getFieldValue should have failed");
-        } catch(IllegalArgumentException e) {
+        } catch (IllegalArgumentException e) {
             assertThat(e.getMessage(), equalTo("field [int] of type [java.lang.Integer] cannot be cast to [java.lang.String]"));
         }
 
         try {
             ingestDocument.getFieldValue("foo", Integer.class);
             fail("getFieldValue should have failed");
-        } catch(IllegalArgumentException e) {
+        } catch (IllegalArgumentException e) {
             assertThat(e.getMessage(), equalTo("field [foo] of type [java.lang.String] cannot be cast to [java.lang.Integer]"));
         }
     }
@@ -146,8 +149,9 @@ public class IngestDocumentTests extends ESTestCase {
     public void testNestedGetFieldValueTypeMismatch() {
         try {
             ingestDocument.getFieldValue("foo.foo.bar", String.class);
-        } catch(IllegalArgumentException e) {
-            assertThat(e.getMessage(), equalTo("cannot resolve [foo] from object of type [java.lang.String] as part of path [foo.foo.bar]"));
+        } catch (IllegalArgumentException e) {
+            assertThat(e.getMessage(),
+                    equalTo("cannot resolve [foo] from object of type [java.lang.String] as part of path [foo.foo.bar]"));
         }
     }
 
@@ -162,7 +166,7 @@ public class IngestDocumentTests extends ESTestCase {
     public void testListGetFieldValueIndexNotNumeric() {
         try {
             ingestDocument.getFieldValue("list.test.field", String.class);
-        } catch(IllegalArgumentException e) {
+        } catch (IllegalArgumentException e) {
             assertThat(e.getMessage(), equalTo("[test] is not an integer, cannot be used as an index as part of path [list.test.field]"));
         }
     }
@@ -170,7 +174,7 @@ public class IngestDocumentTests extends ESTestCase {
     public void testListGetFieldValueIndexOutOfBounds() {
         try {
             ingestDocument.getFieldValue("list.10.field", String.class);
-        } catch(IllegalArgumentException e) {
+        } catch (IllegalArgumentException e) {
             assertThat(e.getMessage(), equalTo("[10] is out of bounds for array with length [2] as part of path [list.10.field]"));
         }
     }
@@ -179,7 +183,7 @@ public class IngestDocumentTests extends ESTestCase {
         try {
             ingestDocument.getFieldValue("not.here", String.class);
             fail("get field value should have failed");
-        } catch(IllegalArgumentException e) {
+        } catch (IllegalArgumentException e) {
             assertThat(e.getMessage(), equalTo("field [not] not present as part of path [not.here]"));
         }
     }
@@ -188,7 +192,7 @@ public class IngestDocumentTests extends ESTestCase {
         try {
             ingestDocument.getFieldValue("fizz.foo_null.not_there", String.class);
             fail("get field value should have failed");
-        } catch(IllegalArgumentException e) {
+        } catch (IllegalArgumentException e) {
             assertThat(e.getMessage(), equalTo("cannot resolve [not_there] from null as part of path [fizz.foo_null.not_there]"));
         }
     }
@@ -197,7 +201,7 @@ public class IngestDocumentTests extends ESTestCase {
         try {
             ingestDocument.getFieldValue(null, String.class);
             fail("get field value should have failed");
-        } catch(IllegalArgumentException e) {
+        } catch (IllegalArgumentException e) {
             assertThat(e.getMessage(), equalTo("path cannot be null nor empty"));
         }
     }
@@ -206,7 +210,7 @@ public class IngestDocumentTests extends ESTestCase {
         try {
             ingestDocument.getFieldValue("", String.class);
             fail("get field value should have failed");
-        } catch(IllegalArgumentException e) {
+        } catch (IllegalArgumentException e) {
             assertThat(e.getMessage(), equalTo("path cannot be null nor empty"));
         }
     }
@@ -261,7 +265,7 @@ public class IngestDocumentTests extends ESTestCase {
         try {
             ingestDocument.hasField(null);
             fail("has field should have failed");
-        } catch(IllegalArgumentException e) {
+        } catch (IllegalArgumentException e) {
             assertThat(e.getMessage(), equalTo("path cannot be null nor empty"));
         }
     }
@@ -274,7 +278,7 @@ public class IngestDocumentTests extends ESTestCase {
         try {
             ingestDocument.hasField("");
             fail("has field should have failed");
-        } catch(IllegalArgumentException e) {
+        } catch (IllegalArgumentException e) {
             assertThat(e.getMessage(), equalTo("path cannot be null nor empty"));
         }
     }
@@ -291,14 +295,14 @@ public class IngestDocumentTests extends ESTestCase {
         try {
             ingestDocument.hasField("_source.");
             fail("has field value should have failed");
-        } catch(IllegalArgumentException e) {
+        } catch (IllegalArgumentException e) {
             assertThat(e.getMessage(), equalTo("path [_source.] is not valid"));
         }
 
         try {
             ingestDocument.hasField("_ingest.");
             fail("has field value should have failed");
-        } catch(IllegalArgumentException e) {
+        } catch (IllegalArgumentException e) {
             assertThat(e.getMessage(), equalTo("path [_ingest.] is not valid"));
         }
     }
@@ -356,8 +360,9 @@ public class IngestDocumentTests extends ESTestCase {
         try {
             ingestDocument.setFieldValue("fizz.buzz.new", "bar");
             fail("add field should have failed");
-        } catch(IllegalArgumentException e) {
-            assertThat(e.getMessage(), equalTo("cannot set [new] with parent object of type [java.lang.String] as part of path [fizz.buzz.new]"));
+        } catch (IllegalArgumentException e) {
+            assertThat(e.getMessage(),
+                    equalTo("cannot set [new] with parent object of type [java.lang.String] as part of path [fizz.buzz.new]"));
         }
     }
 
@@ -365,7 +370,7 @@ public class IngestDocumentTests extends ESTestCase {
         try {
             ingestDocument.setFieldValue("fizz.foo_null.test", "bar");
             fail("add field should have failed");
-        } catch(IllegalArgumentException e) {
+        } catch (IllegalArgumentException e) {
             assertThat(e.getMessage(), equalTo("cannot set [test] with null parent as part of path [fizz.foo_null.test]"));
         }
     }
@@ -374,7 +379,7 @@ public class IngestDocumentTests extends ESTestCase {
         try {
             ingestDocument.setFieldValue(null, "bar");
             fail("add field should have failed");
-        } catch(IllegalArgumentException e) {
+        } catch (IllegalArgumentException e) {
             assertThat(e.getMessage(), equalTo("path cannot be null nor empty"));
         }
     }
@@ -399,14 +404,14 @@ public class IngestDocumentTests extends ESTestCase {
         try {
             ingestDocument.setFieldValue("_source.", "value");
             fail("set field value should have failed");
-        } catch(IllegalArgumentException e) {
+        } catch (IllegalArgumentException e) {
             assertThat(e.getMessage(), equalTo("path [_source.] is not valid"));
         }
 
         try {
             ingestDocument.setFieldValue("_ingest.", "_value");
             fail("set field value should have failed");
-        } catch(IllegalArgumentException e) {
+        } catch (IllegalArgumentException e) {
             assertThat(e.getMessage(), equalTo("path [_ingest.] is not valid"));
         }
     }
@@ -703,13 +708,13 @@ public class IngestDocumentTests extends ESTestCase {
     public void testListSetFieldValueIndexNotNumeric() {
         try {
             ingestDocument.setFieldValue("list.test", "value");
-        } catch(IllegalArgumentException e) {
+        } catch (IllegalArgumentException e) {
             assertThat(e.getMessage(), equalTo("[test] is not an integer, cannot be used as an index as part of path [list.test]"));
         }
 
         try {
             ingestDocument.setFieldValue("list.test.field", "new_value");
-        } catch(IllegalArgumentException e) {
+        } catch (IllegalArgumentException e) {
             assertThat(e.getMessage(), equalTo("[test] is not an integer, cannot be used as an index as part of path [list.test.field]"));
         }
     }
@@ -717,13 +722,13 @@ public class IngestDocumentTests extends ESTestCase {
     public void testListSetFieldValueIndexOutOfBounds() {
         try {
             ingestDocument.setFieldValue("list.10", "value");
-        } catch(IllegalArgumentException e) {
+        } catch (IllegalArgumentException e) {
             assertThat(e.getMessage(), equalTo("[10] is out of bounds for array with length [2] as part of path [list.10]"));
         }
 
         try {
             ingestDocument.setFieldValue("list.10.field", "value");
-        } catch(IllegalArgumentException e) {
+        } catch (IllegalArgumentException e) {
             assertThat(e.getMessage(), equalTo("[10] is out of bounds for array with length [2] as part of path [list.10.field]"));
         }
     }
@@ -732,7 +737,7 @@ public class IngestDocumentTests extends ESTestCase {
         try {
             ingestDocument.setFieldValue("", "bar");
             fail("add field should have failed");
-        } catch(IllegalArgumentException e) {
+        } catch (IllegalArgumentException e) {
             assertThat(e.getMessage(), equalTo("path cannot be null nor empty"));
         }
     }
@@ -782,7 +787,7 @@ public class IngestDocumentTests extends ESTestCase {
         try {
             ingestDocument.removeField("does_not_exist");
             fail("remove field should have failed");
-        } catch(IllegalArgumentException e) {
+        } catch (IllegalArgumentException e) {
             assertThat(e.getMessage(), equalTo("field [does_not_exist] not present as part of path [does_not_exist]"));
         }
     }
@@ -791,8 +796,9 @@ public class IngestDocumentTests extends ESTestCase {
         try {
             ingestDocument.removeField("foo.foo.bar");
             fail("remove field should have failed");
-        } catch(IllegalArgumentException e) {
-            assertThat(e.getMessage(), equalTo("cannot resolve [foo] from object of type [java.lang.String] as part of path [foo.foo.bar]"));
+        } catch (IllegalArgumentException e) {
+            assertThat(e.getMessage(),
+                    equalTo("cannot resolve [foo] from object of type [java.lang.String] as part of path [foo.foo.bar]"));
         }
     }
 
@@ -815,14 +821,14 @@ public class IngestDocumentTests extends ESTestCase {
         try {
             ingestDocument.removeField("_source.");
             fail("set field value should have failed");
-        } catch(IllegalArgumentException e) {
+        } catch (IllegalArgumentException e) {
             assertThat(e.getMessage(), equalTo("path [_source.] is not valid"));
         }
 
         try {
             ingestDocument.removeField("_ingest.");
             fail("set field value should have failed");
-        } catch(IllegalArgumentException e) {
+        } catch (IllegalArgumentException e) {
             assertThat(e.getMessage(), equalTo("path [_ingest.] is not valid"));
         }
     }
@@ -850,7 +856,7 @@ public class IngestDocumentTests extends ESTestCase {
         try {
             ingestDocument.removeField("fizz.foo_null.not_there");
             fail("get field value should have failed");
-        } catch(IllegalArgumentException e) {
+        } catch (IllegalArgumentException e) {
             assertThat(e.getMessage(), equalTo("cannot remove [not_there] from null as part of path [fizz.foo_null.not_there]"));
         }
     }
@@ -858,7 +864,7 @@ public class IngestDocumentTests extends ESTestCase {
     public void testNestedRemoveFieldTypeMismatch() {
         try {
             ingestDocument.removeField("fizz.1.bar");
-        } catch(IllegalArgumentException e) {
+        } catch (IllegalArgumentException e) {
             assertThat(e.getMessage(), equalTo("cannot remove [bar] from object of type [java.lang.String] as part of path [fizz.1.bar]"));
         }
     }
@@ -866,7 +872,7 @@ public class IngestDocumentTests extends ESTestCase {
     public void testListRemoveFieldIndexNotNumeric() {
         try {
             ingestDocument.removeField("list.test");
-        } catch(IllegalArgumentException e) {
+        } catch (IllegalArgumentException e) {
             assertThat(e.getMessage(), equalTo("[test] is not an integer, cannot be used as an index as part of path [list.test]"));
         }
     }
@@ -874,7 +880,7 @@ public class IngestDocumentTests extends ESTestCase {
     public void testListRemoveFieldIndexOutOfBounds() {
         try {
             ingestDocument.removeField("list.10");
-        } catch(IllegalArgumentException e) {
+        } catch (IllegalArgumentException e) {
             assertThat(e.getMessage(), equalTo("[10] is out of bounds for array with length [2] as part of path [list.10]"));
         }
     }
@@ -883,7 +889,7 @@ public class IngestDocumentTests extends ESTestCase {
         try {
             ingestDocument.removeField((String) null);
             fail("remove field should have failed");
-        } catch(IllegalArgumentException e) {
+        } catch (IllegalArgumentException e) {
             assertThat(e.getMessage(), equalTo("path cannot be null nor empty"));
         }
     }
@@ -892,7 +898,7 @@ public class IngestDocumentTests extends ESTestCase {
         try {
             ingestDocument.removeField("");
             fail("remove field should have failed");
-        } catch(IllegalArgumentException e) {
+        } catch (IllegalArgumentException e) {
             assertThat(e.getMessage(), equalTo("path cannot be null nor empty"));
         }
     }
@@ -946,7 +952,8 @@ public class IngestDocumentTests extends ESTestCase {
             assertThat(ingestDocument, equalTo(otherIngestDocument));
             assertThat(otherIngestDocument, equalTo(ingestDocument));
             assertThat(ingestDocument.hashCode(), equalTo(otherIngestDocument.hashCode()));
-            IngestDocument thirdIngestDocument = new IngestDocument(Collections.unmodifiableMap(sourceAndMetadata), Collections.unmodifiableMap(ingestMetadata));
+            IngestDocument thirdIngestDocument = new IngestDocument(Collections.unmodifiableMap(sourceAndMetadata),
+                    Collections.unmodifiableMap(ingestMetadata));
             assertThat(thirdIngestDocument, equalTo(ingestDocument));
             assertThat(ingestDocument, equalTo(thirdIngestDocument));
             assertThat(ingestDocument.hashCode(), equalTo(thirdIngestDocument.hashCode()));
@@ -969,18 +976,34 @@ public class IngestDocumentTests extends ESTestCase {
     public void testCopyConstructor() {
         IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random());
         IngestDocument copy = new IngestDocument(ingestDocument);
-        recursiveEqualsButNotSameCheck(ingestDocument.getSourceAndMetadata(), copy.getSourceAndMetadata());
+        assertThat(ingestDocument.getSourceAndMetadata(), not(sameInstance(copy.getSourceAndMetadata())));
+        assertIngestDocument(ingestDocument, copy);
     }
 
-    private void recursiveEqualsButNotSameCheck(Object a, Object b) {
-        assertThat(a, not(sameInstance(b)));
-        assertThat(a, equalTo(b));
+    public void testSetInvalidSourceField() throws Exception {
+        Map<String, Object> document = new HashMap<>();
+        Object randomObject = randomFrom(new ArrayList<>(), new HashMap<>(), 12, 12.34);
+        document.put("source_field", randomObject);
+
+        IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random(), document);
+        try {
+            ingestDocument.getFieldValueAsBytes("source_field");
+            fail("Expected an exception due to invalid source field, but did not happen");
+        } catch (IllegalArgumentException e) {
+            String expectedClassName = randomObject.getClass().getName();
+
+            assertThat(e.getMessage(),
+                    containsString("field [source_field] of unknown type [" + expectedClassName + "], must be string or byte array"));
+        }
+    }
+
+    public static void assertIngestDocument(Object a, Object b) {
         if (a instanceof Map) {
             Map<?, ?> mapA = (Map<?, ?>) a;
             Map<?, ?> mapB = (Map<?, ?>) b;
             for (Map.Entry<?, ?> entry : mapA.entrySet()) {
                 if (entry.getValue() instanceof List || entry.getValue() instanceof Map) {
-                    recursiveEqualsButNotSameCheck(entry.getValue(), mapB.get(entry.getKey()));
+                    assertIngestDocument(entry.getValue(), mapB.get(entry.getKey()));
                 }
             }
         } else if (a instanceof List) {
@@ -989,11 +1012,19 @@ public class IngestDocumentTests extends ESTestCase {
             for (int i = 0; i < listA.size(); i++) {
                 Object value = listA.get(i);
                 if (value instanceof List || value instanceof Map) {
-                    recursiveEqualsButNotSameCheck(value, listB.get(i));
+                    assertIngestDocument(value, listB.get(i));
                 }
             }
+        } else if (a instanceof byte[]) {
+            assertArrayEquals((byte[]) a, (byte[])b);
+        } else if (a instanceof IngestDocument) {
+            IngestDocument docA = (IngestDocument) a;
+            IngestDocument docB = (IngestDocument) b;
+            assertIngestDocument(docA.getSourceAndMetadata(), docB.getSourceAndMetadata());
+            assertIngestDocument(docA.getIngestMetadata(), docB.getIngestMetadata());
+        } else {
+            String msg = String.format(Locale.ROOT, "Expected %s class to be equal to %s", a.getClass().getName(), b.getClass().getName());
+            assertThat(msg, a, equalTo(b));
         }
-
     }
-
 }

docs/plugins/ingest-attachment.asciidoc

@@ -6,7 +6,10 @@ using the Apache text extraction library http://lucene.apache.org/tika/[Tika].
 
 You can use the ingest attachment plugin as a replacement for the mapper attachment plugin.
 
-The source field must be a base64 encoded binary.
+The source field must be a base64 encoded binary. If you do not want to incur
+the overhead of converting back and forth between base64, you can use the CBOR
+format instead of JSON and specify the field as a bytes array instead of a string
+representation. The processor will skip the base64 decoding then.
 
 [[ingest-attachment-options]]
 .Attachment options

plugins/ingest-attachment/src/main/java/org/elasticsearch/ingest/attachment/AttachmentProcessor.java

@@ -23,7 +23,6 @@ import org.apache.tika.language.LanguageIdentifier;
 import org.apache.tika.metadata.Metadata;
 import org.apache.tika.metadata.TikaCoreProperties;
 import org.elasticsearch.ElasticsearchParseException;
-import org.elasticsearch.common.Base64;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.ingest.core.AbstractProcessor;
 import org.elasticsearch.ingest.core.AbstractProcessorFactory;
@@ -38,7 +37,6 @@ import java.util.Locale;
 import java.util.Map;
 import java.util.Set;
 
-import static java.nio.charset.StandardCharsets.UTF_8;
 import static org.elasticsearch.ingest.core.ConfigurationUtils.newConfigurationException;
 import static org.elasticsearch.ingest.core.ConfigurationUtils.readIntProperty;
 import static org.elasticsearch.ingest.core.ConfigurationUtils.readOptionalList;
@@ -66,13 +64,12 @@ public final class AttachmentProcessor extends AbstractProcessor {
 
     @Override
     public void execute(IngestDocument ingestDocument) {
-        String base64Input = ingestDocument.getFieldValue(sourceField, String.class);
         Map<String, Object> additionalFields = new HashMap<>();
 
         try {
-            byte[] decodedContent = Base64.decode(base64Input.getBytes(UTF_8));
             Metadata metadata = new Metadata();
-            String parsedContent = TikaImpl.parse(decodedContent, metadata, indexedChars);
+            byte[] input = ingestDocument.getFieldValueAsBytes(sourceField);
+            String parsedContent = TikaImpl.parse(input, metadata, indexedChars);
 
             if (fields.contains(Field.CONTENT) && Strings.hasLength(parsedContent)) {
                 // somehow tika seems to append a newline at the end automatically, lets remove that again

plugins/ingest-attachment/src/test/java/org/elasticsearch/ingest/attachment/AttachmentProcessorTests.java

@@ -57,8 +57,7 @@ public class AttachmentProcessorTests extends ESTestCase {
     public void testEnglishTextDocument() throws Exception {
         Map<String, Object> attachmentData = parseDocument("text-in-english.txt", processor);
 
-        assertThat(attachmentData.keySet(), containsInAnyOrder("language", "content", "content_type",
-            "content_length"));
+        assertThat(attachmentData.keySet(), containsInAnyOrder("language", "content", "content_type", "content_length"));
         assertThat(attachmentData.get("language"), is("en"));
         assertThat(attachmentData.get("content"), is("\"God Save the Queen\" (alternatively \"God Save the King\""));
         assertThat(attachmentData.get("content_type").toString(), containsString("text/plain"));
@@ -137,8 +136,8 @@ public class AttachmentProcessorTests extends ESTestCase {
     public void testHtmlDocument() throws Exception {
         Map<String, Object> attachmentData = parseDocument("htmlWithEmptyDateMeta.html", processor);
 
-        assertThat(attachmentData.keySet(), containsInAnyOrder("language", "content", "author", "keywords", "title",
-            "content_type", "content_length"));
+        assertThat(attachmentData.keySet(), containsInAnyOrder("language", "content", "author", "keywords", "title", "content_type",
+            "content_length"));
         assertThat(attachmentData.get("language"), is("en"));
         assertThat(attachmentData.get("content"), is(notNullValue()));
         assertThat(attachmentData.get("content_length"), is(notNullValue()));
@@ -151,16 +150,15 @@ public class AttachmentProcessorTests extends ESTestCase {
     public void testXHtmlDocument() throws Exception {
         Map<String, Object> attachmentData = parseDocument("testXHTML.html", processor);
 
-        assertThat(attachmentData.keySet(), containsInAnyOrder("language", "content", "author", "title",
-            "content_type", "content_length"));
+        assertThat(attachmentData.keySet(), containsInAnyOrder("language", "content", "author", "title", "content_type", "content_length"));
         assertThat(attachmentData.get("content_type").toString(), containsString("application/xhtml+xml"));
     }
 
     public void testEpubDocument() throws Exception {
         Map<String, Object> attachmentData = parseDocument("testEPUB.epub", processor);
 
-        assertThat(attachmentData.keySet(), containsInAnyOrder("language", "content", "author", "title",
-            "content_type", "content_length", "date", "keywords"));
+        assertThat(attachmentData.keySet(), containsInAnyOrder("language", "content", "author", "title", "content_type", "content_length",
+            "date", "keywords"));
         assertThat(attachmentData.get("content_type").toString(), containsString("application/epub+zip"));
     }
 
@@ -168,11 +166,33 @@ public class AttachmentProcessorTests extends ESTestCase {
     public void testAsciidocDocument() throws Exception {
         Map<String, Object> attachmentData = parseDocument("asciidoc.asciidoc", processor);
 
-        assertThat(attachmentData.keySet(), containsInAnyOrder("language", "content_type", "content",
-            "content_length"));
+        assertThat(attachmentData.keySet(), containsInAnyOrder("language", "content_type", "content", "content_length"));
         assertThat(attachmentData.get("content_type").toString(), containsString("text/plain"));
     }
 
+    public void testParseAsBytesArray() throws Exception {
+        String path = "/org/elasticsearch/ingest/attachment/test/sample-files/text-in-english.txt";
+        byte[] bytes;
+        try (InputStream is = AttachmentProcessorTests.class.getResourceAsStream(path)) {
+            bytes = IOUtils.toByteArray(is);
+        }
+
+        Map<String, Object> document = new HashMap<>();
+        document.put("source_field", bytes);
+
+        IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random(), document);
+        processor.execute(ingestDocument);
+
+        @SuppressWarnings("unchecked")
+        Map<String, Object> attachmentData = (Map<String, Object>) ingestDocument.getSourceAndMetadata().get("target_field");
+
+        assertThat(attachmentData.keySet(), containsInAnyOrder("language", "content", "content_type", "content_length"));
+        assertThat(attachmentData.get("language"), is("en"));
+        assertThat(attachmentData.get("content"), is("\"God Save the Queen\" (alternatively \"God Save the King\""));
+        assertThat(attachmentData.get("content_type").toString(), containsString("text/plain"));
+        assertThat(attachmentData.get("content_length"), is(notNullValue()));
+    }
+
     private Map<String, Object> parseDocument(String file, AttachmentProcessor processor) throws Exception {
         Map<String, Object> document = new HashMap<>();
         document.put("source_field", getAsBase64(file));
@@ -181,8 +201,7 @@ public class AttachmentProcessorTests extends ESTestCase {
         processor.execute(ingestDocument);
 
         @SuppressWarnings("unchecked")
-        Map<String, Object> attachmentData = (Map<String, Object>) ingestDocument.getSourceAndMetadata()
-            .get("target_field");
+        Map<String, Object> attachmentData = (Map<String, Object>) ingestDocument.getSourceAndMetadata().get("target_field");
         return attachmentData;
     }
 

test/framework/src/main/java/org/elasticsearch/ingest/RandomDocumentPicks.java

@@ -23,6 +23,7 @@ import com.carrotsearch.randomizedtesting.generators.RandomInts;
 import com.carrotsearch.randomizedtesting.generators.RandomPicks;
 import com.carrotsearch.randomizedtesting.generators.RandomStrings;
 import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.ingest.core.IngestDocument;
 
 import java.util.ArrayList;
@@ -171,7 +172,7 @@ public final class RandomDocumentPicks {
     }
 
     private static Object randomFieldValue(Random random, int currentDepth) {
-        switch(RandomInts.randomIntBetween(random, 0, 8)) {
+        switch(RandomInts.randomIntBetween(random, 0, 9)) {
             case 0:
                 return randomString(random);
             case 1:
@@ -212,6 +213,10 @@ public final class RandomDocumentPicks {
                 Map<String, Object> newNode = new HashMap<>();
                 addRandomFields(random, newNode, ++currentDepth);
                 return newNode;
+            case 9:
+                byte[] byteArray = new byte[RandomInts.randomIntBetween(random, 1, 1024)];
+                random.nextBytes(byteArray);
+                return byteArray;
             default:
                 throw new UnsupportedOperationException();
         }