Add jitter for remote download calls

Signed-off-by: Arpit Bandejiya <[email protected]>

server/src/main/java/org/opensearch/cluster/coordination/UnsafeBootstrapClusterManagerCommand.java

@@ -53,7 +53,7 @@
 import java.util.Locale;
 import java.util.Objects;
 
-import static org.opensearch.gateway.remote.RemoteClusterStateService.REMOTE_CLUSTER_STATE_ENABLED_SETTING;
+import static org.opensearch.gateway.remote.RemoteClusterStateSettings.REMOTE_CLUSTER_STATE_ENABLED_SETTING;
 
 /**
  * Tool to run an unsafe bootstrap

server/src/main/java/org/opensearch/cluster/routing/remote/InternalRemoteRoutingTableService.java

@@ -27,6 +27,8 @@
 import org.opensearch.core.action.ActionListener;
 import org.opensearch.core.compress.Compressor;
 import org.opensearch.gateway.remote.ClusterMetadataManifest;
+import org.opensearch.gateway.remote.DefaultRandomObject;
+import org.opensearch.gateway.remote.RemoteClusterStateSettings;
 import org.opensearch.gateway.remote.RemoteClusterStateUtils;
 import org.opensearch.gateway.remote.RemoteStateTransferException;
 import org.opensearch.gateway.remote.model.RemoteRoutingTableBlobStore;
@@ -45,10 +47,12 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Random;
 import java.util.function.Function;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
 
+import static org.opensearch.gateway.remote.RemoteClusterStateUtils.toPositiveLongAtMost;
 import static org.opensearch.node.remotestore.RemoteStoreNodeAttribute.isRemoteRoutingTableEnabled;
 
 /**
@@ -68,11 +72,14 @@ public class InternalRemoteRoutingTableService extends AbstractLifecycleComponen
     private BlobStoreRepository blobStoreRepository;
     private final ThreadPool threadPool;
     private final String clusterName;
+    private RemoteClusterStateSettings remoteClusterStateSettings;
+    private Random random;
 
     public InternalRemoteRoutingTableService(
         Supplier<RepositoriesService> repositoriesService,
         Settings settings,
         ClusterSettings clusterSettings,
+        RemoteClusterStateSettings remoteClusterStateSettings,
         ThreadPool threadpool,
         String clusterName
     ) {
@@ -82,6 +89,8 @@ public InternalRemoteRoutingTableService(
         this.threadPool = threadpool;
         this.clusterName = clusterName;
         this.clusterSettings = clusterSettings;
+        this.remoteClusterStateSettings = remoteClusterStateSettings;
+        this.random = DefaultRandomObject.INSTANCE;
     }
 
     public List<IndexRoutingTable> getIndicesRouting(RoutingTable routingTable) {
@@ -193,7 +202,10 @@ public void getAsyncIndexRoutingReadAction(
 
         RemoteIndexRoutingTable remoteIndexRoutingTable = new RemoteIndexRoutingTable(uploadedFilename, clusterUUID, compressor);
 
-        remoteIndexRoutingTableStore.readAsync(remoteIndexRoutingTable, actionListener);
+        long maxDelayInMillis = remoteClusterStateSettings.getRemoteStateReadMaxJitter().getMillis();
+        final long delayInMillis = toPositiveLongAtMost(random.nextLong(), maxDelayInMillis);
+        remoteIndexRoutingTableStore.readAsyncWithDelay(delayInMillis, remoteIndexRoutingTable, actionListener);
+
     }
 
     @Override
@@ -208,7 +220,10 @@ public void getAsyncIndexRoutingTableDiffReadAction(
         );
 
         RemoteRoutingTableDiff remoteRoutingTableDiff = new RemoteRoutingTableDiff(uploadedFilename, clusterUUID, compressor);
-        remoteRoutingTableDiffStore.readAsync(remoteRoutingTableDiff, actionListener);
+
+        long maxDelayInMillis = remoteClusterStateSettings.getRemoteStateReadMaxJitter().getMillis();
+        final long delayInMillis = toPositiveLongAtMost(random.nextLong(), maxDelayInMillis);
+        remoteRoutingTableDiffStore.readAsyncWithDelay(delayInMillis, remoteRoutingTableDiff, actionListener);
     }
 
     @Override

server/src/main/java/org/opensearch/cluster/routing/remote/RemoteRoutingTableServiceFactory.java

@@ -10,6 +10,7 @@
 
 import org.opensearch.common.settings.ClusterSettings;
 import org.opensearch.common.settings.Settings;
+import org.opensearch.gateway.remote.RemoteClusterStateSettings;
 import org.opensearch.repositories.RepositoriesService;
 import org.opensearch.threadpool.ThreadPool;
 
@@ -34,11 +35,19 @@ public static RemoteRoutingTableService getService(
         Supplier<RepositoriesService> repositoriesService,
         Settings settings,
         ClusterSettings clusterSettings,
+        RemoteClusterStateSettings remoteClusterStateSettings,
         ThreadPool threadPool,
         String clusterName
     ) {
         if (isRemoteRoutingTableEnabled(settings)) {
-            return new InternalRemoteRoutingTableService(repositoriesService, settings, clusterSettings, threadPool, clusterName);
+            return new InternalRemoteRoutingTableService(
+                repositoriesService,
+                settings,
+                clusterSettings,
+                remoteClusterStateSettings,
+                threadPool,
+                clusterName
+            );
         }
         return new NoopRemoteRoutingTableService();
     }

server/src/main/java/org/opensearch/common/remote/AbstractRemoteWritableEntityManager.java

@@ -8,12 +8,19 @@
 
 package org.opensearch.common.remote;
 
+import org.opensearch.common.settings.ClusterSettings;
+import org.opensearch.common.settings.Settings;
 import org.opensearch.core.action.ActionListener;
 import org.opensearch.gateway.remote.ClusterMetadataManifest;
+import org.opensearch.gateway.remote.DefaultRandomObject;
+import org.opensearch.gateway.remote.RemoteClusterStateSettings;
 import org.opensearch.gateway.remote.model.RemoteReadResult;
 
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Random;
+
+import static org.opensearch.gateway.remote.RemoteClusterStateUtils.toPositiveLongAtMost;
 
 /**
  * An abstract class that provides a base implementation for managing remote entities in the remote store.
@@ -24,6 +31,13 @@ public abstract class AbstractRemoteWritableEntityManager implements RemoteWrita
      */
     protected final Map<String, RemoteWritableEntityStore> remoteWritableEntityStores = new HashMap<>();
 
+    protected RemoteClusterStateSettings remoteClusterStateSettings = new RemoteClusterStateSettings(
+        Settings.EMPTY,
+        new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS)
+    );
+
+    protected final Random random = DefaultRandomObject.INSTANCE;
+
     /**
      * Retrieves the remote writable entity store for the given entity.
      *
@@ -79,6 +93,9 @@ public void writeAsync(
 
     @Override
     public void readAsync(String component, AbstractClusterMetadataWriteableBlobEntity entity, ActionListener<RemoteReadResult> listener) {
-        getStore(entity).readAsync(entity, getWrappedReadListener(component, entity, listener));
+        long maxDelayInMillis = remoteClusterStateSettings.getRemoteStateReadMaxJitter().getMillis();
+        final long delayInMillis = toPositiveLongAtMost(random.nextLong(), maxDelayInMillis);
+        getStore(entity).readAsyncWithDelay(delayInMillis, entity, getWrappedReadListener(component, entity, listener));
     }
+
 }

server/src/main/java/org/opensearch/common/remote/RemoteWritableEntityStore.java

@@ -27,4 +27,6 @@ public interface RemoteWritableEntityStore<T, U extends RemoteWriteableEntity<T>
     public T read(U entity) throws IOException;
 
     public void readAsync(U entity, ActionListener<T> listener);
+
+    public void readAsyncWithDelay(long delayInMillis, U entity, ActionListener<T> listener);
 }

server/src/main/java/org/opensearch/common/remote/RemoteWriteableEntityBlobStore.java

@@ -8,8 +8,11 @@
 
 package org.opensearch.common.remote;
 
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 import org.opensearch.common.blobstore.BlobPath;
 import org.opensearch.common.blobstore.stream.write.WritePriority;
+import org.opensearch.common.unit.TimeValue;
 import org.opensearch.core.action.ActionListener;
 import org.opensearch.index.translog.transfer.BlobStoreTransferService;
 import org.opensearch.repositories.blobstore.BlobStoreRepository;
@@ -29,11 +32,15 @@
  */
 public class RemoteWriteableEntityBlobStore<T, U extends RemoteWriteableBlobEntity<T>> implements RemoteWritableEntityStore<T, U> {
 
+    private static final Logger logger = LogManager.getLogger(RemoteWriteableEntityBlobStore.class);
+
     private final BlobStoreTransferService transferService;
     private final BlobStoreRepository blobStoreRepository;
     private final String clusterName;
+    private final String executor;
     private final ExecutorService executorService;
     private final String pathToken;
+    private final ThreadPool threadPool;
 
     public RemoteWriteableEntityBlobStore(
         final BlobStoreTransferService blobStoreTransferService,
@@ -46,6 +53,8 @@ public RemoteWriteableEntityBlobStore(
         this.transferService = blobStoreTransferService;
         this.blobStoreRepository = blobStoreRepository;
         this.clusterName = clusterName;
+        this.threadPool = threadPool;
+        this.executor = executor;
         this.executorService = threadPool.executor(executor);
         this.pathToken = pathToken;
     }
@@ -89,6 +98,17 @@ public void readAsync(final U entity, final ActionListener<T> listener) {
         });
     }
 
+    public void readAsyncWithDelay(final long delayMillis, final U entity, final ActionListener<T> listener) {
+        Runnable runnable = () -> {
+            try {
+                listener.onResponse(read(entity));
+            } catch (Exception e) {
+                listener.onFailure(e);
+            }
+        };
+        threadPool.scheduleUnlessShuttingDown(TimeValue.timeValueMillis(delayMillis), executor, runnable);
+    }
+
     public String getClusterName() {
         return clusterName;
     }
