Implement parallel shard refresh behind cluster settings

Signed-off-by: Ashish Singh <[email protected]>

Author: ashking94

server/src/internalClusterTest/java/org/opensearch/index/shard/IndexShardIT.java

@@ -719,7 +719,10 @@ public static final IndexShard newIndexShard(
             DefaultRemoteStoreSettings.INSTANCE,
             false,
             IndexShardTestUtils.getFakeDiscoveryNodes(initializingShardRouting),
-            mock(Function.class)
+            mock(Function.class),
+            false,
+            () -> Boolean.FALSE,
+            () -> indexService.getRefreshInterval()
         );
     }
 

server/src/main/java/org/opensearch/common/settings/ClusterSettings.java

@@ -823,7 +823,8 @@ public void apply(Settings value, Settings current, Settings previous) {
                 ),
 
                 // Setting related to refresh optimisations
-                IndicesService.CLUSTER_REFRESH_FIXED_INTERVAL_SCHEDULE_ENABLED_SETTING
+                IndicesService.CLUSTER_REFRESH_FIXED_INTERVAL_SCHEDULE_ENABLED_SETTING,
+                IndicesService.CLUSTER_REFRESH_SHARD_LEVEL_ENABLED_SETTING
             )
         )
     );

server/src/main/java/org/opensearch/index/IndexModule.java

