Java 类org.elasticsearch.action.search.ShardSearchFailure 实例源码

private Response wrap(SearchResponse response) {
    List<SearchFailure> failures;
    if (response.getShardFailures() == null) {
        failures = emptyList();
    } else {
        failures = new ArrayList<>(response.getShardFailures().length);
        for (ShardSearchFailure failure: response.getShardFailures()) {
            String nodeId = failure.shard() == null ? null : failure.shard().getNodeId();
            failures.add(new SearchFailure(failure.getCause(), failure.index(), failure.shardId(), nodeId));
        }
    }
    List<Hit> hits;
    if (response.getHits().getHits() == null || response.getHits().getHits().length == 0) {
        hits = emptyList();
    } else {
        hits = new ArrayList<>(response.getHits().getHits().length);
        for (SearchHit hit: response.getHits().getHits()) {
            hits.add(new ClientHit(hit));
        }
        hits = unmodifiableList(hits);
    }
    return new Response(response.isTimedOut(), failures, response.getHits().getTotalHits(),
            hits, response.getScrollId());
}

private void checkExceptions(Script script) {
    try {
        SearchResponse sr = client().prepareSearch("test").setQuery(QueryBuilders.matchAllQuery()).addScriptField("tvtest", script)
                .execute().actionGet();
        assertThat(sr.getHits().getHits().length, equalTo(0));
        ShardSearchFailure[] shardFails = sr.getShardFailures();
        for (ShardSearchFailure fail : shardFails) {
            assertThat(fail.reason().indexOf("Cannot iterate twice! If you want to iterate more that once, add _CACHE explicitly."),
                    Matchers.greaterThan(-1));
        }
    } catch (SearchPhaseExecutionException ex) {
        assertThat(
                "got " + ex.toString(),
                ex.toString().indexOf("Cannot iterate twice! If you want to iterate more that once, add _CACHE explicitly."),
                Matchers.greaterThan(-1));
    }
}

public void testSearchPhaseExecutionException() throws IOException {
    ShardSearchFailure[] empty = new ShardSearchFailure[0];
    SearchPhaseExecutionException ex = serialize(new SearchPhaseExecutionException("boom", "baam", new NullPointerException(), empty));
    assertEquals("boom", ex.getPhaseName());
    assertEquals("baam", ex.getMessage());
    assertTrue(ex.getCause() instanceof NullPointerException);
    assertEquals(empty.length, ex.shardFailures().length);
    ShardSearchFailure[] one = new ShardSearchFailure[]{
            new ShardSearchFailure(new IllegalArgumentException("nono!"))
    };

    ex = serialize(new SearchPhaseExecutionException("boom", "baam", new NullPointerException(), one));
    assertEquals("boom", ex.getPhaseName());
    assertEquals("baam", ex.getMessage());
    assertTrue(ex.getCause() instanceof NullPointerException);
    assertEquals(one.length, ex.shardFailures().length);
    assertTrue(ex.shardFailures()[0].getCause() instanceof IllegalArgumentException);
}

/**
 * Perform a test search request to validate the element prior to storing it.
 *
 * @param validation validation info
 * @param element the element stored
 * @param task the parent task
 * @param listener the action listener to write to
 * @param onSuccess action ro run when the validation is successfull
 */
private void validate(FeatureValidation validation,
                      StorableElement element,
                      Task task,
                      ActionListener<FeatureStoreResponse> listener,
                      Runnable onSuccess) {
    ValidatingLtrQueryBuilder ltrBuilder = new ValidatingLtrQueryBuilder(element,
            validation, factory);
    SearchRequestBuilder builder = SearchAction.INSTANCE.newRequestBuilder(client);
    builder.setIndices(validation.getIndex());
    builder.setQuery(ltrBuilder);
    builder.setFrom(0);
    builder.setSize(20);
    // Bail out early and don't score the whole index.
    builder.setTerminateAfter(1000);
    builder.request().setParentTask(clusterService.localNode().getId(), task.getId());
    builder.execute(wrap((r) -> {
            if (r.getFailedShards() > 0) {
                ShardSearchFailure failure = r.getShardFailures()[0];
                throw new IllegalArgumentException("Validating the element caused " + r.getFailedShards() +
                        " shard failures, see root cause: " + failure.reason(), failure.getCause());
            }
            onSuccess.run();
        },
        (e) -> listener.onFailure(new IllegalArgumentException("Cannot store element, validation failed.", e))));
}

