Standardize Redact processor metadata field usage (elastic#134607)

Standardizes how the Redact processor writes its
`_ingest._redact._is_redacted` field to always create an intermediate
object in the ingest metadata. This protects users from not being able
to see the field if the processor is run in `flexible` access pattern
mode (and thus would be stored as a dotted field name by default). 

PR also moves some commonly used ingest document test code to a place
where it can be reused.

Author: jbaiera

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

@@ -26,7 +26,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
-import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 import java.util.stream.DoubleStream;
@@ -148,33 +147,7 @@ public void setTestIngestDocument() {
      * @throws Exception Any exception thrown from the provided consumer
      */
     private void doWithAccessPattern(IngestPipelineFieldAccessPattern accessPattern, Consumer<IngestDocument> action) throws Exception {
-        AtomicReference<Exception> exceptionAtomicReference = new AtomicReference<>(null);
-        document.executePipeline(
-            new Pipeline(
-                randomAlphanumericOfLength(10),
-                null,
-                null,
-                null,
-                new CompoundProcessor(new TestProcessor(action)),
-                accessPattern,
-                null,
-                null,
-                null
-            ),
-            (ignored, ex) -> {
-                if (ex != null) {
-                    if (ex instanceof IngestProcessorException ingestProcessorException) {
-                        exceptionAtomicReference.set((Exception) ingestProcessorException.getCause());
-                    } else {
-                        exceptionAtomicReference.set(ex);
-                    }
-                }
-            }
-        );
-        Exception exception = exceptionAtomicReference.get();
-        if (exception != null) {
-            throw exception;
-        }
+        IngestPipelineTestUtils.doWithAccessPattern(accessPattern, document, action);
     }
 
     /**
@@ -184,7 +157,7 @@ private void doWithAccessPattern(IngestPipelineFieldAccessPattern accessPattern,
      * @throws Exception Any exception thrown from the provided consumer
      */
     private void doWithRandomAccessPattern(Consumer<IngestDocument> action) throws Exception {
-        doWithAccessPattern(randomFrom(IngestPipelineFieldAccessPattern.values()), action);
+        IngestPipelineTestUtils.doWithRandomAccessPattern(document, action);
     }
 
     private void assertPathValid(IngestDocument doc, String path) {

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

@@ -29,8 +29,11 @@
 import org.elasticsearch.xcontent.XContentType;
 
 import java.io.IOException;
+import java.util.function.Consumer;
 
 import static org.elasticsearch.test.ESTestCase.TEST_REQUEST_TIMEOUT;
+import static org.elasticsearch.test.ESTestCase.randomAlphanumericOfLength;
+import static org.elasticsearch.test.ESTestCase.randomFrom;
 import static org.elasticsearch.test.ESTestCase.safeGet;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
 import static org.elasticsearch.xcontent.XContentFactory.jsonBuilder;
@@ -140,4 +143,86 @@ public static SimulatePipelineRequest jsonSimulatePipelineRequest(String jsonStr
     public static SimulatePipelineRequest jsonSimulatePipelineRequest(BytesReference jsonBytes) {
         return new SimulatePipelineRequest(ReleasableBytesReference.wrap(jsonBytes), XContentType.JSON);
     }
+
+    /**
+     * Executes an action against an ingest document using a random access pattern. A synthetic pipeline instance with the provided
+     * access pattern is created and executed against the ingest document, thus updating its internal access pattern.
+     * @param document The document to operate on
+     * @param action A consumer which takes the updated ingest document during execution
+     * @throws Exception Any exception thrown from the provided consumer
+     */
+    public static void doWithRandomAccessPattern(IngestDocument document, Consumer<IngestDocument> action) throws Exception {
+        doWithAccessPattern(randomFrom(IngestPipelineFieldAccessPattern.values()), document, action);
+    }
+
+    /**
+     * Executes an action against an ingest document using a random access pattern. A synthetic pipeline instance with the provided
+     * access pattern is created and executed against the ingest document, thus updating its internal access pattern.
+     * @param accessPattern The access pattern to use when executing the block of code
+     * @param document The document to operate on
+     * @param action A consumer which takes the updated ingest document during execution
+     * @throws Exception Any exception thrown from the provided consumer
+     */
+    public static void doWithAccessPattern(
+        IngestPipelineFieldAccessPattern accessPattern,
+        IngestDocument document,
+        Consumer<IngestDocument> action
+    ) throws Exception {
+        runWithAccessPattern(accessPattern, document, new TestProcessor(action));
+    }
+
+    /**
+     * Executes a processor against an ingest document using a random access pattern. A synthetic pipeline instance with the provided
+     * access pattern is created and executed against the ingest document, thus updating its internal access pattern.
+     * @param document The document to operate on
+     * @param processor A processor which takes the updated ingest document during execution
+     * @return the resulting ingest document instance
+     * @throws Exception Any exception thrown from the provided consumer
+     */
+    public static IngestDocument runWithRandomAccessPattern(IngestDocument document, Processor processor) throws Exception {
+        return runWithAccessPattern(randomFrom(IngestPipelineFieldAccessPattern.values()), document, processor);
+    }
+
+    /**
+     * Executes a processor against an ingest document using the provided access pattern. A synthetic pipeline instance with the provided
+     * access pattern is created and executed against the ingest document, thus updating its internal access pattern.
+     * @param accessPattern The access pattern to use when executing the block of code
+     * @param document The document to operate on
+     * @param processor A processor which takes the updated ingest document during execution
+     * @return the resulting ingest document instance
+     * @throws Exception Any exception thrown from the provided consumer
+     */
+    public static IngestDocument runWithAccessPattern(
+        IngestPipelineFieldAccessPattern accessPattern,
+        IngestDocument document,
+        Processor processor
+    ) throws Exception {
+        IngestDocument[] ingestDocumentHolder = new IngestDocument[1];
+        Exception[] exceptionHolder = new Exception[1];
+        document.executePipeline(
+            new Pipeline(
+                randomAlphanumericOfLength(10),
+                null,
+                null,
+                null,
+                new CompoundProcessor(processor),
+                accessPattern,
+                null,
+                null,
+                null
+            ),
+            (result, ex) -> {
+                ingestDocumentHolder[0] = result;
+                exceptionHolder[0] = ex;
+            }
+        );
+        Exception exception = exceptionHolder[0];
+        if (exception != null) {
+            if (exception instanceof IngestProcessorException ingestProcessorException) {
+                exception = ((Exception) ingestProcessorException.getCause());
+            }
+            throw exception;
+        }
+        return ingestDocumentHolder[0];
+    }
 }

x-pack/plugin/redact/src/main/java/org/elasticsearch/xpack/redact/RedactProcessor.java

@@ -31,6 +31,7 @@
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.Comparator;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
@@ -227,7 +228,10 @@ private void updateMetadataIfNecessary(IngestDocument ingestDocument, String fie
 
         // document newly redacted
         if (alreadyRedacted == false && isRedacted) {
-            ingestDocument.setFieldValue(METADATA_PATH_REDACT_IS_REDACTED, true);
+            // Set the field directly in the metadata map to avoid any access pattern related problems
+            Map<String, Object> redactObject = HashMap.newHashMap(1);
+            redactObject.put(IS_REDACTED_KEY, true);
+            ingestDocument.getIngestMetadata().put(REDACT_KEY, redactObject);
         }
     }
 

x-pack/plugin/redact/src/test/java/org/elasticsearch/xpack/redact/RedactProcessorTests.java

@@ -11,6 +11,8 @@
 import org.elasticsearch.grok.MatcherWatchdog;
 import org.elasticsearch.index.VersionType;
 import org.elasticsearch.ingest.IngestDocument;
+import org.elasticsearch.ingest.IngestPipelineFieldAccessPattern;
+import org.elasticsearch.ingest.IngestPipelineTestUtils;
 import org.elasticsearch.license.MockLicenseState;
 import org.elasticsearch.license.TestUtils;
 import org.elasticsearch.license.XPackLicenseState;
@@ -369,6 +371,27 @@ public void testTraceRedact() throws Exception {
             assertTrue(redactMetadata.containsKey(RedactProcessor.IS_REDACTED_KEY));
             assertTrue((Boolean) redactMetadata.get(RedactProcessor.IS_REDACTED_KEY));
         }
+        {
+            var processor = new RedactProcessor.Factory(mockLicenseState(), MatcherWatchdog.noop()).create(
+                null,
+                "t",
+                "d",
+                new HashMap<>(config),
+                null
+            );
+            var ingestDoc = createIngestDoc(Map.of("to_redact", "[email protected] will be redacted"));
+            var redactedDoc = IngestPipelineTestUtils.runWithAccessPattern(IngestPipelineFieldAccessPattern.FLEXIBLE, ingestDoc, processor);
+
+            assertEquals("<REDACTED> will be redacted", redactedDoc.getFieldValue("to_redact", String.class));
+            // validate ingest metadata path correctly resolved in classic mode
+            assertTrue(redactedDoc.getFieldValue(RedactProcessor.METADATA_PATH_REDACT_IS_REDACTED, Boolean.class));
+            // validate ingest metadata structure correct
+            var ingestMeta = redactedDoc.getIngestMetadata();
+            assertTrue(ingestMeta.containsKey(RedactProcessor.REDACT_KEY));
+            var redactMetadata = (HashMap<String, Object>) ingestMeta.get(RedactProcessor.REDACT_KEY);
+            assertTrue(redactMetadata.containsKey(RedactProcessor.IS_REDACTED_KEY));
+            assertTrue((Boolean) redactMetadata.get(RedactProcessor.IS_REDACTED_KEY));
+        }
         {
             var configNoTrace = new HashMap<String, Object>();
             configNoTrace.put("field", "to_redact");