Introduce global checkpoint background sync #26591
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -25,11 +25,15 @@ | |
| import org.apache.lucene.store.AlreadyClosedException; | ||
| import org.apache.lucene.util.Accountable; | ||
| import org.apache.lucene.util.IOUtils; | ||
| import org.elasticsearch.action.ActionListener; | ||
| import org.elasticsearch.client.Client; | ||
| import org.elasticsearch.cluster.metadata.IndexMetaData; | ||
| import org.elasticsearch.cluster.routing.ShardRouting; | ||
| import org.elasticsearch.common.Nullable; | ||
| import org.elasticsearch.common.io.stream.NamedWriteableRegistry; | ||
| import org.elasticsearch.common.lease.Releasable; | ||
| import org.elasticsearch.common.settings.Setting; | ||
| import org.elasticsearch.common.settings.Setting.Property; | ||
| import org.elasticsearch.common.settings.Settings; | ||
| import org.elasticsearch.common.unit.TimeValue; | ||
| import org.elasticsearch.common.util.BigArrays; | ||
|
|
@@ -82,6 +86,7 @@ | |
| import java.util.concurrent.ScheduledFuture; | ||
| import java.util.concurrent.TimeUnit; | ||
| import java.util.concurrent.atomic.AtomicBoolean; | ||
| import java.util.function.Consumer; | ||
| import java.util.function.LongSupplier; | ||
| import java.util.function.Supplier; | ||
|
|
||
|
|
@@ -109,10 +114,11 @@ public class IndexService extends AbstractIndexComponent implements IndicesClust | |
| private final AtomicBoolean closed = new AtomicBoolean(false); | ||
| private final AtomicBoolean deleted = new AtomicBoolean(false); | ||
| private final IndexSettings indexSettings; | ||
| private final List<SearchOperationListener> searchOperationListeners; | ||
| private final List<IndexingOperationListener> indexingOperationListeners; | ||
| private volatile AsyncRefreshTask refreshTask; | ||
| private volatile AsyncTranslogFSync fsyncTask; | ||
| private volatile AsyncGlobalCheckpointTask globalCheckpointTask; | ||
|
|
||
| // don't convert to Setting<> and register... we only set this in tests and register via a plugin | ||
| private final String INDEX_TRANSLOG_RETENTION_CHECK_INTERVAL_SETTING = "index.translog.retention.check_interval"; | ||
|
|
@@ -182,11 +188,12 @@ public IndexService( | |
| this.engineFactory = engineFactory; | ||
| // initialize this last -- otherwise if the wrapper requires any other member to be non-null we fail with an NPE | ||
| this.searcherWrapper = wrapperFactory.newWrapper(this); | ||
| this.searchOperationListeners = Collections.unmodifiableList(searchOperationListeners); | ||
| this.indexingOperationListeners = Collections.unmodifiableList(indexingOperationListeners); | ||
| // kick off async ops for the first shard in this index | ||
| this.refreshTask = new AsyncRefreshTask(this); | ||
| this.trimTranslogTask = new AsyncTrimTranslogTask(this); | ||
| this.globalCheckpointTask = new AsyncGlobalCheckpointTask(this); | ||
| rescheduleFsyncTask(indexSettings.getTranslogDurability()); | ||
| } | ||
|
|
||
|
|
@@ -268,7 +275,15 @@ public synchronized void close(final String reason, boolean delete) throws IOExc | |
| } | ||
| } | ||
| } finally { | ||
| IOUtils.close( | ||
| bitsetFilterCache, | ||
| indexCache, | ||
| indexFieldData, | ||
| mapperService, | ||
| refreshTask, | ||
| fsyncTask, | ||
| trimTranslogTask, | ||
| globalCheckpointTask); | ||
| } | ||
| } | ||
| } | ||
|
|
@@ -293,8 +308,7 @@ private long getAvgShardSizeInBytes() throws IOException { | |
| } | ||
| } | ||
|
|
||
| public synchronized IndexShard createShard(ShardRouting routing, Consumer<ShardId> globalCheckpointSyncer) throws IOException { | ||
| /* | ||
| * TODO: we execute this in parallel but it's a synced method. Yet, we might | ||
| * be able to serialize the execution via the cluster state in the future. for now we just | ||
|
|
@@ -365,7 +379,7 @@ public synchronized IndexShard createShard(ShardRouting routing) throws IOExcept | |
| indexShard = new IndexShard(routing, this.indexSettings, path, store, indexSortSupplier, | ||
| indexCache, mapperService, similarityService, engineFactory, | ||
| eventListener, searcherWrapper, threadPool, bigArrays, engineWarmer, | ||
| searchOperationListeners, indexingOperationListeners, () -> globalCheckpointSyncer.accept(shardId)); | ||
| eventListener.indexShardStateChanged(indexShard, null, indexShard.state(), "shard created"); | ||
| eventListener.afterIndexShardCreated(indexShard); | ||
| shards = newMapBuilder(shards).put(shardId.id(), indexShard).immutableMap(); | ||
|
|
@@ -710,6 +724,44 @@ private void maybeTrimTranslog() { | |
| } | ||
| } | ||
|
|
||
| private void maybeSyncGlobalCheckpoints() { | ||
| for (final IndexShard shard : this.shards.values()) { | ||
| if (shard.routingEntry().active() && shard.routingEntry().primary()) { | ||
| switch (shard.state()) { | ||
| case CLOSED: | ||
| case CREATED: | ||
| case RECOVERING: | ||
| case RELOCATED: | ||
| continue; | ||
| case POST_RECOVERY: | ||
|
There was a problem hiding this comment. routing entry cannot be active while shard state is POST_RECOVERY ;-) There was a problem hiding this comment. Addressed in 9bf875e. |
||
| assert false : "shard " + shard.shardId() + " is in post-recovery but marked as active"; | ||
| continue; | ||
| case STARTED: | ||
| try { | ||
| shard.acquirePrimaryOperationPermit( | ||
| ActionListener.wrap( | ||
| releasable -> { | ||
| try (Releasable ignored = releasable) { | ||
| shard.maybeSyncGlobalCheckpoint("background"); | ||
| } | ||
| }, | ||
| e -> { | ||
| if (!(e instanceof AlreadyClosedException || e instanceof IndexShardClosedException)) { | ||
| logger.info("failed to execute background global checkpoint sync", e); | ||
| } | ||
| }), | ||
| ThreadPool.Names.SAME); | ||
| } catch (final AlreadyClosedException | IndexShardClosedException e) { | ||
| // the shard was closed concurrently, continue | ||
| } | ||
| continue; | ||
| default: | ||
| throw new IllegalStateException("unknown state [" + shard.state() + "]"); | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| abstract static class BaseAsyncTask implements Runnable, Closeable { | ||
| protected final IndexService indexService; | ||
| protected final ThreadPool threadPool; | ||
|
|
@@ -877,6 +929,41 @@ public String toString() { | |
| } | ||
| } | ||
|
|
||
| // this setting is intentionally not registered, it is only used in tests | ||
| public static final Setting<TimeValue> GLOBAL_CHECKPOINT_SYNC_INTERVAL_SETTING = | ||
| Setting.timeSetting( | ||
| "index.global_checkpoint_sync.interval", | ||
| new TimeValue(30, TimeUnit.SECONDS), | ||
| new TimeValue(0, TimeUnit.MILLISECONDS), | ||
| Property.Dynamic, | ||
| Property.IndexScope); | ||
|
|
||
| /** | ||
| * Background task that syncs the global checkpoint to replicas. | ||
| */ | ||
| final class AsyncGlobalCheckpointTask extends BaseAsyncTask { | ||
|
|
||
| AsyncGlobalCheckpointTask(final IndexService indexService) { | ||
| // index.global_checkpoint_sync_interval is not a real setting, it is only registered in tests | ||
| super(indexService, GLOBAL_CHECKPOINT_SYNC_INTERVAL_SETTING.get(indexService.getIndexSettings().getSettings())); | ||
| } | ||
|
|
||
| @Override | ||
| protected void runInternal() { | ||
| indexService.maybeSyncGlobalCheckpoints(); | ||
| } | ||
|
|
||
| @Override | ||
| protected String getThreadPool() { | ||
| return ThreadPool.Names.GENERIC; | ||
| } | ||
|
|
||
| @Override | ||
| public String toString() { | ||
| return "global_checkpoint_sync"; | ||
| } | ||
| } | ||
|
|
||
| AsyncRefreshTask getRefreshTask() { // for tests | ||
| return refreshTask; | ||
| } | ||
|
|
@@ -885,6 +972,10 @@ AsyncTranslogFSync getFsyncTask() { // for tests | |
| return fsyncTask; | ||
| } | ||
|
|
||
| AsyncGlobalCheckpointTask getGlobalCheckpointTask() { | ||
| return globalCheckpointTask; | ||
| } | ||
|
|
||
| /** | ||
| * Clears the caches for the given shard id if the shard is still allocated on this node | ||
| */ | ||
|
|
||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
IDE gone rogue
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I pushed b8adcce.