@@ -631,6 +631,7 @@ public IndexService newIndexService(
         BiFunction<IndexSettings, ShardRouting, TranslogFactory> translogFactorySupplier,
         Supplier<TimeValue> clusterDefaultRefreshIntervalSupplier,
         Supplier<Boolean> fixedRefreshIntervalSchedulingEnabled,
+        Supplier<Boolean> shardLevelRefreshEnabled,
         RecoverySettings recoverySettings,
         RemoteStoreSettings remoteStoreSettings
     ) throws IOException {
@@ -655,6 +656,7 @@ public IndexService newIndexService(
             translogFactorySupplier,
             clusterDefaultRefreshIntervalSupplier,
             fixedRefreshIntervalSchedulingEnabled,
+            shardLevelRefreshEnabled,
             recoverySettings,
             remoteStoreSettings,
             (s) -> {},
@@ -683,6 +685,7 @@ public IndexService newIndexService(
         BiFunction<IndexSettings, ShardRouting, TranslogFactory> translogFactorySupplier,
         Supplier<TimeValue> clusterDefaultRefreshIntervalSupplier,
         Supplier<Boolean> fixedRefreshIntervalSchedulingEnabled,
+        Supplier<Boolean> shardLevelRefreshEnabled,
         RecoverySettings recoverySettings,
         RemoteStoreSettings remoteStoreSettings,
         Consumer<IndexShard> replicator,
@@ -745,6 +748,7 @@ public IndexService newIndexService(
                 translogFactorySupplier,
                 clusterDefaultRefreshIntervalSupplier,
                 fixedRefreshIntervalSchedulingEnabled,
+                shardLevelRefreshEnabled.get(),
                 recoverySettings,
                 remoteStoreSettings,
                 fileCache,

server/src/main/java/org/opensearch/index/IndexService.java

@@ -199,6 +199,9 @@ public class IndexService extends AbstractIndexComponent implements IndicesClust
     private final CompositeIndexSettings compositeIndexSettings;
     private final Consumer<IndexShard> replicator;
     private final Function<ShardId, ReplicationStats> segmentReplicationStatsProvider;
+    private final Object refreshMutex = new Object();
+    private volatile TimeValue refreshInterval;
+    private volatile boolean shardLevelRefreshEnabled;
 
     public IndexService(
         IndexSettings indexSettings,
@@ -234,6 +237,7 @@ public IndexService(
         BiFunction<IndexSettings, ShardRouting, TranslogFactory> translogFactorySupplier,
         Supplier<TimeValue> clusterDefaultRefreshIntervalSupplier,
         Supplier<Boolean> fixedRefreshIntervalSchedulingEnabled,
+        boolean shardLevelRefreshEnabled,
         RecoverySettings recoverySettings,
         RemoteStoreSettings remoteStoreSettings,
         FileCache fileCache,
@@ -310,8 +314,9 @@ public IndexService(
         this.indexingOperationListeners = Collections.unmodifiableList(indexingOperationListeners);
         this.clusterDefaultRefreshIntervalSupplier = clusterDefaultRefreshIntervalSupplier;
         this.fixedRefreshIntervalSchedulingEnabled = fixedRefreshIntervalSchedulingEnabled;
+        this.shardLevelRefreshEnabled = shardLevelRefreshEnabled;
+        this.refreshInterval = getRefreshInterval();
         // kick off async ops for the first shard in this index
-        this.refreshTask = new AsyncRefreshTask(this);
         this.trimTranslogTask = new AsyncTrimTranslogTask(this);
         // disable these checks for ingestion source engine
         if (!indexSettings.getIndexMetadata().useIngestionSource()) {
@@ -329,6 +334,10 @@ public IndexService(
         this.replicator = replicator;
         this.segmentReplicationStatsProvider = segmentReplicationStatsProvider;
         updateFsyncTaskIfNecessary();
+        if (shardLevelRefreshEnabled == false) {
+            logger.debug("shard level refresh is disabled for index [{}]", indexSettings.getIndex().getName());
+            startIndexLevelRefreshTask();
+        }
     }
 
     public IndexService(
@@ -365,6 +374,7 @@ public IndexService(
         BiFunction<IndexSettings, ShardRouting, TranslogFactory> translogFactorySupplier,
         Supplier<TimeValue> clusterDefaultRefreshIntervalSupplier,
         Supplier<Boolean> fixedRefreshIntervalSchedulingEnabled,
+        boolean shardLevelRefreshEnabled,
         RecoverySettings recoverySettings,
         RemoteStoreSettings remoteStoreSettings
     ) {
@@ -402,6 +412,7 @@ public IndexService(
             translogFactorySupplier,
             clusterDefaultRefreshIntervalSupplier,
             fixedRefreshIntervalSchedulingEnabled,
+            shardLevelRefreshEnabled,
             recoverySettings,
             remoteStoreSettings,
             null,
@@ -704,7 +715,10 @@ protected void closeInternal() {
                 remoteStoreSettings,
                 seedRemote,
                 discoveryNodes,
-                segmentReplicationStatsProvider
+                segmentReplicationStatsProvider,
+                shardLevelRefreshEnabled,
+                fixedRefreshIntervalSchedulingEnabled,
+                this::getRefreshInterval
             );
             eventListener.indexShardStateChanged(indexShard, null, indexShard.state(), "shard created");
             eventListener.afterIndexShardCreated(indexShard);
@@ -1133,9 +1147,10 @@ private void updateReplicationTask() {
      * 2. {@code index.refresh_interval} index setting changes.
      */
     public void onRefreshIntervalChange() {
-        if (refreshTask.getInterval().equals(getRefreshInterval())) {
+        if (refreshInterval.equals(getRefreshInterval())) {
             return;
         }
+        refreshInterval = getRefreshInterval();
         // once we change the refresh interval we schedule yet another refresh
         // to ensure we are in a clean and predictable state.
         // it doesn't matter if we move from or to <code>-1</code> in both cases we want
@@ -1177,10 +1192,20 @@ private void updateFsyncTaskIfNecessary() {
     }
 
     private void rescheduleRefreshTasks() {
-        try {
-            refreshTask.close();
-        } finally {
-            refreshTask = new AsyncRefreshTask(this);
+        synchronized (refreshMutex) {
+            if (shardLevelRefreshEnabled) {
+                try {
+                    stopShardLevelRefreshTasks();
+                } finally {
+                    startShardLevelRefreshTasks();
+                }
+            } else {
+                try {
+                    stopIndexLevelRefreshTask();
+                } finally {
+                    startIndexLevelRefreshTask();
+                }
+            }
         }
     }
 
@@ -1304,7 +1329,7 @@ private void sync(final Consumer<IndexShard> sync, final String source) {
      * Gets the refresh interval seen by the index service. Index setting overrides takes the highest precedence.
      * @return the refresh interval.
      */
-    private TimeValue getRefreshInterval() {
+    public TimeValue getRefreshInterval() {
         if (getIndexSettings().isExplicitRefresh()) {
             return getIndexSettings().getRefreshInterval();
         }
@@ -1584,4 +1609,60 @@ public boolean clearCaches(boolean queryCache, boolean fieldDataCache, String...
         return clearedAtLeastOne;
     }
 
+    /**
+     * This method is called when the refresh level is changed from index level to shard level or vice versa.
+     */
+    public void onRefreshLevelChange(boolean newShardLevelRefreshVal) {
+        synchronized (refreshMutex) {
+            if (this.shardLevelRefreshEnabled != newShardLevelRefreshVal) {
+                boolean prevShardLevelRefreshVal = this.shardLevelRefreshEnabled;
+                this.shardLevelRefreshEnabled = newShardLevelRefreshVal;
+                // The refresh mode has changed from index level to shard level
+                logger.info("refresh tasks rescheduled oldVal={} newVal={}", prevShardLevelRefreshVal, newShardLevelRefreshVal);
+                if (newShardLevelRefreshVal) {
+                    try {
+                        stopIndexLevelRefreshTask();
+                    } finally {
+                        startShardLevelRefreshTasks();
+                    }
+                } else {
+                    try {
+                        stopShardLevelRefreshTasks();
+                    } finally {
+                        startIndexLevelRefreshTask();
+                    }
+                }
+            }
+        }
+    }
+
+    private void stopIndexLevelRefreshTask() {
+        // The refresh task is expected to be non-null at this point.
+        assert Objects.nonNull(refreshTask);
+        refreshTask.close();
+        refreshTask = null;
+    }
+
+    private void startShardLevelRefreshTasks() {
+        for (IndexShard shard : this.shards.values()) {
+            shard.startRefreshTask();
+        }
+    }
+
+    private void stopShardLevelRefreshTasks() {
+        for (IndexShard shard : this.shards.values()) {
+            shard.stopRefreshTask();
+        }
+    }
+
+    private void startIndexLevelRefreshTask() {
+        // The refresh task is expected to be null at this point.
+        assert Objects.isNull(refreshTask);
+        refreshTask = new AsyncRefreshTask(this);
+    }
+
+    // Visible for testing
+    boolean isShardLevelRefreshEnabled() {
+        return shardLevelRefreshEnabled;
+    }
 }

server/src/main/java/org/opensearch/index/shard/IndexShard.java

@@ -82,6 +82,7 @@
 import org.opensearch.common.CheckedRunnable;
 import org.opensearch.common.Nullable;
 import org.opensearch.common.SetOnce;
+import org.opensearch.common.annotation.ExperimentalApi;
 import org.opensearch.common.annotation.PublicApi;
 import org.opensearch.common.collect.Tuple;
 import org.opensearch.common.concurrent.GatedCloseable;
@@ -95,6 +96,7 @@
 import org.opensearch.common.settings.Settings;
 import org.opensearch.common.unit.TimeValue;
 import org.opensearch.common.util.BigArrays;
+import org.opensearch.common.util.concurrent.AbstractAsyncTask;
 import org.opensearch.common.util.concurrent.AbstractRunnable;
 import org.opensearch.common.util.concurrent.AsyncIOProcessor;
 import org.opensearch.common.util.concurrent.BufferedAsyncIOProcessor;
@@ -367,6 +369,9 @@ Runnable getGlobalCheckpointSyncer() {
     private final ShardMigrationState shardMigrationState;
     private DiscoveryNodes discoveryNodes;
     private final Function<ShardId, ReplicationStats> segmentReplicationStatsProvider;
+    private final Supplier<Boolean> fixedRefreshIntervalSchedulingEnabled;
+    private final Supplier<TimeValue> refreshInterval;
+    private volatile AsyncShardRefreshTask refreshTask;
 
     public IndexShard(
         final ShardRouting shardRouting,
@@ -398,7 +403,10 @@ public IndexShard(
         final RemoteStoreSettings remoteStoreSettings,
         boolean seedRemote,
         final DiscoveryNodes discoveryNodes,
-        final Function<ShardId, ReplicationStats> segmentReplicationStatsProvider
+        final Function<ShardId, ReplicationStats> segmentReplicationStatsProvider,
+        final boolean shardLevelRefreshEnabled,
+        final Supplier<Boolean> fixedRefreshIntervalSchedulingEnabled,
+        final Supplier<TimeValue> refreshInterval
     ) throws IOException {
         super(shardRouting.shardId(), indexSettings);
         assert shardRouting.initializing();
@@ -501,6 +509,11 @@ public boolean shouldCache(Query query) {
         this.shardMigrationState = getShardMigrationState(indexSettings, seedRemote);
         this.discoveryNodes = discoveryNodes;
         this.segmentReplicationStatsProvider = segmentReplicationStatsProvider;
+        this.fixedRefreshIntervalSchedulingEnabled = fixedRefreshIntervalSchedulingEnabled;
+        this.refreshInterval = refreshInterval;
+        if (shardLevelRefreshEnabled) {
+            startRefreshTask();
+        }
     }
 
     /**
@@ -2099,7 +2112,7 @@ public void close(String reason, boolean flushEngine, boolean deleted) throws IO
                 } finally {
                     // playing safe here and close the engine even if the above succeeds - close can be called multiple times
                     // Also closing refreshListeners to prevent us from accumulating any more listeners
-                    IOUtils.close(engine, globalCheckpointListeners, refreshListeners, pendingReplicationActions);
+                    IOUtils.close(engine, globalCheckpointListeners, refreshListeners, pendingReplicationActions, refreshTask);
 
                     if (deleted && engine != null && isPrimaryMode()) {
                         // Translog Clean up
@@ -5484,4 +5497,57 @@ public ShardIngestionState getIngestionState() {
         IngestionEngine ingestionEngine = (IngestionEngine) engine;
         return ingestionEngine.getIngestionState();
     }
+
+    @ExperimentalApi
+    public final class AsyncShardRefreshTask extends AbstractAsyncTask {
+
+        private final IndexShard indexShard;
+        private final Logger logger;
+
+        public AsyncShardRefreshTask(IndexShard indexShard) {
+            super(indexShard.logger, indexShard.threadPool, refreshInterval.get(), true, fixedRefreshIntervalSchedulingEnabled);
+            this.logger = indexShard.logger;
+            this.indexShard = indexShard;
+            rescheduleIfNecessary();
+        }
+
+        @Override
+        protected boolean mustReschedule() {
+            return indexShard.state != IndexShardState.CLOSED
+                && indexShard.indexSettings.getIndexMetadata().getState() == IndexMetadata.State.OPEN;
+        }
+
+        @Override
+        protected void runInternal() {
+            indexShard.scheduledRefresh();
+        }
+
+        @Override
+        protected String getThreadPool() {
+            return ThreadPool.Names.REFRESH;
+        }
+
+        @Override
+        public String toString() {
+            return "shard_refresh";
+        }
+    }
+
+    public void startRefreshTask() {
+        // The refresh task is expected to be null at this point.
+        assert Objects.isNull(refreshTask);
+        refreshTask = new AsyncShardRefreshTask(this);
+    }
+
+    public void stopRefreshTask() {
+        // The refresh task is expected to be non-null at this point.
+        assert Objects.nonNull(refreshTask);
+        refreshTask.close();
+        refreshTask = null;
+    }
+
+    // Visible for testing
+    public AsyncShardRefreshTask getRefreshTask() {
+        return refreshTask;
+    }
 }