Ensure ILM policies run safely on leader indices (#38140)
Adds a Step to the Shrink and Delete actions which prevents those actions from running on a leader index - all follower indices must first unfollow the leader index before these actions can run. This prevents the loss of history before follower indices are ready, which might otherwise result in the loss of data.
This commit is contained in:
Gordon Brown
GitHub
parent
8bee5b8e06
commit
7a1e89c7ed
|
|
@ -15,7 +15,7 @@ import org.elasticsearch.common.xcontent.XContentParser;
|
|||
import org.elasticsearch.xpack.core.indexlifecycle.Step.StepKey;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.Collections;
|
||||
import java.util.Arrays;
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
|
|
@ -59,13 +59,20 @@ public class DeleteAction implements LifecycleAction {
|
|||
|
||||
@Override
|
||||
public List<Step> toSteps(Client client, String phase, Step.StepKey nextStepKey) {
|
||||
Step.StepKey waitForNoFollowerStepKey = new Step.StepKey(phase, NAME, WaitForNoFollowersStep.NAME);
|
||||
Step.StepKey deleteStepKey = new Step.StepKey(phase, NAME, DeleteStep.NAME);
|
||||
return Collections.singletonList(new DeleteStep(deleteStepKey, nextStepKey, client));
|
||||
|
||||
WaitForNoFollowersStep waitForNoFollowersStep = new WaitForNoFollowersStep(waitForNoFollowerStepKey, deleteStepKey, client);
|
||||
DeleteStep deleteStep = new DeleteStep(deleteStepKey, nextStepKey, client);
|
||||
return Arrays.asList(waitForNoFollowersStep, deleteStep);
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<StepKey> toStepKeys(String phase) {
|
||||
return Collections.singletonList(new Step.StepKey(phase, NAME, DeleteStep.NAME));
|
||||
Step.StepKey waitForNoFollowerStepKey = new Step.StepKey(phase, NAME, WaitForNoFollowersStep.NAME);
|
||||
Step.StepKey deleteStepKey = new Step.StepKey(phase, NAME, DeleteStep.NAME);
|
||||
|
||||
return Arrays.asList(waitForNoFollowerStepKey, deleteStepKey);
|
||||
}
|
||||
|
||||
@Override
|
||||
|
|
|
|||
|
|
@ -86,6 +86,7 @@ public class ShrinkAction implements LifecycleAction {
|
|||
Settings readOnlySettings = Settings.builder().put(IndexMetaData.SETTING_BLOCKS_WRITE, true).build();
|
||||
|
||||
StepKey branchingKey = new StepKey(phase, NAME, BranchingStep.NAME);
|
||||
StepKey waitForNoFollowerStepKey = new StepKey(phase, NAME, WaitForNoFollowersStep.NAME);
|
||||
StepKey readOnlyKey = new StepKey(phase, NAME, ReadOnlyAction.NAME);
|
||||
StepKey setSingleNodeKey = new StepKey(phase, NAME, SetSingleNodeAllocateStep.NAME);
|
||||
StepKey allocationRoutedKey = new StepKey(phase, NAME, CheckShrinkReadyStep.NAME);
|
||||
|
|
@ -95,8 +96,9 @@ public class ShrinkAction implements LifecycleAction {
|
|||
StepKey aliasKey = new StepKey(phase, NAME, ShrinkSetAliasStep.NAME);
|
||||
StepKey isShrunkIndexKey = new StepKey(phase, NAME, ShrunkenIndexCheckStep.NAME);
|
||||
|
||||
BranchingStep conditionalSkipShrinkStep = new BranchingStep(branchingKey, readOnlyKey, nextStepKey,
|
||||
BranchingStep conditionalSkipShrinkStep = new BranchingStep(branchingKey, waitForNoFollowerStepKey, nextStepKey,
|
||||
(index, clusterState) -> clusterState.getMetaData().index(index).getNumberOfShards() == numberOfShards);
|
||||
WaitForNoFollowersStep waitForNoFollowersStep = new WaitForNoFollowersStep(waitForNoFollowerStepKey, readOnlyKey, client);
|
||||
UpdateSettingsStep readOnlyStep = new UpdateSettingsStep(readOnlyKey, setSingleNodeKey, client, readOnlySettings);
|
||||
SetSingleNodeAllocateStep setSingleNodeStep = new SetSingleNodeAllocateStep(setSingleNodeKey, allocationRoutedKey, client);
|
||||
CheckShrinkReadyStep checkShrinkReadyStep = new CheckShrinkReadyStep(allocationRoutedKey, shrinkKey);
|
||||
|
|
@ -105,13 +107,14 @@ public class ShrinkAction implements LifecycleAction {
|
|||
CopyExecutionStateStep copyMetadata = new CopyExecutionStateStep(copyMetadataKey, aliasKey, SHRUNKEN_INDEX_PREFIX);
|
||||
ShrinkSetAliasStep aliasSwapAndDelete = new ShrinkSetAliasStep(aliasKey, isShrunkIndexKey, client, SHRUNKEN_INDEX_PREFIX);
|
||||
ShrunkenIndexCheckStep waitOnShrinkTakeover = new ShrunkenIndexCheckStep(isShrunkIndexKey, nextStepKey, SHRUNKEN_INDEX_PREFIX);
|
||||
return Arrays.asList(conditionalSkipShrinkStep, readOnlyStep, setSingleNodeStep, checkShrinkReadyStep, shrink, allocated,
|
||||
copyMetadata, aliasSwapAndDelete, waitOnShrinkTakeover);
|
||||
return Arrays.asList(conditionalSkipShrinkStep, waitForNoFollowersStep, readOnlyStep, setSingleNodeStep, checkShrinkReadyStep,
|
||||
shrink, allocated, copyMetadata, aliasSwapAndDelete, waitOnShrinkTakeover);
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<StepKey> toStepKeys(String phase) {
|
||||
StepKey conditionalSkipKey = new StepKey(phase, NAME, BranchingStep.NAME);
|
||||
StepKey waitForNoFollowerStepKey = new StepKey(phase, NAME, WaitForNoFollowersStep.NAME);
|
||||
StepKey readOnlyKey = new StepKey(phase, NAME, ReadOnlyAction.NAME);
|
||||
StepKey setSingleNodeKey = new StepKey(phase, NAME, SetSingleNodeAllocateStep.NAME);
|
||||
StepKey checkShrinkReadyKey = new StepKey(phase, NAME, CheckShrinkReadyStep.NAME);
|
||||
|
|
@ -120,8 +123,8 @@ public class ShrinkAction implements LifecycleAction {
|
|||
StepKey copyMetadataKey = new StepKey(phase, NAME, CopyExecutionStateStep.NAME);
|
||||
StepKey aliasKey = new StepKey(phase, NAME, ShrinkSetAliasStep.NAME);
|
||||
StepKey isShrunkIndexKey = new StepKey(phase, NAME, ShrunkenIndexCheckStep.NAME);
|
||||
return Arrays.asList(conditionalSkipKey, readOnlyKey, setSingleNodeKey, checkShrinkReadyKey, shrinkKey, enoughShardsKey,
|
||||
copyMetadataKey, aliasKey, isShrunkIndexKey);
|
||||
return Arrays.asList(conditionalSkipKey, waitForNoFollowerStepKey, readOnlyKey, setSingleNodeKey, checkShrinkReadyKey, shrinkKey,
|
||||
enoughShardsKey, copyMetadataKey, aliasKey, isShrunkIndexKey);
|
||||
}
|
||||
|
||||
@Override
|
||||
|
|
|
|||
|
|
@ -0,0 +1,108 @@
|
|||
/*
|
||||
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
|
||||
* or more contributor license agreements. Licensed under the Elastic License;
|
||||
* you may not use this file except in compliance with the Elastic License.
|
||||
*/
|
||||
|
||||
package org.elasticsearch.xpack.core.indexlifecycle;
|
||||
|
||||
import org.apache.logging.log4j.LogManager;
|
||||
import org.apache.logging.log4j.Logger;
|
||||
import org.elasticsearch.action.ActionListener;
|
||||
import org.elasticsearch.action.admin.indices.stats.IndexStats;
|
||||
import org.elasticsearch.action.admin.indices.stats.IndicesStatsRequest;
|
||||
import org.elasticsearch.action.admin.indices.stats.ShardStats;
|
||||
import org.elasticsearch.client.Client;
|
||||
import org.elasticsearch.cluster.metadata.IndexMetaData;
|
||||
import org.elasticsearch.common.ParseField;
|
||||
import org.elasticsearch.common.xcontent.ToXContentObject;
|
||||
import org.elasticsearch.common.xcontent.XContentBuilder;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.Arrays;
|
||||
import java.util.Objects;
|
||||
|
||||
/**
|
||||
* A step that waits until the index it's used on is no longer a leader index.
|
||||
* This is necessary as there are some actions which are not safe to perform on
|
||||
* a leader index, such as those which delete the index, including Shrink and
|
||||
* Delete.
|
||||
*/
|
||||
public class WaitForNoFollowersStep extends AsyncWaitStep {
|
||||
|
||||
private static final Logger logger = LogManager.getLogger(WaitForNoFollowersStep.class);
|
||||
|
||||
static final String NAME = "wait-for-shard-history-leases";
|
||||
static final String CCR_LEASE_KEY = "ccr";
|
||||
|
||||
WaitForNoFollowersStep(StepKey key, StepKey nextStepKey, Client client) {
|
||||
super(key, nextStepKey, client);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void evaluateCondition(IndexMetaData indexMetaData, Listener listener) {
|
||||
IndicesStatsRequest request = new IndicesStatsRequest();
|
||||
request.clear();
|
||||
String indexName = indexMetaData.getIndex().getName();
|
||||
request.indices(indexName);
|
||||
getClient().admin().indices().stats(request, ActionListener.wrap((response) -> {
|
||||
IndexStats indexStats = response.getIndex(indexName);
|
||||
if (indexStats == null) {
|
||||
// Index was probably deleted
|
||||
logger.debug("got null shard stats for index {}, proceeding on the assumption it has been deleted",
|
||||
indexMetaData.getIndex());
|
||||
listener.onResponse(true, null);
|
||||
return;
|
||||
}
|
||||
|
||||
boolean isCurrentlyLeaderIndex = Arrays.stream(indexStats.getShards())
|
||||
.map(ShardStats::getRetentionLeaseStats)
|
||||
.flatMap(retentionLeaseStats -> retentionLeaseStats.retentionLeases().leases().stream())
|
||||
.anyMatch(lease -> CCR_LEASE_KEY.equals(lease.source()));
|
||||
|
||||
if (isCurrentlyLeaderIndex) {
|
||||
listener.onResponse(false, new Info());
|
||||
} else {
|
||||
listener.onResponse(true, null);
|
||||
}
|
||||
}, listener::onFailure));
|
||||
}
|
||||
|
||||
static final class Info implements ToXContentObject {
|
||||
|
||||
static final ParseField MESSAGE_FIELD = new ParseField("message");
|
||||
|
||||
private static final String message = "this index is a leader index; waiting for all following indices to cease " +
|
||||
"following before proceeding";
|
||||
|
||||
Info() { }
|
||||
|
||||
String getMessage() {
|
||||
return message;
|
||||
}
|
||||
|
||||
@Override
|
||||
public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
|
||||
builder.startObject();
|
||||
builder.field(MESSAGE_FIELD.getPreferredName(), message);
|
||||
builder.endObject();
|
||||
return builder;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean equals(Object o) {
|
||||
if (this == o) {
|
||||
return true;
|
||||
}
|
||||
if (o == null || getClass() != o.getClass()) {
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
@Override
|
||||
public int hashCode() {
|
||||
return Objects.hash(getMessage());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -36,10 +36,14 @@ public class DeleteActionTests extends AbstractActionTestCase<DeleteAction> {
|
|||
randomAlphaOfLengthBetween(1, 10));
|
||||
List<Step> steps = action.toSteps(null, phase, nextStepKey);
|
||||
assertNotNull(steps);
|
||||
assertEquals(1, steps.size());
|
||||
StepKey expectedFirstStepKey = new StepKey(phase, DeleteAction.NAME, DeleteStep.NAME);
|
||||
DeleteStep firstStep = (DeleteStep) steps.get(0);
|
||||
assertEquals(2, steps.size());
|
||||
StepKey expectedFirstStepKey = new StepKey(phase, DeleteAction.NAME, WaitForNoFollowersStep.NAME);
|
||||
StepKey expectedSecondStepKey = new StepKey(phase, DeleteAction.NAME, DeleteStep.NAME);
|
||||
WaitForNoFollowersStep firstStep = (WaitForNoFollowersStep) steps.get(0);
|
||||
DeleteStep secondStep = (DeleteStep) steps.get(1);
|
||||
assertEquals(expectedFirstStepKey, firstStep.getKey());
|
||||
assertEquals(nextStepKey, firstStep.getNextStepKey());
|
||||
assertEquals(expectedSecondStepKey, firstStep.getNextStepKey());
|
||||
assertEquals(expectedSecondStepKey, secondStep.getKey());
|
||||
assertEquals(nextStepKey, secondStep.getNextStepKey());
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -126,16 +126,17 @@ public class ShrinkActionTests extends AbstractActionTestCase<ShrinkAction> {
|
|||
StepKey nextStepKey = new StepKey(randomAlphaOfLengthBetween(1, 10), randomAlphaOfLengthBetween(1, 10),
|
||||
randomAlphaOfLengthBetween(1, 10));
|
||||
List<Step> steps = action.toSteps(null, phase, nextStepKey);
|
||||
assertThat(steps.size(), equalTo(9));
|
||||
assertThat(steps.size(), equalTo(10));
|
||||
StepKey expectedFirstKey = new StepKey(phase, ShrinkAction.NAME, BranchingStep.NAME);
|
||||
StepKey expectedSecondKey = new StepKey(phase, ShrinkAction.NAME, ReadOnlyAction.NAME);
|
||||
StepKey expectedThirdKey = new StepKey(phase, ShrinkAction.NAME, SetSingleNodeAllocateStep.NAME);
|
||||
StepKey expectedFourthKey = new StepKey(phase, ShrinkAction.NAME, CheckShrinkReadyStep.NAME);
|
||||
StepKey expectedFifthKey = new StepKey(phase, ShrinkAction.NAME, ShrinkStep.NAME);
|
||||
StepKey expectedSixthKey = new StepKey(phase, ShrinkAction.NAME, ShrunkShardsAllocatedStep.NAME);
|
||||
StepKey expectedSeventhKey = new StepKey(phase, ShrinkAction.NAME, CopyExecutionStateStep.NAME);
|
||||
StepKey expectedEighthKey = new StepKey(phase, ShrinkAction.NAME, ShrinkSetAliasStep.NAME);
|
||||
StepKey expectedNinthKey = new StepKey(phase, ShrinkAction.NAME, ShrunkenIndexCheckStep.NAME);
|
||||
StepKey expectedSecondKey = new StepKey(phase, ShrinkAction.NAME, WaitForNoFollowersStep.NAME);
|
||||
StepKey expectedThirdKey = new StepKey(phase, ShrinkAction.NAME, ReadOnlyAction.NAME);
|
||||
StepKey expectedFourthKey = new StepKey(phase, ShrinkAction.NAME, SetSingleNodeAllocateStep.NAME);
|
||||
StepKey expectedFifthKey = new StepKey(phase, ShrinkAction.NAME, CheckShrinkReadyStep.NAME);
|
||||
StepKey expectedSixthKey = new StepKey(phase, ShrinkAction.NAME, ShrinkStep.NAME);
|
||||
StepKey expectedSeventhKey = new StepKey(phase, ShrinkAction.NAME, ShrunkShardsAllocatedStep.NAME);
|
||||
StepKey expectedEighthKey = new StepKey(phase, ShrinkAction.NAME, CopyExecutionStateStep.NAME);
|
||||
StepKey expectedNinthKey = new StepKey(phase, ShrinkAction.NAME, ShrinkSetAliasStep.NAME);
|
||||
StepKey expectedTenthKey = new StepKey(phase, ShrinkAction.NAME, ShrunkenIndexCheckStep.NAME);
|
||||
|
||||
assertTrue(steps.get(0) instanceof BranchingStep);
|
||||
assertThat(steps.get(0).getKey(), equalTo(expectedFirstKey));
|
||||
|
|
@ -143,43 +144,47 @@ public class ShrinkActionTests extends AbstractActionTestCase<ShrinkAction> {
|
|||
assertThat(((BranchingStep) steps.get(0)).getNextStepKeyOnFalse(), equalTo(expectedSecondKey));
|
||||
assertThat(((BranchingStep) steps.get(0)).getNextStepKeyOnTrue(), equalTo(nextStepKey));
|
||||
|
||||
assertTrue(steps.get(1) instanceof UpdateSettingsStep);
|
||||
assertTrue(steps.get(1) instanceof WaitForNoFollowersStep);
|
||||
assertThat(steps.get(1).getKey(), equalTo(expectedSecondKey));
|
||||
assertThat(steps.get(1).getNextStepKey(), equalTo(expectedThirdKey));
|
||||
assertTrue(IndexMetaData.INDEX_BLOCKS_WRITE_SETTING.get(((UpdateSettingsStep)steps.get(1)).getSettings()));
|
||||
|
||||
assertTrue(steps.get(2) instanceof SetSingleNodeAllocateStep);
|
||||
assertTrue(steps.get(2) instanceof UpdateSettingsStep);
|
||||
assertThat(steps.get(2).getKey(), equalTo(expectedThirdKey));
|
||||
assertThat(steps.get(2).getNextStepKey(), equalTo(expectedFourthKey));
|
||||
assertTrue(IndexMetaData.INDEX_BLOCKS_WRITE_SETTING.get(((UpdateSettingsStep)steps.get(2)).getSettings()));
|
||||
|
||||
assertTrue(steps.get(3) instanceof CheckShrinkReadyStep);
|
||||
assertTrue(steps.get(3) instanceof SetSingleNodeAllocateStep);
|
||||
assertThat(steps.get(3).getKey(), equalTo(expectedFourthKey));
|
||||
assertThat(steps.get(3).getNextStepKey(), equalTo(expectedFifthKey));
|
||||
|
||||
assertTrue(steps.get(4) instanceof ShrinkStep);
|
||||
assertTrue(steps.get(4) instanceof CheckShrinkReadyStep);
|
||||
assertThat(steps.get(4).getKey(), equalTo(expectedFifthKey));
|
||||
assertThat(steps.get(4).getNextStepKey(), equalTo(expectedSixthKey));
|
||||
assertThat(((ShrinkStep) steps.get(4)).getShrunkIndexPrefix(), equalTo(ShrinkAction.SHRUNKEN_INDEX_PREFIX));
|
||||
|
||||
assertTrue(steps.get(5) instanceof ShrunkShardsAllocatedStep);
|
||||
assertTrue(steps.get(5) instanceof ShrinkStep);
|
||||
assertThat(steps.get(5).getKey(), equalTo(expectedSixthKey));
|
||||
assertThat(steps.get(5).getNextStepKey(), equalTo(expectedSeventhKey));
|
||||
assertThat(((ShrunkShardsAllocatedStep) steps.get(5)).getShrunkIndexPrefix(), equalTo(ShrinkAction.SHRUNKEN_INDEX_PREFIX));
|
||||
assertThat(((ShrinkStep) steps.get(5)).getShrunkIndexPrefix(), equalTo(ShrinkAction.SHRUNKEN_INDEX_PREFIX));
|
||||
|
||||
assertTrue(steps.get(6) instanceof CopyExecutionStateStep);
|
||||
assertTrue(steps.get(6) instanceof ShrunkShardsAllocatedStep);
|
||||
assertThat(steps.get(6).getKey(), equalTo(expectedSeventhKey));
|
||||
assertThat(steps.get(6).getNextStepKey(), equalTo(expectedEighthKey));
|
||||
assertThat(((CopyExecutionStateStep) steps.get(6)).getShrunkIndexPrefix(), equalTo(ShrinkAction.SHRUNKEN_INDEX_PREFIX));
|
||||
assertThat(((ShrunkShardsAllocatedStep) steps.get(6)).getShrunkIndexPrefix(), equalTo(ShrinkAction.SHRUNKEN_INDEX_PREFIX));
|
||||
|
||||
assertTrue(steps.get(7) instanceof ShrinkSetAliasStep);
|
||||
assertTrue(steps.get(7) instanceof CopyExecutionStateStep);
|
||||
assertThat(steps.get(7).getKey(), equalTo(expectedEighthKey));
|
||||
assertThat(steps.get(7).getNextStepKey(), equalTo(expectedNinthKey));
|
||||
assertThat(((ShrinkSetAliasStep) steps.get(7)).getShrunkIndexPrefix(), equalTo(ShrinkAction.SHRUNKEN_INDEX_PREFIX));
|
||||
assertThat(((CopyExecutionStateStep) steps.get(7)).getShrunkIndexPrefix(), equalTo(ShrinkAction.SHRUNKEN_INDEX_PREFIX));
|
||||
|
||||
assertTrue(steps.get(8) instanceof ShrunkenIndexCheckStep);
|
||||
assertTrue(steps.get(8) instanceof ShrinkSetAliasStep);
|
||||
assertThat(steps.get(8).getKey(), equalTo(expectedNinthKey));
|
||||
assertThat(steps.get(8).getNextStepKey(), equalTo(nextStepKey));
|
||||
assertThat(((ShrunkenIndexCheckStep) steps.get(8)).getShrunkIndexPrefix(), equalTo(ShrinkAction.SHRUNKEN_INDEX_PREFIX));
|
||||
assertThat(steps.get(8).getNextStepKey(), equalTo(expectedTenthKey));
|
||||
assertThat(((ShrinkSetAliasStep) steps.get(8)).getShrunkIndexPrefix(), equalTo(ShrinkAction.SHRUNKEN_INDEX_PREFIX));
|
||||
|
||||
assertTrue(steps.get(9) instanceof ShrunkenIndexCheckStep);
|
||||
assertThat(steps.get(9).getKey(), equalTo(expectedTenthKey));
|
||||
assertThat(steps.get(9).getNextStepKey(), equalTo(nextStepKey));
|
||||
assertThat(((ShrunkenIndexCheckStep) steps.get(9)).getShrunkIndexPrefix(), equalTo(ShrinkAction.SHRUNKEN_INDEX_PREFIX));
|
||||
}
|
||||
|
||||
@Override
|
||||
|
|
|
|||
|
|
@ -0,0 +1,254 @@
|
|||
/*
|
||||
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
|
||||
* or more contributor license agreements. Licensed under the Elastic License;
|
||||
* you may not use this file except in compliance with the Elastic License.
|
||||
*/
|
||||
|
||||
package org.elasticsearch.xpack.core.indexlifecycle;
|
||||
|
||||
import org.apache.lucene.util.SetOnce;
|
||||
import org.elasticsearch.Version;
|
||||
import org.elasticsearch.action.ActionListener;
|
||||
import org.elasticsearch.action.admin.indices.stats.IndexStats;
|
||||
import org.elasticsearch.action.admin.indices.stats.IndicesStatsRequest;
|
||||
import org.elasticsearch.action.admin.indices.stats.IndicesStatsResponse;
|
||||
import org.elasticsearch.action.admin.indices.stats.ShardStats;
|
||||
import org.elasticsearch.client.AdminClient;
|
||||
import org.elasticsearch.client.Client;
|
||||
import org.elasticsearch.client.IndicesAdminClient;
|
||||
import org.elasticsearch.cluster.metadata.IndexMetaData;
|
||||
import org.elasticsearch.common.Strings;
|
||||
import org.elasticsearch.common.xcontent.ToXContentObject;
|
||||
import org.elasticsearch.index.seqno.RetentionLease;
|
||||
import org.elasticsearch.index.seqno.RetentionLeaseStats;
|
||||
import org.elasticsearch.index.seqno.RetentionLeases;
|
||||
import org.elasticsearch.index.shard.ShardId;
|
||||
import org.elasticsearch.index.shard.ShardPath;
|
||||
import org.mockito.Mockito;
|
||||
|
||||
import java.nio.file.Path;
|
||||
import java.util.ArrayList;
|
||||
|
||||
import static org.elasticsearch.xpack.core.indexlifecycle.WaitForNoFollowersStep.CCR_LEASE_KEY;
|
||||
import static org.hamcrest.Matchers.containsString;
|
||||
import static org.hamcrest.Matchers.equalTo;
|
||||
import static org.mockito.Matchers.any;
|
||||
import static org.mockito.Mockito.mock;
|
||||
import static org.mockito.Mockito.when;
|
||||
|
||||
public class WaitForNoFollowersStepTests extends AbstractStepTestCase<WaitForNoFollowersStep> {
|
||||
|
||||
|
||||
@Override
|
||||
protected WaitForNoFollowersStep createRandomInstance() {
|
||||
Step.StepKey stepKey = randomStepKey();
|
||||
Step.StepKey nextStepKey = randomStepKey();
|
||||
return new WaitForNoFollowersStep(stepKey, nextStepKey, mock(Client.class));
|
||||
}
|
||||
|
||||
@Override
|
||||
protected WaitForNoFollowersStep mutateInstance(WaitForNoFollowersStep instance) {
|
||||
Step.StepKey key = instance.getKey();
|
||||
Step.StepKey nextKey = instance.getNextStepKey();
|
||||
|
||||
if (randomBoolean()) {
|
||||
key = new Step.StepKey(key.getPhase(), key.getAction(), key.getName() + randomAlphaOfLength(5));
|
||||
} else {
|
||||
nextKey = new Step.StepKey(key.getPhase(), key.getAction(), key.getName() + randomAlphaOfLength(5));
|
||||
}
|
||||
|
||||
return new WaitForNoFollowersStep(key, nextKey, instance.getClient());
|
||||
}
|
||||
|
||||
@Override
|
||||
protected WaitForNoFollowersStep copyInstance(WaitForNoFollowersStep instance) {
|
||||
return new WaitForNoFollowersStep(instance.getKey(), instance.getNextStepKey(), instance.getClient());
|
||||
}
|
||||
|
||||
public void testConditionMet() {
|
||||
WaitForNoFollowersStep step = createRandomInstance();
|
||||
|
||||
String indexName = randomAlphaOfLengthBetween(5,10);
|
||||
|
||||
int numberOfShards = randomIntBetween(1, 100);
|
||||
final IndexMetaData indexMetaData = IndexMetaData.builder(indexName)
|
||||
.settings(settings(Version.CURRENT))
|
||||
.numberOfShards(numberOfShards)
|
||||
.numberOfReplicas(randomIntBetween(1, 10))
|
||||
.build();
|
||||
|
||||
mockIndexStatsCall(step.getClient(), indexName, randomIndexStats(false, numberOfShards));
|
||||
|
||||
final SetOnce<Boolean> conditionMetHolder = new SetOnce<>();
|
||||
final SetOnce<ToXContentObject> stepInfoHolder = new SetOnce<>();
|
||||
step.evaluateCondition(indexMetaData, new AsyncWaitStep.Listener() {
|
||||
@Override
|
||||
public void onResponse(boolean conditionMet, ToXContentObject infomationContext) {
|
||||
conditionMetHolder.set(conditionMet);
|
||||
stepInfoHolder.set(infomationContext);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onFailure(Exception e) {
|
||||
fail("onFailure should not be called in this test, called with exception: " + e.getMessage());
|
||||
}
|
||||
});
|
||||
|
||||
assertTrue(conditionMetHolder.get());
|
||||
assertNull(stepInfoHolder.get());
|
||||
}
|
||||
|
||||
public void testConditionNotMet() {
|
||||
WaitForNoFollowersStep step = createRandomInstance();
|
||||
|
||||
String indexName = randomAlphaOfLengthBetween(5,10);
|
||||
|
||||
int numberOfShards = randomIntBetween(1, 100);
|
||||
final IndexMetaData indexMetaData = IndexMetaData.builder(indexName)
|
||||
.settings(settings(Version.CURRENT))
|
||||
.numberOfShards(numberOfShards)
|
||||
.numberOfReplicas(randomIntBetween(1, 10))
|
||||
.build();
|
||||
|
||||
mockIndexStatsCall(step.getClient(), indexName, randomIndexStats(true, numberOfShards));
|
||||
|
||||
final SetOnce<Boolean> conditionMetHolder = new SetOnce<>();
|
||||
final SetOnce<ToXContentObject> stepInfoHolder = new SetOnce<>();
|
||||
step.evaluateCondition(indexMetaData, new AsyncWaitStep.Listener() {
|
||||
@Override
|
||||
public void onResponse(boolean conditionMet, ToXContentObject infomationContext) {
|
||||
conditionMetHolder.set(conditionMet);
|
||||
stepInfoHolder.set(infomationContext);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onFailure(Exception e) {
|
||||
fail("onFailure should not be called in this test, called with exception: " + e.getMessage());
|
||||
}
|
||||
});
|
||||
|
||||
assertFalse(conditionMetHolder.get());
|
||||
assertThat(Strings.toString(stepInfoHolder.get()),
|
||||
containsString("this index is a leader index; waiting for all following indices to cease following before proceeding"));
|
||||
}
|
||||
|
||||
public void testFailure() {
|
||||
WaitForNoFollowersStep step = createRandomInstance();
|
||||
|
||||
String indexName = randomAlphaOfLengthBetween(5,10);
|
||||
|
||||
int numberOfShards = randomIntBetween(1, 100);
|
||||
IndexMetaData indexMetaData = IndexMetaData.builder(indexName)
|
||||
.settings(settings(Version.CURRENT))
|
||||
.numberOfShards(numberOfShards)
|
||||
.numberOfReplicas(randomIntBetween(1, 10))
|
||||
.build();
|
||||
|
||||
final Exception expectedException = new RuntimeException(randomAlphaOfLength(5));
|
||||
|
||||
Client client = step.getClient();
|
||||
AdminClient adminClient = Mockito.mock(AdminClient.class);
|
||||
IndicesAdminClient indicesClient = Mockito.mock(IndicesAdminClient.class);
|
||||
Mockito.when(client.admin()).thenReturn(adminClient);
|
||||
Mockito.when(adminClient.indices()).thenReturn(indicesClient);
|
||||
Mockito.doAnswer(invocationOnMock -> {
|
||||
@SuppressWarnings("unchecked")
|
||||
ActionListener<IndicesStatsResponse> listener = (ActionListener<IndicesStatsResponse>) invocationOnMock.getArguments()[1];
|
||||
listener.onFailure(expectedException);
|
||||
return null;
|
||||
}).when(indicesClient).stats(any(), any());
|
||||
|
||||
final SetOnce<Exception> exceptionHolder = new SetOnce<>();
|
||||
step.evaluateCondition(indexMetaData, new AsyncWaitStep.Listener() {
|
||||
@Override
|
||||
public void onResponse(boolean conditionMet, ToXContentObject infomationContext) {
|
||||
fail("onResponse should not be called in this test, called with conditionMet: " + conditionMet
|
||||
+ " and stepInfo: " + Strings.toString(infomationContext));
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onFailure(Exception e) {
|
||||
exceptionHolder.set(e);
|
||||
}
|
||||
});
|
||||
|
||||
assertThat(exceptionHolder.get(), equalTo(expectedException));
|
||||
}
|
||||
|
||||
private void mockIndexStatsCall(Client client, String expectedIndexName, IndexStats indexStats) {
|
||||
AdminClient adminClient = Mockito.mock(AdminClient.class);
|
||||
IndicesAdminClient indicesClient = Mockito.mock(IndicesAdminClient.class);
|
||||
Mockito.when(client.admin()).thenReturn(adminClient);
|
||||
Mockito.when(adminClient.indices()).thenReturn(indicesClient);
|
||||
Mockito.doAnswer(invocationOnMock -> {
|
||||
IndicesStatsRequest request = (IndicesStatsRequest) invocationOnMock.getArguments()[0];
|
||||
assertThat(request.indices().length, equalTo(1));
|
||||
assertThat(request.indices()[0], equalTo(expectedIndexName));
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
ActionListener<IndicesStatsResponse> listener = (ActionListener<IndicesStatsResponse>) invocationOnMock.getArguments()[1];
|
||||
|
||||
// Trying to create a real IndicesStatsResponse requires setting up a ShardRouting, so just mock it
|
||||
IndicesStatsResponse response = mock(IndicesStatsResponse.class);
|
||||
when(response.getIndex(expectedIndexName)).thenReturn(indexStats);
|
||||
|
||||
listener.onResponse(response);
|
||||
return null;
|
||||
}).when(indicesClient).stats(any(), any());
|
||||
}
|
||||
|
||||
private IndexStats randomIndexStats(boolean isLeaderIndex, int numOfShards) {
|
||||
ShardStats[] shardStats = new ShardStats[numOfShards];
|
||||
for (int i = 0; i < numOfShards; i++) {
|
||||
shardStats[i] = randomShardStats(isLeaderIndex);
|
||||
}
|
||||
return new IndexStats(randomAlphaOfLength(5), randomAlphaOfLength(10), shardStats);
|
||||
}
|
||||
|
||||
private ShardStats randomShardStats(boolean isLeaderIndex) {
|
||||
return new ShardStats(null,
|
||||
mockShardPath(),
|
||||
null,
|
||||
null,
|
||||
null,
|
||||
randomRetentionLeaseStats(isLeaderIndex)
|
||||
);
|
||||
}
|
||||
|
||||
private RetentionLeaseStats randomRetentionLeaseStats(boolean isLeaderIndex) {
|
||||
int numOfLeases = randomIntBetween(1, 10);
|
||||
|
||||
ArrayList<RetentionLease> leases = new ArrayList<>();
|
||||
for (int i=0; i < numOfLeases; i++) {
|
||||
leases.add(new RetentionLease(randomAlphaOfLength(5), randomNonNegativeLong(), randomNonNegativeLong(),
|
||||
isLeaderIndex ? CCR_LEASE_KEY : randomAlphaOfLength(5)));
|
||||
}
|
||||
return new RetentionLeaseStats(
|
||||
new RetentionLeases(randomLongBetween(1, Long.MAX_VALUE), randomLongBetween(1, Long.MAX_VALUE), leases));
|
||||
}
|
||||
|
||||
private ShardPath mockShardPath() {
|
||||
// Mock paths in a way that pass ShardPath constructor assertions
|
||||
final int shardId = randomIntBetween(0, 10);
|
||||
final Path getFileNameShardId = mock(Path.class);
|
||||
when(getFileNameShardId.toString()).thenReturn(Integer.toString(shardId));
|
||||
|
||||
final String shardUuid = randomAlphaOfLength(5);
|
||||
final Path getFileNameShardUuid = mock(Path.class);
|
||||
when(getFileNameShardUuid.toString()).thenReturn(shardUuid);
|
||||
|
||||
final Path getParent = mock(Path.class);
|
||||
when(getParent.getFileName()).thenReturn(getFileNameShardUuid);
|
||||
|
||||
final Path path = mock(Path.class);
|
||||
when(path.getParent()).thenReturn(getParent);
|
||||
when(path.getFileName()).thenReturn(getFileNameShardId);
|
||||
|
||||
// Mock paths for ShardPath#getRootDataPath()
|
||||
final Path getParentOfParent = mock(Path.class);
|
||||
when(getParent.getParent()).thenReturn(getParentOfParent);
|
||||
when(getParentOfParent.getParent()).thenReturn(mock(Path.class));
|
||||
|
||||
return new ShardPath(false, path, path, new ShardId(randomAlphaOfLength(5), shardUuid, shardId));
|
||||
}
|
||||
}
|
||||
|
|
@ -20,6 +20,7 @@ import org.elasticsearch.common.xcontent.XContentBuilder;
|
|||
import org.elasticsearch.common.xcontent.XContentHelper;
|
||||
import org.elasticsearch.common.xcontent.XContentType;
|
||||
import org.elasticsearch.common.xcontent.json.JsonXContent;
|
||||
import org.elasticsearch.rest.RestStatus;
|
||||
import org.elasticsearch.xpack.ccr.ESCCRRestTestCase;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.LifecycleAction;
|
||||
import org.elasticsearch.xpack.core.indexlifecycle.LifecyclePolicy;
|
||||
|
|
@ -293,50 +294,7 @@ public class CCRIndexLifecycleIT extends ESCCRRestTestCase {
|
|||
ensureGreen(indexName);
|
||||
} else if ("follow".equals(targetCluster)) {
|
||||
// Create a policy with just a Shrink action on the follower
|
||||
final XContentBuilder builder = jsonBuilder();
|
||||
builder.startObject();
|
||||
{
|
||||
builder.startObject("policy");
|
||||
{
|
||||
builder.startObject("phases");
|
||||
{
|
||||
builder.startObject("warm");
|
||||
{
|
||||
builder.startObject("actions");
|
||||
{
|
||||
builder.startObject("shrink");
|
||||
{
|
||||
builder.field("number_of_shards", 1);
|
||||
}
|
||||
builder.endObject();
|
||||
}
|
||||
builder.endObject();
|
||||
}
|
||||
builder.endObject();
|
||||
|
||||
// Sometimes throw in an extraneous unfollow just to check it doesn't break anything
|
||||
if (randomBoolean()) {
|
||||
builder.startObject("cold");
|
||||
{
|
||||
builder.startObject("actions");
|
||||
{
|
||||
builder.startObject("unfollow");
|
||||
builder.endObject();
|
||||
}
|
||||
builder.endObject();
|
||||
}
|
||||
builder.endObject();
|
||||
}
|
||||
}
|
||||
builder.endObject();
|
||||
}
|
||||
builder.endObject();
|
||||
}
|
||||
builder.endObject();
|
||||
|
||||
final Request request = new Request("PUT", "_ilm/policy/" + policyName);
|
||||
request.setJsonEntity(Strings.toString(builder));
|
||||
assertOK(client().performRequest(request));
|
||||
putShrinkOnlyPolicy(client(), policyName);
|
||||
|
||||
// Follow the index
|
||||
followIndex(indexName, indexName);
|
||||
|
|
@ -368,6 +326,73 @@ public class CCRIndexLifecycleIT extends ESCCRRestTestCase {
|
|||
}
|
||||
}
|
||||
|
||||
public void testCannotShrinkLeaderIndex() throws Exception {
|
||||
String indexName = "shrink-leader-test";
|
||||
String shrunkenIndexName = "shrink-" + indexName;
|
||||
|
||||
String policyName = "shrink-leader-test-policy";
|
||||
if ("leader".equals(targetCluster)) {
|
||||
// Set up the policy and index, but don't attach the policy yet,
|
||||
// otherwise it'll proceed through shrink before we can set up the
|
||||
// follower
|
||||
putShrinkOnlyPolicy(client(), policyName);
|
||||
Settings indexSettings = Settings.builder()
|
||||
.put("index.soft_deletes.enabled", true)
|
||||
.put("index.number_of_shards", 2)
|
||||
.put("index.number_of_replicas", 0)
|
||||
.build();
|
||||
createIndex(indexName, indexSettings, "", "");
|
||||
ensureGreen(indexName);
|
||||
} else if ("follow".equals(targetCluster)) {
|
||||
|
||||
try (RestClient leaderClient = buildLeaderClient()) {
|
||||
// Policy with the same name must exist in follower cluster too:
|
||||
putUnfollowOnlyPolicy(client(), policyName);
|
||||
followIndex(indexName, indexName);
|
||||
ensureGreen(indexName);
|
||||
|
||||
// Now we can set up the leader to use the policy
|
||||
Request changePolicyRequest = new Request("PUT", "/" + indexName + "/_settings");
|
||||
final StringEntity changePolicyEntity = new StringEntity("{ \"index.lifecycle.name\": \"" + policyName + "\" }",
|
||||
ContentType.APPLICATION_JSON);
|
||||
changePolicyRequest.setEntity(changePolicyEntity);
|
||||
assertOK(leaderClient.performRequest(changePolicyRequest));
|
||||
|
||||
index(leaderClient, indexName, "1");
|
||||
assertDocumentExists(leaderClient, indexName, "1");
|
||||
|
||||
assertBusy(() -> {
|
||||
assertDocumentExists(client(), indexName, "1");
|
||||
// Sanity check that following_index setting has been set, so that we can verify later that this setting has been unset:
|
||||
assertThat(getIndexSetting(client(), indexName, "index.xpack.ccr.following_index"), equalTo("true"));
|
||||
|
||||
// We should get into a state with these policies where both leader and followers are waiting on each other
|
||||
assertILMPolicy(leaderClient, indexName, policyName, "warm", "shrink", "wait-for-shard-history-leases");
|
||||
assertILMPolicy(client(), indexName, policyName, "hot", "unfollow", "wait-for-indexing-complete");
|
||||
});
|
||||
|
||||
// Manually set this to kick the process
|
||||
updateIndexSettings(leaderClient, indexName, Settings.builder()
|
||||
.put("index.lifecycle.indexing_complete", true)
|
||||
.build()
|
||||
);
|
||||
|
||||
assertBusy(() -> {
|
||||
// The shrunken index should now be created on the leader...
|
||||
Response shrunkenIndexExistsResponse = leaderClient.performRequest(new Request("HEAD", "/" + shrunkenIndexName));
|
||||
assertEquals(RestStatus.OK.getStatus(), shrunkenIndexExistsResponse.getStatusLine().getStatusCode());
|
||||
|
||||
// And both of these should now finish their policies
|
||||
assertILMPolicy(leaderClient, shrunkenIndexName, policyName, "completed");
|
||||
assertILMPolicy(client(), indexName, policyName, "completed");
|
||||
});
|
||||
}
|
||||
} else {
|
||||
fail("unexpected target cluster [" + targetCluster + "]");
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
private static void putILMPolicy(String name, String maxSize, Integer maxDocs, TimeValue maxAge) throws IOException {
|
||||
final Request request = new Request("PUT", "_ilm/policy/" + name);
|
||||
XContentBuilder builder = jsonBuilder();
|
||||
|
|
@ -436,6 +461,83 @@ public class CCRIndexLifecycleIT extends ESCCRRestTestCase {
|
|||
assertOK(client().performRequest(request));
|
||||
}
|
||||
|
||||
private void putShrinkOnlyPolicy(RestClient client, String policyName) throws IOException {
|
||||
final XContentBuilder builder = jsonBuilder();
|
||||
builder.startObject();
|
||||
{
|
||||
builder.startObject("policy");
|
||||
{
|
||||
builder.startObject("phases");
|
||||
{
|
||||
builder.startObject("warm");
|
||||
{
|
||||
builder.startObject("actions");
|
||||
{
|
||||
builder.startObject("shrink");
|
||||
{
|
||||
builder.field("number_of_shards", 1);
|
||||
}
|
||||
builder.endObject();
|
||||
}
|
||||
builder.endObject();
|
||||
}
|
||||
builder.endObject();
|
||||
|
||||
// Sometimes throw in an extraneous unfollow just to check it doesn't break anything
|
||||
if (randomBoolean()) {
|
||||
builder.startObject("cold");
|
||||
{
|
||||
builder.startObject("actions");
|
||||
{
|
||||
builder.startObject("unfollow");
|
||||
builder.endObject();
|
||||
}
|
||||
builder.endObject();
|
||||
}
|
||||
builder.endObject();
|
||||
}
|
||||
}
|
||||
builder.endObject();
|
||||
}
|
||||
builder.endObject();
|
||||
}
|
||||
builder.endObject();
|
||||
|
||||
final Request request = new Request("PUT", "_ilm/policy/" + policyName);
|
||||
request.setJsonEntity(Strings.toString(builder));
|
||||
assertOK(client.performRequest(request));
|
||||
}
|
||||
|
||||
private void putUnfollowOnlyPolicy(RestClient client, String policyName) throws Exception {
|
||||
final XContentBuilder builder = jsonBuilder();
|
||||
builder.startObject();
|
||||
{
|
||||
builder.startObject("policy");
|
||||
{
|
||||
builder.startObject("phases");
|
||||
{
|
||||
builder.startObject("hot");
|
||||
{
|
||||
builder.startObject("actions");
|
||||
{
|
||||
builder.startObject("unfollow");
|
||||
builder.endObject();
|
||||
}
|
||||
builder.endObject();
|
||||
}
|
||||
builder.endObject();
|
||||
}
|
||||
builder.endObject();
|
||||
}
|
||||
builder.endObject();
|
||||
}
|
||||
builder.endObject();
|
||||
|
||||
final Request request = new Request("PUT", "_ilm/policy/" + policyName);
|
||||
request.setJsonEntity(Strings.toString(builder));
|
||||
assertOK(client.performRequest(request));
|
||||
}
|
||||
|
||||
private static void assertILMPolicy(RestClient client, String index, String policy, String expectedPhase) throws IOException {
|
||||
assertILMPolicy(client, index, policy, expectedPhase, null, null);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -111,10 +111,10 @@ public class PermissionsIT extends ESRestTestCase {
|
|||
Map<String, Object> indexExplain = (Map<String, Object>) ((Map<String, Object>) mapResponse.get("indices")).get("not-ilm");
|
||||
assertThat(indexExplain.get("managed"), equalTo(true));
|
||||
assertThat(indexExplain.get("step"), equalTo("ERROR"));
|
||||
assertThat(indexExplain.get("failed_step"), equalTo("delete"));
|
||||
assertThat(indexExplain.get("failed_step"), equalTo("wait-for-shard-history-leases"));
|
||||
Map<String, String> stepInfo = (Map<String, String>) indexExplain.get("step_info");
|
||||
assertThat(stepInfo.get("type"), equalTo("security_exception"));
|
||||
assertThat(stepInfo.get("reason"), equalTo("action [indices:admin/delete] is unauthorized for user [test_ilm]"));
|
||||
assertThat(stepInfo.get("reason"), equalTo("action [indices:monitor/stats] is unauthorized for user [test_ilm]"));
|
||||
}
|
||||
});
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue