add serialization options for allocation commands
This commit is contained in:
Shay Banon
parent
2bd9b3aed0
commit
ef9974ce2c
|
|
@ -21,12 +21,19 @@ package org.elasticsearch.cluster.routing.allocation.command;
|
|||
|
||||
import org.elasticsearch.ElasticSearchException;
|
||||
import org.elasticsearch.ElasticSearchIllegalArgumentException;
|
||||
import org.elasticsearch.ElasticSearchParseException;
|
||||
import org.elasticsearch.cluster.node.DiscoveryNode;
|
||||
import org.elasticsearch.cluster.routing.MutableShardRouting;
|
||||
import org.elasticsearch.cluster.routing.RoutingNode;
|
||||
import org.elasticsearch.cluster.routing.allocation.RoutingAllocation;
|
||||
import org.elasticsearch.common.io.stream.StreamInput;
|
||||
import org.elasticsearch.common.io.stream.StreamOutput;
|
||||
import org.elasticsearch.common.xcontent.ToXContent;
|
||||
import org.elasticsearch.common.xcontent.XContentBuilder;
|
||||
import org.elasticsearch.common.xcontent.XContentParser;
|
||||
import org.elasticsearch.index.shard.ShardId;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.Iterator;
|
||||
|
||||
/**
|
||||
|
|
@ -36,19 +43,103 @@ import java.util.Iterator;
|
|||
*/
|
||||
public class AllocateAllocationCommand implements AllocationCommand {
|
||||
|
||||
public static final String NAME = "allocate";
|
||||
|
||||
public static class Factory implements AllocationCommand.Factory<AllocateAllocationCommand> {
|
||||
|
||||
@Override
|
||||
public AllocateAllocationCommand readFrom(StreamInput in) throws IOException {
|
||||
return new AllocateAllocationCommand(ShardId.readShardId(in), in.readString(), in.readBoolean());
|
||||
}
|
||||
|
||||
@Override
|
||||
public void writeTo(AllocateAllocationCommand command, StreamOutput out) throws IOException {
|
||||
command.shardId().writeTo(out);
|
||||
out.writeString(command.node());
|
||||
out.writeBoolean(command.allowPrimary());
|
||||
}
|
||||
|
||||
@Override
|
||||
public AllocateAllocationCommand fromXContent(XContentParser parser) throws IOException {
|
||||
String index = null;
|
||||
int shardId = -1;
|
||||
String nodeId = null;
|
||||
boolean allowPrimary = false;
|
||||
|
||||
String currentFieldName = null;
|
||||
XContentParser.Token token;
|
||||
while ((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) {
|
||||
if (token == XContentParser.Token.FIELD_NAME) {
|
||||
currentFieldName = parser.currentName();
|
||||
} else if (token.isValue()) {
|
||||
if ("index".equals(currentFieldName)) {
|
||||
index = parser.text();
|
||||
} else if ("shard".equals(currentFieldName)) {
|
||||
shardId = parser.intValue();
|
||||
} else if ("node".equals(currentFieldName)) {
|
||||
nodeId = parser.text();
|
||||
} else if ("allow_primary".equals(currentFieldName) || "allowPrimary".equals(currentFieldName)) {
|
||||
allowPrimary = parser.booleanValue();
|
||||
} else {
|
||||
throw new ElasticSearchParseException("[allocate] command does not support field [" + currentFieldName + "]");
|
||||
}
|
||||
} else {
|
||||
throw new ElasticSearchParseException("[allocate] command does not support complex json tokens [" + token + "]");
|
||||
}
|
||||
}
|
||||
if (index == null) {
|
||||
throw new ElasticSearchParseException("[allocate] command missing the index parameter");
|
||||
}
|
||||
if (shardId == -1) {
|
||||
throw new ElasticSearchParseException("[allocate] command missing the shard parameter");
|
||||
}
|
||||
if (nodeId == null) {
|
||||
throw new ElasticSearchParseException("[allocate] command missing the node parameter");
|
||||
}
|
||||
return new AllocateAllocationCommand(new ShardId(index, shardId), nodeId, allowPrimary);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void toXContent(AllocateAllocationCommand command, XContentBuilder builder, ToXContent.Params params) throws IOException {
|
||||
builder.startObject();
|
||||
builder.field("index", command.shardId().index());
|
||||
builder.field("shard", command.shardId().id());
|
||||
builder.field("node", command.node());
|
||||
builder.field("allow_primary", command.allowPrimary());
|
||||
builder.endObject();
|
||||
}
|
||||
}
|
||||
|
||||
private final ShardId shardId;
|
||||
private final String nodeId;
|
||||
private final String node;
|
||||
private final boolean allowPrimary;
|
||||
|
||||
public AllocateAllocationCommand(ShardId shardId, String nodeId, boolean allowPrimary) {
|
||||
public AllocateAllocationCommand(ShardId shardId, String node, boolean allowPrimary) {
|
||||
this.shardId = shardId;
|
||||
this.nodeId = nodeId;
|
||||
this.node = node;
|
||||
this.allowPrimary = allowPrimary;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String name() {
|
||||
return NAME;
|
||||
}
|
||||
|
||||
public ShardId shardId() {
|
||||
return this.shardId;
|
||||
}
|
||||
|
||||
public String node() {
|
||||
return this.node;
|
||||
}
|
||||
|
||||
public boolean allowPrimary() {
|
||||
return this.allowPrimary;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void execute(RoutingAllocation allocation) throws ElasticSearchException {
|
||||
DiscoveryNode node = allocation.nodes().resolveNode(nodeId);
|
||||
DiscoveryNode discoNode = allocation.nodes().resolveNode(node);
|
||||
|
||||
MutableShardRouting shardRouting = null;
|
||||
for (MutableShardRouting routing : allocation.routingNodes().unassigned()) {
|
||||
|
|
@ -68,10 +159,10 @@ public class AllocateAllocationCommand implements AllocationCommand {
|
|||
throw new ElasticSearchIllegalArgumentException("[allocate] trying to allocate a primary shard " + shardId + "], which is disabled");
|
||||
}
|
||||
|
||||
RoutingNode routingNode = allocation.routingNodes().node(node.id());
|
||||
RoutingNode routingNode = allocation.routingNodes().node(discoNode.id());
|
||||
allocation.addIgnoreDisable(shardRouting.shardId(), routingNode.nodeId());
|
||||
if (!allocation.deciders().canAllocate(shardRouting, routingNode, allocation).allowed()) {
|
||||
throw new ElasticSearchIllegalArgumentException("[allocate] allocation of " + shardId + " on node " + node + " is not allowed");
|
||||
throw new ElasticSearchIllegalArgumentException("[allocate] allocation of " + shardId + " on node " + discoNode + " is not allowed");
|
||||
}
|
||||
// go over and remove it from the unassigned
|
||||
for (Iterator<MutableShardRouting> it = allocation.routingNodes().unassigned().iterator(); it.hasNext(); ) {
|
||||
|
|
|
|||
|
|
@ -21,10 +21,30 @@ package org.elasticsearch.cluster.routing.allocation.command;
|
|||
|
||||
import org.elasticsearch.ElasticSearchException;
|
||||
import org.elasticsearch.cluster.routing.allocation.RoutingAllocation;
|
||||
import org.elasticsearch.common.io.stream.StreamInput;
|
||||
import org.elasticsearch.common.io.stream.StreamOutput;
|
||||
import org.elasticsearch.common.xcontent.ToXContent;
|
||||
import org.elasticsearch.common.xcontent.XContentBuilder;
|
||||
import org.elasticsearch.common.xcontent.XContentParser;
|
||||
|
||||
import java.io.IOException;
|
||||
|
||||
/**
|
||||
*/
|
||||
public interface AllocationCommand {
|
||||
|
||||
interface Factory<T extends AllocationCommand> {
|
||||
|
||||
T readFrom(StreamInput in) throws IOException;
|
||||
|
||||
void writeTo(T command, StreamOutput out) throws IOException;
|
||||
|
||||
T fromXContent(XContentParser parser) throws IOException;
|
||||
|
||||
void toXContent(T command, XContentBuilder builder, ToXContent.Params params) throws IOException;
|
||||
}
|
||||
|
||||
String name();
|
||||
|
||||
void execute(RoutingAllocation allocation) throws ElasticSearchException;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -21,15 +21,56 @@ package org.elasticsearch.cluster.routing.allocation.command;
|
|||
|
||||
import com.google.common.collect.Lists;
|
||||
import org.elasticsearch.ElasticSearchException;
|
||||
import org.elasticsearch.ElasticSearchIllegalArgumentException;
|
||||
import org.elasticsearch.ElasticSearchParseException;
|
||||
import org.elasticsearch.cluster.routing.allocation.RoutingAllocation;
|
||||
import org.elasticsearch.common.Nullable;
|
||||
import org.elasticsearch.common.io.stream.StreamInput;
|
||||
import org.elasticsearch.common.io.stream.StreamOutput;
|
||||
import org.elasticsearch.common.xcontent.ToXContent;
|
||||
import org.elasticsearch.common.xcontent.XContentBuilder;
|
||||
import org.elasticsearch.common.xcontent.XContentParser;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.Arrays;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
*/
|
||||
public class AllocationCommands {
|
||||
|
||||
private static Map<String, AllocationCommand.Factory> factories = new HashMap<String, AllocationCommand.Factory>();
|
||||
|
||||
/**
|
||||
* Register a custom index meta data factory. Make sure to call it from a static block.
|
||||
*/
|
||||
public static void registerFactory(String type, AllocationCommand.Factory factory) {
|
||||
factories.put(type, factory);
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
@Nullable
|
||||
public static <T extends AllocationCommand> AllocationCommand.Factory<T> lookupFactory(String name) {
|
||||
return factories.get(name);
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
public static <T extends AllocationCommand> AllocationCommand.Factory<T> lookupFactorySafe(String name) throws ElasticSearchIllegalArgumentException {
|
||||
AllocationCommand.Factory<T> factory = factories.get(name);
|
||||
if (factory == null) {
|
||||
throw new ElasticSearchIllegalArgumentException("No allocation command factory registered for name [" + name + "]");
|
||||
}
|
||||
return factory;
|
||||
}
|
||||
|
||||
static {
|
||||
registerFactory(AllocateAllocationCommand.NAME, new AllocateAllocationCommand.Factory());
|
||||
registerFactory(CancelAllocationCommand.NAME, new CancelAllocationCommand.Factory());
|
||||
registerFactory(MoveAllocationCommand.NAME, new MoveAllocationCommand.Factory());
|
||||
}
|
||||
|
||||
private final List<AllocationCommand> commands = Lists.newArrayList();
|
||||
|
||||
public AllocationCommands(AllocationCommand... commands) {
|
||||
|
|
@ -45,9 +86,94 @@ public class AllocationCommands {
|
|||
return this;
|
||||
}
|
||||
|
||||
public List<AllocationCommand> commands() {
|
||||
return this.commands;
|
||||
}
|
||||
|
||||
public void execute(RoutingAllocation allocation) throws ElasticSearchException {
|
||||
for (AllocationCommand command : commands) {
|
||||
command.execute(allocation);
|
||||
}
|
||||
}
|
||||
|
||||
public static AllocationCommands readFrom(StreamInput in) throws IOException {
|
||||
AllocationCommands commands = new AllocationCommands();
|
||||
int size = in.readVInt();
|
||||
for (int i = 0; i < size; i++) {
|
||||
String name = in.readString();
|
||||
commands.add(lookupFactorySafe(name).readFrom(in));
|
||||
}
|
||||
return commands;
|
||||
}
|
||||
|
||||
public static void writeTo(AllocationCommands commands, StreamOutput out) throws IOException {
|
||||
out.writeVInt(commands.commands.size());
|
||||
for (AllocationCommand command : commands.commands) {
|
||||
out.writeString(command.name());
|
||||
lookupFactorySafe(command.name()).writeTo(command, out);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* <pre>
|
||||
* {
|
||||
* "commands" : [
|
||||
* {"allocate" : {"index" : "test", "shard" : 0, "node" : "test"}}
|
||||
* ]
|
||||
* }
|
||||
* </pre>
|
||||
*/
|
||||
public static AllocationCommands fromXContent(XContentParser parser) throws IOException {
|
||||
AllocationCommands commands = new AllocationCommands();
|
||||
|
||||
XContentParser.Token token = parser.nextToken();
|
||||
if (token == null) {
|
||||
throw new ElasticSearchParseException("No commands");
|
||||
}
|
||||
if (token != XContentParser.Token.START_OBJECT) {
|
||||
throw new ElasticSearchParseException("No start object, got " + token);
|
||||
}
|
||||
|
||||
token = parser.nextToken();
|
||||
if (token != XContentParser.Token.FIELD_NAME) {
|
||||
throw new ElasticSearchParseException("expected the field name `commands` to exists, got " + token);
|
||||
}
|
||||
if (!parser.currentName().equals("commands")) {
|
||||
throw new ElasticSearchParseException("expected field name to be named `commands`, got " + parser.currentName());
|
||||
}
|
||||
|
||||
token = parser.nextToken();
|
||||
if (token != XContentParser.Token.START_ARRAY) {
|
||||
throw new ElasticSearchParseException("commands should follow with an array element");
|
||||
}
|
||||
while ((token = parser.nextToken()) != XContentParser.Token.END_ARRAY) {
|
||||
if (token == XContentParser.Token.START_OBJECT) {
|
||||
// move to the command name
|
||||
token = parser.nextToken();
|
||||
String commandName = parser.currentName();
|
||||
token = parser.nextToken();
|
||||
commands.add(AllocationCommands.lookupFactorySafe(commandName).fromXContent(parser));
|
||||
// move to the end object one
|
||||
if (parser.nextToken() != XContentParser.Token.END_OBJECT) {
|
||||
throw new ElasticSearchParseException("allocation command is malformed, done parsing a command, but didn't get END_OBJECT, got " + token);
|
||||
}
|
||||
} else {
|
||||
throw new ElasticSearchParseException("allocation command is malformed, got token " + token);
|
||||
}
|
||||
}
|
||||
return commands;
|
||||
}
|
||||
|
||||
public static void toXContent(AllocationCommands commands, XContentBuilder builder, ToXContent.Params params) throws IOException {
|
||||
builder.startObject();
|
||||
builder.startArray("commands");
|
||||
for (AllocationCommand command : commands.commands) {
|
||||
builder.startObject();
|
||||
builder.field(command.name());
|
||||
AllocationCommands.lookupFactorySafe(command.name()).toXContent(command, builder, params);
|
||||
builder.endObject();
|
||||
}
|
||||
builder.endArray();
|
||||
builder.endObject();
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -21,13 +21,20 @@ package org.elasticsearch.cluster.routing.allocation.command;
|
|||
|
||||
import org.elasticsearch.ElasticSearchException;
|
||||
import org.elasticsearch.ElasticSearchIllegalArgumentException;
|
||||
import org.elasticsearch.ElasticSearchParseException;
|
||||
import org.elasticsearch.cluster.node.DiscoveryNode;
|
||||
import org.elasticsearch.cluster.routing.MutableShardRouting;
|
||||
import org.elasticsearch.cluster.routing.RoutingNode;
|
||||
import org.elasticsearch.cluster.routing.ShardRoutingState;
|
||||
import org.elasticsearch.cluster.routing.allocation.RoutingAllocation;
|
||||
import org.elasticsearch.common.io.stream.StreamInput;
|
||||
import org.elasticsearch.common.io.stream.StreamOutput;
|
||||
import org.elasticsearch.common.xcontent.ToXContent;
|
||||
import org.elasticsearch.common.xcontent.XContentBuilder;
|
||||
import org.elasticsearch.common.xcontent.XContentParser;
|
||||
import org.elasticsearch.index.shard.ShardId;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.Iterator;
|
||||
|
||||
import static org.elasticsearch.cluster.routing.ShardRoutingState.INITIALIZING;
|
||||
|
|
@ -38,21 +45,96 @@ import static org.elasticsearch.cluster.routing.ShardRoutingState.RELOCATING;
|
|||
*/
|
||||
public class CancelAllocationCommand implements AllocationCommand {
|
||||
|
||||
private final ShardId shardId;
|
||||
public static final String NAME = "cancel";
|
||||
|
||||
private final String nodeId;
|
||||
public static class Factory implements AllocationCommand.Factory<CancelAllocationCommand> {
|
||||
|
||||
@Override
|
||||
public CancelAllocationCommand readFrom(StreamInput in) throws IOException {
|
||||
return new CancelAllocationCommand(ShardId.readShardId(in), in.readString());
|
||||
}
|
||||
|
||||
@Override
|
||||
public void writeTo(CancelAllocationCommand command, StreamOutput out) throws IOException {
|
||||
command.shardId().writeTo(out);
|
||||
out.writeString(command.node());
|
||||
}
|
||||
|
||||
@Override
|
||||
public CancelAllocationCommand fromXContent(XContentParser parser) throws IOException {
|
||||
String index = null;
|
||||
int shardId = -1;
|
||||
String nodeId = null;
|
||||
|
||||
String currentFieldName = null;
|
||||
XContentParser.Token token;
|
||||
while ((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) {
|
||||
if (token == XContentParser.Token.FIELD_NAME) {
|
||||
currentFieldName = parser.currentName();
|
||||
} else if (token.isValue()) {
|
||||
if ("index".equals(currentFieldName)) {
|
||||
index = parser.text();
|
||||
} else if ("shard".equals(currentFieldName)) {
|
||||
shardId = parser.intValue();
|
||||
} else if ("node".equals(currentFieldName)) {
|
||||
nodeId = parser.text();
|
||||
} else {
|
||||
throw new ElasticSearchParseException("[cancel] command does not support field [" + currentFieldName + "]");
|
||||
}
|
||||
} else {
|
||||
throw new ElasticSearchParseException("[cancel] command does not support complex json tokens [" + token + "]");
|
||||
}
|
||||
}
|
||||
if (index == null) {
|
||||
throw new ElasticSearchParseException("[cancel] command missing the index parameter");
|
||||
}
|
||||
if (shardId == -1) {
|
||||
throw new ElasticSearchParseException("[cancel] command missing the shard parameter");
|
||||
}
|
||||
if (nodeId == null) {
|
||||
throw new ElasticSearchParseException("[cancel] command missing the node parameter");
|
||||
}
|
||||
return new CancelAllocationCommand(new ShardId(index, shardId), nodeId);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void toXContent(CancelAllocationCommand command, XContentBuilder builder, ToXContent.Params params) throws IOException {
|
||||
builder.startObject();
|
||||
builder.field("index", command.shardId().index());
|
||||
builder.field("shard", command.shardId().id());
|
||||
builder.field("node", command.node());
|
||||
builder.endObject();
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
private final ShardId shardId;
|
||||
private final String node;
|
||||
|
||||
public CancelAllocationCommand(ShardId shardId, String node) {
|
||||
this.shardId = shardId;
|
||||
this.nodeId = node;
|
||||
this.node = node;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String name() {
|
||||
return NAME;
|
||||
}
|
||||
|
||||
public ShardId shardId() {
|
||||
return this.shardId;
|
||||
}
|
||||
|
||||
public String node() {
|
||||
return this.node;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void execute(RoutingAllocation allocation) throws ElasticSearchException {
|
||||
DiscoveryNode node = allocation.nodes().resolveNode(nodeId);
|
||||
DiscoveryNode discoNode = allocation.nodes().resolveNode(node);
|
||||
|
||||
boolean found = false;
|
||||
for (Iterator<MutableShardRouting> it = allocation.routingNodes().node(node.id()).iterator(); it.hasNext(); ) {
|
||||
for (Iterator<MutableShardRouting> it = allocation.routingNodes().node(discoNode.id()).iterator(); it.hasNext(); ) {
|
||||
MutableShardRouting shardRouting = it.next();
|
||||
if (!shardRouting.shardId().equals(shardId)) {
|
||||
continue;
|
||||
|
|
@ -77,7 +159,7 @@ public class CancelAllocationCommand implements AllocationCommand {
|
|||
// the shard is relocating to another node, cancel the recovery on the other node, and deallocate this one
|
||||
if (shardRouting.primary()) {
|
||||
// can't cancel a primary shard being initialized
|
||||
throw new ElasticSearchIllegalArgumentException("[cancel_allocation] can't cancel " + shardId + " on node " + node + ", shard is primary and initializing its state");
|
||||
throw new ElasticSearchIllegalArgumentException("[cancel_allocation] can't cancel " + shardId + " on node " + discoNode + ", shard is primary and initializing its state");
|
||||
}
|
||||
it.remove();
|
||||
allocation.routingNodes().unassigned().add(new MutableShardRouting(shardRouting.index(), shardRouting.id(),
|
||||
|
|
@ -99,7 +181,7 @@ public class CancelAllocationCommand implements AllocationCommand {
|
|||
// the shard is not relocating, its either started, or initializing, just cancel it and move on...
|
||||
if (shardRouting.primary()) {
|
||||
// can't cancel a primary shard being initialized
|
||||
throw new ElasticSearchIllegalArgumentException("[cancel_allocation] can't cancel " + shardId + " on node " + node + ", shard is primary and initializing its state");
|
||||
throw new ElasticSearchIllegalArgumentException("[cancel_allocation] can't cancel " + shardId + " on node " + discoNode + ", shard is primary and initializing its state");
|
||||
}
|
||||
it.remove();
|
||||
allocation.routingNodes().unassigned().add(new MutableShardRouting(shardRouting.index(), shardRouting.id(),
|
||||
|
|
@ -108,7 +190,7 @@ public class CancelAllocationCommand implements AllocationCommand {
|
|||
}
|
||||
|
||||
if (!found) {
|
||||
throw new ElasticSearchIllegalArgumentException("[cancel_allocation] can't cancel " + shardId + ", failed to find it on node " + node);
|
||||
throw new ElasticSearchIllegalArgumentException("[cancel_allocation] can't cancel " + shardId + ", failed to find it on node " + discoNode);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -21,39 +21,132 @@ package org.elasticsearch.cluster.routing.allocation.command;
|
|||
|
||||
import org.elasticsearch.ElasticSearchException;
|
||||
import org.elasticsearch.ElasticSearchIllegalArgumentException;
|
||||
import org.elasticsearch.ElasticSearchParseException;
|
||||
import org.elasticsearch.cluster.node.DiscoveryNode;
|
||||
import org.elasticsearch.cluster.routing.MutableShardRouting;
|
||||
import org.elasticsearch.cluster.routing.RoutingNode;
|
||||
import org.elasticsearch.cluster.routing.ShardRoutingState;
|
||||
import org.elasticsearch.cluster.routing.allocation.RoutingAllocation;
|
||||
import org.elasticsearch.cluster.routing.allocation.decider.AllocationDecider;
|
||||
import org.elasticsearch.common.Nullable;
|
||||
import org.elasticsearch.common.io.stream.StreamInput;
|
||||
import org.elasticsearch.common.io.stream.StreamOutput;
|
||||
import org.elasticsearch.common.xcontent.ToXContent;
|
||||
import org.elasticsearch.common.xcontent.XContentBuilder;
|
||||
import org.elasticsearch.common.xcontent.XContentParser;
|
||||
import org.elasticsearch.index.shard.ShardId;
|
||||
|
||||
import java.io.IOException;
|
||||
|
||||
/**
|
||||
* A command that moves a shard from a specific node to another node. Note, the shards
|
||||
* need to be in "started" state in order to be moved if from is specified.
|
||||
*/
|
||||
public class MoveAllocationCommand implements AllocationCommand {
|
||||
|
||||
public static final String NAME = "move";
|
||||
|
||||
public static class Factory implements AllocationCommand.Factory<MoveAllocationCommand> {
|
||||
|
||||
@Override
|
||||
public MoveAllocationCommand readFrom(StreamInput in) throws IOException {
|
||||
return new MoveAllocationCommand(ShardId.readShardId(in), in.readString(), in.readString());
|
||||
}
|
||||
|
||||
@Override
|
||||
public void writeTo(MoveAllocationCommand command, StreamOutput out) throws IOException {
|
||||
command.shardId().writeTo(out);
|
||||
out.writeString(command.fromNode());
|
||||
out.writeString(command.toNode());
|
||||
}
|
||||
|
||||
@Override
|
||||
public MoveAllocationCommand fromXContent(XContentParser parser) throws IOException {
|
||||
String index = null;
|
||||
int shardId = -1;
|
||||
String fromNode = null;
|
||||
String toNode = null;
|
||||
|
||||
String currentFieldName = null;
|
||||
XContentParser.Token token;
|
||||
while ((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) {
|
||||
if (token == XContentParser.Token.FIELD_NAME) {
|
||||
currentFieldName = parser.currentName();
|
||||
} else if (token.isValue()) {
|
||||
if ("index".equals(currentFieldName)) {
|
||||
index = parser.text();
|
||||
} else if ("shard".equals(currentFieldName)) {
|
||||
shardId = parser.intValue();
|
||||
} else if ("from_node".equals(currentFieldName) || "fromNode".equals(currentFieldName)) {
|
||||
fromNode = parser.text();
|
||||
} else if ("to_node".equals(currentFieldName) || "toNode".equals(currentFieldName)) {
|
||||
toNode = parser.text();
|
||||
} else {
|
||||
throw new ElasticSearchParseException("[move] command does not support field [" + currentFieldName + "]");
|
||||
}
|
||||
} else {
|
||||
throw new ElasticSearchParseException("[move] command does not support complex json tokens [" + token + "]");
|
||||
}
|
||||
}
|
||||
if (index == null) {
|
||||
throw new ElasticSearchParseException("[move] command missing the index parameter");
|
||||
}
|
||||
if (shardId == -1) {
|
||||
throw new ElasticSearchParseException("[move] command missing the shard parameter");
|
||||
}
|
||||
if (fromNode == null) {
|
||||
throw new ElasticSearchParseException("[move] command missing the from_node parameter");
|
||||
}
|
||||
if (toNode == null) {
|
||||
throw new ElasticSearchParseException("[move] command missing the to_node parameter");
|
||||
}
|
||||
return new MoveAllocationCommand(new ShardId(index, shardId), fromNode, toNode);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void toXContent(MoveAllocationCommand command, XContentBuilder builder, ToXContent.Params params) throws IOException {
|
||||
builder.startObject();
|
||||
builder.field("index", command.shardId().index());
|
||||
builder.field("shard", command.shardId().id());
|
||||
builder.field("from_node", command.fromNode());
|
||||
builder.field("to_node", command.toNode());
|
||||
builder.endObject();
|
||||
}
|
||||
}
|
||||
|
||||
private final ShardId shardId;
|
||||
@Nullable
|
||||
private final String fromNode;
|
||||
private final String toNode;
|
||||
|
||||
public MoveAllocationCommand(ShardId shardId, @Nullable String fromNode, String toNode) {
|
||||
public MoveAllocationCommand(ShardId shardId, String fromNode, String toNode) {
|
||||
this.shardId = shardId;
|
||||
this.fromNode = fromNode;
|
||||
this.toNode = toNode;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String name() {
|
||||
return NAME;
|
||||
}
|
||||
|
||||
public ShardId shardId() {
|
||||
return this.shardId;
|
||||
}
|
||||
|
||||
public String fromNode() {
|
||||
return this.fromNode;
|
||||
}
|
||||
|
||||
public String toNode() {
|
||||
return this.toNode;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void execute(RoutingAllocation allocation) throws ElasticSearchException {
|
||||
DiscoveryNode from = allocation.nodes().resolveNode(fromNode);
|
||||
DiscoveryNode to = allocation.nodes().resolveNode(toNode);
|
||||
DiscoveryNode fromDiscoNode = allocation.nodes().resolveNode(fromNode);
|
||||
DiscoveryNode toDiscoNode = allocation.nodes().resolveNode(toNode);
|
||||
|
||||
boolean found = false;
|
||||
for (MutableShardRouting shardRouting : allocation.routingNodes().node(from.id())) {
|
||||
for (MutableShardRouting shardRouting : allocation.routingNodes().node(fromDiscoNode.id())) {
|
||||
if (!shardRouting.shardId().equals(shardId)) {
|
||||
continue;
|
||||
}
|
||||
|
|
@ -64,10 +157,10 @@ public class MoveAllocationCommand implements AllocationCommand {
|
|||
throw new ElasticSearchIllegalArgumentException("[move_allocation] can't move " + shardId + ", shard is not started (state = " + shardRouting.state() + "]");
|
||||
}
|
||||
|
||||
RoutingNode toRoutingNode = allocation.routingNodes().node(to.id());
|
||||
RoutingNode toRoutingNode = allocation.routingNodes().node(toDiscoNode.id());
|
||||
AllocationDecider.Decision decision = allocation.deciders().canAllocate(shardRouting, toRoutingNode, allocation);
|
||||
if (!decision.allowed()) {
|
||||
throw new ElasticSearchIllegalArgumentException("[move_allocation] can't move " + shardId + ", from " + from + ", to " + to + ", since its not allowed");
|
||||
throw new ElasticSearchIllegalArgumentException("[move_allocation] can't move " + shardId + ", from " + fromDiscoNode + ", to " + toDiscoNode + ", since its not allowed");
|
||||
}
|
||||
if (!decision.allocate()) {
|
||||
// its being throttled, maybe have a flag to take it into account and fail? for now, just do it since the "user" wants it...
|
||||
|
|
@ -81,7 +174,7 @@ public class MoveAllocationCommand implements AllocationCommand {
|
|||
}
|
||||
|
||||
if (!found) {
|
||||
throw new ElasticSearchIllegalArgumentException("[move_allocation] can't move " + shardId + ", failed to find it on node " + from);
|
||||
throw new ElasticSearchIllegalArgumentException("[move_allocation] can't move " + shardId + ", failed to find it on node " + fromDiscoNode);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -30,8 +30,12 @@ import org.elasticsearch.cluster.routing.allocation.command.AllocateAllocationCo
|
|||
import org.elasticsearch.cluster.routing.allocation.command.AllocationCommands;
|
||||
import org.elasticsearch.cluster.routing.allocation.command.CancelAllocationCommand;
|
||||
import org.elasticsearch.cluster.routing.allocation.command.MoveAllocationCommand;
|
||||
import org.elasticsearch.common.io.stream.BytesStreamInput;
|
||||
import org.elasticsearch.common.io.stream.BytesStreamOutput;
|
||||
import org.elasticsearch.common.logging.ESLogger;
|
||||
import org.elasticsearch.common.logging.Loggers;
|
||||
import org.elasticsearch.common.xcontent.XContentFactory;
|
||||
import org.elasticsearch.common.xcontent.XContentType;
|
||||
import org.elasticsearch.index.shard.ShardId;
|
||||
import org.testng.annotations.Test;
|
||||
|
||||
|
|
@ -284,4 +288,52 @@ public class AllocationCommandsTests {
|
|||
assertThat(clusterState.routingNodes().node("node2").shards().size(), equalTo(0));
|
||||
assertThat(clusterState.routingNodes().node("node3").shards().size(), equalTo(0));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void serialization() throws Exception {
|
||||
AllocationCommands commands = new AllocationCommands(
|
||||
new AllocateAllocationCommand(new ShardId("test", 1), "node1", true),
|
||||
new MoveAllocationCommand(new ShardId("test", 3), "node2", "node3"),
|
||||
new CancelAllocationCommand(new ShardId("test", 4), "node5")
|
||||
);
|
||||
BytesStreamOutput bytes = new BytesStreamOutput();
|
||||
AllocationCommands.writeTo(commands, bytes);
|
||||
AllocationCommands sCommands = AllocationCommands.readFrom(new BytesStreamInput(bytes.bytes()));
|
||||
|
||||
assertThat(sCommands.commands().size(), equalTo(3));
|
||||
assertThat(((AllocateAllocationCommand) (sCommands.commands().get(0))).shardId(), equalTo(new ShardId("test", 1)));
|
||||
assertThat(((AllocateAllocationCommand) (sCommands.commands().get(0))).node(), equalTo("node1"));
|
||||
assertThat(((AllocateAllocationCommand) (sCommands.commands().get(0))).allowPrimary(), equalTo(true));
|
||||
|
||||
assertThat(((MoveAllocationCommand) (sCommands.commands().get(1))).shardId(), equalTo(new ShardId("test", 3)));
|
||||
assertThat(((MoveAllocationCommand) (sCommands.commands().get(1))).fromNode(), equalTo("node2"));
|
||||
assertThat(((MoveAllocationCommand) (sCommands.commands().get(1))).toNode(), equalTo("node3"));
|
||||
|
||||
assertThat(((CancelAllocationCommand) (sCommands.commands().get(2))).shardId(), equalTo(new ShardId("test", 4)));
|
||||
assertThat(((CancelAllocationCommand) (sCommands.commands().get(2))).node(), equalTo("node5"));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void xContent() throws Exception {
|
||||
String commands = "{\n" +
|
||||
" \"commands\" : [\n" +
|
||||
" {\"allocate\" : {\"index\" : \"test\", \"shard\" : 1, \"node\" : \"node1\", \"allow_primary\" : true}}\n" +
|
||||
" ,{\"move\" : {\"index\" : \"test\", \"shard\" : 3, \"from_node\" : \"node2\", \"to_node\" : \"node3\"}} \n" +
|
||||
" ,{\"cancel\" : {\"index\" : \"test\", \"shard\" : 4, \"node\" : \"node5\"}} \n" +
|
||||
" ]\n" +
|
||||
"}\n";
|
||||
AllocationCommands sCommands = AllocationCommands.fromXContent(XContentFactory.xContent(XContentType.JSON).createParser(commands));
|
||||
|
||||
assertThat(sCommands.commands().size(), equalTo(3));
|
||||
assertThat(((AllocateAllocationCommand) (sCommands.commands().get(0))).shardId(), equalTo(new ShardId("test", 1)));
|
||||
assertThat(((AllocateAllocationCommand) (sCommands.commands().get(0))).node(), equalTo("node1"));
|
||||
assertThat(((AllocateAllocationCommand) (sCommands.commands().get(0))).allowPrimary(), equalTo(true));
|
||||
|
||||
assertThat(((MoveAllocationCommand) (sCommands.commands().get(1))).shardId(), equalTo(new ShardId("test", 3)));
|
||||
assertThat(((MoveAllocationCommand) (sCommands.commands().get(1))).fromNode(), equalTo("node2"));
|
||||
assertThat(((MoveAllocationCommand) (sCommands.commands().get(1))).toNode(), equalTo("node3"));
|
||||
|
||||
assertThat(((CancelAllocationCommand) (sCommands.commands().get(2))).shardId(), equalTo(new ShardId("test", 4)));
|
||||
assertThat(((CancelAllocationCommand) (sCommands.commands().get(2))).node(), equalTo("node5"));
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue