Add subsidiary failures to search responses

Author: thecoop

client/client-benchmark-noop-api-plugin/src/main/java/org/elasticsearch/plugin/noop/action/search/TransportNoopSearchAction.java

@@ -12,9 +12,9 @@
 import org.elasticsearch.action.search.SearchRequest;
 import org.elasticsearch.action.search.SearchResponse;
 import org.elasticsearch.action.search.ShardSearchFailure;
+import org.elasticsearch.action.search.SubsidiaryFailure;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.HandledTransportAction;
-import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.injection.guice.Inject;
 import org.elasticsearch.plugin.noop.NoopPlugin;
@@ -34,7 +34,7 @@ public TransportNoopSearchAction(TransportService transportService, ActionFilter
             NoopPlugin.NOOP_SEARCH_ACTION.name(),
             transportService,
             actionFilters,
-            (Writeable.Reader<SearchRequest>) SearchRequest::new,
+            SearchRequest::new,
             EsExecutors.DIRECT_EXECUTOR_SERVICE
         );
     }
@@ -56,6 +56,7 @@ protected void doExecute(Task task, SearchRequest request, ActionListener<Search
                 0,
                 0,
                 ShardSearchFailure.EMPTY_ARRAY,
+                SubsidiaryFailure.EMPTY_ARRAY,
                 SearchResponse.Clusters.EMPTY
             )
         );

docs/changelog/122788.yaml

@@ -0,0 +1,6 @@
+pr: 122788
+summary: Add subsidiary failures to search responses
+area: Ranking
+type: enhancement
+issues:
+ - 116796

modules/lang-mustache/src/test/java/org/elasticsearch/script/mustache/SearchTemplateResponseTests.java

@@ -12,6 +12,7 @@
 import org.apache.lucene.search.TotalHits;
 import org.elasticsearch.action.search.SearchResponse;
 import org.elasticsearch.action.search.ShardSearchFailure;
+import org.elasticsearch.action.search.SubsidiaryFailure;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.search.SearchHit;
@@ -201,6 +202,7 @@ public void testSearchResponseToXContent() throws IOException {
             0,
             0,
             ShardSearchFailure.EMPTY_ARRAY,
+            SubsidiaryFailure.EMPTY_ARRAY,
             SearchResponse.Clusters.EMPTY
         );
 

server/src/main/java/org/elasticsearch/TransportVersions.java

