Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Retry SLM retention after currently running snapshot completes #45802

Merged
merged 5 commits into from
Aug 22, 2019
Merged

Retry SLM retention after currently running snapshot completes #45802

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
cfb516c
Retry SLM retention after currently running snapshot completes
dakrone Aug 14, 2019
6a231db
Increase timeout waiting for snapshot completion
dakrone Aug 21, 2019
be42c68
Apply patch
dakrone Aug 21, 2019
ef19253
Merge remote-tracking branch 'origin/slm-retention' into slm-retry-on…
dakrone Aug 21, 2019
1f9d5c0
Rename test variables
dakrone Aug 21, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 8 additions & 4 deletions ...g/elasticsearch/action/admin/cluster/snapshots/status/TransportSnapshotsStatusAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
import com.carrotsearch.hppc.cursors.ObjectCursor;
import com.carrotsearch.hppc.cursors.ObjectObjectCursor;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.ActionRunnable;
import org.elasticsearch.action.support.ActionFilters;
import org.elasticsearch.action.support.master.TransportMasterNodeAction;
import org.elasticsearch.client.node.NodeClient;
Expand Down Expand Up @@ -121,10 +122,13 @@ protected void masterOperation(Task task, final SnapshotsStatusRequest request,
new TransportNodesSnapshotsStatus.Request(nodesIds.toArray(new String[nodesIds.size()]))
.snapshots(snapshots).timeout(request.masterNodeTimeout());
client.executeLocally(TransportNodesSnapshotsStatus.TYPE, nodesRequest,
ActionListener.map(
listener, nodeSnapshotStatuses ->
buildResponse(request, snapshotsService.currentSnapshots(request.repository(), Arrays.asList(request.snapshots())),
nodeSnapshotStatuses)));
ActionListener.wrap(
nodeSnapshotStatuses ->
threadPool.executor(ThreadPool.Names.GENERIC).execute(
ActionRunnable.wrap(listener, l -> l.onResponse(
buildResponse(
request, snapshotsService.currentSnapshots(request.repository(), Arrays.asList(request.snapshots())),
nodeSnapshotStatuses)))), listener::onFailure));
} else {
// We don't have any in-progress shards, just return current stats
listener.onResponse(buildResponse(request, currentSnapshots, null));
Expand Down
214 changes: 162 additions & 52 deletions ...ugin/ilm/qa/multi-node/src/test/java/org/elasticsearch/xpack/slm/SnapshotLifecycleIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -208,38 +208,33 @@ public void testPolicyManualExecution() throws Exception {
assertThat(EntityUtils.toString(badResp.getResponse().getEntity()),
containsString("no such snapshot lifecycle policy [" + policyName + "-bad]"));

Response goodResp = client().performRequest(new Request("PUT", "/_slm/policy/" + policyName + "/_execute"));
final String snapshotName = executePolicy(policyName);

try (XContentParser parser = JsonXContent.jsonXContent.createParser(NamedXContentRegistry.EMPTY,
DeprecationHandler.THROW_UNSUPPORTED_OPERATION, EntityUtils.toByteArray(goodResp.getEntity()))) {
final String snapshotName = parser.mapStrings().get("snapshot_name");

// Check that the executed snapshot is created
assertBusy(() -> {
try {
Response response = client().performRequest(new Request("GET", "/_snapshot/" + repoId + "/" + snapshotName));
Map<String, Object> snapshotResponseMap;
try (InputStream is = response.getEntity().getContent()) {
snapshotResponseMap = XContentHelper.convertToMap(XContentType.JSON.xContent(), is, true);
}
assertThat(snapshotResponseMap.size(), greaterThan(0));
final Map<String, Object> metadata = extractMetadata(snapshotResponseMap, snapshotName);
assertNotNull(metadata);
assertThat(metadata.get("policy"), equalTo(policyName));
assertHistoryIsPresent(policyName, true, repoId, CREATE_OPERATION);
} catch (ResponseException e) {
fail("expected snapshot to exist but it does not: " + EntityUtils.toString(e.getResponse().getEntity()));
// Check that the executed snapshot is created
assertBusy(() -> {
try {
Response response = client().performRequest(new Request("GET", "/_snapshot/" + repoId + "/" + snapshotName));
Map<String, Object> snapshotResponseMap;
try (InputStream is = response.getEntity().getContent()) {
snapshotResponseMap = XContentHelper.convertToMap(XContentType.JSON.xContent(), is, true);
}
assertThat(snapshotResponseMap.size(), greaterThan(0));
final Map<String, Object> metadata = extractMetadata(snapshotResponseMap, snapshotName);
assertNotNull(metadata);
assertThat(metadata.get("policy"), equalTo(policyName));
assertHistoryIsPresent(policyName, true, repoId, CREATE_OPERATION);
} catch (ResponseException e) {
fail("expected snapshot to exist but it does not: " + EntityUtils.toString(e.getResponse().getEntity()));
}

Map<String, Object> stats = getSLMStats();
Map<String, Object> policyStats = (Map<String, Object>) stats.get(SnapshotLifecycleStats.POLICY_STATS.getPreferredName());
Map<String, Object> policyIdStats = (Map<String, Object>) policyStats.get(policyName);
int snapsTaken = (int) policyIdStats.get(SnapshotLifecycleStats.SnapshotPolicyStats.SNAPSHOTS_TAKEN.getPreferredName());
int totalTaken = (int) stats.get(SnapshotLifecycleStats.TOTAL_TAKEN.getPreferredName());
assertThat(snapsTaken, equalTo(1));
assertThat(totalTaken, equalTo(1));
});
}
Map<String, Object> stats = getSLMStats();
Map<String, Object> policyStats = (Map<String, Object>) stats.get(SnapshotLifecycleStats.POLICY_STATS.getPreferredName());
Map<String, Object> policyIdStats = (Map<String, Object>) policyStats.get(policyName);
int snapsTaken = (int) policyIdStats.get(SnapshotLifecycleStats.SnapshotPolicyStats.SNAPSHOTS_TAKEN.getPreferredName());
int totalTaken = (int) stats.get(SnapshotLifecycleStats.TOTAL_TAKEN.getPreferredName());
assertThat(snapsTaken, equalTo(1));
assertThat(totalTaken, equalTo(1));
});
}

@SuppressWarnings("unchecked")
Expand All @@ -261,31 +256,25 @@ public void testBasicTimeBasedRetenion() throws Exception {
new SnapshotRetentionConfiguration(TimeValue.timeValueMillis(1), null, null));

// Manually create a snapshot
Response executeResp = client().performRequest(new Request("PUT", "/_slm/policy/" + policyName + "/_execute"));
final String snapshotName = executePolicy(policyName);

final String snapshotName;
try (XContentParser parser = JsonXContent.jsonXContent.createParser(NamedXContentRegistry.EMPTY,
DeprecationHandler.THROW_UNSUPPORTED_OPERATION, EntityUtils.toByteArray(executeResp.getEntity()))) {
snapshotName = parser.mapStrings().get("snapshot_name");

// Check that the executed snapshot is created
assertBusy(() -> {
try {
Response response = client().performRequest(new Request("GET", "/_snapshot/" + repoId + "/" + snapshotName));
Map<String, Object> snapshotResponseMap;
try (InputStream is = response.getEntity().getContent()) {
snapshotResponseMap = XContentHelper.convertToMap(XContentType.JSON.xContent(), is, true);
}
assertThat(snapshotResponseMap.size(), greaterThan(0));
final Map<String, Object> metadata = extractMetadata(snapshotResponseMap, snapshotName);
assertNotNull(metadata);
assertThat(metadata.get("policy"), equalTo(policyName));
assertHistoryIsPresent(policyName, true, repoId, CREATE_OPERATION);
} catch (ResponseException e) {
fail("expected snapshot to exist but it does not: " + EntityUtils.toString(e.getResponse().getEntity()));
// Check that the executed snapshot is created
assertBusy(() -> {
try {
Response response = client().performRequest(new Request("GET", "/_snapshot/" + repoId + "/" + snapshotName));
Map<String, Object> snapshotResponseMap;
try (InputStream is = response.getEntity().getContent()) {
snapshotResponseMap = XContentHelper.convertToMap(XContentType.JSON.xContent(), is, true);
}
});
}
assertThat(snapshotResponseMap.size(), greaterThan(0));
final Map<String, Object> metadata = extractMetadata(snapshotResponseMap, snapshotName);
assertNotNull(metadata);
assertThat(metadata.get("policy"), equalTo(policyName));
assertHistoryIsPresent(policyName, true, repoId, CREATE_OPERATION);
} catch (ResponseException e) {
fail("expected snapshot to exist but it does not: " + EntityUtils.toString(e.getResponse().getEntity()));
}
});

// Run retention every second
ClusterUpdateSettingsRequest req = new ClusterUpdateSettingsRequest();
Expand Down Expand Up @@ -391,6 +380,127 @@ public void testSnapshotInProgress() throws Exception {
}
}

@SuppressWarnings("unchecked")
public void testRetentionWhileSnapshotInProgress() throws Exception {
final String indexName = "test";
final String slowPolicy = "slow";
final String fastPolicy = "fast";
final String slowRepo = "slow-repo";
final String fastRepo = "fast-repo";
int docCount = 20;
for (int i = 0; i < docCount; i++) {
index(client(), indexName, "" + i, "foo", "bar");
}

// Create snapshot repos, one fast and one slow
initializeRepo(slowRepo, "1b");
initializeRepo(fastRepo, "10mb");

createSnapshotPolicy(slowPolicy, "snap", "1 2 3 4 5 ?", slowRepo, indexName, true,
new SnapshotRetentionConfiguration(TimeValue.timeValueSeconds(0), null, null));
createSnapshotPolicy(fastPolicy, "snap", "1 2 3 4 5 ?", fastRepo, indexName, true,
new SnapshotRetentionConfiguration(TimeValue.timeValueSeconds(0), null, null));

// Create a snapshot and wait for it to be complete (need something that can be deleted)
final String completedSnapshotName = executePolicy(fastPolicy);
assertBusy(() -> {
try {
Response getResp = client().performRequest(new Request("GET",
"/_snapshot/" + fastRepo + "/" + completedSnapshotName + "/_status"));
try (InputStream content = getResp.getEntity().getContent()) {
Map<String, Object> snaps = XContentHelper.convertToMap(XContentType.JSON.xContent(), content, true);
logger.info("--> waiting for snapshot {} to be successful, got: {}", completedSnapshotName, snaps);
List<Map<String, Object>> snaps2 = (List<Map<String, Object>>) snaps.get("snapshots");
assertThat(snaps2.get(0).get("state"), equalTo("SUCCESS"));
}
} catch (NullPointerException | ResponseException e) {
fail("unable to retrieve completed snapshot: " + e);
}
}, 60, TimeUnit.SECONDS);

// Take another snapshot
final String slowSnapshotName = executePolicy(slowPolicy);

// Check that the executed snapshot shows up in the SLM output as in_progress
assertBusy(() -> {
try {
Response response = client().performRequest(new Request("GET", "/_slm/policy" + (randomBoolean() ? "" : "?human")));
Map<String, Object> policyResponseMap;
try (InputStream content = response.getEntity().getContent()) {
policyResponseMap = XContentHelper.convertToMap(XContentType.JSON.xContent(), content, true);
}
assertThat(policyResponseMap.size(), greaterThan(0));
Optional<Map<String, Object>> inProgress = Optional.ofNullable((Map<String, Object>) policyResponseMap.get(slowPolicy))
.map(policy -> (Map<String, Object>) policy.get("in_progress"));

if (inProgress.isPresent()) {
Map<String, Object> inProgressMap = inProgress.get();
assertThat(inProgressMap.get("name"), equalTo(slowSnapshotName));
assertNotNull(inProgressMap.get("uuid"));
assertThat(inProgressMap.get("state"), equalTo("STARTED"));
assertThat((long) inProgressMap.get("start_time_millis"), greaterThan(0L));
assertNull(inProgressMap.get("failure"));
} else {
fail("expected in_progress to contain a running snapshot, but the response was " + policyResponseMap);
}
} catch (ResponseException e) {
fail("expected policy to exist but it does not: " + EntityUtils.toString(e.getResponse().getEntity()));
}
});

// Run retention every second
ClusterUpdateSettingsRequest req = new ClusterUpdateSettingsRequest();
req.transientSettings(Settings.builder().put(LifecycleSettings.SLM_RETENTION_SCHEDULE, "*/1 * * * * ?"));
try (XContentBuilder builder = jsonBuilder()) {
req.toXContent(builder, ToXContent.EMPTY_PARAMS);
Request r = new Request("PUT", "/_cluster/settings");
r.setJsonEntity(Strings.toString(builder));
client().performRequest(r);
}

// Cancel the snapshot since it is not going to complete quickly, do it in a thread because
// cancelling the snapshot can take a long time and we might as well check retention while
// its deleting
Thread t = new Thread(() -> {
try {
assertOK(client().performRequest(new Request("DELETE", "/_snapshot/" + slowRepo + "/" + slowSnapshotName)));
} catch (IOException e) {
fail("should not have thrown " + e);
}
});
t.start();

// Check that the snapshot created by the policy has been removed by retention
assertBusy(() -> {
// We expect a failed response because the snapshot should not exist
try {
logger.info("--> checking to see if snapshot has been deleted...");
Response response = client().performRequest(new Request("GET", "/_snapshot/" + slowRepo + "/" + completedSnapshotName));
assertThat(EntityUtils.toString(response.getEntity()), containsString("snapshot_missing_exception"));
} catch (ResponseException e) {
assertThat(EntityUtils.toString(e.getResponse().getEntity()), containsString("snapshot_missing_exception"));
}
}, 60, TimeUnit.SECONDS);

t.join(5000);
}

/**
* Execute the given policy and return the generated snapshot name
*/
private String executePolicy(String policyId) {
try {
Response executeRepsonse = client().performRequest(new Request("PUT", "/_slm/policy/" + policyId + "/_execute"));
try (XContentParser parser = JsonXContent.jsonXContent.createParser(NamedXContentRegistry.EMPTY,
DeprecationHandler.THROW_UNSUPPORTED_OPERATION, EntityUtils.toByteArray(executeRepsonse.getEntity()))) {
return parser.mapStrings().get("snapshot_name");
}
} catch (Exception e) {
fail("failed to execute policy " + policyId + " - got: " + e);
throw new RuntimeException(e);
}
}

@SuppressWarnings("unchecked")
private static Map<String, Object> extractMetadata(Map<String, Object> snapshotResponseMap, String snapshotPrefix) {
List<Map<String, Object>> snapResponse = ((List<Map<String, Object>>) snapshotResponseMap.get("responses")).stream()
Expand Down
2 changes: 1 addition & 1 deletion x-pack/plugin/ilm/src/main/java/org/elasticsearch/xpack/ilm/IndexLifecycle.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -161,7 +161,7 @@ public Collection<Object> createComponents(Client client, ClusterService cluster
snapshotLifecycleService.set(new SnapshotLifecycleService(settings,
() -> new SnapshotLifecycleTask(client, clusterService, snapshotHistoryStore.get()), clusterService, getClock()));
snapshotRetentionService.set(new SnapshotRetentionService(settings,
() -> new SnapshotRetentionTask(client, clusterService, System::nanoTime, snapshotHistoryStore.get()),
() -> new SnapshotRetentionTask(client, clusterService, System::nanoTime, snapshotHistoryStore.get(), threadPool),
clusterService, getClock()));
return Arrays.asList(indexLifecycleInitialisationService.get(), snapshotLifecycleService.get(), snapshotHistoryStore.get(),
snapshotRetentionService.get());
Expand Down
Loading