[Bugfix] Better exception handling in search pipelines (#7735)

* [Bugfix] Better exception handling in search pipelines Thanks to @noCharger for reporting a failing negative test case. Since we were rethrowing exceptions when resolving search pipelines and processing search requests, that could end up killing the listener thread. Also, we want to make sure that any exception thrown from search pipelines are wrapped in SearchPipelineProcessingException. Signed-off-by: Michael Froh <froh@amazon.com> * Add changelog entry and unit tests Signed-off-by: Michael Froh <froh@amazon.com> * Add check on error message for negative test Signed-off-by: Michael Froh <froh@amazon.com> * Fix misleading commment in test Signed-off-by: Michael Froh <froh@amazon.com> --------- Signed-off-by: Michael Froh <froh@amazon.com> (cherry picked from commit cf02b96) Signed-off-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>
opensearch-project · May 25, 2023 · 0e4683a · 0e4683a
1 parent f8b4734
commit 0e4683a
Show file tree

Hide file tree

Showing 5 changed files with 97 additions and 5 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -68,6 +68,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Add more index blocks check for resize APIs ([#6774](https://github.com/opensearch-project/OpenSearch/pull/6774))
 - Replaces ZipInputStream with ZipFile to fix Zip Slip vulnerability ([#7230](https://github.com/opensearch-project/OpenSearch/pull/7230))
 - Add missing validation/parsing of SearchBackpressureMode of SearchBackpressureSettings ([#7541](https://github.com/opensearch-project/OpenSearch/pull/7541))
+- [Search Pipelines] Better exception handling in search pipelines ([#7735](https://github.com/opensearch-project/OpenSearch/pull/7735))
 
 ### Security
 

diff --git a/...-common/src/yamlRestTest/resources/rest-api-spec/test/search_pipeline/30_filter_query.yml b/...-common/src/yamlRestTest/resources/rest-api-spec/test/search_pipeline/30_filter_query.yml
@@ -122,3 +122,27 @@ teardown:
         index: test
         body: { }
   - match: { hits.total.value: 2 }
+---
+"Test invalid inline query":
+  - do:
+      catch: bad_request
+      search:
+        index: test
+        body: {
+          search_pipeline: {
+            "request_processors": [
+              {
+                "filter_query": {
+                  "query": {
+                    "woozlewuzzle": {
+                      "field": "foo"
+                    }
+                  }
+                }
+              }
+            ]
+          }
+        }
+  - match: { status: 400 }
+  - match: { error.type: "parsing_exception"}
+  - match: { error.reason: "unknown query [woozlewuzzle]"}
diff --git a/server/src/main/java/org/opensearch/action/search/TransportSearchAction.java b/server/src/main/java/org/opensearch/action/search/TransportSearchAction.java
@@ -401,7 +401,7 @@ private void executeRequest(
             );
         } catch (Exception e) {
             originalListener.onFailure(e);
-            throw new RuntimeException(e);
+            return;
         }
 
         ActionListener<SearchSourceBuilder> rewriteListener = ActionListener.wrap(source -> {

diff --git a/server/src/main/java/org/opensearch/search/pipeline/SearchPipelineService.java b/server/src/main/java/org/opensearch/search/pipeline/SearchPipelineService.java
@@ -400,8 +400,12 @@ public PipelinedRequest resolvePipeline(SearchRequest searchRequest) throws Exce
                 pipeline = pipelineHolder.pipeline;
             }
         }
-        SearchRequest transformedRequest = pipeline.transformRequest(searchRequest);
-        return new PipelinedRequest(pipeline, transformedRequest);
+        try {
+            SearchRequest transformedRequest = pipeline.transformRequest(searchRequest);
+            return new PipelinedRequest(pipeline, transformedRequest);
+        } catch (Exception e) {
+            throw new SearchPipelineProcessingException(e);
+        }
     }
 
     Map<String, Processor.Factory<SearchRequestProcessor>> getRequestProcessorFactories() {

diff --git a/server/src/test/java/org/opensearch/search/pipeline/SearchPipelineServiceTests.java b/server/src/test/java/org/opensearch/search/pipeline/SearchPipelineServiceTests.java
@@ -680,12 +680,12 @@ public void testInlinePipeline() throws Exception {
         requestProcessorConfig.put("scale", 2);
         Map<String, Object> requestProcessorObject = new HashMap<>();
         requestProcessorObject.put("scale_request_size", requestProcessorConfig);
-        pipelineSourceMap.put("request_processors", List.of(requestProcessorObject));
+        pipelineSourceMap.put(Pipeline.REQUEST_PROCESSORS_KEY, List.of(requestProcessorObject));
         Map<String, Object> responseProcessorConfig = new HashMap<>();
         responseProcessorConfig.put("score", 2);
         Map<String, Object> responseProcessorObject = new HashMap<>();
         responseProcessorObject.put("fixed_score", responseProcessorConfig);
-        pipelineSourceMap.put("response_processors", List.of(responseProcessorObject));
+        pipelineSourceMap.put(Pipeline.RESPONSE_PROCESSORS_KEY, List.of(responseProcessorObject));
 
         SearchSourceBuilder sourceBuilder = SearchSourceBuilder.searchSource().size(100).searchPipelineSource(pipelineSourceMap);
         SearchRequest searchRequest = new SearchRequest().source(sourceBuilder);
@@ -723,4 +723,67 @@ public void testInfo() {
         assertTrue(info.containsProcessor(Pipeline.REQUEST_PROCESSORS_KEY, "scale_request_size"));
         assertTrue(info.containsProcessor(Pipeline.RESPONSE_PROCESSORS_KEY, "fixed_score"));
     }
+
+    public void testExceptionOnPipelineCreation() {
+        Map<String, Processor.Factory<SearchRequestProcessor>> badFactory = Map.of(
+            "bad_factory",
+            (pf, t, f, c) -> { throw new RuntimeException(); }
+        );
+        SearchPipelineService searchPipelineService = createWithProcessors(badFactory, Collections.emptyMap());
+
+        Map<String, Object> pipelineSourceMap = new HashMap<>();
+        pipelineSourceMap.put(Pipeline.REQUEST_PROCESSORS_KEY, List.of(Map.of("bad_factory", Collections.emptyMap())));
+
+        SearchSourceBuilder sourceBuilder = SearchSourceBuilder.searchSource().searchPipelineSource(pipelineSourceMap);
+        SearchRequest searchRequest = new SearchRequest().source(sourceBuilder);
+
+        // Exception thrown when creating the pipeline
+        expectThrows(SearchPipelineProcessingException.class, () -> searchPipelineService.resolvePipeline(searchRequest));
+
+    }
+
+    public void testExceptionOnRequestProcessing() {
+        SearchRequestProcessor throwingRequestProcessor = new FakeRequestProcessor("throwing_request", null, null, r -> {
+            throw new RuntimeException();
+        });
+        Map<String, Processor.Factory<SearchRequestProcessor>> throwingRequestProcessorFactory = Map.of(
+            "throwing_request",
+            (pf, t, f, c) -> throwingRequestProcessor
+        );
+
+        SearchPipelineService searchPipelineService = createWithProcessors(throwingRequestProcessorFactory, Collections.emptyMap());
+
+        Map<String, Object> pipelineSourceMap = new HashMap<>();
+        pipelineSourceMap.put(Pipeline.REQUEST_PROCESSORS_KEY, List.of(Map.of("throwing_request", Collections.emptyMap())));
+
+        SearchSourceBuilder sourceBuilder = SearchSourceBuilder.searchSource().searchPipelineSource(pipelineSourceMap);
+        SearchRequest searchRequest = new SearchRequest().source(sourceBuilder);
+
+        // Exception thrown when processing the request
+        expectThrows(SearchPipelineProcessingException.class, () -> searchPipelineService.resolvePipeline(searchRequest));
+    }
+
+    public void testExceptionOnResponseProcessing() throws Exception {
+        SearchResponseProcessor throwingResponseProcessor = new FakeResponseProcessor("throwing_response", null, null, r -> {
+            throw new RuntimeException();
+        });
+        Map<String, Processor.Factory<SearchResponseProcessor>> throwingResponseProcessorFactory = Map.of(
+            "throwing_response",
+            (pf, t, f, c) -> throwingResponseProcessor
+        );
+
+        SearchPipelineService searchPipelineService = createWithProcessors(Collections.emptyMap(), throwingResponseProcessorFactory);
+
+        Map<String, Object> pipelineSourceMap = new HashMap<>();
+        pipelineSourceMap.put(Pipeline.RESPONSE_PROCESSORS_KEY, List.of(Map.of("throwing_response", Collections.emptyMap())));
+
+        SearchSourceBuilder sourceBuilder = SearchSourceBuilder.searchSource().size(100).searchPipelineSource(pipelineSourceMap);
+        SearchRequest searchRequest = new SearchRequest().source(sourceBuilder);
+
+        PipelinedRequest pipelinedRequest = searchPipelineService.resolvePipeline(searchRequest);
+
+        SearchResponse response = new SearchResponse(null, null, 0, 0, 0, 0, null, null);
+        // Exception thrown when processing response
+        expectThrows(SearchPipelineProcessingException.class, () -> pipelinedRequest.transformResponse(response));
+    }
 }