protected void validateRespose(final SearchResponse response) {
    final int totalShards = response.getTotalShards();
    final int successfulShards = response.getSuccessfulShards();
    if (totalShards != successfulShards) {
        throw new MissingShardsException(totalShards - successfulShards
                + " shards are failed.");
    }
    final ShardSearchFailure[] failures = response.getShardFailures();
    if (failures.length > 0) {
        final StringBuilder buf = new StringBuilder();
        for (final ShardOperationFailedException failure : failures) {
            buf.append('\n').append(failure.toString());
        }
        throw new OperationFailedException("Search Operation Failed: "
                + buf.toString());
    }
}

@Override
protected void doExecute(SearchRequest request, ActionListener<SearchResponse> listener) {
    listener.onResponse(new SearchResponse(new InternalSearchResponse(
        new SearchHits(
            new SearchHit[0], 0L, 0.0f),
        new InternalAggregations(Collections.emptyList()),
        new Suggest(Collections.emptyList()),
        new SearchProfileShardResults(Collections.emptyMap()), false, false, 1), "", 1, 1, 0, new ShardSearchFailure[0]));
}

public static String formatShardStatus(SearchResponse response) {
    String msg = " Total shards: " + response.getTotalShards() + " Successful shards: " + response.getSuccessfulShards() + " & "
            + response.getFailedShards() + " shard failures:";
    for (ShardSearchFailure failure : response.getShardFailures()) {
        msg += "\n " + failure.toString();
    }
    return msg;
}

/**
 * Finalizes the shard in repository and then removes it from cluster state
 * <p>
 * This is non-blocking method that runs on a thread from SNAPSHOT thread pool
 *
 * @param entry   snapshot
 * @param failure failure reason or null if snapshot was successful
 */
private void endSnapshot(final SnapshotsInProgress.Entry entry, final String failure) {
    threadPool.executor(ThreadPool.Names.SNAPSHOT).execute(new Runnable() {
        @Override
        public void run() {
            final Snapshot snapshot = entry.snapshot();
            try {
                final Repository repository = repositoriesService.repository(snapshot.getRepository());
                logger.trace("[{}] finalizing snapshot in repository, state: [{}], failure[{}]", snapshot, entry.state(), failure);
                ArrayList<ShardSearchFailure> failures = new ArrayList<>();
                ArrayList<SnapshotShardFailure> shardFailures = new ArrayList<>();
                for (ObjectObjectCursor<ShardId, ShardSnapshotStatus> shardStatus : entry.shards()) {
                    ShardId shardId = shardStatus.key;
                    ShardSnapshotStatus status = shardStatus.value;
                    if (status.state().failed()) {
                        failures.add(new ShardSearchFailure(status.reason(), new SearchShardTarget(status.nodeId(), shardId)));
                        shardFailures.add(new SnapshotShardFailure(status.nodeId(), shardId, status.reason()));
                    }
                }
                SnapshotInfo snapshotInfo = repository.finalizeSnapshot(
                    snapshot.getSnapshotId(),
                    entry.indices(),
                    entry.startTime(),
                    failure,
                    entry.shards().size(),
                    Collections.unmodifiableList(shardFailures),
                    entry.getRepositoryStateId());
                removeSnapshotFromClusterState(snapshot, snapshotInfo, null);
            } catch (Exception e) {
                logger.warn((Supplier<?>) () -> new ParameterizedMessage("[{}] failed to finalize snapshot", snapshot), e);
                removeSnapshotFromClusterState(snapshot, null, e);
            }
        }
    });
}

private void assertShardExecutionState(SearchResponse response, int expectedFailures) throws Exception {
    ShardSearchFailure[] failures = response.getShardFailures();
    if (failures.length != expectedFailures) {
        for (ShardSearchFailure failure : failures) {
            logger.error((Supplier<?>) () -> new ParameterizedMessage("Shard Failure: {}", failure), failure.getCause());
        }
        fail("Unexpected shard failures!");
    }
    assertThat("Not all shards are initialized", response.getSuccessfulShards(), equalTo(response.getTotalShards()));
}

