From ffeb52742df181dbf26544ecdc4515f7a3301830 Mon Sep 17 00:00:00 2001 From: Armin Braun Date: Wed, 4 Aug 2021 11:19:46 +0200 Subject: [PATCH] Ensure Node Shutdown Waits for Running Restores to Complete We must wait for ongoing restores to complete before shutting down the repositories service. Otherwise we may leak file descriptors because tasks for releasing the store are submitted to the `SNAPSHOT` or some searchable snapshot pools that quietly accept but never reject/fail tasks after shutdown. same as #46178 where we had the same bug in recoveries closes #75686 --- .../repositories/FilterRepository.java | 5 ++ .../repositories/RepositoriesService.java | 3 ++ .../repositories/Repository.java | 9 ++++ .../blobstore/BlobStoreRepository.java | 51 ++++++++++++++++++- .../RepositoriesServiceTests.java | 3 ++ .../index/shard/RestoreOnlyRepository.java | 4 ++ .../xpack/ccr/repository/CcrRepository.java | 4 ++ 7 files changed, 78 insertions(+), 1 deletion(-) diff --git a/server/src/main/java/org/elasticsearch/repositories/FilterRepository.java b/server/src/main/java/org/elasticsearch/repositories/FilterRepository.java index 3a87611e27dfc..40926109a53ed 100644 --- a/server/src/main/java/org/elasticsearch/repositories/FilterRepository.java +++ b/server/src/main/java/org/elasticsearch/repositories/FilterRepository.java @@ -181,6 +181,11 @@ public void cloneShardSnapshot( in.cloneShardSnapshot(source, target, shardId, shardGeneration, listener); } + @Override + public void awaitIdle() { + in.awaitIdle(); + } + @Override public Lifecycle.State lifecycleState() { return in.lifecycleState(); diff --git a/server/src/main/java/org/elasticsearch/repositories/RepositoriesService.java b/server/src/main/java/org/elasticsearch/repositories/RepositoriesService.java index f18cf651f1457..d0b0ce99d16ec 100644 --- a/server/src/main/java/org/elasticsearch/repositories/RepositoriesService.java +++ b/server/src/main/java/org/elasticsearch/repositories/RepositoriesService.java @@ -771,5 +771,8 @@ protected void doClose() throws IOException { repos.addAll(internalRepositories.values()); repos.addAll(repositories.values()); IOUtils.close(repos); + for (Repository repo : repos) { + repo.awaitIdle(); + } } } diff --git a/server/src/main/java/org/elasticsearch/repositories/Repository.java b/server/src/main/java/org/elasticsearch/repositories/Repository.java index 0bb267f325559..5d5d280e477da 100644 --- a/server/src/main/java/org/elasticsearch/repositories/Repository.java +++ b/server/src/main/java/org/elasticsearch/repositories/Repository.java @@ -321,6 +321,15 @@ default Map adaptUserMetadata(Map userMetadata) return userMetadata; } + /** + * Block until all in-flight operations for this repository have completed. Must only be called after this instance has been closed + * by a call to stop {@link #close()}. + * Waiting for ongoing operations should be implemented here instead of in {@link #stop()} or {@link #close()} hooks of this interface + * as these are expected to be called on the cluster state applier thread (which must not block) if a repository is removed from the + * cluster. This method is intended to be called on node shutdown instead as a means to ensure no repository operations are leaked. + */ + void awaitIdle(); + static boolean assertSnapshotMetaThread() { final String threadName = Thread.currentThread().getName(); assert threadName.contains('[' + ThreadPool.Names.SNAPSHOT_META + ']') || threadName.startsWith("TEST-") diff --git a/server/src/main/java/org/elasticsearch/repositories/blobstore/BlobStoreRepository.java b/server/src/main/java/org/elasticsearch/repositories/blobstore/BlobStoreRepository.java index 5582b875aa220..c9da2e78baea2 100644 --- a/server/src/main/java/org/elasticsearch/repositories/blobstore/BlobStoreRepository.java +++ b/server/src/main/java/org/elasticsearch/repositories/blobstore/BlobStoreRepository.java @@ -30,6 +30,7 @@ import org.elasticsearch.action.StepListener; import org.elasticsearch.action.support.GroupedActionListener; import org.elasticsearch.action.support.ListenableActionFuture; +import org.elasticsearch.action.support.PlainActionFuture; import org.elasticsearch.action.support.ThreadedActionListener; import org.elasticsearch.cluster.ClusterState; import org.elasticsearch.cluster.ClusterStateUpdateTask; @@ -66,6 +67,7 @@ import org.elasticsearch.common.util.BigArrays; import org.elasticsearch.common.util.concurrent.AbstractRunnable; import org.elasticsearch.common.util.concurrent.ConcurrentCollections; +import org.elasticsearch.common.util.concurrent.FutureUtils; import org.elasticsearch.common.xcontent.LoggingDeprecationHandler; import org.elasticsearch.common.xcontent.NamedXContentRegistry; import org.elasticsearch.common.xcontent.XContentBuilder; @@ -422,6 +424,30 @@ protected void doClose() { } } + // listeners to invoke when a restore completes and there are no more restores running + @Nullable + private List> emptyListeners; + + // Set of shard ids that this repository is currently restoring + private final Set ongoingRestores = new HashSet<>(); + + @Override + public void awaitIdle() { + assert lifecycle.stoppedOrClosed(); + final PlainActionFuture future; + synchronized (ongoingRestores) { + if (ongoingRestores.isEmpty()) { + return; + } + future = new PlainActionFuture<>(); + if (emptyListeners == null) { + emptyListeners = new ArrayList<>(); + } + emptyListeners.add(future); + } + FutureUtils.get(future); + } + @Override public void executeConsistentStateUpdate( Function createUpdateTask, @@ -2875,7 +2901,30 @@ public void restoreShard( ); final Executor executor = threadPool.executor(ThreadPool.Names.SNAPSHOT); final BlobContainer container = shardContainer(indexId, snapshotShardId); - executor.execute(ActionRunnable.wrap(restoreListener, l -> { + synchronized (ongoingRestores) { + if (store.isClosing()) { + restoreListener.onFailure(new AlreadyClosedException("store is closing")); + return; + } + if (lifecycle.started() == false) { + restoreListener.onFailure(new AlreadyClosedException("repository [" + metadata.name() + "] closed")); + return; + } + final boolean added = ongoingRestores.add(shardId); + assert added; + } + executor.execute(ActionRunnable.wrap(ActionListener.runAfter(restoreListener, () -> { + final List> onEmptyListeners; + synchronized (ongoingRestores) { + if (ongoingRestores.remove(shardId) && ongoingRestores.isEmpty() && emptyListeners != null) { + onEmptyListeners = emptyListeners; + emptyListeners = null; + } else { + return; + } + ActionListener.onResponse(onEmptyListeners, null); + } + }), l -> { final BlobStoreIndexShardSnapshot snapshot = loadShardSnapshot(container, snapshotId); final SnapshotFiles snapshotFiles = new SnapshotFiles(snapshot.snapshot(), snapshot.indexFiles(), null); new FileRestoreContext(metadata.name(), shardId, snapshotId, recoveryState) { diff --git a/server/src/test/java/org/elasticsearch/repositories/RepositoriesServiceTests.java b/server/src/test/java/org/elasticsearch/repositories/RepositoriesServiceTests.java index 88ebac243fa0f..de758d9234005 100644 --- a/server/src/test/java/org/elasticsearch/repositories/RepositoriesServiceTests.java +++ b/server/src/test/java/org/elasticsearch/repositories/RepositoriesServiceTests.java @@ -317,6 +317,9 @@ public void cloneShardSnapshot( } + @Override + public void awaitIdle() {} + @Override public Lifecycle.State lifecycleState() { return null; diff --git a/test/framework/src/main/java/org/elasticsearch/index/shard/RestoreOnlyRepository.java b/test/framework/src/main/java/org/elasticsearch/index/shard/RestoreOnlyRepository.java index 25c07e8b7a689..7bd97956f6605 100644 --- a/test/framework/src/main/java/org/elasticsearch/index/shard/RestoreOnlyRepository.java +++ b/test/framework/src/main/java/org/elasticsearch/index/shard/RestoreOnlyRepository.java @@ -147,6 +147,10 @@ public void verify(String verificationToken, DiscoveryNode localNode) { public void updateState(final ClusterState state) { } + @Override + public void awaitIdle() { + } + @Override public void executeConsistentStateUpdate(Function createUpdateTask, String source, Consumer onFailure) { diff --git a/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/repository/CcrRepository.java b/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/repository/CcrRepository.java index 8a146aefe42fd..061362bc00dce 100644 --- a/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/repository/CcrRepository.java +++ b/x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/repository/CcrRepository.java @@ -491,6 +491,10 @@ public void cloneShardSnapshot(SnapshotId source, SnapshotId target, RepositoryS throw new UnsupportedOperationException("Unsupported for repository of type: " + TYPE); } + @Override + public void awaitIdle() { + } + private void updateMappings(Client leaderClient, Index leaderIndex, long leaderMappingVersion, Client followerClient, Index followerIndex) { final PlainActionFuture indexMetadataFuture = new PlainActionFuture<>();