Skip to content

Commit 1668e10

Browse files
kasundra07kasundra07
committed
Addressed PR comments
Signed-off-by: Bansi Kasundra <kasundra@amazon.com>
1 parent 4be4ea0 commit 1668e10

File tree

14 files changed

+397
-94
lines changed

14 files changed

+397
-94
lines changed

‎CHANGELOG.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -124,6 +124,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
124124
- Compress and cache cluster state during validate join request ([#7321](https://github.com/opensearch-project/OpenSearch/pull/7321))
125125
- [Snapshot Interop] Add Changes in Create Snapshot Flow for remote store interoperability. ([#7118](https://github.com/opensearch-project/OpenSearch/pull/7118))
126126
- Allow insecure string settings to warn-log usage and advise to migration of a newer secure variant ([#5496](https://github.com/opensearch-project/OpenSearch/pull/5496))
127+
- [Snapshot Interop] Add Changes in Delete Snapshot Flow for remote store interoperability. ([#7497](https://github.com/opensearch-project/OpenSearch/pull/7497))
127128

128129
### Deprecated
129130

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

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -51,7 +51,6 @@
5151
import org.opensearch.common.unit.ByteSizeValue;
5252
import org.opensearch.core.common.Strings;
5353
import org.opensearch.core.xcontent.NamedXContentRegistry;
54-
import org.opensearch.index.store.lockmanager.RemoteStoreLockManagerFactory;
5554
import org.opensearch.indices.recovery.RecoverySettings;
5655
import org.opensearch.monitor.jvm.JvmInfo;
5756
import org.opensearch.repositories.RepositoryData;
@@ -308,10 +307,9 @@ public void deleteSnapshots(
308307
Collection<SnapshotId> snapshotIds,
309308
long repositoryStateId,
310309
Version repositoryMetaVersion,
311-
RemoteStoreLockManagerFactory remoteStoreLockManagerFactory,
312310
ActionListener<RepositoryData> listener
313311
) {
314-
super.deleteSnapshots(snapshotIds, repositoryStateId, repositoryMetaVersion, remoteStoreLockManagerFactory, listener);
312+
super.deleteSnapshots(snapshotIds, repositoryStateId, repositoryMetaVersion, listener);
315313
}
316314

317315
@Override

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

Lines changed: 84 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,84 @@
1+
/*
2+
* SPDX-License-Identifier: Apache-2.0
3+
*
4+
* The OpenSearch Contributors require contributions made to
5+
* this file be licensed under the Apache-2.0 license or a
6+
* compatible open source license.
7+
*/
8+
9+
package org.opensearch.snapshots;
10+
11+
import org.opensearch.common.settings.Settings;
12+
import org.opensearch.common.util.FeatureFlags;
13+
import org.opensearch.test.FeatureFlagSetter;
14+
import org.opensearch.test.OpenSearchIntegTestCase;
15+
16+
import java.nio.file.Path;
17+
18+
import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertAcked;
19+
20+
@OpenSearchIntegTestCase.ClusterScope(scope = OpenSearchIntegTestCase.Scope.TEST, numDataNodes = 0)
21+
public class DeleteSnapshotIT extends AbstractSnapshotIntegTestCase {
22+
23+
public void testDeleteSnapshot() throws Exception {
24+
disableRepoConsistencyCheck("Remote store repository is being used in the test");
25+
FeatureFlagSetter.set(FeatureFlags.REMOTE_STORE);
26+
internalCluster().startClusterManagerOnlyNode();
27+
internalCluster().startDataOnlyNode();
28+
29+
final String snapshotRepoName = "snapshot-repo-name";
30+
final Path snapshotRepoPath = randomRepoPath();
31+
createRepository(snapshotRepoName, "fs", snapshotRepoPath);
32+
33+
final Path remoteStoreRepoPath = randomRepoPath();
34+
final String remoteStoreRepoName = "remote-store-repo-name";
35+
createRepository(remoteStoreRepoName, "fs", remoteStoreRepoPath);
36+
37+
final String indexName = "index-1";
38+
createIndexWithRandomDocs(indexName, randomIntBetween(5, 10));
39+
40+
final String remoteStoreEnabledIndexName = "remote-index-1";
41+
final Settings remoteStoreEnabledIndexSettings = getRemoteStoreBackedIndexSettings(remoteStoreRepoName);
42+
createIndex(remoteStoreEnabledIndexName, remoteStoreEnabledIndexSettings);
43+
indexRandomDocs(remoteStoreEnabledIndexName, randomIntBetween(5, 10));
44+
45+
final String snapshot = "snapshot";
46+
createFullSnapshot(snapshotRepoName, snapshot);
47+
assert (getLockFilesInRemoteStore(remoteStoreEnabledIndexName, remoteStoreRepoName).length == 0);
48+
assert (getRepositoryData(snapshotRepoName).getSnapshotIds().size() == 1);
49+
50+
assertAcked(startDeleteSnapshot(snapshotRepoName, snapshot).get());
51+
assert (getRepositoryData(snapshotRepoName).getSnapshotIds().size() == 0);
52+
}
53+
54+
public void testDeleteShallowCopySnapshot() throws Exception {
55+
disableRepoConsistencyCheck("Remote store repository is being used in the test");
56+
FeatureFlagSetter.set(FeatureFlags.REMOTE_STORE);
57+
internalCluster().startClusterManagerOnlyNode();
58+
internalCluster().startDataOnlyNode();
59+
60+
final String snapshotRepoName = "snapshot-repo-name";
61+
createRepository(snapshotRepoName, "fs", snapshotRepoSettingsForShallowCopy());
62+
63+
final Path remoteStoreRepoPath = randomRepoPath();
64+
final String remoteStoreRepoName = "remote-store-repo-name";
65+
createRepository(remoteStoreRepoName, "fs", remoteStoreRepoPath);
66+
67+
final String indexName = "index-1";
68+
createIndexWithRandomDocs(indexName, randomIntBetween(5, 10));
69+
70+
final String remoteStoreEnabledIndexName = "remote-index-1";
71+
final Settings remoteStoreEnabledIndexSettings = getRemoteStoreBackedIndexSettings(remoteStoreRepoName);
72+
createIndex(remoteStoreEnabledIndexName, remoteStoreEnabledIndexSettings);
73+
indexRandomDocs(remoteStoreEnabledIndexName, randomIntBetween(5, 10));
74+
75+
final String shallowSnapshot = "shallow-snapshot";
76+
createFullSnapshot(snapshotRepoName, shallowSnapshot);
77+
assert (getLockFilesInRemoteStore(remoteStoreEnabledIndexName, remoteStoreRepoName).length == 1);
78+
assert (getRepositoryData(snapshotRepoName).getSnapshotIds().size() == 1);
79+
80+
assertAcked(startDeleteSnapshot(snapshotRepoName, shallowSnapshot).get());
81+
assert (getRepositoryData(snapshotRepoName).getSnapshotIds().size() == 0);
82+
assert (getLockFilesInRemoteStore(remoteStoreEnabledIndexName, remoteStoreRepoName).length == 0);
83+
}
84+
}

‎server/src/main/java/org/opensearch/action/admin/cluster/repositories/cleanup/TransportCleanupRepositoryAction.java‎

Lines changed: 30 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,7 @@
3838
import org.opensearch.action.ActionRunnable;
3939
import org.opensearch.action.StepListener;
4040
import org.opensearch.action.support.ActionFilters;
41+
import org.opensearch.action.support.GroupedActionListener;
4142
import org.opensearch.action.support.clustermanager.TransportClusterManagerNodeAction;
4243
import org.opensearch.cluster.ClusterState;
4344
import org.opensearch.cluster.ClusterStateUpdateTask;
@@ -50,8 +51,10 @@
5051
import org.opensearch.cluster.node.DiscoveryNode;
5152
import org.opensearch.cluster.service.ClusterService;
5253
import org.opensearch.common.Nullable;
54+
import org.opensearch.common.blobstore.DeleteResult;
5355
import org.opensearch.common.inject.Inject;
5456
import org.opensearch.common.io.stream.StreamInput;
57+
import org.opensearch.index.store.lockmanager.RemoteStoreLockManagerFactory;
5558
import org.opensearch.repositories.RepositoriesService;
5659
import org.opensearch.repositories.Repository;
5760
import org.opensearch.repositories.RepositoryCleanupResult;
@@ -93,6 +96,8 @@ public final class TransportCleanupRepositoryAction extends TransportClusterMana
9396

9497
private final SnapshotsService snapshotsService;
9598

99+
private final RemoteStoreLockManagerFactory remoteStoreLockManagerFactory;
100+
96101
@Override
97102
protected String executor() {
98103
return ThreadPool.Names.SAME;
@@ -119,6 +124,7 @@ public TransportCleanupRepositoryAction(
119124
);
120125
this.repositoriesService = repositoriesService;
121126
this.snapshotsService = snapshotsService;
127+
this.remoteStoreLockManagerFactory = new RemoteStoreLockManagerFactory(() -> repositoriesService);
122128
// We add a state applier that will remove any dangling repository cleanup actions on cluster-manager failover.
123129
// This is safe to do since cleanups will increment the repository state id before executing any operations to prevent concurrent
124130
// operations from corrupting the repository. This is the same safety mechanism used by snapshot deletes.
@@ -260,17 +266,31 @@ public void onFailure(String source, Exception e) {
260266
public void clusterStateProcessed(String source, ClusterState oldState, ClusterState newState) {
261267
startedCleanup = true;
262268
logger.debug("Initialized repository cleanup in cluster state for [{}][{}]", repositoryName, repositoryStateId);
263-
threadPool.executor(ThreadPool.Names.SNAPSHOT)
264-
.execute(
265-
ActionRunnable.wrap(
266-
listener,
267-
l -> blobStoreRepository.cleanup(
268-
repositoryStateId,
269-
snapshotsService.minCompatibleVersion(newState.nodes().getMinNodeVersion(), repositoryData, null),
270-
ActionListener.wrap(result -> after(null, result), e -> after(e, null))
271-
)
272-
)
269+
threadPool.executor(ThreadPool.Names.SNAPSHOT).execute(ActionRunnable.wrap(listener, l -> {
270+
final GroupedActionListener<RepositoryCleanupResult> groupedListener = new GroupedActionListener<>(
271+
ActionListener.wrap(repositoryCleanupResults -> {
272+
long blobs = 0;
273+
long bytes = 0;
274+
for (RepositoryCleanupResult result : repositoryCleanupResults) {
275+
blobs += result.blobs();
276+
bytes += result.bytes();
277+
}
278+
RepositoryCleanupResult result = new RepositoryCleanupResult(new DeleteResult(blobs, bytes));
279+
after(null, result);
280+
}, e -> { after(e, null); }),
281+
2
282+
);
283+
blobStoreRepository.cleanupRemoteStoreLockFiles(
284+
repositoryStateId,
285+
remoteStoreLockManagerFactory,
286+
ActionListener.wrap(result -> groupedListener.onResponse(result), e -> groupedListener.onFailure(e))
287+
);
288+
blobStoreRepository.cleanup(
289+
repositoryStateId,
290+
snapshotsService.minCompatibleVersion(newState.nodes().getMinNodeVersion(), repositoryData, null),
291+
ActionListener.wrap(result -> groupedListener.onResponse(result), e -> groupedListener.onFailure(e))
273292
);
293+
}));
274294
}
275295

276296
private void after(@Nullable Exception failure, @Nullable RepositoryCleanupResult result) {

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

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -46,7 +46,6 @@
4646
import org.opensearch.index.shard.ShardId;
4747
import org.opensearch.index.snapshots.IndexShardSnapshotStatus;
4848
import org.opensearch.index.store.Store;
49-
import org.opensearch.index.store.lockmanager.RemoteStoreLockManagerFactory;
5049
import org.opensearch.indices.recovery.RecoveryState;
5150
import org.opensearch.snapshots.SnapshotId;
5251
import org.opensearch.snapshots.SnapshotInfo;
@@ -121,10 +120,9 @@ public void deleteSnapshots(
121120
Collection<SnapshotId> snapshotIds,
122121
long repositoryStateId,
123122
Version repositoryMetaVersion,
124-
RemoteStoreLockManagerFactory remoteStoreLockManagerFactory,
125123
ActionListener<RepositoryData> listener
126124
) {
127-
in.deleteSnapshots(snapshotIds, repositoryStateId, repositoryMetaVersion, remoteStoreLockManagerFactory, listener);
125+
in.deleteSnapshots(snapshotIds, repositoryStateId, repositoryMetaVersion, listener);
128126
}
129127

130128
@Override

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,7 @@
4242
import org.opensearch.cluster.metadata.RepositoryMetadata;
4343
import org.opensearch.cluster.node.DiscoveryNode;
4444
import org.opensearch.common.Nullable;
45+
import org.opensearch.common.blobstore.DeleteResult;
4546
import org.opensearch.common.component.LifecycleComponent;
4647
import org.opensearch.index.mapper.MapperService;
4748
import org.opensearch.index.shard.ShardId;
@@ -165,7 +166,6 @@ void deleteSnapshots(
165166
Collection<SnapshotId> snapshotIds,
166167
long repositoryStateId,
167168
Version repositoryMetaVersion,
168-
RemoteStoreLockManagerFactory remoteStoreLockManagerFactory,
169169
ActionListener<RepositoryData> listener
170170
);
171171

0 commit comments

Comments
 (0)