@@ -201,6 +201,8 @@ static TransportVersion def(int id) {
     public static final TransportVersion REMOVE_REPOSITORY_CONFLICT_MESSAGE = def(9_012_0_00);
     public static final TransportVersion RERANKER_FAILURES_ALLOWED = def(9_013_0_00);
     public static final TransportVersion VOYAGE_AI_INTEGRATION_ADDED = def(9_014_0_00);
+    public static final TransportVersion SEARCH_SUBSIDIARY_FAILURES = def(9_015_0_00);
+
     /*
      * STOP! READ THIS FIRST! No, really,
      *        ____ _____ ___  ____  _        ____  _____    _    ____    _____ _   _ ___ ____    _____ ___ ____  ____ _____ _

server/src/main/java/org/elasticsearch/action/search/AbstractSearchAsyncAction.java

@@ -43,6 +43,7 @@
 
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
@@ -91,6 +92,7 @@ abstract class AbstractSearchAsyncAction<Result extends SearchPhaseResult> exten
     private final AtomicInteger successfulOps = new AtomicInteger();
     private final SearchTimeProvider timeProvider;
     private final SearchResponse.Clusters clusters;
+    private final List<SubsidiaryFailure> subsidiaryFailures = Collections.synchronizedList(new ArrayList<>());
 
     protected final List<SearchShardIterator> toSkipShardsIts;
     protected final List<SearchShardIterator> shardsIts;
@@ -573,6 +575,7 @@ public boolean isPartOfPointInTime(ShardSearchContextId contextId) {
     private SearchResponse buildSearchResponse(
         SearchResponseSections internalSearchResponse,
         ShardSearchFailure[] failures,
+        SubsidiaryFailure[] subsidiaryFailures,
         String scrollId,
         BytesReference searchContextId
     ) {
@@ -588,6 +591,7 @@ private SearchResponse buildSearchResponse(
             toSkipShardsIts.size(),
             buildTookInMillis(),
             failures,
+            subsidiaryFailures,
             clusters,
             searchContextId
         );
@@ -623,7 +627,13 @@ public void sendSearchResponse(SearchResponseSections internalSearchResponse, At
                     searchContextId = null;
                 }
             }
-            ActionListener.respondAndRelease(listener, buildSearchResponse(internalSearchResponse, failures, scrollId, searchContextId));
+
+            SubsidiaryFailure[] subFailures = subsidiaryFailures.toArray(SubsidiaryFailure.EMPTY_ARRAY);
+
+            ActionListener.respondAndRelease(
+                listener,
+                buildSearchResponse(internalSearchResponse, failures, subFailures, scrollId, searchContextId)
+            );
         }
     }
 
@@ -661,6 +671,10 @@ private void raisePhaseFailure(SearchPhaseExecutionException exception) {
         listener.onFailure(exception);
     }
 
+    public void addSubsidiaryFailure(String description, Exception exception) {
+        subsidiaryFailures.add(new SubsidiaryFailure(description, exception));
+    }
+
     /**
       * Releases a search context with the given context ID on the node the given connection is connected to.
       * @see org.elasticsearch.search.query.QuerySearchResult#getContextId()

server/src/main/java/org/elasticsearch/action/search/RankFeaturePhase.java

@@ -199,9 +199,10 @@ public void onResponse(RankFeatureDoc[] docsWithUpdatedScores) {
                 @Override
                 public void onFailure(Exception e) {
                     if (rankFeaturePhaseRankCoordinatorContext.failuresAllowed()) {
-                        // TODO: handle the exception somewhere
                         // don't want to log the entire stack trace, it's not helpful here
                         logger.warn("Exception computing updated ranks, continuing with existing ranks: {}", e.toString());
+                        context.addSubsidiaryFailure("reranking", e);
+
                         // use the existing score docs as-is
                         // downstream things expect every doc to have a score, so we need to infer a score here
                         // if the doc doesn't otherwise have one. We can use the rank to infer a possible score instead (1/rank).

server/src/main/java/org/elasticsearch/action/search/SearchResponse.java

@@ -19,6 +19,7 @@
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.lucene.Lucene;
+import org.elasticsearch.common.util.CollectionUtils;
 import org.elasticsearch.common.util.concurrent.ConcurrentCollections;
 import org.elasticsearch.common.xcontent.ChunkedToXContentHelper;
 import org.elasticsearch.common.xcontent.ChunkedToXContentObject;
@@ -53,8 +54,6 @@
 import java.util.function.Predicate;
 import java.util.function.Supplier;
 
-import static org.elasticsearch.action.search.ShardSearchFailure.readShardSearchFailure;
-
 /**
  * A response of a search request.
  */
@@ -70,6 +69,7 @@ public class SearchResponse extends ActionResponse implements ChunkedToXContentO
     public static final ParseField TIMED_OUT = new ParseField("timed_out");
     public static final ParseField TERMINATED_EARLY = new ParseField("terminated_early");
     public static final ParseField NUM_REDUCE_PHASES = new ParseField("num_reduce_phases");
+    public static final ParseField SUBSIDIARY_FAILURES = new ParseField("subsidiaryFailures");
 
     private final SearchHits hits;
     private final InternalAggregations aggregations;
@@ -84,6 +84,7 @@ public class SearchResponse extends ActionResponse implements ChunkedToXContentO
     private final int successfulShards;
     private final int skippedShards;
     private final ShardSearchFailure[] shardFailures;
+    private final SubsidiaryFailure[] subsidiaryFailures;
     private final Clusters clusters;
     private final long tookInMillis;
 
@@ -100,14 +101,17 @@ public SearchResponse(StreamInput in) throws IOException {
         this.numReducePhases = in.readVInt();
         totalShards = in.readVInt();
         successfulShards = in.readVInt();
-        int size = in.readVInt();
-        if (size == 0) {
-            shardFailures = ShardSearchFailure.EMPTY_ARRAY;
+        shardFailures = in.readArray(
+            ShardSearchFailure::readShardSearchFailure,
+            s -> s == 0 ? ShardSearchFailure.EMPTY_ARRAY : new ShardSearchFailure[s]
+        );
+        if (in.getTransportVersion().onOrAfter(TransportVersions.SEARCH_SUBSIDIARY_FAILURES)) {
+            subsidiaryFailures = in.readArray(
+                SubsidiaryFailure::new,
+                s -> s == 0 ? SubsidiaryFailure.EMPTY_ARRAY : new SubsidiaryFailure[s]
+            );
         } else {
-            shardFailures = new ShardSearchFailure[size];
-            for (int i = 0; i < shardFailures.length; i++) {
-                shardFailures[i] = readShardSearchFailure(in);
-            }
+            subsidiaryFailures = SubsidiaryFailure.EMPTY_ARRAY;
         }
         clusters = new Clusters(in);
         scrollId = in.readOptionalString();
@@ -130,6 +134,7 @@ public SearchResponse(
         int skippedShards,
         long tookInMillis,
         ShardSearchFailure[] shardFailures,
+        SubsidiaryFailure[] subsidiaryFailures,
         Clusters clusters
     ) {
         this(
@@ -146,6 +151,7 @@ public SearchResponse(
             skippedShards,
             tookInMillis,
             shardFailures,
+            subsidiaryFailures,
             clusters,
             null
         );
@@ -159,6 +165,7 @@ public SearchResponse(
         int skippedShards,
         long tookInMillis,
         ShardSearchFailure[] shardFailures,
+        SubsidiaryFailure[] subsidiaryFailures,
         Clusters clusters,
         BytesReference pointInTimeId
     ) {
@@ -176,6 +183,7 @@ public SearchResponse(
             skippedShards,
             tookInMillis,
             shardFailures,
+            subsidiaryFailures,
             clusters,
             pointInTimeId
         );
@@ -195,6 +203,7 @@ public SearchResponse(
         int skippedShards,
         long tookInMillis,
         ShardSearchFailure[] shardFailures,
+        SubsidiaryFailure[] subsidiaryFailures,
         Clusters clusters,
         BytesReference pointInTimeId
     ) {
@@ -214,6 +223,7 @@ public SearchResponse(
         this.skippedShards = skippedShards;
         this.tookInMillis = tookInMillis;
         this.shardFailures = shardFailures;
+        this.subsidiaryFailures = subsidiaryFailures;
         assert skippedShards <= totalShards : "skipped: " + skippedShards + " total: " + totalShards;
         assert scrollId == null || pointInTimeId == null
             : "SearchResponse can't have both scrollId [" + scrollId + "] and searchContextId [" + pointInTimeId + "]";
@@ -339,7 +349,14 @@ public int getFailedShards() {
      * The failures that occurred during the search.
      */
     public ShardSearchFailure[] getShardFailures() {
-        return this.shardFailures;
+        return shardFailures;
+    }
+
+    /**
+     * The failures that occurred that did not stop results from being returned.
+     */
+    public SubsidiaryFailure[] getSubsidiaryFailures() {
+        return subsidiaryFailures;
     }
 
     /**
@@ -392,6 +409,9 @@ public Iterator<? extends ToXContent> toXContentChunked(ToXContent.Params params
     public Iterator<? extends ToXContent> innerToXContentChunked(ToXContent.Params params) {
         return Iterators.concat(
             ChunkedToXContentHelper.chunk(SearchResponse.this::headerToXContent),
+            CollectionUtils.isEmpty(subsidiaryFailures)
+                ? Collections.emptyIterator()
+                : ChunkedToXContentHelper.array(SUBSIDIARY_FAILURES.getPreferredName(), Iterators.forArray(subsidiaryFailures)),
             Iterators.single(clusters),
             Iterators.concat(
                 hits.toXContentChunked(params),
@@ -444,10 +464,9 @@ public void writeTo(StreamOutput out) throws IOException {
         out.writeVInt(numReducePhases);
         out.writeVInt(totalShards);
         out.writeVInt(successfulShards);
-
-        out.writeVInt(shardFailures.length);
-        for (ShardSearchFailure shardSearchFailure : shardFailures) {
-            shardSearchFailure.writeTo(out);
+        out.writeArray(shardFailures);
+        if (out.getTransportVersion().onOrAfter(TransportVersions.SEARCH_SUBSIDIARY_FAILURES)) {
+            out.writeArray(subsidiaryFailures);
         }
         clusters.writeTo(out);
         out.writeOptionalString(scrollId);
@@ -1162,6 +1181,7 @@ public static SearchResponse empty(Supplier<Long> tookInMillisSupplier, Clusters
             0,
             tookInMillisSupplier.get(),
             ShardSearchFailure.EMPTY_ARRAY,
+            SubsidiaryFailure.EMPTY_ARRAY,
             clusters,
             null
         );

server/src/main/java/org/elasticsearch/action/search/SearchResponseMerger.java

@@ -129,6 +129,7 @@ public SearchResponse getMergedResponse(Clusters clusters) {
         // the current reduce phase counts as one
         int numReducePhases = 1;
         List<ShardSearchFailure> failures = new ArrayList<>();
+        List<SubsidiaryFailure> subsidiaryFailures = new ArrayList<>();
         Map<String, SearchProfileShardResult> profileResults = new HashMap<>();
         List<InternalAggregations> aggs = new ArrayList<>();
         Map<ShardIdAndClusterAlias, Integer> shards = new TreeMap<>();
@@ -145,6 +146,7 @@ public SearchResponse getMergedResponse(Clusters clusters) {
             numReducePhases += searchResponse.getNumReducePhases();
 
             Collections.addAll(failures, searchResponse.getShardFailures());
+            Collections.addAll(subsidiaryFailures, searchResponse.getSubsidiaryFailures());
 
             profileResults.putAll(searchResponse.getProfileResults());
 
@@ -213,6 +215,7 @@ public SearchResponse getMergedResponse(Clusters clusters) {
                 ? InternalAggregations.EMPTY
                 : InternalAggregations.topLevelReduce(aggs, aggReduceContextBuilder.forFinalReduction());
             ShardSearchFailure[] shardFailures = failures.toArray(ShardSearchFailure.EMPTY_ARRAY);
+            SubsidiaryFailure[] subFailures = subsidiaryFailures.toArray(SubsidiaryFailure.EMPTY_ARRAY);
             SearchProfileResults profileShardResults = profileResults.isEmpty() ? null : new SearchProfileResults(profileResults);
             // make failures ordering consistent between ordinary search and CCS by looking at the shard they come from
             Arrays.sort(shardFailures, FAILURES_COMPARATOR);
@@ -231,6 +234,7 @@ public SearchResponse getMergedResponse(Clusters clusters) {
                 skippedShards,
                 tookInMillis,
                 shardFailures,
+                subFailures,
                 clusters,
                 null
             );

server/src/main/java/org/elasticsearch/action/search/SearchScrollAsyncAction.java

@@ -55,6 +55,7 @@ abstract class SearchScrollAsyncAction<T extends SearchPhaseResult> {
     protected final SearchTransportService searchTransportService;
     private final long startTime;
     private final List<ShardSearchFailure> shardFailures = new ArrayList<>();
+    private final List<SubsidiaryFailure> subsidiaryFailures = new ArrayList<>();
     private final AtomicInteger successfulOps;
 
     protected SearchScrollAsyncAction(
@@ -213,6 +214,10 @@ synchronized ShardSearchFailure[] buildShardFailures() { // pkg private for test
         return shardFailures.toArray(ShardSearchFailure.EMPTY_ARRAY);
     }
 
+    synchronized SubsidiaryFailure[] buildSubsidiaryFailures() {
+        return subsidiaryFailures.toArray(SubsidiaryFailure.EMPTY_ARRAY);
+    }
+
     // we do our best to return the shard failures, but its ok if its not fully concurrently safe
     // we simply try and return as much as possible
     private synchronized void addShardFailure(ShardSearchFailure failure) {
@@ -263,6 +268,7 @@ protected final void sendResponse(
                         0,
                         buildTookInMillis(),
                         buildShardFailures(),
+                        buildSubsidiaryFailures(),
                         SearchResponse.Clusters.EMPTY,
                         null
                     )

server/src/main/java/org/elasticsearch/action/search/SubsidiaryFailure.java

@@ -0,0 +1,47 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.action.search;
+
+import org.elasticsearch.ElasticsearchException;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.xcontent.ToXContentObject;
+import org.elasticsearch.xcontent.XContentBuilder;
+
+import java.io.IOException;
+
+public record SubsidiaryFailure(String phase, Exception failure) implements ToXContentObject, Writeable {
+    public static final String PHASE_FIELD = "phase";
+    public static final String FAILURE_FIELD = "failure";
+
+    public static final SubsidiaryFailure[] EMPTY_ARRAY = new SubsidiaryFailure[0];
+
+    public SubsidiaryFailure(StreamInput in) throws IOException {
+        this(in.readString(), in.readException());
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        out.writeString(phase);
+        out.writeException(failure);
+    }
+
+    @Override
+    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+        builder.startObject();
+        builder.field(PHASE_FIELD, phase);
+        builder.startObject(FAILURE_FIELD);
+        ElasticsearchException.generateThrowableXContent(builder, params, failure);
+        builder.endObject();
+        builder.endObject();
+        return builder;
+    }
+}