From 6bf110a6015c47104308060ee93265b5ef09207c Mon Sep 17 00:00:00 2001 From: Henning Andersen <33268011+henningandersen@users.noreply.github.com> Date: Wed, 13 May 2020 09:41:02 +0200 Subject: [PATCH] Ensure search contexts are removed on index delete (#56335) (#56617) In a race condition, a search context could remain enlisted in SearchService when an index is deleted, potentially causing the index folder to not be cleaned up (for either lengthy searches or scrolls with timeouts > 30 minutes or if the scroll is kept active). --- .../elasticsearch/search/SearchService.java | 3 + .../search/SearchServiceTests.java | 58 ++++++++++++++++++- .../search/MockSearchService.java | 10 +++- 3 files changed, 69 insertions(+), 2 deletions(-) diff --git a/server/src/main/java/org/elasticsearch/search/SearchService.java b/server/src/main/java/org/elasticsearch/search/SearchService.java index 07bb92815484a..2c8d36cf09232 100644 --- a/server/src/main/java/org/elasticsearch/search/SearchService.java +++ b/server/src/main/java/org/elasticsearch/search/SearchService.java @@ -607,6 +607,9 @@ final SearchContext createAndPutContext(ShardSearchRequest request) throws IOExc boolean success = false; try { putContext(context); + // ensure that if we race against afterIndexRemoved, we free the context here. + // this is important to ensure store can be cleaned up, in particular if the search is a scroll with a long timeout. + indicesService.indexServiceSafe(request.shardId().getIndex()); success = true; return context; } finally { diff --git a/server/src/test/java/org/elasticsearch/search/SearchServiceTests.java b/server/src/test/java/org/elasticsearch/search/SearchServiceTests.java index a904c67a5c4be..05e6cb6e286ca 100644 --- a/server/src/test/java/org/elasticsearch/search/SearchServiceTests.java +++ b/server/src/test/java/org/elasticsearch/search/SearchServiceTests.java @@ -42,6 +42,7 @@ import org.elasticsearch.common.xcontent.XContentBuilder; import org.elasticsearch.index.Index; import org.elasticsearch.index.IndexModule; +import org.elasticsearch.index.IndexNotFoundException; import org.elasticsearch.index.IndexService; import org.elasticsearch.index.IndexSettings; import org.elasticsearch.index.query.AbstractQueryBuilder; @@ -113,7 +114,8 @@ protected boolean resetNodeAfterTest() { @Override protected Collection> getPlugins() { - return pluginList(FailOnRewriteQueryPlugin.class, CustomScriptPlugin.class, InternalOrPrivateSettingsPlugin.class); + return pluginList(FailOnRewriteQueryPlugin.class, CustomScriptPlugin.class, InternalOrPrivateSettingsPlugin.class, + MockSearchService.TestPlugin.class); } public static class CustomScriptPlugin extends MockScriptPlugin { @@ -288,6 +290,7 @@ public void onFailure(Exception e) { service.executeFetchPhase(req, new SearchTask(123L, "", "", "", null, Collections.emptyMap()), listener); listener.get(); if (useScroll) { + // have to free context since this test does not remove the index from IndicesService. service.freeContext(searchPhaseResult.getRequestId()); } } catch (ExecutionException ex) { @@ -316,6 +319,59 @@ public void onFailure(Exception e) { assertEquals(0, totalStats.getFetchCurrent()); } + public void testSearchWhileIndexDeletedDoesNotLeakSearchContext() throws ExecutionException, InterruptedException { + createIndex("index"); + client().prepareIndex("index", "type", "1").setSource("field", "value").setRefreshPolicy(IMMEDIATE).get(); + + IndicesService indicesService = getInstanceFromNode(IndicesService.class); + IndexService indexService = indicesService.indexServiceSafe(resolveIndex("index")); + IndexShard indexShard = indexService.getShard(0); + + MockSearchService service = (MockSearchService) getInstanceFromNode(SearchService.class); + service.setOnPutContext( + context -> { + if (context.indexShard() == indexShard) { + assertAcked(client().admin().indices().prepareDelete("index")); + } + } + ); + + SearchRequest searchRequest = new SearchRequest().allowPartialSearchResults(true); + SearchRequest scrollSearchRequest = new SearchRequest().allowPartialSearchResults(true) + .scroll(new Scroll(TimeValue.timeValueMinutes(1))); + + // the scrolls are not explicitly freed, but should all be gone when the test finished. + // for completeness, we also randomly test the regular search path. + final boolean useScroll = randomBoolean(); + PlainActionFuture result = new PlainActionFuture<>(); + ShardSearchLocalRequest shardRequest; + if (useScroll) { + shardRequest = new ShardScrollRequestTest(indexShard.shardId()); + } else { + shardRequest = new ShardSearchLocalRequest(indexShard.shardId(), 1, SearchType.DEFAULT, + new SearchSourceBuilder(), new String[0], false, new AliasFilter(null, Strings.EMPTY_ARRAY), 1.0f, + true, null, null); + } + service.executeQueryPhase( + shardRequest, + new SearchTask(123L, "", "", "", null, Collections.emptyMap()), result); + + try { + result.get(); + } catch (Exception e) { + // ok + } + + expectThrows(IndexNotFoundException.class, () -> client().admin().indices().prepareGetIndex().setIndices("index").get()); + + assertEquals(0, service.getActiveContexts()); + + SearchStats.Stats totalStats = indexShard.searchStats().getTotal(); + assertEquals(0, totalStats.getQueryCurrent()); + assertEquals(0, totalStats.getScrollCurrent()); + assertEquals(0, totalStats.getFetchCurrent()); + } + public void testTimeout() throws IOException { createIndex("index"); final SearchService service = getInstanceFromNode(SearchService.class); diff --git a/test/framework/src/main/java/org/elasticsearch/search/MockSearchService.java b/test/framework/src/main/java/org/elasticsearch/search/MockSearchService.java index b9d9ff3cfc9bb..59d872184170b 100644 --- a/test/framework/src/main/java/org/elasticsearch/search/MockSearchService.java +++ b/test/framework/src/main/java/org/elasticsearch/search/MockSearchService.java @@ -32,6 +32,7 @@ import java.util.HashMap; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; +import java.util.function.Consumer; public class MockSearchService extends SearchService { /** @@ -41,6 +42,8 @@ public static class TestPlugin extends Plugin {} private static final Map ACTIVE_SEARCH_CONTEXTS = new ConcurrentHashMap<>(); + private Consumer onPutContext = context -> {}; + /** Throw an {@link AssertionError} if there are still in-flight contexts. */ public static void assertNoInFlightContext() { final Map copy = new HashMap<>(ACTIVE_SEARCH_CONTEXTS); @@ -74,8 +77,9 @@ public MockSearchService(ClusterService clusterService, @Override protected void putContext(SearchContext context) { - super.putContext(context); + onPutContext.accept(context); addActiveContext(context); + super.putContext(context); } @Override @@ -86,4 +90,8 @@ protected SearchContext removeContext(long id) { } return removed; } + + public void setOnPutContext(Consumer onPutContext) { + this.onPutContext = onPutContext; + } }