Skip to content

Commit 23cba28

Browse files
anshu1106Anshu Agarwal
anshu1106
and
Anshu Agarwal
authored
[SnapshotV2] Support centralize snapshot creation (opensearch-project#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>
1 parent c771bdd commit 23cba28

File tree

21 files changed

+1025
-81
lines changed

21 files changed

+1025
-81
lines changed

‎CHANGELOG.md

+1
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
2626
- Add allowlist setting for ingest-geoip and ingest-useragent ([#15325](https://github.com/opensearch-project/OpenSearch/pull/15325))
2727
- Adding access to noSubMatches and noOverlappingMatches in Hyphenation ([#13895](https://github.com/opensearch-project/OpenSearch/pull/13895))
2828
- Add support for index level max slice count setting for concurrent segment search ([#15336](https://github.com/opensearch-project/OpenSearch/pull/15336))
29+
- Add support for centralize snapshot creation with pinned timestamp ([#15124](https://github.com/opensearch-project/OpenSearch/pull/15124))
2930
- Add concurrent search support for Derived Fields ([#15326](https://github.com/opensearch-project/OpenSearch/pull/15326))
3031

3132
### Dependencies

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

+3
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,7 @@
4242
import org.opensearch.cluster.metadata.Metadata;
4343
import org.opensearch.cluster.metadata.RepositoryMetadata;
4444
import org.opensearch.cluster.service.ClusterService;
45+
import org.opensearch.common.Priority;
4546
import org.opensearch.common.blobstore.BlobPath;
4647
import org.opensearch.common.blobstore.BlobStore;
4748
import org.opensearch.common.blobstore.BlobStoreException;
@@ -391,6 +392,7 @@ public void finalizeSnapshot(
391392
SnapshotInfo snapshotInfo,
392393
Version repositoryMetaVersion,
393394
Function<ClusterState, ClusterState> stateTransformer,
395+
Priority repositoryUpdatePriority,
394396
ActionListener<RepositoryData> listener
395397
) {
396398
super.finalizeSnapshot(
@@ -400,6 +402,7 @@ public void finalizeSnapshot(
400402
snapshotInfo,
401403
repositoryMetaVersion,
402404
stateTransformer,
405+
repositoryUpdatePriority,
403406
listener
404407
);
405408
}

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

+602
Large diffs are not rendered by default.

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

+3
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

+11-1
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
@@ -103,7 +111,9 @@ protected void clusterManagerOperation(
103111
ClusterState state,
104112
final ActionListener<CreateSnapshotResponse> listener
105113
) {
106-
if (request.waitForCompletion()) {
114+
Repository repository = repositoriesService.repository(request.repository());
115+
boolean isSnapshotV2 = SHALLOW_SNAPSHOT_V2.get(repository.getMetadata().settings());
116+
if (request.waitForCompletion() || isSnapshotV2) {
107117
snapshotsService.executeSnapshot(request, ActionListener.map(listener, CreateSnapshotResponse::new));
108118
} else {
109119
snapshotsService.createSnapshot(request, ActionListener.map(listener, snapshot -> new CreateSnapshotResponse()));

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

+3
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;
@@ -104,6 +105,7 @@ public void finalizeSnapshot(
104105
SnapshotInfo snapshotInfo,
105106
Version repositoryMetaVersion,
106107
Function<ClusterState, ClusterState> stateTransformer,
108+
Priority repositoryUpdatePriority,
107109
ActionListener<RepositoryData> listener
108110
) {
109111
in.finalizeSnapshot(
@@ -113,6 +115,7 @@ public void finalizeSnapshot(
113115
snapshotInfo,
114116
repositoryMetaVersion,
115117
stateTransformer,
118+
repositoryUpdatePriority,
116119
listener
117120
);
118121
}

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

+3
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;
@@ -150,6 +151,7 @@ default Repository create(RepositoryMetadata metadata, Function<String, Reposito
150151
* @param repositoryMetaVersion version of the updated repository metadata to write
151152
* @param stateTransformer a function that filters the last cluster state update that the snapshot finalization will execute and
152153
* is used to remove any state tracked for the in-progress snapshot from the cluster state
154+
* @param repositoryUpdatePriority priority for the cluster state update task
153155
* @param listener listener to be invoked with the new {@link RepositoryData} after completing the snapshot
154156
*/
155157
void finalizeSnapshot(
@@ -159,6 +161,7 @@ void finalizeSnapshot(
159161
SnapshotInfo snapshotInfo,
160162
Version repositoryMetaVersion,
161163
Function<ClusterState, ClusterState> stateTransformer,
164+
Priority repositoryUpdatePriority,
162165
ActionListener<RepositoryData> listener
163166
);
164167

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

+15-6
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;
@@ -266,6 +267,8 @@ public abstract class BlobStoreRepository extends AbstractLifecycleComponent imp
266267

267268
public static final Setting<Boolean> REMOTE_STORE_INDEX_SHALLOW_COPY = Setting.boolSetting("remote_store_index_shallow_copy", false);
268269

270+
public static final Setting<Boolean> SHALLOW_SNAPSHOT_V2 = Setting.boolSetting("shallow_snapshot_v2", false);
271+
269272
/**
270273
* Setting to set batch size of stale snapshot shard blobs that will be deleted by snapshot workers as part of snapshot deletion.
271274
* 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
@@ -1046,6 +1049,7 @@ private void doDeleteShardSnapshots(
10461049
repositoryStateId,
10471050
repoMetaVersion,
10481051
Function.identity(),
1052+
Priority.NORMAL,
10491053
ActionListener.wrap(writeUpdatedRepoDataStep::onResponse, listener::onFailure)
10501054
);
10511055
}, listener::onFailure);
@@ -1520,6 +1524,7 @@ public void cleanup(
15201524
repositoryStateId,
15211525
repositoryMetaVersion,
15221526
Function.identity(),
1527+
Priority.NORMAL,
15231528
ActionListener.wrap(
15241529
v -> cleanupStaleBlobs(
15251530
Collections.emptyList(),
@@ -1723,6 +1728,7 @@ public void finalizeSnapshot(
17231728
SnapshotInfo snapshotInfo,
17241729
Version repositoryMetaVersion,
17251730
Function<ClusterState, ClusterState> stateTransformer,
1731+
Priority repositoryUpdatePriority,
17261732
final ActionListener<RepositoryData> listener
17271733
) {
17281734
assert repositoryStateId > RepositoryData.UNKNOWN_REPO_GEN : "Must finalize based on a valid repository generation but received ["
@@ -1759,6 +1765,7 @@ public void finalizeSnapshot(
17591765
repositoryStateId,
17601766
repositoryMetaVersion,
17611767
stateTransformer,
1768+
repositoryUpdatePriority,
17621769
ActionListener.wrap(newRepoData -> {
17631770
cleanupOldShardGens(existingRepositoryData, updatedRepositoryData);
17641771
listener.onResponse(newRepoData);
@@ -2280,17 +2287,19 @@ public boolean isSystemRepository() {
22802287
* Lastly, the {@link RepositoryMetadata} entry for this repository is updated to the new generation {@code P + 1} and thus
22812288
* pending and safe generation are set to the same value marking the end of the update of the repository data.
22822289
*
2283-
* @param repositoryData RepositoryData to write
2284-
* @param expectedGen expected repository generation at the start of the operation
2285-
* @param version version of the repository metadata to write
2286-
* @param stateFilter filter for the last cluster state update executed by this method
2290+
* @param repositoryData RepositoryData to write
2291+
* @param expectedGen expected repository generation at the start of the operation
2292+
* @param version version of the repository metadata to write
2293+
* @param stateFilter filter for the last cluster state update executed by this method
2294+
* @param repositoryUpdatePriority priority for the cluster state update task
22872295
* @param listener completion listener
22882296
*/
22892297
protected void writeIndexGen(
22902298
RepositoryData repositoryData,
22912299
long expectedGen,
22922300
Version version,
22932301
Function<ClusterState, ClusterState> stateFilter,
2302+
Priority repositoryUpdatePriority,
22942303
ActionListener<RepositoryData> listener
22952304
) {
22962305
assert isReadOnly() == false; // can not write to a read only repository
@@ -2315,7 +2324,7 @@ protected void writeIndexGen(
23152324
final StepListener<Long> setPendingStep = new StepListener<>();
23162325
clusterService.submitStateUpdateTask(
23172326
"set pending repository generation [" + metadata.name() + "][" + expectedGen + "]",
2318-
new ClusterStateUpdateTask() {
2327+
new ClusterStateUpdateTask(repositoryUpdatePriority) {
23192328

23202329
private long newGen;
23212330

@@ -2453,7 +2462,7 @@ public void onFailure(Exception e) {
24532462
// Step 3: Update CS to reflect new repository generation.
24542463
clusterService.submitStateUpdateTask(
24552464
"set safe repository generation [" + metadata.name() + "][" + newGen + "]",
2456-
new ClusterStateUpdateTask() {
2465+
new ClusterStateUpdateTask(repositoryUpdatePriority) {
24572466
@Override
24582467
public ClusterState execute(ClusterState currentState) {
24592468
final RepositoryMetadata meta = getRepoMetadata(currentState);

0 commit comments

Comments
 (0)