@@ -121,5 +141,4 @@ public BlobPath getBlobPathForDownload(final RemoteWriteableBlobEntity<T> obj) {
     private static String encodeString(String content) {
         return Base64.getUrlEncoder().withoutPadding().encodeToString(content.getBytes(StandardCharsets.UTF_8));
     }
-
 }

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

@@ -107,7 +107,6 @@
 import org.opensearch.gateway.PersistedClusterStateService;
 import org.opensearch.gateway.ShardsBatchGatewayAllocator;
 import org.opensearch.gateway.remote.RemoteClusterStateCleanupManager;
-import org.opensearch.gateway.remote.RemoteClusterStateService;
 import org.opensearch.gateway.remote.RemoteIndexMetadataManager;
 import org.opensearch.gateway.remote.model.RemoteRoutingTableBlobStore;
 import org.opensearch.http.HttpTransportSettings;
@@ -186,6 +185,11 @@
 import java.util.Set;
 import java.util.function.Predicate;
 
+import static org.opensearch.gateway.remote.RemoteClusterStateSettings.CLUSTER_REMOTE_STORE_STATE_PATH_PREFIX;
+import static org.opensearch.gateway.remote.RemoteClusterStateSettings.REMOTE_CLUSTER_STATE_CHECKSUM_VALIDATION_MODE_SETTING;
+import static org.opensearch.gateway.remote.RemoteClusterStateSettings.REMOTE_CLUSTER_STATE_ENABLED_SETTING;
+import static org.opensearch.gateway.remote.RemoteClusterStateSettings.REMOTE_STATE_READ_MAX_JITTER;
+import static org.opensearch.gateway.remote.RemoteClusterStateSettings.REMOTE_STATE_READ_TIMEOUT_SETTING;
 import static org.opensearch.gateway.remote.RemoteGlobalMetadataManager.GLOBAL_METADATA_UPLOAD_TIMEOUT_SETTING;
 import static org.opensearch.gateway.remote.RemoteIndexMetadataManager.INDEX_METADATA_UPLOAD_TIMEOUT_SETTING;
 import static org.opensearch.gateway.remote.RemoteManifestManager.METADATA_MANIFEST_UPLOAD_TIMEOUT_SETTING;
@@ -732,12 +736,12 @@ public void apply(Settings value, Settings current, Settings previous) {
 
                 // Remote cluster state settings
                 RemoteClusterStateCleanupManager.REMOTE_CLUSTER_STATE_CLEANUP_INTERVAL_SETTING,
-                RemoteClusterStateService.REMOTE_CLUSTER_STATE_ENABLED_SETTING,
+                REMOTE_CLUSTER_STATE_ENABLED_SETTING,
                 INDEX_METADATA_UPLOAD_TIMEOUT_SETTING,
                 GLOBAL_METADATA_UPLOAD_TIMEOUT_SETTING,
                 METADATA_MANIFEST_UPLOAD_TIMEOUT_SETTING,
-                RemoteClusterStateService.REMOTE_STATE_READ_TIMEOUT_SETTING,
-                RemoteClusterStateService.CLUSTER_REMOTE_STORE_STATE_PATH_PREFIX,
+                REMOTE_STATE_READ_TIMEOUT_SETTING,
+                CLUSTER_REMOTE_STORE_STATE_PATH_PREFIX,
                 RemoteIndexMetadataManager.REMOTE_INDEX_METADATA_PATH_TYPE_SETTING,
                 RemoteIndexMetadataManager.REMOTE_INDEX_METADATA_PATH_HASH_ALGO_SETTING,
                 RemoteStoreNodeService.REMOTE_STORE_COMPATIBILITY_MODE_SETTING,
@@ -746,7 +750,8 @@ public void apply(Settings value, Settings current, Settings previous) {
                 IndicesService.CLUSTER_INDEX_RESTRICT_REPLICATION_TYPE_SETTING,
                 RemoteRoutingTableBlobStore.REMOTE_ROUTING_TABLE_PATH_TYPE_SETTING,
                 RemoteRoutingTableBlobStore.REMOTE_ROUTING_TABLE_PATH_HASH_ALGO_SETTING,
-                RemoteClusterStateService.REMOTE_CLUSTER_STATE_CHECKSUM_VALIDATION_MODE_SETTING,
+                REMOTE_CLUSTER_STATE_CHECKSUM_VALIDATION_MODE_SETTING,
+                REMOTE_STATE_READ_MAX_JITTER,
                 RemoteRoutingTableBlobStore.CLUSTER_REMOTE_STORE_ROUTING_TABLE_PATH_PREFIX,
 
                 // Admission Control Settings

server/src/main/java/org/opensearch/gateway/remote/DefaultRandomObject.java

@@ -0,0 +1,17 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.gateway.remote;
+
+import org.opensearch.common.Randomness;
+
+import java.util.Random;
+
+public class DefaultRandomObject {
+    public static final Random INSTANCE = new Random(Randomness.get().nextLong());
+}