Skip to content

Commit 296b0c9

Browse files
anshu1106Anshu Agarwal
anshu1106
and
Anshu Agarwal
committed
[SnapshotV2] Support centralize snapshot creation (#15124)
* Initial Commit to support centralize snapshot creation and implicit locking mechanism Signed-off-by: Anshu Agarwal <anshukag@amazon.com> * Fix deserilization error Signed-off-by: Anshu Agarwal <anshukag@amazon.com> * Fix gradle spotless check Signed-off-by: Anshu Agarwal <anshukag@amazon.com> * Fix listener Signed-off-by: Anshu Agarwal <anshukag@amazon.com> * Fix test Signed-off-by: Anshu Agarwal <anshukag@amazon.com> * Fix snapshot generation Signed-off-by: Anshu Agarwal <anshukag@amazon.com> * Modify cluster setting name Signed-off-by: Anshu Agarwal <anshukag@amazon.com> * Add more tests Signed-off-by: Anshu Agarwal <anshukag@amazon.com> * Uncomment pin timestamp code Signed-off-by: Anshu Agarwal <anshukag@amazon.com> * Modify log messages Signed-off-by: Anshu Agarwal <anshukag@amazon.com> * Add spotless check failure fix Signed-off-by: Anshu Agarwal <anshukag@amazon.com> * Fix completion listener for snapshot v2 Signed-off-by: Anshu Agarwal <anshukag@amazon.com> * Elevate cluster state update priority for repository metadata update task Signed-off-by: Anshu Agarwal <anshukag@amazon.com> * Add more integ tests Signed-off-by: Anshu Agarwal <anshukag@amazon.com> * Add priority as IMMEDIATE for cluster state repo update task only for v2 snapshots Signed-off-by: Anshu Agarwal <anshukag@amazon.com> * Fix build error Signed-off-by: Anshu Agarwal <anshukag@amazon.com> * Fix spotless error Signed-off-by: Anshu Agarwal <anshukag@amazon.com> * Add repository setting for snapshot v2 Signed-off-by: Anshu Agarwal <anshukag@amazon.com> * Address review comments Signed-off-by: Anshu Agarwal <anshukag@amazon.com> * Add integ test to verify snapshot creation if shallow copy repo setting is disabled Signed-off-by: Anshu Agarwal <anshukag@amazon.com> * Fix spotless vilation error Signed-off-by: Anshu Agarwal <anshukag@amazon.com> * Address review comment Signed-off-by: Anshu Agarwal <anshukag@amazon.com> * Address review comments Signed-off-by: Anshu Agarwal <anshukag@amazon.com> * Add min version check for backward compatibility Signed-off-by: Anshu Agarwal <anshukag@amazon.com> * address review comments Signed-off-by: Anshu Agarwal <anshukag@amazon.com> * add integ test for master failover scenario Signed-off-by: Anshu Agarwal <anshukag@amazon.com> * Add more integ tests Signed-off-by: Anshu Agarwal <anshukag@amazon.com> * refactor code Signed-off-by: Anshu Agarwal <anshukag@amazon.com> * add changelog Signed-off-by: Anshu Agarwal <anshukag@amazon.com> * Add pinned timestamp setting in integ tests Signed-off-by: Anshu Agarwal <anshukag@amazon.com> --------- Signed-off-by: Anshu Agarwal <anshukag@amazon.com> Signed-off-by: Anshu Agarwal <anshuagarwal11@gmail.com> Co-authored-by: Anshu Agarwal <anshukag@amazon.com> (cherry picked from commit 23cba28) Signed-off-by: Anshu Agarwal <anshukag@amazon.com>
1 parent 5653ed6 commit 296b0c9

File tree

21 files changed

+1073
-112
lines changed

21 files changed

+1073
-112
lines changed

‎CHANGELOG.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
2828
- Star tree mapping changes ([#14605](https://github.com/opensearch-project/OpenSearch/pull/14605))
2929
- Add support for index level max slice count setting for concurrent segment search ([#15336](https://github.com/opensearch-project/OpenSearch/pull/15336))
3030
- [Streaming Indexing] Introduce bulk HTTP API streaming flavor ([#15381](https://github.com/opensearch-project/OpenSearch/pull/15381))
31+
- Add support for centralize snapshot creation with pinned timestamp ([#15124](https://github.com/opensearch-project/OpenSearch/pull/15124))
3132
- Add concurrent search support for Derived Fields ([#15326](https://github.com/opensearch-project/OpenSearch/pull/15326))
3233
- [Workload Management] Add query group stats constructs ([#15343](https://github.com/opensearch-project/OpenSearch/pull/15343)))
3334
- Add runAs to Subject interface and introduce IdentityAwarePlugin extension point ([#14630](https://github.com/opensearch-project/OpenSearch/pull/14630))

‎plugins/repository-s3/src/main/java/org/opensearch/repositories/s3/S3Repository.java‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -44,6 +44,7 @@
4444
import org.opensearch.cluster.metadata.Metadata;
4545
import org.opensearch.cluster.metadata.RepositoryMetadata;
4646
import org.opensearch.cluster.service.ClusterService;
47+
import org.opensearch.common.Priority;
4748
import org.opensearch.common.blobstore.BlobPath;
4849
import org.opensearch.common.blobstore.BlobStore;
4950
import org.opensearch.common.blobstore.BlobStoreException;
@@ -424,6 +425,7 @@ public void finalizeSnapshot(
424425
SnapshotInfo snapshotInfo,
425426
Version repositoryMetaVersion,
426427
Function<ClusterState, ClusterState> stateTransformer,
428+
Priority repositoryUpdatePriority,
427429
ActionListener<RepositoryData> listener
428430
) {
429431
if (SnapshotsService.useShardGenerations(repositoryMetaVersion) == false) {
@@ -436,6 +438,7 @@ public void finalizeSnapshot(
436438
snapshotInfo,
437439
repositoryMetaVersion,
438440
stateTransformer,
441+
repositoryUpdatePriority,
439442
listener
440443
);
441444
}

‎server/src/internalClusterTest/java/org/opensearch/remotestore/RemoteRestoreSnapshotIT.java‎

Lines changed: 602 additions & 0 deletions
Large diffs are not rendered by default.

‎server/src/internalClusterTest/java/org/opensearch/snapshots/RepositoryFilterUserMetadataIT.java‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,7 @@
3636
import org.opensearch.cluster.ClusterState;
3737
import org.opensearch.cluster.metadata.Metadata;
3838
import org.opensearch.cluster.service.ClusterService;
39+
import org.opensearch.common.Priority;
3940
import org.opensearch.common.settings.Settings;
4041
import org.opensearch.core.action.ActionListener;
4142
import org.opensearch.core.xcontent.NamedXContentRegistry;
@@ -127,6 +128,7 @@ public void finalizeSnapshot(
127128
SnapshotInfo snapshotInfo,
128129
Version repositoryMetaVersion,
129130
Function<ClusterState, ClusterState> stateTransformer,
131+
Priority repositoryUpdatePriority,
130132
ActionListener<RepositoryData> listener
131133
) {
132134
super.finalizeSnapshot(
@@ -136,6 +138,7 @@ public void finalizeSnapshot(
136138
snapshotInfo,
137139
repositoryMetaVersion,
138140
stateTransformer,
141+
repositoryUpdatePriority,
139142
listener
140143
);
141144
}

‎server/src/main/java/org/opensearch/action/admin/cluster/snapshots/create/TransportCreateSnapshotAction.java‎

Lines changed: 12 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -42,12 +42,16 @@
4242
import org.opensearch.common.inject.Inject;
4343
import org.opensearch.core.action.ActionListener;
4444
import org.opensearch.core.common.io.stream.StreamInput;
45+
import org.opensearch.repositories.RepositoriesService;
46+
import org.opensearch.repositories.Repository;
4547
import org.opensearch.snapshots.SnapshotsService;
4648
import org.opensearch.threadpool.ThreadPool;
4749
import org.opensearch.transport.TransportService;
4850

4951
import java.io.IOException;
5052

53+
import static org.opensearch.repositories.blobstore.BlobStoreRepository.SHALLOW_SNAPSHOT_V2;
54+
5155
/**
5256
* Transport action for create snapshot operation
5357
*
@@ -56,12 +60,15 @@
5660
public class TransportCreateSnapshotAction extends TransportClusterManagerNodeAction<CreateSnapshotRequest, CreateSnapshotResponse> {
5761
private final SnapshotsService snapshotsService;
5862

63+
private final RepositoriesService repositoriesService;
64+
5965
@Inject
6066
public TransportCreateSnapshotAction(
6167
TransportService transportService,
6268
ClusterService clusterService,
6369
ThreadPool threadPool,
6470
SnapshotsService snapshotsService,
71+
RepositoriesService repositoriesService,
6572
ActionFilters actionFilters,
6673
IndexNameExpressionResolver indexNameExpressionResolver
6774
) {
@@ -75,6 +82,7 @@ public TransportCreateSnapshotAction(
7582
indexNameExpressionResolver
7683
);
7784
this.snapshotsService = snapshotsService;
85+
this.repositoriesService = repositoriesService;
7886
}
7987

8088
@Override
@@ -110,7 +118,10 @@ protected void clusterManagerOperation(
110118
snapshotsService.createSnapshotLegacy(request, ActionListener.map(listener, snapshot -> new CreateSnapshotResponse()));
111119
}
112120
} else {
113-
if (request.waitForCompletion()) {
121+
Repository repository = repositoriesService.repository(request.repository());
122+
boolean isSnapshotV2 = SHALLOW_SNAPSHOT_V2.get(repository.getMetadata().settings());
123+
124+
if (request.waitForCompletion() || isSnapshotV2) {
114125
snapshotsService.executeSnapshot(request, ActionListener.map(listener, CreateSnapshotResponse::new));
115126
} else {
116127
snapshotsService.createSnapshot(request, ActionListener.map(listener, snapshot -> new CreateSnapshotResponse()));

‎server/src/main/java/org/opensearch/repositories/FilterRepository.java‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,7 @@
3939
import org.opensearch.cluster.metadata.Metadata;
4040
import org.opensearch.cluster.metadata.RepositoryMetadata;
4141
import org.opensearch.cluster.node.DiscoveryNode;
42+
import org.opensearch.common.Priority;
4243
import org.opensearch.common.lifecycle.Lifecycle;
4344
import org.opensearch.common.lifecycle.LifecycleListener;
4445
import org.opensearch.core.action.ActionListener;
@@ -110,6 +111,7 @@ public void finalizeSnapshot(
110111
SnapshotInfo snapshotInfo,
111112
Version repositoryMetaVersion,
112113
Function<ClusterState, ClusterState> stateTransformer,
114+
Priority repositoryUpdatePriority,
113115
ActionListener<RepositoryData> listener
114116
) {
115117
in.finalizeSnapshot(
@@ -119,6 +121,7 @@ public void finalizeSnapshot(
119121
snapshotInfo,
120122
repositoryMetaVersion,
121123
stateTransformer,
124+
repositoryUpdatePriority,
122125
listener
123126
);
124127
}

‎server/src/main/java/org/opensearch/repositories/Repository.java‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -41,6 +41,7 @@
4141
import org.opensearch.cluster.metadata.RepositoryMetadata;
4242
import org.opensearch.cluster.node.DiscoveryNode;
4343
import org.opensearch.common.Nullable;
44+
import org.opensearch.common.Priority;
4445
import org.opensearch.common.annotation.PublicApi;
4546
import org.opensearch.common.lifecycle.LifecycleComponent;
4647
import org.opensearch.common.settings.Setting;
@@ -163,6 +164,7 @@ default Repository create(RepositoryMetadata metadata, Function<String, Reposito
163164
* @param repositoryMetaVersion version of the updated repository metadata to write
164165
* @param stateTransformer a function that filters the last cluster state update that the snapshot finalization will execute and
165166
* is used to remove any state tracked for the in-progress snapshot from the cluster state
167+
* @param repositoryUpdatePriority priority for the cluster state update task
166168
* @param listener listener to be invoked with the new {@link RepositoryData} after completing the snapshot
167169
*/
168170
void finalizeSnapshot(
@@ -172,6 +174,7 @@ void finalizeSnapshot(
172174
SnapshotInfo snapshotInfo,
173175
Version repositoryMetaVersion,
174176
Function<ClusterState, ClusterState> stateTransformer,
177+
Priority repositoryUpdatePriority,
175178
ActionListener<RepositoryData> listener
176179
);
177180

‎server/src/main/java/org/opensearch/repositories/blobstore/BlobStoreRepository.java‎

Lines changed: 52 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -65,6 +65,7 @@
6565
import org.opensearch.cluster.service.ClusterService;
6666
import org.opensearch.common.Nullable;
6767
import org.opensearch.common.Numbers;
68+
import org.opensearch.common.Priority;
6869
import org.opensearch.common.SetOnce;
6970
import org.opensearch.common.UUIDs;
7071
import 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(),
@@ -1786,6 +1798,7 @@ public void finalizeSnapshot(
17861798
SnapshotInfo snapshotInfo,
17871799
Version repositoryMetaVersion,
17881800
Function<ClusterState, ClusterState> stateTransformer,
1801+
Priority repositoryUpdatePriority,
17891802
final ActionListener<RepositoryData> listener
17901803
) {
17911804
assert repositoryStateId > RepositoryData.UNKNOWN_REPO_GEN : "Must finalize based on a valid repository generation but received ["
@@ -1834,6 +1847,7 @@ public void finalizeSnapshot(
18341847
repositoryStateId,
18351848
repositoryMetaVersion,
18361849
stateTransformer,
1850+
repositoryUpdatePriority,
18371851
ActionListener.wrap(newRepoData -> {
18381852
if (writeShardGens) {
18391853
cleanupOldShardGens(existingRepositoryData, updatedRepositoryData);
@@ -2367,17 +2381,19 @@ public boolean isSystemRepository() {
23672381
* Lastly, the {@link RepositoryMetadata} entry for this repository is updated to the new generation {@code P + 1} and thus
23682382
* pending and safe generation are set to the same value marking the end of the update of the repository data.
23692383
*
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
2384+
* @param repositoryData RepositoryData to write
2385+
* @param expectedGen expected repository generation at the start of the operation
2386+
* @param version version of the repository metadata to write
2387+
* @param stateFilter filter for the last cluster state update executed by this method
2388+
* @param repositoryUpdatePriority priority for the cluster state update task
23742389
* @param listener completion listener
23752390
*/
23762391
protected void writeIndexGen(
23772392
RepositoryData repositoryData,
23782393
long expectedGen,
23792394
Version version,
23802395
Function<ClusterState, ClusterState> stateFilter,
2396+
Priority repositoryUpdatePriority,
23812397
ActionListener<RepositoryData> listener
23822398
) {
23832399
assert isReadOnly() == false; // can not write to a read only repository
@@ -2402,7 +2418,7 @@ protected void writeIndexGen(
24022418
final StepListener<Long> setPendingStep = new StepListener<>();
24032419
clusterService.submitStateUpdateTask(
24042420
"set pending repository generation [" + metadata.name() + "][" + expectedGen + "]",
2405-
new ClusterStateUpdateTask() {
2421+
new ClusterStateUpdateTask(repositoryUpdatePriority) {
24062422

24072423
private long newGen;
24082424

@@ -2540,7 +2556,7 @@ public void onFailure(Exception e) {
25402556
// Step 3: Update CS to reflect new repository generation.
25412557
clusterService.submitStateUpdateTask(
25422558
"set safe repository generation [" + metadata.name() + "][" + newGen + "]",
2543-
new ClusterStateUpdateTask() {
2559+
new ClusterStateUpdateTask(repositoryUpdatePriority) {
25442560
@Override
25452561
public ClusterState execute(ClusterState currentState) {
25462562
final RepositoryMetadata meta = getRepoMetadata(currentState);

0 commit comments

Comments
 (0)