public void testIgnoreUnmapped() throws Exception {
    createIndex("test");

    client().prepareIndex("test", "type1", "1").setSource(jsonBuilder().startObject()
            .field("id", "1")
            .field("i_value", -1)
            .field("d_value", -1.1)
            .endObject()).execute().actionGet();

    logger.info("--> sort with an unmapped field, verify it fails");
    try {
        SearchResponse result = client().prepareSearch()
                .setQuery(matchAllQuery())
                .addSort(SortBuilders.fieldSort("kkk"))
                .execute().actionGet();
        assertThat("Expected exception but returned with", result, nullValue());
    } catch (SearchPhaseExecutionException e) {
        //we check that it's a parse failure rather than a different shard failure
        for (ShardSearchFailure shardSearchFailure : e.shardFailures()) {
            assertThat(shardSearchFailure.toString(), containsString("[No mapping found for [kkk] in order to sort on]"));
        }
    }

    SearchResponse searchResponse = client().prepareSearch()
            .setQuery(matchAllQuery())
            .addSort(SortBuilders.fieldSort("kkk").unmappedType("keyword"))
            .execute().actionGet();
    assertNoFailures(searchResponse);
}

public void testConvert() throws IOException {
    RestRequest request = new FakeRestRequest();
    RestChannel channel = new DetailedExceptionRestChannel(request);
    ShardSearchFailure failure = new ShardSearchFailure(new ParsingException(1, 2, "foobar", null),
            new SearchShardTarget("node_1", new Index("foo", "_na_"), 1));
    ShardSearchFailure failure1 = new ShardSearchFailure(new ParsingException(1, 2, "foobar", null),
            new SearchShardTarget("node_1", new Index("foo", "_na_"), 2));
    SearchPhaseExecutionException ex = new SearchPhaseExecutionException("search", "all shards failed",  new ShardSearchFailure[] {failure, failure1});
    BytesRestResponse response = new BytesRestResponse(channel, new RemoteTransportException("foo", ex));
    String text = response.content().utf8ToString();
    String expected = "{\"error\":{\"root_cause\":[{\"type\":\"parsing_exception\",\"reason\":\"foobar\",\"line\":1,\"col\":2}],\"type\":\"search_phase_execution_exception\",\"reason\":\"all shards failed\",\"phase\":\"search\",\"grouped\":true,\"failed_shards\":[{\"shard\":1,\"index\":\"foo\",\"node\":\"node_1\",\"reason\":{\"type\":\"parsing_exception\",\"reason\":\"foobar\",\"line\":1,\"col\":2}}]},\"status\":400}";
    assertEquals(expected.trim(), text.trim());
    String stackTrace = ExceptionsHelper.stackTrace(ex);
    assertTrue(stackTrace.contains("Caused by: ParsingException[foobar]"));
}

/**
 * Finalizes the shard in repository and then removes it from cluster state
 * <p>
 * This is non-blocking method that runs on a thread from SNAPSHOT thread pool
 *
 * @param entry   snapshot
 * @param failure failure reason or null if snapshot was successful
 */
private void endSnapshot(final SnapshotsInProgress.Entry entry, final String failure) {
    threadPool.executor(ThreadPool.Names.SNAPSHOT).execute(new Runnable() {
        @Override
        public void run() {
            SnapshotId snapshotId = entry.snapshotId();
            try {
                final Repository repository = repositoriesService.repository(snapshotId.getRepository());
                logger.trace("[{}] finalizing snapshot in repository, state: [{}], failure[{}]", snapshotId, entry.state(), failure);
                ArrayList<ShardSearchFailure> failures = new ArrayList<>();
                ArrayList<SnapshotShardFailure> shardFailures = new ArrayList<>();
                for (Map.Entry<ShardId, ShardSnapshotStatus> shardStatus : entry.shards().entrySet()) {
                    ShardId shardId = shardStatus.getKey();
                    ShardSnapshotStatus status = shardStatus.getValue();
                    if (status.state().failed()) {
                        failures.add(new ShardSearchFailure(status.reason(), new SearchShardTarget(status.nodeId(), shardId.getIndex(), shardId.id())));
                        shardFailures.add(new SnapshotShardFailure(status.nodeId(), shardId.getIndex(), shardId.id(), status.reason()));
                    }
                }
                Snapshot snapshot = repository.finalizeSnapshot(snapshotId, entry.indices(), entry.startTime(), failure, entry.shards().size(), Collections.unmodifiableList(shardFailures));
                removeSnapshotFromClusterState(snapshotId, new SnapshotInfo(snapshot), null);
            } catch (Throwable t) {
                logger.warn("[{}] failed to finalize snapshot", t, snapshotId);
                removeSnapshotFromClusterState(snapshotId, null, t);
            }
        }
    });
}

