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.

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

Add dedicated step for checking shrink allocation status #35161

Merged
merged 6 commits into from
Nov 5, 2018
Merged
Show file tree
Hide file tree
Changes from 5 commits
Commits
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
Original file line number Diff line number Diff line change
Expand Up @@ -145,7 +145,7 @@ public List<Step> toSteps(Client client, String phase, StepKey nextStepKey) {
exclude.forEach((key, value) -> newSettings.put(IndexMetaData.INDEX_ROUTING_EXCLUDE_GROUP_SETTING.getKey() + key, value));
require.forEach((key, value) -> newSettings.put(IndexMetaData.INDEX_ROUTING_REQUIRE_GROUP_SETTING.getKey() + key, value));
UpdateSettingsStep allocateStep = new UpdateSettingsStep(allocateKey, allocationRoutedKey, client, newSettings.build());
AllocationRoutedStep routedCheckStep = new AllocationRoutedStep(allocationRoutedKey, nextStepKey, true);
AllocationRoutedStep routedCheckStep = new AllocationRoutedStep(allocationRoutedKey, nextStepKey);
return Arrays.asList(allocateStep, routedCheckStep);
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -40,15 +40,8 @@ public class AllocationRoutedStep extends ClusterStateWaitStep {
private static final AllocationDeciders ALLOCATION_DECIDERS = new AllocationDeciders(Collections.singletonList(
new FilterAllocationDecider(Settings.EMPTY, new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS))));

private boolean waitOnAllShardCopies;

AllocationRoutedStep(StepKey key, StepKey nextStepKey, boolean waitOnAllShardCopies) {
AllocationRoutedStep(StepKey key, StepKey nextStepKey) {
super(key, nextStepKey);
this.waitOnAllShardCopies = waitOnAllShardCopies;
}

public boolean getWaitOnAllShardCopies() {
return waitOnAllShardCopies;
}

@Override
Expand All @@ -68,42 +61,35 @@ public Result isConditionMet(Index index, ClusterState clusterState) {
// if the allocation has happened
RoutingAllocation allocation = new RoutingAllocation(ALLOCATION_DECIDERS, clusterState.getRoutingNodes(), clusterState, null,
System.nanoTime());

int allocationPendingAllShards = 0;

ImmutableOpenIntMap<IndexShardRoutingTable> allShards = clusterState.getRoutingTable().index(index).getShards();
for (ObjectCursor<IndexShardRoutingTable> shardRoutingTable : allShards.values()) {
int allocationPendingThisShard = 0;
int shardCopiesThisShard = shardRoutingTable.value.size();
for (ShardRouting shardRouting : shardRoutingTable.value.shards()) {
String currentNodeId = shardRouting.currentNodeId();
boolean canRemainOnCurrentNode = ALLOCATION_DECIDERS
.canRemain(shardRouting, clusterState.getRoutingNodes().node(currentNodeId), allocation)
.type() == Decision.Type.YES;
if (canRemainOnCurrentNode == false) {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We need to add a check for the shard being started here too otherwise we will end up with the same issue on the allocate action but it will be harder to diagnose

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'd like to address that in a separate PR, since I think it warrants more testing for the full ramifications, would that be okay with you?

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ok, could you create an issue so we can track the extra work and it doesn't get missed?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Certainly, opened #35258

allocationPendingThisShard++;
allocationPendingAllShards++;
}
}

if (waitOnAllShardCopies) {
allocationPendingAllShards += allocationPendingThisShard;
} else if (shardCopiesThisShard - allocationPendingThisShard == 0) {
allocationPendingAllShards++;
}
}

if (allocationPendingAllShards > 0) {
logger.debug(
"[{}] lifecycle action for index [{}] waiting for [{}] shards " + "to be allocated to nodes matching the given filters",
getKey().getAction(), index, allocationPendingAllShards);
logger.debug("{} lifecycle action [{}] waiting for [{}] shards to be allocated to nodes matching the given filters",
index, getKey().getAction(), allocationPendingAllShards);
return new Result(false, new Info(idxMeta.getNumberOfReplicas(), allocationPendingAllShards, true));
} else {
logger.debug("[{}] lifecycle action for index [{}] complete", getKey().getAction(), index);
logger.debug("{} lifecycle action for [{}] complete", index, getKey().getAction());
return new Result(true, null);
}
}

@Override
public int hashCode() {
return Objects.hash(super.hashCode(), waitOnAllShardCopies);
return 611;
}

@Override
Expand All @@ -114,9 +100,7 @@ public boolean equals(Object obj) {
if (getClass() != obj.getClass()) {
return false;
}
AllocationRoutedStep other = (AllocationRoutedStep) obj;
return super.equals(obj) &&
Objects.equals(waitOnAllShardCopies, other.waitOnAllShardCopies);
return super.equals(obj);
}

public static final class Info implements ToXContentObject {
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,175 @@
/*
* 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.support.ActiveShardCount;
import org.elasticsearch.cluster.ClusterState;
import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.cluster.routing.IndexRoutingTable;
import org.elasticsearch.cluster.routing.ShardRouting;
import org.elasticsearch.cluster.routing.ShardRoutingState;
import org.elasticsearch.common.ParseField;
import org.elasticsearch.common.Strings;
import org.elasticsearch.common.xcontent.ConstructingObjectParser;
import org.elasticsearch.common.xcontent.ToXContentObject;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.index.Index;

import java.io.IOException;
import java.util.Locale;
import java.util.Objects;

/**
* This step is used prior to running a shrink step in order to ensure that the index being shrunk
* has a copy of each shard allocated on one particular node (the node used by the require
* parameter) and that the shards are not relocating.
*/
public class CheckShrinkReadyStep extends ClusterStateWaitStep {
public static final String NAME = "check-shrink-allocation";

private static final Logger logger = LogManager.getLogger(CheckShrinkReadyStep.class);

CheckShrinkReadyStep(StepKey key, StepKey nextStepKey) {
super(key, nextStepKey);
}

@Override
public Result isConditionMet(Index index, ClusterState clusterState) {
IndexMetaData idxMeta = clusterState.metaData().index(index);

if (idxMeta == null) {
// Index must have been since deleted, ignore it
logger.debug("[{}] lifecycle action for index [{}] executed but index no longer exists",
getKey().getAction(), index.getName());
return new Result(false, null);
}

// How many shards the node should have
int expectedShardCount = idxMeta.getNumberOfShards();

if (ActiveShardCount.ALL.enoughShardsActive(clusterState, index.getName()) == false) {
logger.debug("[{}] shrink action for [{}] cannot make progress because not all shards are active",
getKey().getAction(), index.getName());
return new Result(false, new CheckShrinkReadyStep.Info("", expectedShardCount, -1));
}

// The id of the node the shards should be on
final String idShardsShouldBeOn = idxMeta.getSettings().get(IndexMetaData.INDEX_ROUTING_REQUIRE_GROUP_PREFIX + "._id");
if (idShardsShouldBeOn == null) {
throw new IllegalStateException("Cannot check shrink allocation as there are no allocation rules by _id");
}

final IndexRoutingTable routingTable = clusterState.getRoutingTable().index(index);
int foundShards = 0;
for (ShardRouting shard : routingTable.shardsWithState(ShardRoutingState.STARTED)) {
final String currentNodeId = shard.currentNodeId();
if (idShardsShouldBeOn.equals(currentNodeId) && shard.relocating() == false) {
foundShards++;
}
}

logger.trace("{} checking for shrink readiness on [{}], found {} shards and need {}",
index, idShardsShouldBeOn, foundShards, expectedShardCount);

if (foundShards == expectedShardCount) {
logger.trace("{} successfully found {} allocated shards for shrink readiness on node [{}] ({})",
index, expectedShardCount, idShardsShouldBeOn, getKey().getAction());
return new Result(true, null);
} else {
logger.trace("{} failed to find {} allocated shards (found {}) on node [{}] for shrink readiness ({})",
index, expectedShardCount, foundShards, idShardsShouldBeOn, getKey().getAction());
return new Result(false, new CheckShrinkReadyStep.Info(idShardsShouldBeOn, expectedShardCount,
expectedShardCount - foundShards));
}
}

@Override
public int hashCode() {
return 612;
}

@Override
public boolean equals(Object obj) {
if (obj == null) {
return false;
}
if (getClass() != obj.getClass()) {
return false;
}
return super.equals(obj);
}

public static final class Info implements ToXContentObject {

private final String nodeId;
private final long actualReplicas;
private final long numberShardsLeftToAllocate;
private final String message;

static final ParseField NODE_ID = new ParseField("node_id");
static final ParseField EXPECTED_SHARDS = new ParseField("expected_shards");
static final ParseField SHARDS_TO_ALLOCATE = new ParseField("shards_left_to_allocate");
static final ParseField MESSAGE = new ParseField("message");
static final ConstructingObjectParser<CheckShrinkReadyStep.Info, Void> PARSER = new ConstructingObjectParser<>(
"check_shrink_ready_step_info", a -> new CheckShrinkReadyStep.Info((String) a[0], (long) a[1], (long) a[2]));
static {
PARSER.declareString(ConstructingObjectParser.constructorArg(), NODE_ID);
PARSER.declareLong(ConstructingObjectParser.constructorArg(), EXPECTED_SHARDS);
PARSER.declareLong(ConstructingObjectParser.constructorArg(), SHARDS_TO_ALLOCATE);
PARSER.declareString((i, s) -> {}, MESSAGE);
}

public Info(String nodeId, long expectedShards, long numberShardsLeftToAllocate) {
this.nodeId = nodeId;
this.actualReplicas = expectedShards;
this.numberShardsLeftToAllocate = numberShardsLeftToAllocate;
if (numberShardsLeftToAllocate < 0) {
this.message = "Waiting for all shards to become active";
} else {
this.message = String.format(Locale.ROOT, "Waiting for node [%s] to contain [%d] shards, found [%d], remaining [%d]",
nodeId, expectedShards, expectedShards - numberShardsLeftToAllocate, numberShardsLeftToAllocate);
}
}

@Override
public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
builder.startObject();
builder.field(MESSAGE.getPreferredName(), message);
builder.field(NODE_ID.getPreferredName(), nodeId);
builder.field(SHARDS_TO_ALLOCATE.getPreferredName(), numberShardsLeftToAllocate);
builder.field(EXPECTED_SHARDS.getPreferredName(), actualReplicas);
builder.endObject();
return builder;
}

@Override
public int hashCode() {
return Objects.hash(nodeId, actualReplicas, numberShardsLeftToAllocate);
}

@Override
public boolean equals(Object obj) {
if (obj == null) {
return false;
}
if (getClass() != obj.getClass()) {
return false;
}
CheckShrinkReadyStep.Info other = (CheckShrinkReadyStep.Info) obj;
return Objects.equals(actualReplicas, other.actualReplicas) &&
Objects.equals(numberShardsLeftToAllocate, other.numberShardsLeftToAllocate) &&
Objects.equals(nodeId, other.nodeId);
}

@Override
public String toString() {
return Strings.toString(this);
}
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -87,7 +87,7 @@ public List<Step> toSteps(Client client, String phase, Step.StepKey nextStepKey)

StepKey readOnlyKey = new StepKey(phase, NAME, ReadOnlyAction.NAME);
StepKey setSingleNodeKey = new StepKey(phase, NAME, SetSingleNodeAllocateStep.NAME);
StepKey allocationRoutedKey = new StepKey(phase, NAME, AllocationRoutedStep.NAME);
StepKey allocationRoutedKey = new StepKey(phase, NAME, CheckShrinkReadyStep.NAME);
StepKey shrinkKey = new StepKey(phase, NAME, ShrinkStep.NAME);
StepKey enoughShardsKey = new StepKey(phase, NAME, ShrunkShardsAllocatedStep.NAME);
StepKey copyMetadataKey = new StepKey(phase, NAME, CopyExecutionStateStep.NAME);
Expand All @@ -96,27 +96,27 @@ public List<Step> toSteps(Client client, String phase, Step.StepKey nextStepKey)

UpdateSettingsStep readOnlyStep = new UpdateSettingsStep(readOnlyKey, setSingleNodeKey, client, readOnlySettings);
SetSingleNodeAllocateStep setSingleNodeStep = new SetSingleNodeAllocateStep(setSingleNodeKey, allocationRoutedKey, client);
AllocationRoutedStep allocationStep = new AllocationRoutedStep(allocationRoutedKey, shrinkKey, false);
CheckShrinkReadyStep checkShrinkReadyStep = new CheckShrinkReadyStep(allocationRoutedKey, shrinkKey);
ShrinkStep shrink = new ShrinkStep(shrinkKey, enoughShardsKey, client, numberOfShards, SHRUNKEN_INDEX_PREFIX);
ShrunkShardsAllocatedStep allocated = new ShrunkShardsAllocatedStep(enoughShardsKey, copyMetadataKey, SHRUNKEN_INDEX_PREFIX);
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(readOnlyStep, setSingleNodeStep, allocationStep, shrink, allocated, copyMetadata,
return Arrays.asList(readOnlyStep, setSingleNodeStep, checkShrinkReadyStep, shrink, allocated, copyMetadata,
aliasSwapAndDelete, waitOnShrinkTakeover);
}

@Override
public List<StepKey> toStepKeys(String phase) {
StepKey readOnlyKey = new StepKey(phase, NAME, ReadOnlyAction.NAME);
StepKey setSingleNodeKey = new StepKey(phase, NAME, SetSingleNodeAllocateStep.NAME);
StepKey allocationRoutedKey = new StepKey(phase, NAME, AllocationRoutedStep.NAME);
StepKey checkShrinkReadyKey = new StepKey(phase, NAME, CheckShrinkReadyStep.NAME);
StepKey shrinkKey = new StepKey(phase, NAME, ShrinkStep.NAME);
StepKey enoughShardsKey = new StepKey(phase, NAME, ShrunkShardsAllocatedStep.NAME);
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(readOnlyKey, setSingleNodeKey, allocationRoutedKey, shrinkKey, enoughShardsKey,
return Arrays.asList(readOnlyKey, setSingleNodeKey, checkShrinkReadyKey, shrinkKey, enoughShardsKey,
copyMetadataKey, aliasKey, isShrunkIndexKey);
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -36,37 +36,32 @@ public class AllocationRoutedStepTests extends AbstractStepTestCase<AllocationRo
public AllocationRoutedStep createRandomInstance() {
StepKey stepKey = randomStepKey();
StepKey nextStepKey = randomStepKey();
boolean waitOnAllShardCopies = randomBoolean();

return new AllocationRoutedStep(stepKey, nextStepKey, waitOnAllShardCopies);
return new AllocationRoutedStep(stepKey, nextStepKey);
}

@Override
public AllocationRoutedStep mutateInstance(AllocationRoutedStep instance) {
StepKey key = instance.getKey();
StepKey nextKey = instance.getNextStepKey();
boolean waitOnAllShardCopies = instance.getWaitOnAllShardCopies();

switch (between(0, 2)) {
switch (between(0, 1)) {
case 0:
key = new StepKey(key.getPhase(), key.getAction(), key.getName() + randomAlphaOfLength(5));
break;
case 1:
nextKey = new StepKey(key.getPhase(), key.getAction(), key.getName() + randomAlphaOfLength(5));
break;
case 2:
waitOnAllShardCopies = waitOnAllShardCopies == false;
break;
default:
throw new AssertionError("Illegal randomisation branch");
}

return new AllocationRoutedStep(key, nextKey, waitOnAllShardCopies);
return new AllocationRoutedStep(key, nextKey);
}

@Override
public AllocationRoutedStep copyInstance(AllocationRoutedStep instance) {
return new AllocationRoutedStep(instance.getKey(), instance.getNextStepKey(), instance.getWaitOnAllShardCopies());
return new AllocationRoutedStep(instance.getKey(), instance.getNextStepKey());
}

public void testConditionMet() {
Expand Down Expand Up @@ -132,9 +127,9 @@ public void testConditionMetOnlyOneCopyAllocated() {
.addShard(TestShardRouting.newShardRouting(new ShardId(index, 0), "node2", primaryOnNode1 == false,
ShardRoutingState.STARTED));

AllocationRoutedStep step = new AllocationRoutedStep(randomStepKey(), randomStepKey(), false);
AllocationRoutedStep step = new AllocationRoutedStep(randomStepKey(), randomStepKey());
assertAllocateStatus(index, 1, 0, step, existingSettings, node1Settings, node2Settings, indexRoutingTable,
new ClusterStateWaitStep.Result(true, null));
new ClusterStateWaitStep.Result(false, new AllocationRoutedStep.Info(0, 1, true)));
}

public void testExecuteAllocateNotComplete() throws Exception {
Expand Down Expand Up @@ -202,7 +197,7 @@ public void testExecuteAllocateNotCompleteOnlyOneCopyAllocated() throws Exceptio
.addShard(TestShardRouting.newShardRouting(new ShardId(index, 0), "node2", primaryOnNode1 == false,
ShardRoutingState.STARTED));

AllocationRoutedStep step = new AllocationRoutedStep(randomStepKey(), randomStepKey(), true);
AllocationRoutedStep step = new AllocationRoutedStep(randomStepKey(), randomStepKey());
assertAllocateStatus(index, 2, 0, step, existingSettings, node1Settings, node2Settings, indexRoutingTable,
new ClusterStateWaitStep.Result(false, new AllocationRoutedStep.Info(0, 1, true)));
}
Expand Down
Loading