Optimise clone operation for incremental full cluster snapshots

CHANGELOG.md

@@ -55,7 +55,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Remove Identity FeatureFlag ([#16024](https:/opensearch-project/OpenSearch/pull/16024))
 - Ensure RestHandler.Wrapper delegates all implementations to the wrapped handler ([#16154](https:/opensearch-project/OpenSearch/pull/16154))
 - Code cleanup: Remove ApproximateIndexOrDocValuesQuery ([#16273](https:/opensearch-project/OpenSearch/pull/16273))
-
+- Optimise clone operation for incremental full cluster snapshots ([#16296](https:/opensearch-project/OpenSearch/pull/16296))
 
 ### Deprecated
 

server/src/main/java/org/opensearch/snapshots/SnapshotsService.java

@@ -1327,7 +1327,8 @@
 
  private final Set<RepositoryShardId> currentlyCloning = Collections.synchronizedSet(new HashSet<>());
 
- private void runReadyClone(
+ // Made to package private to be able to test the method in UTs
+ void runReadyClone(
  Snapshot target,
  SnapshotId sourceSnapshot,
  ShardSnapshotStatus shardStatusBefore,
@@ -1351,69 +1352,75 @@
  @Override
  protected void doRun() {
  final String localNodeId = clusterService.localNode().getId();
- repository.getRepositoryData(ActionListener.wrap(repositoryData -> {
- try {
- final IndexMetadata indexMetadata = repository.getSnapshotIndexMetaData(
- repositoryData,
+ if (remoteStoreIndexShallowCopy == false) {
+ executeClone(localNodeId, false);
+ } else {
+ repository.getRepositoryData(ActionListener.wrap(repositoryData -> {
+ try {
+ final IndexMetadata indexMetadata = repository.getSnapshotIndexMetaData(
+ repositoryData,
+ sourceSnapshot,
+ repoShardId.index()
+ );
+ final boolean cloneRemoteStoreIndexShardSnapshot = indexMetadata.getSettings()
+ .getAsBoolean(IndexMetadata.SETTING_REMOTE_STORE_ENABLED, false);
+ executeClone(localNodeId, cloneRemoteStoreIndexShardSnapshot);
+ } catch (IOException e) {
+ logger.warn("Failed to get index-metadata from repository data for index [{}]", repoShardId.index().getName());
+ failCloneShardAndUpdateClusterState(target, sourceSnapshot, repoShardId);
+ }
+ }, this::onFailure));
+ }
+ }
+
+ private void executeClone(String localNodeId, boolean cloneRemoteStoreIndexShardSnapshot) {
+ if (currentlyCloning.add(repoShardId)) {
+ if (cloneRemoteStoreIndexShardSnapshot) {
+ repository.cloneRemoteStoreIndexShardSnapshot(
  sourceSnapshot,
- repoShardId.index()
+ target.getSnapshotId(),
+ repoShardId,
+ shardStatusBefore.generation(),
+ remoteStoreLockManagerFactory,
+ getCloneCompletionListener(localNodeId)
  );
- final boolean cloneRemoteStoreIndexShardSnapshot = remoteStoreIndexShallowCopy
- && indexMetadata.getSettings().getAsBoolean(IndexMetadata.SETTING_REMOTE_STORE_ENABLED, false);
- final SnapshotId targetSnapshot = target.getSnapshotId();
- final ActionListener<String> listener = ActionListener.wrap(
- generation -> innerUpdateSnapshotState(
- new ShardSnapshotUpdate(
- target,
- repoShardId,
- new ShardSnapshotStatus(localNodeId, ShardState.SUCCESS, generation)
- ),
- ActionListener.runBefore(
- ActionListener.wrap(
- v -> logger.trace(
- "Marked [{}] as successfully cloned from [{}] to [{}]",
- repoShardId,
- sourceSnapshot,
- targetSnapshot
- ),
- e -> {
- logger.warn("Cluster state update after successful shard clone [{}] failed", repoShardId);
- failAllListenersOnMasterFailOver(e);
- }
- ),
- () -> currentlyCloning.remove(repoShardId)
- )
- ),
- e -> {
- logger.warn("Exception [{}] while trying to clone shard [{}]", e, repoShardId);
- failCloneShardAndUpdateClusterState(target, sourceSnapshot, repoShardId);
- }
+ } else {
+ repository.cloneShardSnapshot(
+ sourceSnapshot,
+ target.getSnapshotId(),
+ repoShardId,
+ shardStatusBefore.generation(),
+ getCloneCompletionListener(localNodeId)
  );
- if (currentlyCloning.add(repoShardId)) {
- if (cloneRemoteStoreIndexShardSnapshot) {
- repository.cloneRemoteStoreIndexShardSnapshot(
- sourceSnapshot,
- targetSnapshot,
+ }
+ }
+ }
+
+ private ActionListener<String> getCloneCompletionListener(String localNodeId) {
+ return ActionListener.wrap(
+ generation -> innerUpdateSnapshotState(
+ new ShardSnapshotUpdate(target, repoShardId, new ShardSnapshotStatus(localNodeId, ShardState.SUCCESS, generation)),
+ ActionListener.runBefore(
+ ActionListener.wrap(
+ v -> logger.trace(
+ "Marked [{}] as successfully cloned from [{}] to [{}]",
  repoShardId,
- shardStatusBefore.generation(),
- remoteStoreLockManagerFactory,
- listener
- );
- } else {
- repository.cloneShardSnapshot(
  sourceSnapshot,
- targetSnapshot,
- repoShardId,
- shardStatusBefore.generation(),
- listener
- );
- }
- }
- } catch (IOException e) {
- logger.warn("Failed to get index-metadata from repository data for index [{}]", repoShardId.index().getName());
+ target.getSnapshotId()
+ ),
+ e -> {
+ logger.warn("Cluster state update after successful shard clone [{}] failed", repoShardId);
+ failAllListenersOnMasterFailOver(e);
+ }
+ ),
+ () -> currentlyCloning.remove(repoShardId)
+ )
+ ),
+ e -> {
+ logger.warn("Exception [{}] while trying to clone shard [{}]", e, repoShardId);
  failCloneShardAndUpdateClusterState(target, sourceSnapshot, repoShardId);
  }
- }, this::onFailure));
+ );
  }
  });
  }