Simulate API: Return 400 on invalid processor(s)

e.g. missing processor or invalid grok pattern

Closes elastic#120731

Author: szybia

modules/ingest-common/build.gradle

@@ -28,7 +28,7 @@ dependencies {
 
 restResources {
   restApi {
-    include '_common', 'ingest', 'cluster', 'indices', 'index', 'bulk', 'nodes', 'get', 'update', 'cat', 'mget', 'search'
+    include '_common', 'ingest', 'cluster', 'indices', 'index', 'bulk', 'nodes', 'get', 'update', 'cat', 'mget', 'search', 'simulate'
   }
 }
 

modules/ingest-common/src/yamlRestTest/resources/rest-api-spec/test/ingest/120_grok.yml

@@ -154,3 +154,45 @@ teardown:
       ingest.processor_grok: {}
   - length: { patterns: 318 }
   - match: { patterns.PATH: "(?:%{UNIXPATH}|%{WINPATH})" }
+
+
+---
+"Test simulate with invalid GROK pattern":
+  - skip:
+        features: headers
+  - do:
+      catch: bad_request
+      headers:
+        Content-Type: application/json
+      simulate.ingest:
+        pipeline: "invalid-grok"
+        body: >
+          {
+            "docs": [
+              {
+                "_index": "index-1",
+                "_source": {
+                  "foo": "bar"
+                }
+              }
+            ],
+            "pipeline_substitutions": {
+              "invalid-grok": {
+                "description": "invalid grok pattern",
+                "processors": [
+                  {
+                    "grok": {
+                      "field": "field",
+                      "patterns": [
+                        "%{INVALID_PATTERN:field}"
+                      ]
+                    }
+                  }
+                ]
+              }
+            }
+          }
+  - match: { status: 400 }
+  - match: { error.reason: "[patterns] Invalid regex pattern found in: [%{INVALID_PATTERN:field}]. Unable to find pattern [INVALID_PATTERN] in Grok's pattern dictionary" }
+  - match: { error.property_name: "patterns" }
+  - match: { error.processor_type: "grok" }

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

@@ -628,7 +628,7 @@ setup:
               default_pipeline: "my-pipeline"
 
   - do:
-      catch: "request"
+      catch: bad_request
       headers:
         Content-Type: application/json
       simulate.ingest:
@@ -661,7 +661,8 @@ setup:
               }
             }
           }
-  - match: { status: 500 }
+  - match: { status: 400 }
+  - match: { error.reason: "No processor type exists with name [non-existent-processor]" }
 
 ---
 "Test index in path":

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

@@ -9,6 +9,7 @@
 
 package org.elasticsearch.ingest;
 
+import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.action.bulk.BulkRequest;
 import org.elasticsearch.action.bulk.SimulateBulkRequest;
 import org.elasticsearch.cluster.metadata.ProjectId;
@@ -28,6 +29,8 @@ public SimulateIngestService(IngestService ingestService, BulkRequest request) {
         if (request instanceof SimulateBulkRequest simulateBulkRequest) {
             try {
                 pipelineSubstitutions = getPipelineSubstitutions(simulateBulkRequest.getPipelineSubstitutions(), ingestService);
+            } catch (ElasticsearchException elasticEx) {
+                throw elasticEx;
             } catch (Exception e) {
                 throw new RuntimeException(e);
             }

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

@@ -9,6 +9,7 @@
 
 package org.elasticsearch.ingest;
 
+import org.elasticsearch.ElasticsearchParseException;
 import org.elasticsearch.action.bulk.FailureStoreMetrics;
 import org.elasticsearch.action.bulk.SimulateBulkRequest;
 import org.elasticsearch.client.internal.Client;
@@ -34,6 +35,8 @@
 import static org.elasticsearch.test.LambdaMatchers.transformedMatch;
 import static org.hamcrest.Matchers.contains;
 import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.is;
+import static org.junit.Assert.assertThrows;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
@@ -46,6 +49,10 @@ private static <K, V> Map<K, V> newHashMap(K key, V value) {
         return map;
     }
 
+    private static <K, V> Map<K, V> emptyMutableHashMap() {
+        return new HashMap<>(0);
+    }
+
     public void testGetPipeline() {
         PipelineConfiguration pipelineConfiguration = new PipelineConfiguration("pipeline1", new BytesArray("""
             {"processors": [{"processor1" : {}}]}"""), XContentType.JSON);
@@ -118,6 +125,31 @@ public void testGetPipeline() {
         }
     }
 
+    public void testRethrowingOfElasticParseExceptionFromProcessors() {
+        final PipelineConfiguration pipelineConfiguration = new PipelineConfiguration("pipeline1", new BytesArray("""
+            {"processors": []}"""), XContentType.JSON);
+        final IngestMetadata ingestMetadata = new IngestMetadata(Map.of("pipeline1", pipelineConfiguration));
+        final Processor.Factory factoryThatThrowsElasticParseException = (factory, tag, description, config, projectId) -> {
+            throw new ElasticsearchParseException("house: it's never lupus");
+        };
+        final Map<String, Processor.Factory> processors = Map.of("parse_exception_processor", factoryThatThrowsElasticParseException);
+        final var projectId = randomProjectIdOrDefault();
+        IngestService ingestService = createWithProcessors(projectId, processors);
+        ingestService.innerUpdatePipelines(projectId, ingestMetadata);
+        SimulateBulkRequest simulateBulkRequest = new SimulateBulkRequest(
+            newHashMap("pipeline1", newHashMap("processors", List.of(Map.of("parse_exception_processor", emptyMutableHashMap())))),
+            Map.of(),
+            Map.of(),
+            Map.of()
+        );
+
+        final ElasticsearchParseException ex = assertThrows(
+            ElasticsearchParseException.class,
+            () -> new SimulateIngestService(ingestService, simulateBulkRequest)
+        );
+        assertThat(ex.getMessage(), is("house: it's never lupus"));
+    }
+
     private static IngestService createWithProcessors(ProjectId projectId, Map<String, Processor.Factory> processors) {
         Client client = mock(Client.class);
         ThreadPool threadPool = mock(ThreadPool.class);