@Override
public SearchResponse search(String searchTerms, List<String> references, List<String> siteIds, int start, int end, Map<String,String> additionalSearchInfromation) {
    return new SearchResponse(
            new InternalSearchResponse(new InternalSearchHits(new InternalSearchHit[0], 0, 0.0f), new InternalFacets(Collections.EMPTY_LIST), new InternalAggregations(Collections.EMPTY_LIST), new Suggest(), false, false),
            "no-op",
            1,
            1,
            1,
            new ShardSearchFailure[0]
    );
}

/**
 * Handle individual shard failures that can occur even when the response is OK.  These
 * can indicate misconfiguration of the search indices.
 * @param request The search request.
 * @param response  The search response.
 */
private void handleShardFailures(
        org.elasticsearch.action.search.SearchRequest request,
        org.elasticsearch.action.search.SearchResponse response) {
  /*
   * shard failures are only logged.  the search itself is not failed.  this approach
   * assumes that a user is interested in partial search results, even if the
   * entire search result set cannot be produced.
   *
   * for example, assume the user adds an additional sensor and the telemetry
   * is indexed into a new search index.  if that search index is misconfigured,
   * it can result in partial shard failures.  rather than failing the entire search,
   * we log the error and allow the results to be returned from shards that
   * are correctly configured.
   */
  int errors = ArrayUtils.getLength(response.getShardFailures());
  LOG.error("Search resulted in {}/{} shards failing; errors={}, search={}",
          response.getFailedShards(),
          response.getTotalShards(),
          errors,
          ElasticsearchUtils.toJSON(request).orElse("???"));

  // log each reported failure
  int failureCount=1;
  for(ShardSearchFailure fail: response.getShardFailures()) {
    String msg = String.format(
            "Shard search failure [%s/%s]; reason=%s, index=%s, shard=%s, status=%s, nodeId=%s",
            failureCount,
            errors,
            ExceptionUtils.getRootCauseMessage(fail.getCause()),
            fail.index(),
            fail.shardId(),
            fail.status(),
            fail.shard().getNodeId());
    LOG.error(msg, fail.getCause());
  }
}

@Test
public void searchShouldHandleShardFailure() throws InvalidSearchException {
  // mocks
  SearchResponse response = mock(SearchResponse.class);
  SearchRequest request = new SearchRequest();
  ShardSearchFailure fail = mock(ShardSearchFailure.class);
  SearchShardTarget target = new SearchShardTarget("node1", mock(Index.class), 1, "metron");

  // response will have status of OK
  when(response.status()).thenReturn(RestStatus.OK);

  // the response will report shard failures
  when(response.getFailedShards()).thenReturn(1);
  when(response.getTotalShards()).thenReturn(2);

  // the response will return the failures
  ShardSearchFailure[] failures = { fail };
  when(response.getShardFailures()).thenReturn(failures);

  // shard failure needs to report the node
  when(fail.shard()).thenReturn(target);

  // shard failure needs to report details of failure
  when(fail.index()).thenReturn("bro_index_2017-10-11");
  when(fail.shardId()).thenReturn(1);

  // search should succeed, even with failed shards
  ElasticsearchRequestSubmitter submitter = setup(response);
  SearchResponse actual = submitter.submitSearch(request);
  assertNotNull(actual);
}

private SearchResponse readFromCache(final BytesReference value)
        throws IOException {
    final long startTime = System.nanoTime();
    final StreamInput in = value.streamInput();
    Map<String, Object> headers = null;
    if (in.readBoolean()) {
        headers = in.readMap();
    }
    final InternalSearchResponse internalResponse = new InternalSearchResponse(
            null, null, null, null, false, null);
    internalResponse.readFrom(in);
    final int totalShards = in.readVInt();
    final int successfulShards = in.readVInt();
    final int size = in.readVInt();
    ShardSearchFailure[] shardFailures;
    if (size == 0) {
        shardFailures = ShardSearchFailure.EMPTY_ARRAY;
    } else {
        shardFailures = new ShardSearchFailure[size];
        for (int i = 0; i < shardFailures.length; i++) {
            shardFailures[i] = readShardSearchFailure(in);
        }
    }
    final String scrollId = in.readOptionalString();
    final long tookInMillis = (System.nanoTime() - startTime) / 1000000;
    final SearchResponse response = new SearchResponse(internalResponse,
            scrollId, totalShards, successfulShards, tookInMillis,
            shardFailures);
    if (headers != null) {
        for (final Map.Entry<String, Object> entry : headers.entrySet()) {
            response.putHeader(entry.getKey(), entry.getValue());
        }
    }
    return response;
}

