Skip to content

Commit 952c13c

Browse files
original-brownbearoriginal-brownbear
authored
Dry up Snapshot Integ Tests some More (#62856)
Just some obvious drying up of these super complex tests.
1 parent 0bcf783 commit 952c13c

File tree

9 files changed

+268
-381
lines changed

9 files changed

+268
-381
lines changed

‎server/src/internalClusterTest/java/org/elasticsearch/repositories/blobstore/BlobStoreRepositoryCleanupIT.java

+4-11
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,6 @@
3131

3232
import java.io.ByteArrayInputStream;
3333
import java.util.concurrent.ExecutionException;
34-
import java.util.concurrent.TimeUnit;
3534

3635
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertFutureThrows;
3736
import static org.hamcrest.Matchers.is;
@@ -49,11 +48,8 @@ public void testMasterFailoverDuringCleanup() throws Exception {
4948
ensureStableCluster(nodeCount - 1);
5049

5150
logger.info("--> wait for cleanup to finish and disappear from cluster state");
52-
assertBusy(() -> {
53-
RepositoryCleanupInProgress cleanupInProgress =
54-
client().admin().cluster().prepareState().get().getState().custom(RepositoryCleanupInProgress.TYPE);
55-
assertFalse(cleanupInProgress.hasCleanupInProgress());
56-
}, 30, TimeUnit.SECONDS);
51+
awaitClusterState(state ->
52+
state.custom(RepositoryCleanupInProgress.TYPE, RepositoryCleanupInProgress.EMPTY).hasCleanupInProgress() == false);
5753
}
5854

5955
public void testRepeatCleanupsDontRemove() throws Exception {
@@ -71,11 +67,8 @@ public void testRepeatCleanupsDontRemove() throws Exception {
7167
unblockNode("test-repo", masterNode);
7268

7369
logger.info("--> wait for cleanup to finish and disappear from cluster state");
74-
assertBusy(() -> {
75-
RepositoryCleanupInProgress cleanupInProgress =
76-
client().admin().cluster().prepareState().get().getState().custom(RepositoryCleanupInProgress.TYPE);
77-
assertFalse(cleanupInProgress.hasCleanupInProgress());
78-
}, 30, TimeUnit.SECONDS);
70+
awaitClusterState(state ->
71+
state.custom(RepositoryCleanupInProgress.TYPE, RepositoryCleanupInProgress.EMPTY).hasCleanupInProgress() == false);
7972
}
8073

8174
private String startBlockedCleanup(String repoName) throws Exception {

‎server/src/internalClusterTest/java/org/elasticsearch/snapshots/CorruptedBlobStoreRepositoryIT.java

+4-25
Original file line numberDiff line numberDiff line change
@@ -24,10 +24,8 @@
2424
import org.elasticsearch.action.support.PlainActionFuture;
2525
import org.elasticsearch.client.Client;
2626
import org.elasticsearch.cluster.ClusterState;
27-
import org.elasticsearch.cluster.ClusterStateUpdateTask;
2827
import org.elasticsearch.cluster.metadata.Metadata;
2928
import org.elasticsearch.cluster.metadata.RepositoriesMetadata;
30-
import org.elasticsearch.cluster.service.ClusterService;
3129
import org.elasticsearch.common.bytes.BytesReference;
3230
import org.elasticsearch.common.settings.Settings;
3331
import org.elasticsearch.common.unit.ByteSizeUnit;
@@ -149,29 +147,10 @@ public void testFindDanglingLatestGeneration() throws Exception {
149147
Files.move(repo.resolve("index-" + beforeMoveGen), repo.resolve("index-" + (beforeMoveGen + 1)));
150148

151149
logger.info("--> set next generation as pending in the cluster state");
152-
final PlainActionFuture<Void> csUpdateFuture = PlainActionFuture.newFuture();
153-
internalCluster().getCurrentMasterNodeInstance(ClusterService.class).submitStateUpdateTask("set pending generation",
154-
new ClusterStateUpdateTask() {
155-
@Override
156-
public ClusterState execute(ClusterState currentState) {
157-
return ClusterState.builder(currentState).metadata(Metadata.builder(currentState.getMetadata())
158-
.putCustom(RepositoriesMetadata.TYPE,
159-
currentState.metadata().<RepositoriesMetadata>custom(RepositoriesMetadata.TYPE).withUpdatedGeneration(
160-
repository.getMetadata().name(), beforeMoveGen, beforeMoveGen + 1)).build()).build();
161-
}
162-
163-
@Override
164-
public void onFailure(String source, Exception e) {
165-
csUpdateFuture.onFailure(e);
166-
}
167-
168-
@Override
169-
public void clusterStateProcessed(String source, ClusterState oldState, ClusterState newState) {
170-
csUpdateFuture.onResponse(null);
171-
}
172-
}
173-
);
174-
csUpdateFuture.get();
150+
updateClusterState(currentState -> ClusterState.builder(currentState).metadata(Metadata.builder(currentState.getMetadata())
151+
.putCustom(RepositoriesMetadata.TYPE,
152+
currentState.metadata().<RepositoriesMetadata>custom(RepositoriesMetadata.TYPE).withUpdatedGeneration(
153+
repository.getMetadata().name(), beforeMoveGen, beforeMoveGen + 1)).build()).build());
175154

176155
logger.info("--> full cluster restart");
177156
internalCluster().fullRestart();

‎server/src/internalClusterTest/java/org/elasticsearch/snapshots/DedicatedClusterSnapshotRestoreIT.java

+59-121
Large diffs are not rendered by default.

‎server/src/internalClusterTest/java/org/elasticsearch/snapshots/SharedClusterSnapshotRestoreIT.java

+141-181
Large diffs are not rendered by default.

‎server/src/internalClusterTest/java/org/elasticsearch/snapshots/SnapshotShardsServiceIT.java

+2-7
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,6 @@
1919

2020
package org.elasticsearch.snapshots;
2121

22-
import org.elasticsearch.action.admin.cluster.snapshots.get.GetSnapshotsResponse;
2322
import org.elasticsearch.common.unit.TimeValue;
2423
import org.elasticsearch.index.snapshots.IndexShardSnapshotStatus;
2524
import org.elasticsearch.plugins.Plugin;
@@ -66,8 +65,7 @@ public void testRetryPostingSnapshotStatusMessages() throws Exception {
6665
.get();
6766
waitForBlock(blockedNode, "test-repo", TimeValue.timeValueSeconds(60));
6867

69-
final SnapshotId snapshotId = client().admin().cluster().prepareGetSnapshots("test-repo").setSnapshots("test-snap")
70-
.get().getSnapshots("test-repo").get(0).snapshotId();
68+
final SnapshotId snapshotId = getSnapshot("test-repo", "test-snap").snapshotId();
7169

7270
logger.info("--> start disrupting cluster");
7371
final NetworkDisruption networkDisruption = isolateMasterDisruption(NetworkDisruption.NetworkDelay.random(random()));
@@ -92,10 +90,7 @@ public void testRetryPostingSnapshotStatusMessages() throws Exception {
9290
internalCluster().clearDisruptionScheme(true);
9391

9492
assertBusy(() -> {
95-
GetSnapshotsResponse snapshotsStatusResponse = client().admin().cluster()
96-
.prepareGetSnapshots("test-repo")
97-
.setSnapshots("test-snap").get();
98-
SnapshotInfo snapshotInfo = snapshotsStatusResponse.getSnapshots("test-repo").get(0);
93+
SnapshotInfo snapshotInfo = getSnapshot("test-repo", "test-snap");
9994
logger.info("Snapshot status [{}], successfulShards [{}]", snapshotInfo.state(), snapshotInfo.successfulShards());
10095
assertThat(snapshotInfo.state(), equalTo(SnapshotState.SUCCESS));
10196
assertThat(snapshotInfo.successfulShards(), equalTo(shards));

‎server/src/internalClusterTest/java/org/elasticsearch/snapshots/SnapshotStatusApisIT.java

+4-11
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,6 @@
2828
import org.elasticsearch.action.admin.cluster.snapshots.status.SnapshotStats;
2929
import org.elasticsearch.action.admin.cluster.snapshots.status.SnapshotStatus;
3030
import org.elasticsearch.action.admin.cluster.snapshots.status.SnapshotsStatusRequest;
31-
import org.elasticsearch.client.Client;
3231
import org.elasticsearch.cluster.SnapshotsInProgress;
3332
import org.elasticsearch.common.settings.Settings;
3433
import org.elasticsearch.common.unit.TimeValue;
@@ -49,8 +48,6 @@
4948
public class SnapshotStatusApisIT extends AbstractSnapshotIntegTestCase {
5049

5150
public void testStatusApiConsistency() {
52-
Client client = client();
53-
5451
createRepository("test-repo", "fs");
5552

5653
createIndex("test-idx-1", "test-idx-2", "test-idx-3");
@@ -66,14 +63,13 @@ public void testStatusApiConsistency() {
6663

6764
createFullSnapshot("test-repo", "test-snap");
6865

69-
List<SnapshotInfo> snapshotInfos =
70-
client.admin().cluster().prepareGetSnapshots("test-repo").get().getSnapshots("test-repo");
66+
List<SnapshotInfo> snapshotInfos = clusterAdmin().prepareGetSnapshots("test-repo").get().getSnapshots("test-repo");
7167
assertThat(snapshotInfos.size(), equalTo(1));
7268
SnapshotInfo snapshotInfo = snapshotInfos.get(0);
7369
assertThat(snapshotInfo.state(), equalTo(SnapshotState.SUCCESS));
7470
assertThat(snapshotInfo.version(), equalTo(Version.CURRENT));
7571

76-
final List<SnapshotStatus> snapshotStatus = client.admin().cluster().snapshotsStatus(
72+
final List<SnapshotStatus> snapshotStatus = clusterAdmin().snapshotsStatus(
7773
new SnapshotsStatusRequest("test-repo", new String[]{"test-snap"})).actionGet().getSnapshots();
7874
assertThat(snapshotStatus.size(), equalTo(1));
7975
final SnapshotStatus snStatus = snapshotStatus.get(0);
@@ -82,8 +78,6 @@ public void testStatusApiConsistency() {
8278
}
8379

8480
public void testStatusAPICallInProgressSnapshot() throws Exception {
85-
Client client = client();
86-
8781
createRepository("test-repo", "mock", Settings.builder().put("location", randomRepoPath()).put("block_on_data", true));
8882

8983
createIndex("test-idx-1");
@@ -96,13 +90,12 @@ public void testStatusAPICallInProgressSnapshot() throws Exception {
9690
refresh();
9791

9892
logger.info("--> snapshot");
99-
ActionFuture<CreateSnapshotResponse> createSnapshotResponseActionFuture =
100-
client.admin().cluster().prepareCreateSnapshot("test-repo", "test-snap").setWaitForCompletion(true).execute();
93+
ActionFuture<CreateSnapshotResponse> createSnapshotResponseActionFuture = startFullSnapshot("test-repo", "test-snap");
10194

10295
logger.info("--> wait for data nodes to get blocked");
10396
waitForBlockOnAnyDataNode("test-repo", TimeValue.timeValueMinutes(1));
10497
awaitNumberOfSnapshotsInProgress(1);
105-
assertEquals(SnapshotsInProgress.State.STARTED, client.admin().cluster().prepareSnapshotStatus("test-repo")
98+
assertEquals(SnapshotsInProgress.State.STARTED, client().admin().cluster().prepareSnapshotStatus("test-repo")
10699
.setSnapshots("test-snap").get().getSnapshots().get(0).getState());
107100

108101
logger.info("--> unblock all data nodes");

‎test/framework/src/main/java/org/elasticsearch/snapshots/AbstractSnapshotIntegTestCase.java

+44-16
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@
2828
import org.elasticsearch.action.support.master.AcknowledgedResponse;
2929
import org.elasticsearch.cluster.ClusterState;
3030
import org.elasticsearch.cluster.ClusterStateObserver;
31+
import org.elasticsearch.cluster.ClusterStateUpdateTask;
3132
import org.elasticsearch.cluster.SnapshotDeletionsInProgress;
3233
import org.elasticsearch.cluster.SnapshotsInProgress;
3334
import org.elasticsearch.cluster.metadata.IndexMetadata;
@@ -82,7 +83,7 @@
8283

8384
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
8485
import static org.hamcrest.Matchers.empty;
85-
import static org.hamcrest.Matchers.equalTo;
86+
import static org.hamcrest.Matchers.hasSize;
8687
import static org.hamcrest.Matchers.is;
8788

8889
public abstract class AbstractSnapshotIntegTestCase extends ESIntegTestCase {
@@ -122,11 +123,11 @@ public void verifyNoLeakedListeners() throws Exception {
122123
@After
123124
public void assertRepoConsistency() {
124125
if (skipRepoConsistencyCheckReason == null) {
125-
client().admin().cluster().prepareGetRepositories().get().repositories().forEach(repositoryMetadata -> {
126+
clusterAdmin().prepareGetRepositories().get().repositories().forEach(repositoryMetadata -> {
126127
final String name = repositoryMetadata.name();
127128
if (repositoryMetadata.settings().getAsBoolean("readonly", false) == false) {
128-
client().admin().cluster().prepareDeleteSnapshot(name, OLD_VERSION_SNAPSHOT_PREFIX + "*").get();
129-
client().admin().cluster().prepareCleanupRepository(name).get();
129+
clusterAdmin().prepareDeleteSnapshot(name, OLD_VERSION_SNAPSHOT_PREFIX + "*").get();
130+
clusterAdmin().prepareCleanupRepository(name).get();
130131
}
131132
BlobStoreTestUtil.assertRepoConsistency(internalCluster(), name);
132133
});
@@ -202,12 +203,10 @@ public static void waitForBlock(String node, String repository, TimeValue timeou
202203
public SnapshotInfo waitForCompletion(String repository, String snapshotName, TimeValue timeout) throws InterruptedException {
203204
long start = System.currentTimeMillis();
204205
while (System.currentTimeMillis() - start < timeout.millis()) {
205-
List<SnapshotInfo> snapshotInfos = client().admin().cluster().prepareGetSnapshots(repository).setSnapshots(snapshotName)
206-
.get().getSnapshots(repository);
207-
assertThat(snapshotInfos.size(), equalTo(1));
208-
if (snapshotInfos.get(0).state().completed()) {
206+
final SnapshotInfo snapshotInfo = getSnapshot(repository, snapshotName);
207+
if (snapshotInfo.state().completed()) {
209208
// Make sure that snapshot clean up operations are finished
210-
ClusterStateResponse stateResponse = client().admin().cluster().prepareState().get();
209+
ClusterStateResponse stateResponse = clusterAdmin().prepareState().get();
211210
boolean found = false;
212211
for (SnapshotsInProgress.Entry entry :
213212
stateResponse.getState().custom(SnapshotsInProgress.TYPE, SnapshotsInProgress.EMPTY).entries()) {
@@ -218,7 +217,7 @@ public SnapshotInfo waitForCompletion(String repository, String snapshotName, Ti
218217
}
219218
}
220219
if (found == false) {
221-
return snapshotInfos.get(0);
220+
return snapshotInfo;
222221
}
223222
}
224223
Thread.sleep(100);
@@ -307,7 +306,7 @@ public void unblockNode(final String repository, final String node) {
307306

308307
protected void createRepository(String repoName, String type, Settings.Builder settings) {
309308
logger.info("--> creating repository [{}] [{}]", repoName, type);
310-
assertAcked(client().admin().cluster().preparePutRepository(repoName)
309+
assertAcked(clusterAdmin().preparePutRepository(repoName)
311310
.setType(type).setSettings(settings));
312311
}
313312

@@ -349,7 +348,7 @@ protected void maybeInitWithOldSnapshotVersion(String repoName, Path repoPath) t
349348
protected String initWithSnapshotVersion(String repoName, Path repoPath, Version version) throws IOException {
350349
assertThat("This hack only works on an empty repository", getRepositoryData(repoName).getSnapshotIds(), empty());
351350
final String oldVersionSnapshot = OLD_VERSION_SNAPSHOT_PREFIX + version.id;
352-
final CreateSnapshotResponse createSnapshotResponse = client().admin().cluster()
351+
final CreateSnapshotResponse createSnapshotResponse = clusterAdmin()
353352
.prepareCreateSnapshot(repoName, oldVersionSnapshot).setIndices("does-not-exist-for-sure-*")
354353
.setWaitForCompletion(true).get();
355354
assertThat(createSnapshotResponse.getSnapshotInfo().totalShards(), is(0));
@@ -372,7 +371,7 @@ protected String initWithSnapshotVersion(String repoName, Path repoPath, Version
372371

373372
protected SnapshotInfo createFullSnapshot(String repoName, String snapshotName) {
374373
logger.info("--> creating full snapshot [{}] in [{}]", snapshotName, repoName);
375-
CreateSnapshotResponse createSnapshotResponse = client().admin().cluster().prepareCreateSnapshot(repoName, snapshotName)
374+
CreateSnapshotResponse createSnapshotResponse = clusterAdmin().prepareCreateSnapshot(repoName, snapshotName)
376375
.setIncludeGlobalState(true)
377376
.setWaitForCompletion(true)
378377
.get();
@@ -416,7 +415,7 @@ protected void assertDocCount(String index, long count) {
416415
* @param metadata snapshot metadata to write (as returned by {@link SnapshotInfo#userMetadata()})
417416
*/
418417
protected void addBwCFailedSnapshot(String repoName, String snapshotName, Map<String, Object> metadata) throws Exception {
419-
final ClusterState state = client().admin().cluster().prepareState().get().getState();
418+
final ClusterState state = clusterAdmin().prepareState().get().getState();
420419
final RepositoriesMetadata repositoriesMetadata = state.metadata().custom(RepositoriesMetadata.TYPE);
421420
assertNotNull(repositoriesMetadata);
422421
final RepositoryMetadata initialRepoMetadata = repositoriesMetadata.repository(repoName);
@@ -486,7 +485,7 @@ protected ActionFuture<CreateSnapshotResponse> startFullSnapshot(String repoName
486485

487486
protected ActionFuture<CreateSnapshotResponse> startFullSnapshot(String repoName, String snapshotName, boolean partial) {
488487
logger.info("--> creating full snapshot [{}] to repo [{}]", snapshotName, repoName);
489-
return client().admin().cluster().prepareCreateSnapshot(repoName, snapshotName).setWaitForCompletion(true)
488+
return clusterAdmin().prepareCreateSnapshot(repoName, snapshotName).setWaitForCompletion(true)
490489
.setPartial(partial).execute();
491490
}
492491

@@ -517,6 +516,35 @@ protected void createIndexWithContent(String indexName, Settings indexSettings)
517516

518517
protected ActionFuture<AcknowledgedResponse> startDeleteSnapshot(String repoName, String snapshotName) {
519518
logger.info("--> deleting snapshot [{}] from repo [{}]", snapshotName, repoName);
520-
return client().admin().cluster().prepareDeleteSnapshot(repoName, snapshotName).execute();
519+
return clusterAdmin().prepareDeleteSnapshot(repoName, snapshotName).execute();
520+
}
521+
522+
protected void updateClusterState(final Function<ClusterState, ClusterState> updater) throws Exception {
523+
final PlainActionFuture<Void> future = PlainActionFuture.newFuture();
524+
final ClusterService clusterService = internalCluster().getCurrentMasterNodeInstance(ClusterService.class);
525+
clusterService.submitStateUpdateTask("test", new ClusterStateUpdateTask() {
526+
@Override
527+
public ClusterState execute(ClusterState currentState) {
528+
return updater.apply(currentState);
529+
}
530+
531+
@Override
532+
public void onFailure(String source, Exception e) {
533+
future.onFailure(e);
534+
}
535+
536+
@Override
537+
public void clusterStateProcessed(String source, ClusterState oldState, ClusterState newState) {
538+
future.onResponse(null);
539+
}
540+
});
541+
future.get();
542+
}
543+
544+
protected SnapshotInfo getSnapshot(String repository, String snapshot) {
545+
final List<SnapshotInfo> snapshotInfos = clusterAdmin().prepareGetSnapshots(repository).setSnapshots(snapshot)
546+
.get().getSnapshots(repository);
547+
assertThat(snapshotInfos, hasSize(1));
548+
return snapshotInfos.get(0);
521549
}
522550
}

‎test/framework/src/main/java/org/elasticsearch/test/ESIntegTestCase.java

+8
Original file line numberDiff line numberDiff line change
@@ -58,6 +58,7 @@
5858
import org.elasticsearch.action.support.IndicesOptions;
5959
import org.elasticsearch.client.AdminClient;
6060
import org.elasticsearch.client.Client;
61+
import org.elasticsearch.client.ClusterAdminClient;
6162
import org.elasticsearch.client.Requests;
6263
import org.elasticsearch.client.RestClient;
6364
import org.elasticsearch.client.RestClientBuilder;
@@ -1329,6 +1330,13 @@ protected AdminClient admin() {
13291330
return client().admin();
13301331
}
13311332

1333+
/**
1334+
* Returns a random cluster admin client. This client can be pointing to any of the nodes in the cluster.
1335+
*/
1336+
protected ClusterAdminClient clusterAdmin() {
1337+
return admin().cluster();
1338+
}
1339+
13321340
/**
13331341
* Convenience method that forwards to {@link #indexRandom(boolean, List)}.
13341342
*/

0 commit comments

Comments
 (0)