Fixed wrong failure calcualtions and related tests

Signed-off-by: Martin Gaievski <[email protected]>

Author: martin-gaievski

server/src/main/java/org/opensearch/search/pipeline/Pipeline.java

@@ -290,12 +290,14 @@ <Result extends SearchPhaseResult> void runSearchPhaseResultsTransformer(
     ) throws SearchPipelineProcessingException {
         long pipelineStart = relativeTimeSupplier.getAsLong();
         beforeTransformPhaseResults();
+        boolean hasMatchingProcessor = false;
         try {
             for (SearchPhaseResultsProcessor searchPhaseResultsProcessor : searchPhaseResultsProcessors) {
                 beforePhaseResultsProcessor(searchPhaseResultsProcessor);
                 long start = relativeTimeSupplier.getAsLong();
                 if (currentPhase.equals(searchPhaseResultsProcessor.getBeforePhase().getName())
                     && nextPhase.equals(searchPhaseResultsProcessor.getAfterPhase().getName())) {
+                    hasMatchingProcessor = true;
                     try {
                         searchPhaseResultsProcessor.process(searchPhaseResult, context, requestContext);
                         long took = TimeUnit.NANOSECONDS.toMillis(relativeTimeSupplier.getAsLong() - start);
@@ -314,14 +316,15 @@ <Result extends SearchPhaseResult> void runSearchPhaseResultsTransformer(
                                 e
                             );
                         } else {
+                            // Only track pipeline-level failure when processor failure is not ignored
+                            onTransformPhaseResultsFailure();
                             throw e;
                         }
                     }
 
                 }
             }
-        } catch (RuntimeException e) {
-            onTransformPhaseResultsFailure();
+        } catch (Exception e) {
             throw new SearchPipelineProcessingException(e);
         } finally {
             long took = TimeUnit.NANOSECONDS.toMillis(relativeTimeSupplier.getAsLong() - pipelineStart);

server/src/main/java/org/opensearch/search/pipeline/PipelineWithMetrics.java

@@ -249,18 +249,24 @@ protected void onResponseProcessorFailed(Processor processor) {
     protected void beforeTransformPhaseResults() {
         super.beforeTransformPhaseResults();
         totalPhaseResultsMetrics.before();
+        // Phase results processing is also tracked as part of individual pipeline request processing
+        pipelineRequestMetrics.before();
     }
 
     @Override
     protected void afterTransformPhaseResults(long timeInNanos) {
         super.afterTransformPhaseResults(timeInNanos);
         totalPhaseResultsMetrics.after(timeInNanos);
+        // Phase results processing is also tracked as part of individual pipeline request processing
+        pipelineRequestMetrics.after(timeInNanos);
     }
 
     @Override
     protected void onTransformPhaseResultsFailure() {
         super.onTransformPhaseResultsFailure();
         totalPhaseResultsMetrics.failed();
+        // Phase results processing failures are also tracked as part of individual pipeline request processing
+        pipelineRequestMetrics.failed();
     }
 
     protected void beforePhaseResultsProcessor(Processor processor) {

server/src/main/java/org/opensearch/search/pipeline/SearchPipelineStats.java

@@ -37,24 +37,28 @@ public class SearchPipelineStats implements Writeable, ToXContentFragment {
 
     private final OperationStats totalRequestStats;
     private final OperationStats totalResponseStats;
+    private final OperationStats totalPhaseResultsStats;
     private final List<PerPipelineStats> perPipelineStats;
     private final Map<String, PipelineDetailStats> perPipelineProcessorStats;
 
     public SearchPipelineStats(
         OperationStats totalRequestStats,
         OperationStats totalResponseStats,
+        OperationStats totalPhaseResultsStats,
         List<PerPipelineStats> perPipelineStats,
         Map<String, PipelineDetailStats> perPipelineProcessorStats
     ) {
         this.totalRequestStats = totalRequestStats;
         this.totalResponseStats = totalResponseStats;
+        this.totalPhaseResultsStats = totalPhaseResultsStats;
         this.perPipelineStats = perPipelineStats;
         this.perPipelineProcessorStats = perPipelineProcessorStats;
     }
 
     public SearchPipelineStats(StreamInput in) throws IOException {
         this.totalRequestStats = new OperationStats(in);
         this.totalResponseStats = new OperationStats(in);
+        this.totalPhaseResultsStats = new OperationStats(in);
         int size = in.readVInt();
         List<PerPipelineStats> perPipelineStats = new ArrayList<>(size);
         Map<String, PipelineDetailStats> pipelineDetailStatsMap = new TreeMap<>();
@@ -152,6 +156,7 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
     public void writeTo(StreamOutput out) throws IOException {
         totalRequestStats.writeTo(out);
         totalResponseStats.writeTo(out);
+        totalPhaseResultsStats.writeTo(out);
         out.writeVInt(perPipelineStats.size());
         for (PerPipelineStats pipelineStat : perPipelineStats) {
             out.writeString(pipelineStat.pipelineId);
@@ -259,6 +264,7 @@ SearchPipelineStats build() {
             return new SearchPipelineStats(
                 totalRequestStats,
                 totalResponseStats,
+                totalPhaseResultsStats,
                 unmodifiableList(perPipelineStats),
                 unmodifiableMap(pipelineDetailStatsMap)
             );
@@ -401,6 +407,10 @@ OperationStats getTotalResponseStats() {
         return totalResponseStats;
     }
 
+    OperationStats getTotalPhaseResultsStats() {
+        return totalPhaseResultsStats;
+    }
+
     List<PerPipelineStats> getPipelineStats() {
         return perPipelineStats;
     }

server/src/test/java/org/opensearch/search/pipeline/SearchPipelineServiceTests.java

@@ -1315,8 +1315,9 @@ public void testStats() throws Exception {
         );
 
         SearchPipelineStats stats = searchPipelineService.stats();
-        assertPipelineStats(stats.getTotalRequestStats(), 4, 2);
+        assertPipelineStats(stats.getTotalRequestStats(), 2, 1);
         assertPipelineStats(stats.getTotalResponseStats(), 2, 1);
+        assertPipelineStats(stats.getTotalPhaseResultsStats(), 2, 1);
         for (SearchPipelineStats.PerPipelineStats perPipelineStats : stats.getPipelineStats()) {
             SearchPipelineStats.PipelineDetailStats detailStats = stats.getPerPipelineProcessorStats()
                 .get(perPipelineStats.getPipelineId());
@@ -1470,8 +1471,9 @@ public void testStatsEnabledIgnoreFailure() throws Exception {
 
         // when ignoreFailure enabled, the search pipelines will all succeed.
         SearchPipelineStats stats = searchPipelineService.stats();
-        assertPipelineStats(stats.getTotalRequestStats(), 4, 0);
+        assertPipelineStats(stats.getTotalRequestStats(), 2, 0);
         assertPipelineStats(stats.getTotalResponseStats(), 2, 0);
+        assertPipelineStats(stats.getTotalPhaseResultsStats(), 2, 0);
 
         for (SearchPipelineStats.PerPipelineStats perPipelineStats : stats.getPipelineStats()) {
             SearchPipelineStats.PipelineDetailStats detailStats = stats.getPerPipelineProcessorStats()

server/src/test/java/org/opensearch/search/pipeline/SearchPipelineStatsTests.java

@@ -42,6 +42,7 @@ private static SearchPipelineStats createStats() {
         return new SearchPipelineStats(
             new OperationStats(1, 2, 3, 4),
             new OperationStats(5, 6, 7, 8),
+            new OperationStats(9, 10, 11, 12),
             List.of(
                 new SearchPipelineStats.PerPipelineStats("p1", new OperationStats(9, 10, 11, 12), new OperationStats(13, 14, 15, 16)),
                 new SearchPipelineStats.PerPipelineStats("p2", new OperationStats(17, 18, 19, 20), new OperationStats(21, 22, 23, 24))