6565import org .opensearch .cluster .service .ClusterService ;
6666import org .opensearch .common .Nullable ;
6767import org .opensearch .common .Numbers ;
68+ import org .opensearch .common .Priority ;
6869import org .opensearch .common .SetOnce ;
6970import org .opensearch .common .UUIDs ;
7071import org .opensearch .common .blobstore .BlobContainer ;
@@ -267,6 +268,8 @@ public abstract class BlobStoreRepository extends AbstractLifecycleComponent imp
267268
268269 public static final Setting <Boolean > REMOTE_STORE_INDEX_SHALLOW_COPY = Setting .boolSetting ("remote_store_index_shallow_copy" , false );
269270
271+ public static final Setting <Boolean > SHALLOW_SNAPSHOT_V2 = Setting .boolSetting ("shallow_snapshot_v2" , false );
272+
270273 /**
271274 * Setting to set batch size of stale snapshot shard blobs that will be deleted by snapshot workers as part of snapshot deletion.
272275 * For optimal performance the value of the setting should be equal to or close to repository's max # of keys that can be deleted in single operation
@@ -1072,6 +1075,7 @@ private void doDeleteShardSnapshots(
10721075 repositoryStateId ,
10731076 repoMetaVersion ,
10741077 Function .identity (),
1078+ Priority .NORMAL ,
10751079 ActionListener .wrap (writeUpdatedRepoDataStep ::onResponse , listener ::onFailure )
10761080 );
10771081 }, listener ::onFailure );
@@ -1101,39 +1105,46 @@ private void doDeleteShardSnapshots(
11011105 } else {
11021106 // Write the new repository data first (with the removed snapshot), using no shard generations
11031107 final RepositoryData updatedRepoData = repositoryData .removeSnapshots (snapshotIds , ShardGenerations .EMPTY );
1104- writeIndexGen (updatedRepoData , repositoryStateId , repoMetaVersion , Function .identity (), ActionListener .wrap (newRepoData -> {
1105- // Run unreferenced blobs cleanup in parallel to shard-level snapshot deletion
1106- final ActionListener <Void > afterCleanupsListener = new GroupedActionListener <>(
1107- ActionListener .wrap (() -> listener .onResponse (newRepoData )),
1108- 2
1109- );
1110- cleanupUnlinkedRootAndIndicesBlobs (
1111- snapshotIds ,
1112- foundIndices ,
1113- rootBlobs ,
1114- newRepoData ,
1115- remoteStoreLockManagerFactory ,
1116- afterCleanupsListener
1117- );
1118- final StepListener <Collection <ShardSnapshotMetaDeleteResult >> writeMetaAndComputeDeletesStep = new StepListener <>();
1119- writeUpdatedShardMetaDataAndComputeDeletes (
1120- snapshotIds ,
1121- repositoryData ,
1122- false ,
1123- remoteStoreLockManagerFactory ,
1124- writeMetaAndComputeDeletesStep
1125- );
1126- writeMetaAndComputeDeletesStep .whenComplete (
1127- deleteResults -> asyncCleanupUnlinkedShardLevelBlobs (
1128- repositoryData ,
1108+ writeIndexGen (
1109+ updatedRepoData ,
1110+ repositoryStateId ,
1111+ repoMetaVersion ,
1112+ Function .identity (),
1113+ Priority .NORMAL ,
1114+ ActionListener .wrap (newRepoData -> {
1115+ // Run unreferenced blobs cleanup in parallel to shard-level snapshot deletion
1116+ final ActionListener <Void > afterCleanupsListener = new GroupedActionListener <>(
1117+ ActionListener .wrap (() -> listener .onResponse (newRepoData )),
1118+ 2
1119+ );
1120+ cleanupUnlinkedRootAndIndicesBlobs (
11291121 snapshotIds ,
1130- deleteResults ,
1122+ foundIndices ,
1123+ rootBlobs ,
1124+ newRepoData ,
11311125 remoteStoreLockManagerFactory ,
11321126 afterCleanupsListener
1133- ),
1134- afterCleanupsListener ::onFailure
1135- );
1136- }, listener ::onFailure ));
1127+ );
1128+ final StepListener <Collection <ShardSnapshotMetaDeleteResult >> writeMetaAndComputeDeletesStep = new StepListener <>();
1129+ writeUpdatedShardMetaDataAndComputeDeletes (
1130+ snapshotIds ,
1131+ repositoryData ,
1132+ false ,
1133+ remoteStoreLockManagerFactory ,
1134+ writeMetaAndComputeDeletesStep
1135+ );
1136+ writeMetaAndComputeDeletesStep .whenComplete (
1137+ deleteResults -> asyncCleanupUnlinkedShardLevelBlobs (
1138+ repositoryData ,
1139+ snapshotIds ,
1140+ deleteResults ,
1141+ remoteStoreLockManagerFactory ,
1142+ afterCleanupsListener
1143+ ),
1144+ afterCleanupsListener ::onFailure
1145+ );
1146+ }, listener ::onFailure )
1147+ );
11371148 }
11381149 }
11391150
@@ -1583,6 +1594,7 @@ public void cleanup(
15831594 repositoryStateId ,
15841595 repositoryMetaVersion ,
15851596 Function .identity (),
1597+ Priority .NORMAL ,
15861598 ActionListener .wrap (
15871599 v -> cleanupStaleBlobs (
15881600 Collections .emptyList (),
@@ -1787,6 +1799,29 @@ public void finalizeSnapshot(
17871799 Version repositoryMetaVersion ,
17881800 Function <ClusterState , ClusterState > stateTransformer ,
17891801 final ActionListener <RepositoryData > listener
1802+ ) {
1803+ finalizeSnapshot (
1804+ shardGenerations ,
1805+ repositoryStateId ,
1806+ clusterMetadata ,
1807+ snapshotInfo ,
1808+ repositoryMetaVersion ,
1809+ stateTransformer ,
1810+ Priority .NORMAL ,
1811+ listener
1812+ );
1813+ }
1814+
1815+ @ Override
1816+ public void finalizeSnapshot (
1817+ final ShardGenerations shardGenerations ,
1818+ final long repositoryStateId ,
1819+ final Metadata clusterMetadata ,
1820+ SnapshotInfo snapshotInfo ,
1821+ Version repositoryMetaVersion ,
1822+ Function <ClusterState , ClusterState > stateTransformer ,
1823+ Priority repositoryUpdatePriority ,
1824+ final ActionListener <RepositoryData > listener
17901825 ) {
17911826 assert repositoryStateId > RepositoryData .UNKNOWN_REPO_GEN : "Must finalize based on a valid repository generation but received ["
17921827 + repositoryStateId
@@ -1834,6 +1869,7 @@ public void finalizeSnapshot(
18341869 repositoryStateId ,
18351870 repositoryMetaVersion ,
18361871 stateTransformer ,
1872+ repositoryUpdatePriority ,
18371873 ActionListener .wrap (newRepoData -> {
18381874 if (writeShardGens ) {
18391875 cleanupOldShardGens (existingRepositoryData , updatedRepositoryData );
@@ -2367,17 +2403,19 @@ public boolean isSystemRepository() {
23672403 * Lastly, the {@link RepositoryMetadata} entry for this repository is updated to the new generation {@code P + 1} and thus
23682404 * pending and safe generation are set to the same value marking the end of the update of the repository data.
23692405 *
2370- * @param repositoryData RepositoryData to write
2371- * @param expectedGen expected repository generation at the start of the operation
2372- * @param version version of the repository metadata to write
2373- * @param stateFilter filter for the last cluster state update executed by this method
2406+ * @param repositoryData RepositoryData to write
2407+ * @param expectedGen expected repository generation at the start of the operation
2408+ * @param version version of the repository metadata to write
2409+ * @param stateFilter filter for the last cluster state update executed by this method
2410+ * @param repositoryUpdatePriority priority for the cluster state update task
23742411 * @param listener completion listener
23752412 */
23762413 protected void writeIndexGen (
23772414 RepositoryData repositoryData ,
23782415 long expectedGen ,
23792416 Version version ,
23802417 Function <ClusterState , ClusterState > stateFilter ,
2418+ Priority repositoryUpdatePriority ,
23812419 ActionListener <RepositoryData > listener
23822420 ) {
23832421 assert isReadOnly () == false ; // can not write to a read only repository
@@ -2402,7 +2440,7 @@ protected void writeIndexGen(
24022440 final StepListener <Long > setPendingStep = new StepListener <>();
24032441 clusterService .submitStateUpdateTask (
24042442 "set pending repository generation [" + metadata .name () + "][" + expectedGen + "]" ,
2405- new ClusterStateUpdateTask () {
2443+ new ClusterStateUpdateTask (repositoryUpdatePriority ) {
24062444
24072445 private long newGen ;
24082446
@@ -2540,7 +2578,7 @@ public void onFailure(Exception e) {
25402578 // Step 3: Update CS to reflect new repository generation.
25412579 clusterService .submitStateUpdateTask (
25422580 "set safe repository generation [" + metadata .name () + "][" + newGen + "]" ,
2543- new ClusterStateUpdateTask () {
2581+ new ClusterStateUpdateTask (repositoryUpdatePriority ) {
25442582 @ Override
25452583 public ClusterState execute (ClusterState currentState ) {
25462584 final RepositoryMetadata meta = getRepoMetadata (currentState );
0 commit comments