Skip to content

Commit

Permalink
[SnapshotV2] Snapshot Status API changes (#15409)
Browse files Browse the repository at this point in the history 
---------

Signed-off-by: Lakshya Taragi <lakshya.taragi@gmail.com>
  • Loading branch information
@ltaragi
ltaragi authored Sep 3, 2024
1 parent 41ba00a commit 925f41b
Show file tree
Hide file tree
Showing 15 changed files with 657 additions and 51 deletions.
1 change: 1 addition & 0 deletions CHANGELOG.md
View file
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
- [Workload Management] Add query group level failure tracking ([#15227](https://github.com/opensearch-project/OpenSearch/pull/15527))
- Add support to upload snapshot shard blobs with hashed prefix ([#15426](https://github.com/opensearch-project/OpenSearch/pull/15426))
- [Remote Publication] Add remote download stats ([#15291](https://github.com/opensearch-project/OpenSearch/pull/15291)))
- Add support for comma-separated list of index names to be used with Snapshot Status API ([#15409](https://github.com/opensearch-project/OpenSearch/pull/15409))

### Dependencies
- Bump `netty` from 4.1.111.Final to 4.1.112.Final ([#15081](https://github.com/opensearch-project/OpenSearch/pull/15081))
Expand Down
3 changes: 2 additions & 1 deletion ...t/rest-high-level/src/test/java/org/opensearch/client/SnapshotRequestConvertersTests.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -230,14 +230,15 @@ public void testSnapshotsStatus() {
Map<String, String> expectedParams = new HashMap<>();
String repository = RequestConvertersTests.randomIndicesNames(1, 1)[0];
String[] snapshots = RequestConvertersTests.randomIndicesNames(1, 5);
String[] indices = RequestConvertersTests.randomIndicesNames(1, 5);
StringBuilder snapshotNames = new StringBuilder(snapshots[0]);
for (int idx = 1; idx < snapshots.length; idx++) {
snapshotNames.append(",").append(snapshots[idx]);
}
boolean ignoreUnavailable = randomBoolean();
String endpoint = "/_snapshot/" + repository + "/" + snapshotNames.toString() + "/_status";

SnapshotsStatusRequest snapshotsStatusRequest = new SnapshotsStatusRequest(repository, snapshots);
SnapshotsStatusRequest snapshotsStatusRequest = new SnapshotsStatusRequest(repository, snapshots, indices);
RequestConvertersTests.setRandomClusterManagerTimeout(snapshotsStatusRequest, expectedParams);
snapshotsStatusRequest.ignoreUnavailable(ignoreUnavailable);
expectedParams.put("ignore_unavailable", Boolean.toString(ignoreUnavailable));
Expand Down
18 changes: 10 additions & 8 deletions ...y-multi-version/src/test/java/org/opensearch/upgrades/MultiVersionRepositoryAccessIT.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@

package org.opensearch.upgrades;

import com.sun.jna.StringArray;
import org.opensearch.action.admin.cluster.repositories.put.PutRepositoryRequest;
import org.opensearch.action.admin.cluster.snapshots.delete.DeleteSnapshotRequest;
import org.opensearch.action.admin.cluster.snapshots.status.SnapshotStatus;
Expand All @@ -44,6 +45,7 @@
import org.opensearch.client.RestClient;
import org.opensearch.client.RestHighLevelClient;
import org.opensearch.common.settings.Settings;
import org.opensearch.core.common.Strings;
import org.opensearch.core.xcontent.DeprecationHandler;
import org.opensearch.core.xcontent.XContentParser;
import org.opensearch.common.xcontent.json.JsonXContent;
Expand Down Expand Up @@ -141,14 +143,14 @@ public void testCreateAndRestoreSnapshot() throws IOException {
case STEP2_NEW_CLUSTER:
case STEP4_NEW_CLUSTER:
assertSnapshotStatusSuccessful(client, repoName,
snapshots.stream().map(sn -> (String) sn.get("snapshot")).toArray(String[]::new));
snapshots.stream().map(sn -> (String) sn.get("snapshot")).toArray(String[]::new), Strings.EMPTY_ARRAY);
break;
case STEP1_OLD_CLUSTER:
assertSnapshotStatusSuccessful(client, repoName, "snapshot-" + TEST_STEP);
assertSnapshotStatusSuccessful(client, repoName, new String[] {"snapshot-" + TEST_STEP}, Strings.EMPTY_ARRAY);
break;
case STEP3_OLD_CLUSTER:
assertSnapshotStatusSuccessful(
client, repoName, "snapshot-" + TEST_STEP, "snapshot-" + TestStep.STEP3_OLD_CLUSTER);
client, repoName, new String[] {"snapshot-" + TEST_STEP, "snapshot-" + TestStep.STEP3_OLD_CLUSTER}, Strings.EMPTY_ARRAY);
break;
}
if (TEST_STEP == TestStep.STEP3_OLD_CLUSTER) {
Expand Down Expand Up @@ -186,10 +188,10 @@ public void testReadOnlyRepo() throws IOException {
break;
}
if (TEST_STEP == TestStep.STEP1_OLD_CLUSTER || TEST_STEP == TestStep.STEP3_OLD_CLUSTER) {
assertSnapshotStatusSuccessful(client, repoName, "snapshot-" + TestStep.STEP1_OLD_CLUSTER);
assertSnapshotStatusSuccessful(client, repoName, new String[] {"snapshot-" + TestStep.STEP1_OLD_CLUSTER}, Strings.EMPTY_ARRAY);
} else {
assertSnapshotStatusSuccessful(client, repoName,
"snapshot-" + TestStep.STEP1_OLD_CLUSTER, "snapshot-" + TestStep.STEP2_NEW_CLUSTER);
new String[] {"snapshot-" + TestStep.STEP1_OLD_CLUSTER, "snapshot-" + TestStep.STEP2_NEW_CLUSTER}, Strings.EMPTY_ARRAY);
}
if (TEST_STEP == TestStep.STEP3_OLD_CLUSTER) {
ensureSnapshotRestoreWorks(repoName, "snapshot-" + TestStep.STEP1_OLD_CLUSTER, shards);
Expand All @@ -214,7 +216,7 @@ public void testUpgradeMovesRepoToNewMetaVersion() throws IOException {
// Every step creates one snapshot
assertThat(snapshots, hasSize(TEST_STEP.ordinal() + 1));
assertSnapshotStatusSuccessful(client, repoName,
snapshots.stream().map(sn -> (String) sn.get("snapshot")).toArray(String[]::new));
snapshots.stream().map(sn -> (String) sn.get("snapshot")).toArray(String[]::new), Strings.EMPTY_ARRAY);
if (TEST_STEP == TestStep.STEP1_OLD_CLUSTER) {
ensureSnapshotRestoreWorks(repoName, "snapshot-" + TestStep.STEP1_OLD_CLUSTER, shards);
} else {
Expand All @@ -239,9 +241,9 @@ public void testUpgradeMovesRepoToNewMetaVersion() throws IOException {
}

private static void assertSnapshotStatusSuccessful(RestHighLevelClient client, String repoName,
String... snapshots) throws IOException {
String[] snapshots, String[] indices) throws IOException {
final SnapshotsStatusResponse statusResponse = client.snapshot()
.status(new SnapshotsStatusRequest(repoName, snapshots), RequestOptions.DEFAULT);
.status(new SnapshotsStatusRequest(repoName, snapshots, indices), RequestOptions.DEFAULT);
for (SnapshotStatus status : statusResponse.getSnapshots()) {
assertThat(status.getShardsStats().getFailedShards(), is(0));
}
Expand Down
22 changes: 21 additions & 1 deletion rest-api-spec/src/main/resources/rest-api-spec/api/snapshot.status.json
View file
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,26 @@
"description":"A comma-separated list of snapshot names"
}
}
},
{
"path":"/_snapshot/{repository}/{snapshot}/{index}/_status",
"methods":[
"GET"
],
"parts":{
"repository":{
"type":"string",
"description":"A repository name"
},
"snapshot":{
"type":"string",
"description":"A snapshot name"
},
"index":{
"type": "list",
"description":"A comma-separated list of index names"
}
}
}
]
},
Expand All @@ -58,7 +78,7 @@
},
"ignore_unavailable":{
"type":"boolean",
"description":"Whether to ignore unavailable snapshots, defaults to false which means a SnapshotMissingException is thrown"
"description":"Whether to ignore unavailable snapshots and indices, defaults to false which means a SnapshotMissingException or IndexNotFoundException is thrown"
}
}
}
Expand Down
112 changes: 112 additions & 0 deletions ...src/internalClusterTest/java/org/opensearch/snapshots/RemoteIndexSnapshotStatusApiIT.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -32,20 +32,28 @@

package org.opensearch.snapshots;

import org.opensearch.action.admin.cluster.settings.ClusterUpdateSettingsRequest;
import org.opensearch.action.admin.cluster.snapshots.create.CreateSnapshotResponse;
import org.opensearch.action.admin.cluster.snapshots.status.SnapshotIndexShardStage;
import org.opensearch.action.admin.cluster.snapshots.status.SnapshotIndexShardStatus;
import org.opensearch.action.admin.cluster.snapshots.status.SnapshotIndexStatus;
import org.opensearch.action.admin.cluster.snapshots.status.SnapshotStatus;
import org.opensearch.cluster.SnapshotsInProgress;
import org.opensearch.common.action.ActionFuture;
import org.opensearch.common.settings.Settings;
import org.opensearch.indices.RemoteStoreSettings;
import org.opensearch.repositories.blobstore.BlobStoreRepository;
import org.opensearch.test.OpenSearchIntegTestCase;
import org.opensearch.threadpool.ThreadPool;
import org.junit.Before;

import java.nio.file.Path;
import java.util.Map;
import java.util.concurrent.TimeUnit;

import static org.opensearch.remotestore.RemoteStoreBaseIntegTestCase.remoteStoreClusterSettings;
import static org.opensearch.snapshots.SnapshotsService.MAX_SHARDS_ALLOWED_IN_STATUS_API;
import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertAcked;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.greaterThan;
import static org.hamcrest.Matchers.is;
Expand Down Expand Up @@ -192,6 +200,110 @@ public void testStatusAPICallInProgressShallowSnapshot() throws Exception {
createSnapshotResponseActionFuture.actionGet();
}

public void testStatusAPICallForShallowV2Snapshot() throws Exception {
disableRepoConsistencyCheck("Remote store repository is being used for the test");
Settings pinnedTimestampSettings = Settings.builder()
.put(RemoteStoreSettings.CLUSTER_REMOTE_STORE_PINNED_TIMESTAMP_ENABLED.getKey(), true)
.build();
internalCluster().startClusterManagerOnlyNode(pinnedTimestampSettings);
internalCluster().startDataOnlyNodes(2, pinnedTimestampSettings);

final String index1 = "remote-index-1";
final String index2 = "remote-index-2";
final String index3 = "remote-index-3";
final String snapshotRepoName = "snapshot-repo-name";
final String snapshot = "snapshot";

logger.info("Create repository for shallow V2 snapshots");
Settings.Builder snapshotV2RepoSettings = snapshotRepoSettingsForShallowCopy().put(
BlobStoreRepository.SHALLOW_SNAPSHOT_V2.getKey(),
Boolean.TRUE
);
createRepository(snapshotRepoName, "fs", snapshotV2RepoSettings);

final Settings remoteStoreEnabledIndexSettings = getRemoteStoreBackedIndexSettings();
createIndex(index1, remoteStoreEnabledIndexSettings);
createIndex(index2, remoteStoreEnabledIndexSettings);
createIndex(index3, remoteStoreEnabledIndexSettings);
ensureGreen();

logger.info("Indexing some data");
for (int i = 0; i < 50; i++) {
index(index1, "_doc", Integer.toString(i), "foo", "bar" + i);
index(index2, "_doc", Integer.toString(i), "foo", "bar" + i);
index(index3, "_doc", Integer.toString(i), "foo", "bar" + i);
}
refresh();

SnapshotInfo snapshotInfo = createFullSnapshot(snapshotRepoName, snapshot);
assertTrue(snapshotInfo.getPinnedTimestamp() > 0); // to assert creation of a shallow v2 snapshot

logger.info("Set MAX_SHARDS_ALLOWED_IN_STATUS_API to a low value");
ClusterUpdateSettingsRequest updateSettingsRequest = new ClusterUpdateSettingsRequest();
updateSettingsRequest.persistentSettings(Settings.builder().put(MAX_SHARDS_ALLOWED_IN_STATUS_API.getKey(), 2));
assertAcked(client().admin().cluster().updateSettings(updateSettingsRequest).actionGet());

assertBusy(() -> {
// without index filter
// although no. of shards in snapshot (3) is greater than the max value allowed in a status api call, the request does not fail
SnapshotStatus snapshotStatusWithoutIndexFilter = client().admin()
.cluster()
.prepareSnapshotStatus(snapshotRepoName)
.setSnapshots(snapshot)
.execute()
.actionGet()
.getSnapshots()
.get(0);

assertShallowV2SnapshotStatus(snapshotStatusWithoutIndexFilter, false);

// with index filter
SnapshotStatus snapshotStatusWithIndexFilter = client().admin()
.cluster()
.prepareSnapshotStatus(snapshotRepoName)
.setSnapshots(snapshot)
.setIndices(index1, index2)
.execute()
.actionGet()
.getSnapshots()
.get(0);

assertShallowV2SnapshotStatus(snapshotStatusWithIndexFilter, true);

}, 1, TimeUnit.MINUTES);

}

private void assertShallowV2SnapshotStatus(SnapshotStatus snapshotStatus, boolean hasIndexFilter) {
if (hasIndexFilter) {
assertEquals(0, snapshotStatus.getStats().getTotalSize());
} else {
// TODO: after adding primary store size at the snapshot level, total size here should be > 0
}
// assert that total and incremental values of file count and size_in_bytes are 0 at index and shard levels
assertEquals(0, snapshotStatus.getStats().getTotalFileCount());
assertEquals(0, snapshotStatus.getStats().getIncrementalSize());
assertEquals(0, snapshotStatus.getStats().getIncrementalFileCount());

for (Map.Entry<String, SnapshotIndexStatus> entry : snapshotStatus.getIndices().entrySet()) {
// index level
SnapshotIndexStatus snapshotIndexStatus = entry.getValue();
assertEquals(0, snapshotIndexStatus.getStats().getTotalSize());
assertEquals(0, snapshotIndexStatus.getStats().getTotalFileCount());
assertEquals(0, snapshotIndexStatus.getStats().getIncrementalSize());
assertEquals(0, snapshotIndexStatus.getStats().getIncrementalFileCount());

for (SnapshotIndexShardStatus snapshotIndexShardStatus : snapshotStatus.getShards()) {
// shard level
assertEquals(0, snapshotIndexShardStatus.getStats().getTotalSize());
assertEquals(0, snapshotIndexShardStatus.getStats().getTotalFileCount());
assertEquals(0, snapshotIndexShardStatus.getStats().getIncrementalSize());
assertEquals(0, snapshotIndexShardStatus.getStats().getIncrementalFileCount());
assertEquals(SnapshotIndexShardStage.DONE, snapshotIndexShardStatus.getStage());
}
}
}

private static SnapshotIndexShardStatus stateFirstShard(SnapshotStatus snapshotStatus, String indexName) {
return snapshotStatus.getIndices().get(indexName).getShards().get(0);
}
Expand Down
Loading

0 comments on commit 925f41b

Please sign in to comment.