Enables the ability to inject serialized json fields into root of document. (#22179)

The JSON processor has an optional field called "target_field".
If you don't specify target_field then target_field becomes what you specified as "field".
There isn't anyway to add the fields to the root of a document. By
setting `add_to_root`, now serialized fields will be inserted into the
top-level fields of the ingest document.

Closes #21898.

docs/reference/ingest/ingest-node.asciidoc

@@ -1473,6 +1473,7 @@ Converts a JSON string into a structured JSON object.
 | Name           | Required  | Default  | Description
 | `field`        | yes       | -        | The field to be parsed
 | `target_field` | no        | `field`  | The field to insert the converted structured object into
+| `add_to_root`  | no        | false    | Flag that forces the serialized json to be injected into the top level of the document. `target_field` must not be set when this option is chosen.
 |======
 
 [source,js]

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

@@ -28,6 +28,8 @@ import org.elasticsearch.ingest.Processor;
 
 import java.util.Map;
 
+import static org.elasticsearch.ingest.ConfigurationUtils.newConfigurationException;
+
 /**
  * Processor that serializes a string-valued field into a
  * map of maps.
@@ -38,11 +40,13 @@ public final class JsonProcessor extends AbstractProcessor {
 
     private final String field;
     private final String targetField;
+    private final boolean addToRoot;
 
-    JsonProcessor(String tag, String field, String targetField) {
+    JsonProcessor(String tag, String field, String targetField, boolean addToRoot) {
         super(tag);
         this.field = field;
         this.targetField = targetField;
+        this.addToRoot = addToRoot;
     }
 
     public String getField() {
@@ -53,12 +57,22 @@ public final class JsonProcessor extends AbstractProcessor {
         return targetField;
     }
 
+    boolean isAddToRoot() {
+        return addToRoot;
+    }
+
     @Override
     public void execute(IngestDocument document) throws Exception {
         String stringValue = document.getFieldValue(field, String.class);
         try {
             Map<String, Object> mapValue = JsonXContent.jsonXContent.createParser(stringValue).map();
-            document.setFieldValue(targetField, mapValue);
+            if (addToRoot) {
+                for (Map.Entry<String, Object> entry : mapValue.entrySet()) {
+                    document.setFieldValue(entry.getKey(), entry.getValue());
+                }
+            } else {
+                document.setFieldValue(targetField, mapValue);
+            }
         } catch (JsonParseException e) {
             throw new IllegalArgumentException(e);
         }
@@ -74,8 +88,19 @@ public final class JsonProcessor extends AbstractProcessor {
         public JsonProcessor create(Map<String, Processor.Factory> registry, String processorTag,
                                     Map<String, Object> config) throws Exception {
             String field = ConfigurationUtils.readStringProperty(TYPE, processorTag, config, "field");
-            String targetField = ConfigurationUtils.readStringProperty(TYPE, processorTag, config, "target_field", field);
+            String targetField = ConfigurationUtils.readOptionalStringProperty(TYPE, processorTag, config, "target_field");
-            return new JsonProcessor(processorTag, field, targetField);
+            boolean addToRoot = ConfigurationUtils.readBooleanProperty(TYPE, processorTag, config, "add_to_root", false);
+
+            if (addToRoot && targetField != null) {
+                throw newConfigurationException(TYPE, processorTag, "target_field",
+                    "Cannot set a target field while also setting `add_to_root` to true");
+            }
+
+            if (targetField == null) {
+                targetField = field;
+            }
+
+            return new JsonProcessor(processorTag, field, targetField, addToRoot);
         }
     }
 }

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

@@ -48,6 +48,19 @@ public class JsonProcessorFactoryTests extends ESTestCase {
         assertThat(jsonProcessor.getTargetField(), equalTo(randomTargetField));
     }
 
+    public void testCreateWithAddToRoot() throws Exception {
+        String processorTag = randomAsciiOfLength(10);
+        String randomField = randomAsciiOfLength(10);
+        Map<String, Object> config = new HashMap<>();
+        config.put("field", randomField);
+        config.put("add_to_root", true);
+        JsonProcessor jsonProcessor = FACTORY.create(null, processorTag, config);
+        assertThat(jsonProcessor.getTag(), equalTo(processorTag));
+        assertThat(jsonProcessor.getField(), equalTo(randomField));
+        assertThat(jsonProcessor.getTargetField(), equalTo(randomField));
+        assertTrue(jsonProcessor.isAddToRoot());
+    }
+
     public void testCreateWithDefaultTarget() throws Exception {
         String processorTag = randomAsciiOfLength(10);
         String randomField = randomAsciiOfLength(10);
@@ -66,4 +79,16 @@ public class JsonProcessorFactoryTests extends ESTestCase {
             () -> FACTORY.create(null, processorTag, config));
         assertThat(exception.getMessage(), equalTo("[field] required property is missing"));
     }
+
+    public void testCreateWithBothTargetFieldAndAddToRoot() throws Exception {
+        String randomField = randomAsciiOfLength(10);
+        String randomTargetField = randomAsciiOfLength(5);
+        Map<String, Object> config = new HashMap<>();
+        config.put("field", randomField);
+        config.put("target_field", randomTargetField);
+        config.put("add_to_root", true);
+        ElasticsearchException exception = expectThrows(ElasticsearchParseException.class,
+            () -> FACTORY.create(null, randomAsciiOfLength(10), config));
+        assertThat(exception.getMessage(), equalTo("[target_field] Cannot set a target field while also setting `add_to_root` to true"));
+    }
 }

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

@@ -39,7 +39,7 @@ public class JsonProcessorTests extends ESTestCase {
         String processorTag = randomAsciiOfLength(3);
         String randomField = randomAsciiOfLength(3);
         String randomTargetField = randomAsciiOfLength(2);
-        JsonProcessor jsonProcessor = new JsonProcessor(processorTag, randomField, randomTargetField);
+        JsonProcessor jsonProcessor = new JsonProcessor(processorTag, randomField, randomTargetField, false);
         Map<String, Object> document = new HashMap<>();
 
         Map<String, Object> randomJsonMap = RandomDocumentPicks.randomSource(random());
@@ -54,7 +54,7 @@ public class JsonProcessorTests extends ESTestCase {
     }
 
     public void testInvalidJson() {
-        JsonProcessor jsonProcessor = new JsonProcessor("tag", "field", "target_field");
+        JsonProcessor jsonProcessor = new JsonProcessor("tag", "field", "target_field", false);
         Map<String, Object> document = new HashMap<>();
         document.put("field", "invalid json");
         IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random(), document);
@@ -66,11 +66,34 @@ public class JsonProcessorTests extends ESTestCase {
     }
 
     public void testFieldMissing() {
-        JsonProcessor jsonProcessor = new JsonProcessor("tag", "field", "target_field");
+        JsonProcessor jsonProcessor = new JsonProcessor("tag", "field", "target_field", false);
         Map<String, Object> document = new HashMap<>();
         IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random(), document);
 
         Exception exception = expectThrows(IllegalArgumentException.class, () -> jsonProcessor.execute(ingestDocument));
         assertThat(exception.getMessage(), equalTo("field [field] not present as part of path [field]"));
     }
+
+    @SuppressWarnings("unchecked")
+    public void testAddToRoot() throws Exception {
+        String processorTag = randomAsciiOfLength(3);
+        String randomTargetField = randomAsciiOfLength(2);
+        JsonProcessor jsonProcessor = new JsonProcessor(processorTag, "a", randomTargetField, true);
+        Map<String, Object> document = new HashMap<>();
+
+        String json = "{\"a\": 1, \"b\": 2}";
+        document.put("a", json);
+        document.put("c", "see");
+
+        IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random(), document);
+        jsonProcessor.execute(ingestDocument);
+
+        Map<String, Object> expected = new HashMap<>();
+        expected.put("a", 1);
+        expected.put("b", 2);
+        expected.put("c", "see");
+        IngestDocument expectedIngestDocument = RandomDocumentPicks.randomIngestDocument(random(), expected);
+
+        assertIngestDocument(ingestDocument, expectedIngestDocument);
+    }
 }