Adding search phase pipeline stats

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

Author: martin-gaievski

release-notes/opensearch.release-notes-3.0.0-alpha1.md

@@ -55,6 +55,7 @@
 - Fix Bug - Handle unsigned long in sorting order assertion of LongHashSet ([#17207](https://github.com/opensearch-project/OpenSearch/pull/17207))
 - Implemented computation of segment replication stats at shard level ([#17055](https://github.com/opensearch-project/OpenSearch/pull/17055))
 - [Rule Based Auto-tagging] Add in-memory attribute value store ([#17342](https://github.com/opensearch-project/OpenSearch/pull/17342))
+- Added pipeline level stats for search phase result processor ([#17559](https://github.com/opensearch-project/OpenSearch/pull/17559))
 
 
 ### Dependencies

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

@@ -323,7 +323,7 @@ <Result extends SearchPhaseResult> void runSearchPhaseResultsTransformer(
         } catch (RuntimeException e) {
             onTransformPhaseResultsFailure();
             throw new SearchPipelineProcessingException(e);
-        }  finally {
+        } finally {
             long took = TimeUnit.NANOSECONDS.toMillis(relativeTimeSupplier.getAsLong() - pipelineStart);
             afterTransformPhaseResults(took);
         }

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

@@ -316,7 +316,12 @@ void populateStats(SearchPipelineStats.Builder statsBuilder) {
         }
         for (Processor phaseResultsProcessor : getSearchPhaseResultsProcessors()) {
             String key = getProcessorKey(phaseResultsProcessor);
-            statsBuilder.addPhaseResultsProcessorStats(getId(), key, phaseResultsProcessor.getType(), phaseResultsProcessorMetrics.get(key));
+            statsBuilder.addPhaseResultsProcessorStats(
+                getId(),
+                key,
+                phaseResultsProcessor.getType(),
+                phaseResultsProcessorMetrics.get(key)
+            );
         }
     }
 }

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

@@ -89,7 +89,11 @@ public SearchPipelineStats(StreamInput in) throws IOException {
             }
             pipelineDetailStatsMap.put(
                 pipelineId,
-                new PipelineDetailStats(unmodifiableList(requestProcessorStats), unmodifiableList(phaseResultsProcessorStats), unmodifiableList(responseProcessorStats))
+                new PipelineDetailStats(
+                    unmodifiableList(requestProcessorStats),
+                    unmodifiableList(phaseResultsProcessorStats),
+                    unmodifiableList(responseProcessorStats)
+                )
             );
         }
         this.perPipelineStats = unmodifiableList(perPipelineStats);
@@ -184,7 +188,11 @@ static class Builder {
         private final Map<String, List<ProcessorStats>> responseProcessorStatsPerPipeline = new HashMap<>();
         private final Map<String, List<ProcessorStats>> phaseResultsProcessorStatsPerPipeline = new HashMap<>();
 
-        Builder withTotalStats(OperationMetrics totalRequestMetrics, OperationMetrics totalResponseMetrics, OperationMetrics totalPhaseProcessingMetrics) {
+        Builder withTotalStats(
+            OperationMetrics totalRequestMetrics,
+            OperationMetrics totalResponseMetrics,
+            OperationMetrics totalPhaseProcessingMetrics
+        ) {
             this.totalRequestStats = totalRequestMetrics.createStats();
             this.totalResponseStats = totalResponseMetrics.createStats();
             this.totalPhaseResultsStats = totalPhaseProcessingMetrics.createStats();
@@ -204,7 +212,12 @@ Builder addRequestProcessorStats(String pipelineId, String processorName, String
             return this;
         }
 
-        Builder addPhaseResultsProcessorStats(String pipelineId, String processorName, String processorType, OperationMetrics processorMetrics) {
+        Builder addPhaseResultsProcessorStats(
+            String pipelineId,
+            String processorName,
+            String processorType,
+            OperationMetrics processorMetrics
+        ) {
             this.phaseResultsProcessorStatsPerPipeline.computeIfAbsent(pipelineId, k -> new ArrayList<>())
                 .add(new ProcessorStats(processorName, processorType, processorMetrics.createStats()));
             return this;
@@ -294,7 +307,11 @@ static class PipelineDetailStats {
         private final List<ProcessorStats> phaseResultsProcessorStats;
         private final List<ProcessorStats> responseProcessorStats;
 
-        public PipelineDetailStats(List<ProcessorStats> requestProcessorStats, List<ProcessorStats> phaseResultsProcessorStats, List<ProcessorStats> responseProcessorStats) {
+        public PipelineDetailStats(
+            List<ProcessorStats> requestProcessorStats,
+            List<ProcessorStats> phaseResultsProcessorStats,
+            List<ProcessorStats> responseProcessorStats
+        ) {
             this.requestProcessorStats = requestProcessorStats;
             this.phaseResultsProcessorStats = phaseResultsProcessorStats;
             this.responseProcessorStats = responseProcessorStats;
@@ -317,7 +334,9 @@ public boolean equals(Object o) {
             if (this == o) return true;
             if (o == null || getClass() != o.getClass()) return false;
             PipelineDetailStats that = (PipelineDetailStats) o;
-            return requestProcessorStats.equals(that.requestProcessorStats) && responseProcessorStats.equals(that.responseProcessorStats) && phaseResultsProcessorStats.equals(that.phaseResultsProcessorStats);
+            return requestProcessorStats.equals(that.requestProcessorStats)
+                && responseProcessorStats.equals(that.responseProcessorStats)
+                && phaseResultsProcessorStats.equals(that.phaseResultsProcessorStats);
         }
 
         @Override