@Override
public SearchResponse search(String searchTerms, List<String> references, List<String> siteIds, int start, int end, Map<String,String> additionalSearchInfromation) {
    return new SearchResponse(
            new InternalSearchResponse(new InternalSearchHits(new InternalSearchHit[0], 0, 0.0f), new InternalFacets(Collections.EMPTY_LIST), new InternalAggregations(Collections.EMPTY_LIST), new Suggest(), false, false),
            "no-op",
            1,
            1,
            1,
            new ShardSearchFailure[0]
    );
}

private SearchResponse checkResponseForErrors(String query, SearchResponse response) {
    for (ShardSearchFailure failure : response.getShardFailures()) {
        LOG.error("Shard failure with query {}: {}", query, failure.reason());
    }
    // Possibly throw an exception here.
    return response;
}

@Override
public ShardSearchFailure[] getShardFailures() {
  return searchResponse.getShardFailures();
}

private SearchResponse createSearchResponseWithScrollId(String scrollId) {
  return new SearchResponse(InternalSearchResponse.empty(), scrollId, 1, 1, 1, new ShardSearchFailure[0]);
}

/**
 * Write ES search statistics record about successful search.
 * 
 * @param type of search performed
 * @param responseUuid UUID of response (also returned over search REST API)
 * @param resp response from search attempt
 * @param dateInMillis timestamp when search was performed
 * @param querySettings performed
 */
public void writeStatisticsRecord(StatsRecordType type, String responseUuid, SearchResponse resp, long dateInMillis,
        QuerySettings querySettings) {

    if (!statsConfiguration.enabled()) {
        return;
    }

    if (resp == null) {
        return;
    }

    Map<String, Object> source = new HashMap<String, Object>();

    source.put(FIELD_RESPONSE_UUID, responseUuid);
    source.put("took", resp.getTookInMillis());
    source.put("timed_out", resp.isTimedOut());
    source.put("total_hits", resp.getHits().totalHits());
    source.put("max_score", resp.getHits().maxScore());
    source.put("shards_successful", resp.getSuccessfulShards());
    source.put("shards_failed", resp.getFailedShards());
    source.put(FIELD_STATUS, resp.status().name());
    if (resp.getFailedShards() > 0) {
        for (ShardSearchFailure ssf : resp.getShardFailures()) {
            source.put("shard_failure", ssf.reason());
        }
    }

    addQuery(source, querySettings);

    if (resp.getHits().totalHits() > 0) {
        List<String> hitIds = new ArrayList<String>();
        for (SearchHit hit : resp.getHits().getHits()) {
            hitIds.add(hit.getId());
        }
        source.put("returned_hits", hitIds.size());
        source.put(FIELD_HITS_ID, hitIds);

    }

    writeStatisticsRecord(type, dateInMillis, source);
}

/**
 * Write ES search statistics record about successful search.
 *
 * @param type          of search performed
 * @param responseUuid  UUID of response (also returned over search REST API)
 * @param resp          response from search attempt
 * @param dateInMillis  timestamp when search was performed
 * @param querySettings performed
 */
public void writeStatisticsRecord(StatsRecordType type, String responseUuid, SearchResponse resp, long dateInMillis,
                                  QuerySettings querySettings) {

    if (!statsConfiguration.enabled()) {
        return;
    }

    if (resp == null) {
        return;
    }

    Map<String, Object> source = new HashMap<>();

    source.put(FIELD_RESPONSE_UUID, responseUuid);
    source.put("took", resp.getTookInMillis());
    source.put("timed_out", resp.isTimedOut());
    source.put("total_hits", resp.getHits().totalHits());
    source.put("max_score", resp.getHits().maxScore());
    source.put("shards_successful", resp.getSuccessfulShards());
    source.put("shards_failed", resp.getFailedShards());
    source.put(FIELD_STATUS, resp.status().name());
    if (resp.getFailedShards() > 0) {
        for (ShardSearchFailure ssf : resp.getShardFailures()) {
            source.put("shard_failure", ssf.reason());
        }
    }

    addQuery(source, querySettings);

    if (resp.getHits().totalHits() > 0) {
        List<String> hitIds = new ArrayList<>();
        for (SearchHit hit : resp.getHits().getHits()) {
            hitIds.add(hit.getId());
        }
        source.put("returned_hits", hitIds.size());
        source.put(FIELD_HITS_ID, hitIds);

    }

    writeStatisticsRecord(type, dateInMillis, source);
}