Block storing invalid pipeline configs caused by nested processors

rest-api-spec/src/yamlRestTest/resources/rest-api-spec/test/ingest/10_basic.yml

@@ -153,6 +153,35 @@
             "invalid_field" : {}
           }
 
+---
+"Test valid config with sequential processors":
+  - do:
+      ingest.put_pipeline:
+        id: "my_pipeline"
+        body:  >
+          {
+            "description": "_description",
+            "processors": [{}, {}, {}]
+          }
+  - match: { acknowledged: true }
+
+---
+"Test invalid config with nested processors":
+  - skip:
+      version: " - 8.6.99"
+      reason: "bug fixed in 8.7.0"
+  - do:
+      catch: /illegal_argument_exception/
+      ingest.put_pipeline:
+        id: "my_pipeline"
+        body: >
+          {
+            "description": "_description",
+            "processors": [{"grok": {}}, {"grok": {}, "date": {}}]
+          }
+  - match: { error.reason: "[processors] contains nested objects but should be a list of single-entry objects" }
+
+
 ---
 "Test Get Summarized Pipelines":
   - skip:

server/src/main/java/org/elasticsearch/action/ingest/SimulatePipelineRequest.java

@@ -36,6 +36,8 @@
 import java.util.Objects;
 
 public class SimulatePipelineRequest extends ActionRequest implements ToXContentObject {
+    public static final String PROCESSORS_KEY = "processors";
+
     private static final DeprecationLogger deprecationLogger = DeprecationLogger.getLogger(SimulatePipelineRequest.class);
     private String id;
     private boolean verbose;
@@ -147,6 +149,12 @@ static Parsed parseWithPipelineId(
     static Parsed parse(Map<String, Object> config, boolean verbose, IngestService ingestService, RestApiVersion restApiVersion)
         throws Exception {
         Map<String, Object> pipelineConfig = ConfigurationUtils.readMap(null, null, config, Fields.PIPELINE);
+
+        // check for nested objects in processor configs
+        if (Pipeline.containsNestedProcessors(pipelineConfig)) {
+            throw new IllegalArgumentException("[processors] contains nested objects but should be a list of single-entry objects");
+        }
+
         Pipeline pipeline = Pipeline.create(
             SIMULATED_PIPELINE_ID,
             pipelineConfig,

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

@@ -68,6 +68,15 @@ public Pipeline(
         this.relativeTimeProvider = relativeTimeProvider;
     }
 
+    public static boolean containsNestedProcessors(Map<String, Object> config) {
+        boolean result;
+
+        List<Map<String, Object>> processorConfigs = ConfigurationUtils.readList(null, null, config, PROCESSORS_KEY);
+        result = processorConfigs.stream().anyMatch(processor -> processor.keySet().size() > 1);
+        config.put(PROCESSORS_KEY, processorConfigs);
+        return result;
+    }
+
     public static Pipeline create(
         String id,
         Map<String, Object> config,

server/src/main/java/org/elasticsearch/rest/action/ingest/RestPutPipelineAction.java

@@ -11,7 +11,9 @@
 import org.elasticsearch.action.ingest.PutPipelineRequest;
 import org.elasticsearch.client.internal.node.NodeClient;
 import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.core.Tuple;
+import org.elasticsearch.ingest.Pipeline;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.rest.action.RestToXContentListener;
@@ -20,10 +22,12 @@
 import java.io.IOException;
 import java.util.List;
 import java.util.Locale;
+import java.util.Map;
 
 import static org.elasticsearch.rest.RestRequest.Method.PUT;
 
 public class RestPutPipelineAction extends BaseRestHandler {
+    public static final String PROCESSORS_KEY = "processors";
 
     @Override
     public List<Route> routes() {
@@ -51,6 +55,13 @@ public RestChannelConsumer prepareRequest(RestRequest restRequest, NodeClient cl
 
         Tuple<XContentType, BytesReference> sourceTuple = restRequest.contentOrSourceParam();
         PutPipelineRequest request = new PutPipelineRequest(restRequest.param("id"), sourceTuple.v2(), sourceTuple.v1(), ifVersion);
+
+        // check for nested processors in the config
+        Map<String, Object> config = XContentHelper.convertToMap(request.getSource(), false, request.getXContentType()).v2();
+        if (Pipeline.containsNestedProcessors(config)) {
+            throw new IllegalArgumentException("[processors] contains nested objects but should be a list of single-entry objects");
+        }
+
         request.masterNodeTimeout(restRequest.paramAsTime("master_timeout", request.masterNodeTimeout()));
         request.timeout(restRequest.paramAsTime("timeout", request.timeout()));
         return channel -> client.admin().cluster().putPipeline(request, new RestToXContentListener<>(channel));

server/src/test/java/org/elasticsearch/action/ingest/SimulatePipelineRequestParsingTests.java

@@ -211,6 +211,38 @@ public void testParseWithProvidedPipeline() throws Exception {
         assertThat(actualRequest.pipeline().getProcessors().size(), equalTo(numProcessors));
     }
 
+    public void testParseWithNestedProcessors() throws Exception {
+        Map<String, Object> requestContent = new HashMap<>();
+        Map<String, Object> pipelineConfig = new HashMap<>();
+        List<Map<String, Object>> processors = new ArrayList<>();
+        Map<String, Object> processorConfig = new HashMap<>();
+        List<Map<String, Object>> docs = new ArrayList<>();
+        requestContent.put(Fields.DOCS, docs);
+        String fieldName = randomAlphaOfLengthBetween(1, 10);
+        String fieldValue = randomAlphaOfLengthBetween(1, 10);
+        Map<String, Object> doc = new HashMap<>();
+        doc.put(Fields.SOURCE, Collections.singletonMap(fieldName, fieldValue));
+        docs.add(doc);
+
+        // first processor
+        processors.add(Collections.singletonMap("mock_processor", new HashMap<>()));
+
+        // second processor, which is a grok and a date in the same object
+        processorConfig.put("mock_grok", new HashMap<>());
+        processorConfig.put("mock_date", new HashMap<>());
+        processors.add(processorConfig);
+
+        pipelineConfig.put("processors", processors);
+        requestContent.put(Fields.PIPELINE, pipelineConfig);
+
+        // parse and expect failure
+        Exception e = expectThrows(
+            IllegalArgumentException.class,
+            () -> SimulatePipelineRequest.parse(requestContent, false, ingestService, RestApiVersion.current())
+        );
+        assertThat(e.getMessage(), equalTo("[processors] contains nested objects but should be a list of single-entry objects"));
+    }
+
     public void testNullPipelineId() {
         Map<String, Object> requestContent = new HashMap<>();
         List<Map<String, Object>> docs = new ArrayList<>();

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

@@ -53,6 +53,32 @@ public void testCreate() throws Exception {
         assertThat(pipeline.getProcessors().get(1).getTag(), nullValue());
     }
 
+    public void testContainsNestedProcessors() throws Exception {
+        // create a pipeline with non-nested processors
+        Map<String, Object> processorConfig0 = new HashMap<>();
+        Map<String, Object> processorConfig1 = new HashMap<>();
+        processorConfig0.put(ConfigurationUtils.TAG_KEY, "first-processor");
+        Map<String, Object> pipelineConfig = new HashMap<>();
+        pipelineConfig.put(Pipeline.DESCRIPTION_KEY, "_description");
+        pipelineConfig.put(Pipeline.VERSION_KEY, versionString);
+        if (metadata != null) {
+            pipelineConfig.put(Pipeline.META_KEY, metadata);
+        }
+        pipelineConfig.put(Pipeline.PROCESSORS_KEY, List.of(Map.of("test", processorConfig0), Map.of("test", processorConfig1)));
+
+        assertFalse(Pipeline.containsNestedProcessors(pipelineConfig));
+
+        // now add one with nested processors
+        Map<String, Object> processorConfigNested = new HashMap<>();
+        processorConfigNested.put("mock_grok", new HashMap<>());
+        processorConfigNested.put("mock_date", new HashMap<>());
+        pipelineConfig.put(
+            Pipeline.PROCESSORS_KEY,
+            List.of(Map.of("test", processorConfig0), Map.of("test", processorConfig1), processorConfigNested)
+        );
+        assertTrue(Pipeline.containsNestedProcessors(pipelineConfig));
+    }
+
     public void testCreateWithNoProcessorsField() throws Exception {
         Map<String, Object> pipelineConfig = new HashMap<>();
         pipelineConfig.put(Pipeline.DESCRIPTION_KEY, "_description");

server/src/test/java/org/elasticsearch/rest/action/ingest/RestPutPipelineActionTests.java

@@ -46,7 +46,7 @@ public void testInvalidIfVersionValue() {
 
         RestRequest request = new FakeRestRequest.Builder(xContentRegistry()).withMethod(RestRequest.Method.PUT)
             .withPath("/_ingest/pipeline/my_pipeline")
-            .withContent(new BytesArray("{\"processors\":{}}"), XContentType.JSON)
+            .withContent(new BytesArray("{\"processors\":[]}"), XContentType.JSON)
             .withParams(params)
             .build();
 
@@ -60,7 +60,7 @@ public void testMissingIfVersionValue() {
 
         RestRequest request = new FakeRestRequest.Builder(xContentRegistry()).withMethod(RestRequest.Method.PUT)
             .withPath("/_ingest/pipeline/my_pipeline")
-            .withContent(new BytesArray("{\"processors\":{}}"), XContentType.JSON)
+            .withContent(new BytesArray("{\"processors\":[]}"), XContentType.JSON)
             .withParams(params)
             .build();
 
@@ -94,7 +94,7 @@ public void testNumericIfVersionValue() {
 
         RestRequest request = new FakeRestRequest.Builder(xContentRegistry()).withMethod(RestRequest.Method.PUT)
             .withPath("/_ingest/pipeline/my_pipeline")
-            .withContent(new BytesArray("{\"processors\":{}}"), XContentType.JSON)
+            .withContent(new BytesArray("{\"processors\":[]}"), XContentType.JSON)
             .withParams(params)
             .build();