mirror of
https://github.com/honeymoose/OpenSearch.git
synced 2025-02-25 06:16:40 +00:00
Make cluster health classes immutable and have them implement Writeable
instead of Streamable Closes #18673
This commit is contained in:
Ali Beyad
parent
dac322a32d
commit
f12b10c48a
@ -182,7 +182,7 @@ public class ClusterHealthResponse extends ActionResponse implements StatusToXCo
|
||||
super.readFrom(in);
|
||||
clusterName = in.readString();
|
||||
clusterHealthStatus = ClusterHealthStatus.fromValue(in.readByte());
|
||||
clusterStateHealth = ClusterStateHealth.readClusterHealth(in);
|
||||
clusterStateHealth = new ClusterStateHealth(in);
|
||||
numberOfPendingTasks = in.readInt();
|
||||
timedOut = in.readBoolean();
|
||||
numberOfInFlightFetch = in.readInt();
|
||||
@ -222,50 +222,48 @@ public class ClusterHealthResponse extends ActionResponse implements StatusToXCo
|
||||
return isTimedOut() ? RestStatus.REQUEST_TIMEOUT : RestStatus.OK;
|
||||
}
|
||||
|
||||
static final class Fields {
|
||||
static final String CLUSTER_NAME = "cluster_name";
|
||||
static final String STATUS = "status";
|
||||
static final String TIMED_OUT = "timed_out";
|
||||
static final String NUMBER_OF_NODES = "number_of_nodes";
|
||||
static final String NUMBER_OF_DATA_NODES = "number_of_data_nodes";
|
||||
static final String NUMBER_OF_PENDING_TASKS = "number_of_pending_tasks";
|
||||
static final String NUMBER_OF_IN_FLIGHT_FETCH = "number_of_in_flight_fetch";
|
||||
static final String DELAYED_UNASSIGNED_SHARDS = "delayed_unassigned_shards";
|
||||
static final String TASK_MAX_WAIT_TIME_IN_QUEUE = "task_max_waiting_in_queue";
|
||||
static final String TASK_MAX_WAIT_TIME_IN_QUEUE_IN_MILLIS = "task_max_waiting_in_queue_millis";
|
||||
static final String ACTIVE_SHARDS_PERCENT_AS_NUMBER = "active_shards_percent_as_number";
|
||||
static final String ACTIVE_SHARDS_PERCENT = "active_shards_percent";
|
||||
static final String ACTIVE_PRIMARY_SHARDS = "active_primary_shards";
|
||||
static final String ACTIVE_SHARDS = "active_shards";
|
||||
static final String RELOCATING_SHARDS = "relocating_shards";
|
||||
static final String INITIALIZING_SHARDS = "initializing_shards";
|
||||
static final String UNASSIGNED_SHARDS = "unassigned_shards";
|
||||
static final String INDICES = "indices";
|
||||
}
|
||||
private static final String CLUSTER_NAME = "cluster_name";
|
||||
private static final String STATUS = "status";
|
||||
private static final String TIMED_OUT = "timed_out";
|
||||
private static final String NUMBER_OF_NODES = "number_of_nodes";
|
||||
private static final String NUMBER_OF_DATA_NODES = "number_of_data_nodes";
|
||||
private static final String NUMBER_OF_PENDING_TASKS = "number_of_pending_tasks";
|
||||
private static final String NUMBER_OF_IN_FLIGHT_FETCH = "number_of_in_flight_fetch";
|
||||
private static final String DELAYED_UNASSIGNED_SHARDS = "delayed_unassigned_shards";
|
||||
private static final String TASK_MAX_WAIT_TIME_IN_QUEUE = "task_max_waiting_in_queue";
|
||||
private static final String TASK_MAX_WAIT_TIME_IN_QUEUE_IN_MILLIS = "task_max_waiting_in_queue_millis";
|
||||
private static final String ACTIVE_SHARDS_PERCENT_AS_NUMBER = "active_shards_percent_as_number";
|
||||
private static final String ACTIVE_SHARDS_PERCENT = "active_shards_percent";
|
||||
private static final String ACTIVE_PRIMARY_SHARDS = "active_primary_shards";
|
||||
private static final String ACTIVE_SHARDS = "active_shards";
|
||||
private static final String RELOCATING_SHARDS = "relocating_shards";
|
||||
private static final String INITIALIZING_SHARDS = "initializing_shards";
|
||||
private static final String UNASSIGNED_SHARDS = "unassigned_shards";
|
||||
private static final String INDICES = "indices";
|
||||
|
||||
@Override
|
||||
public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
|
||||
builder.field(Fields.CLUSTER_NAME, getClusterName());
|
||||
builder.field(Fields.STATUS, getStatus().name().toLowerCase(Locale.ROOT));
|
||||
builder.field(Fields.TIMED_OUT, isTimedOut());
|
||||
builder.field(Fields.NUMBER_OF_NODES, getNumberOfNodes());
|
||||
builder.field(Fields.NUMBER_OF_DATA_NODES, getNumberOfDataNodes());
|
||||
builder.field(Fields.ACTIVE_PRIMARY_SHARDS, getActivePrimaryShards());
|
||||
builder.field(Fields.ACTIVE_SHARDS, getActiveShards());
|
||||
builder.field(Fields.RELOCATING_SHARDS, getRelocatingShards());
|
||||
builder.field(Fields.INITIALIZING_SHARDS, getInitializingShards());
|
||||
builder.field(Fields.UNASSIGNED_SHARDS, getUnassignedShards());
|
||||
builder.field(Fields.DELAYED_UNASSIGNED_SHARDS, getDelayedUnassignedShards());
|
||||
builder.field(Fields.NUMBER_OF_PENDING_TASKS, getNumberOfPendingTasks());
|
||||
builder.field(Fields.NUMBER_OF_IN_FLIGHT_FETCH, getNumberOfInFlightFetch());
|
||||
builder.timeValueField(Fields.TASK_MAX_WAIT_TIME_IN_QUEUE_IN_MILLIS, Fields.TASK_MAX_WAIT_TIME_IN_QUEUE, getTaskMaxWaitingTime());
|
||||
builder.percentageField(Fields.ACTIVE_SHARDS_PERCENT_AS_NUMBER, Fields.ACTIVE_SHARDS_PERCENT, getActiveShardsPercent());
|
||||
builder.field(CLUSTER_NAME, getClusterName());
|
||||
builder.field(STATUS, getStatus().name().toLowerCase(Locale.ROOT));
|
||||
builder.field(TIMED_OUT, isTimedOut());
|
||||
builder.field(NUMBER_OF_NODES, getNumberOfNodes());
|
||||
builder.field(NUMBER_OF_DATA_NODES, getNumberOfDataNodes());
|
||||
builder.field(ACTIVE_PRIMARY_SHARDS, getActivePrimaryShards());
|
||||
builder.field(ACTIVE_SHARDS, getActiveShards());
|
||||
builder.field(RELOCATING_SHARDS, getRelocatingShards());
|
||||
builder.field(INITIALIZING_SHARDS, getInitializingShards());
|
||||
builder.field(UNASSIGNED_SHARDS, getUnassignedShards());
|
||||
builder.field(DELAYED_UNASSIGNED_SHARDS, getDelayedUnassignedShards());
|
||||
builder.field(NUMBER_OF_PENDING_TASKS, getNumberOfPendingTasks());
|
||||
builder.field(NUMBER_OF_IN_FLIGHT_FETCH, getNumberOfInFlightFetch());
|
||||
builder.timeValueField(TASK_MAX_WAIT_TIME_IN_QUEUE_IN_MILLIS, TASK_MAX_WAIT_TIME_IN_QUEUE, getTaskMaxWaitingTime());
|
||||
builder.percentageField(ACTIVE_SHARDS_PERCENT_AS_NUMBER, ACTIVE_SHARDS_PERCENT, getActiveShardsPercent());
|
||||
|
||||
String level = params.param("level", "cluster");
|
||||
boolean outputIndices = "indices".equals(level) || "shards".equals(level);
|
||||
|
||||
if (outputIndices) {
|
||||
builder.startObject(Fields.INDICES);
|
||||
builder.startObject(INDICES);
|
||||
for (ClusterIndexHealth indexHealth : clusterStateHealth.getIndices().values()) {
|
||||
builder.startObject(indexHealth.getIndex());
|
||||
indexHealth.toXContent(builder, params);
|
||||
|
||||
@ -24,7 +24,7 @@ import org.elasticsearch.cluster.routing.IndexRoutingTable;
|
||||
import org.elasticsearch.cluster.routing.IndexShardRoutingTable;
|
||||
import org.elasticsearch.common.io.stream.StreamInput;
|
||||
import org.elasticsearch.common.io.stream.StreamOutput;
|
||||
import org.elasticsearch.common.io.stream.Streamable;
|
||||
import org.elasticsearch.common.io.stream.Writeable;
|
||||
import org.elasticsearch.common.xcontent.ToXContent;
|
||||
import org.elasticsearch.common.xcontent.XContentBuilder;
|
||||
|
||||
@ -34,34 +34,20 @@ import java.util.Iterator;
|
||||
import java.util.Locale;
|
||||
import java.util.Map;
|
||||
|
||||
import static org.elasticsearch.cluster.health.ClusterShardHealth.readClusterShardHealth;
|
||||
|
||||
public final class ClusterIndexHealth implements Iterable<ClusterShardHealth>, Streamable, ToXContent {
|
||||
|
||||
private String index;
|
||||
|
||||
private int numberOfShards;
|
||||
|
||||
private int numberOfReplicas;
|
||||
|
||||
private int activeShards = 0;
|
||||
|
||||
private int relocatingShards = 0;
|
||||
|
||||
private int initializingShards = 0;
|
||||
|
||||
private int unassignedShards = 0;
|
||||
|
||||
private int activePrimaryShards = 0;
|
||||
|
||||
private ClusterHealthStatus status = ClusterHealthStatus.RED;
|
||||
public final class ClusterIndexHealth implements Iterable<ClusterShardHealth>, Writeable, ToXContent {
|
||||
|
||||
private final String index;
|
||||
private final int numberOfShards;
|
||||
private final int numberOfReplicas;
|
||||
private final int activeShards;
|
||||
private final int relocatingShards;
|
||||
private final int initializingShards;
|
||||
private final int unassignedShards;
|
||||
private final int activePrimaryShards;
|
||||
private final ClusterHealthStatus status;
|
||||
private final Map<Integer, ClusterShardHealth> shards = new HashMap<>();
|
||||
|
||||
private ClusterIndexHealth() {
|
||||
}
|
||||
|
||||
public ClusterIndexHealth(IndexMetaData indexMetaData, IndexRoutingTable indexRoutingTable) {
|
||||
public ClusterIndexHealth(final IndexMetaData indexMetaData, final IndexRoutingTable indexRoutingTable) {
|
||||
this.index = indexMetaData.getIndex().getName();
|
||||
this.numberOfShards = indexMetaData.getNumberOfShards();
|
||||
this.numberOfReplicas = indexMetaData.getNumberOfReplicas();
|
||||
@ -72,26 +58,55 @@ public final class ClusterIndexHealth implements Iterable<ClusterShardHealth>, S
|
||||
}
|
||||
|
||||
// update the index status
|
||||
status = ClusterHealthStatus.GREEN;
|
||||
|
||||
ClusterHealthStatus computeStatus = ClusterHealthStatus.GREEN;
|
||||
int computeActivePrimaryShards = 0;
|
||||
int computeActiveShards = 0;
|
||||
int computeRelocatingShards = 0;
|
||||
int computeInitializingShards = 0;
|
||||
int computeUnassignedShards = 0;
|
||||
for (ClusterShardHealth shardHealth : shards.values()) {
|
||||
if (shardHealth.isPrimaryActive()) {
|
||||
activePrimaryShards++;
|
||||
computeActivePrimaryShards++;
|
||||
}
|
||||
activeShards += shardHealth.getActiveShards();
|
||||
relocatingShards += shardHealth.getRelocatingShards();
|
||||
initializingShards += shardHealth.getInitializingShards();
|
||||
unassignedShards += shardHealth.getUnassignedShards();
|
||||
computeActiveShards += shardHealth.getActiveShards();
|
||||
computeRelocatingShards += shardHealth.getRelocatingShards();
|
||||
computeInitializingShards += shardHealth.getInitializingShards();
|
||||
computeUnassignedShards += shardHealth.getUnassignedShards();
|
||||
|
||||
if (shardHealth.getStatus() == ClusterHealthStatus.RED) {
|
||||
status = ClusterHealthStatus.RED;
|
||||
} else if (shardHealth.getStatus() == ClusterHealthStatus.YELLOW && status != ClusterHealthStatus.RED) {
|
||||
computeStatus = ClusterHealthStatus.RED;
|
||||
} else if (shardHealth.getStatus() == ClusterHealthStatus.YELLOW && computeStatus != ClusterHealthStatus.RED) {
|
||||
// do not override an existing red
|
||||
status = ClusterHealthStatus.YELLOW;
|
||||
computeStatus = ClusterHealthStatus.YELLOW;
|
||||
}
|
||||
}
|
||||
if (shards.isEmpty()) { // might be since none has been created yet (two phase index creation)
|
||||
status = ClusterHealthStatus.RED;
|
||||
computeStatus = ClusterHealthStatus.RED;
|
||||
}
|
||||
|
||||
this.status = computeStatus;
|
||||
this.activePrimaryShards = computeActivePrimaryShards;
|
||||
this.activeShards = computeActiveShards;
|
||||
this.relocatingShards = computeRelocatingShards;
|
||||
this.initializingShards = computeInitializingShards;
|
||||
this.unassignedShards = computeUnassignedShards;
|
||||
}
|
||||
|
||||
public ClusterIndexHealth(final StreamInput in) throws IOException {
|
||||
index = in.readString();
|
||||
numberOfShards = in.readVInt();
|
||||
numberOfReplicas = in.readVInt();
|
||||
activePrimaryShards = in.readVInt();
|
||||
activeShards = in.readVInt();
|
||||
relocatingShards = in.readVInt();
|
||||
initializingShards = in.readVInt();
|
||||
unassignedShards = in.readVInt();
|
||||
status = ClusterHealthStatus.fromValue(in.readByte());
|
||||
|
||||
int size = in.readVInt();
|
||||
for (int i = 0; i < size; i++) {
|
||||
ClusterShardHealth shardHealth = new ClusterShardHealth(in);
|
||||
shards.put(shardHealth.getId(), shardHealth);
|
||||
}
|
||||
}
|
||||
|
||||
@ -140,33 +155,8 @@ public final class ClusterIndexHealth implements Iterable<ClusterShardHealth>, S
|
||||
return shards.values().iterator();
|
||||
}
|
||||
|
||||
public static ClusterIndexHealth readClusterIndexHealth(StreamInput in) throws IOException {
|
||||
ClusterIndexHealth indexHealth = new ClusterIndexHealth();
|
||||
indexHealth.readFrom(in);
|
||||
return indexHealth;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void readFrom(StreamInput in) throws IOException {
|
||||
index = in.readString();
|
||||
numberOfShards = in.readVInt();
|
||||
numberOfReplicas = in.readVInt();
|
||||
activePrimaryShards = in.readVInt();
|
||||
activeShards = in.readVInt();
|
||||
relocatingShards = in.readVInt();
|
||||
initializingShards = in.readVInt();
|
||||
unassignedShards = in.readVInt();
|
||||
status = ClusterHealthStatus.fromValue(in.readByte());
|
||||
|
||||
int size = in.readVInt();
|
||||
for (int i = 0; i < size; i++) {
|
||||
ClusterShardHealth shardHealth = readClusterShardHealth(in);
|
||||
shards.put(shardHealth.getId(), shardHealth);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void writeTo(StreamOutput out) throws IOException {
|
||||
public void writeTo(final StreamOutput out) throws IOException {
|
||||
out.writeString(index);
|
||||
out.writeVInt(numberOfShards);
|
||||
out.writeVInt(numberOfReplicas);
|
||||
@ -183,42 +173,40 @@ public final class ClusterIndexHealth implements Iterable<ClusterShardHealth>, S
|
||||
}
|
||||
}
|
||||
|
||||
static final class Fields {
|
||||
static final String STATUS = "status";
|
||||
static final String NUMBER_OF_SHARDS = "number_of_shards";
|
||||
static final String NUMBER_OF_REPLICAS = "number_of_replicas";
|
||||
static final String ACTIVE_PRIMARY_SHARDS = "active_primary_shards";
|
||||
static final String ACTIVE_SHARDS = "active_shards";
|
||||
static final String RELOCATING_SHARDS = "relocating_shards";
|
||||
static final String INITIALIZING_SHARDS = "initializing_shards";
|
||||
static final String UNASSIGNED_SHARDS = "unassigned_shards";
|
||||
static final String SHARDS = "shards";
|
||||
static final String PRIMARY_ACTIVE = "primary_active";
|
||||
}
|
||||
private static final String STATUS = "status";
|
||||
private static final String NUMBER_OF_SHARDS = "number_of_shards";
|
||||
private static final String NUMBER_OF_REPLICAS = "number_of_replicas";
|
||||
private static final String ACTIVE_PRIMARY_SHARDS = "active_primary_shards";
|
||||
private static final String ACTIVE_SHARDS = "active_shards";
|
||||
private static final String RELOCATING_SHARDS = "relocating_shards";
|
||||
private static final String INITIALIZING_SHARDS = "initializing_shards";
|
||||
private static final String UNASSIGNED_SHARDS = "unassigned_shards";
|
||||
private static final String SHARDS = "shards";
|
||||
private static final String PRIMARY_ACTIVE = "primary_active";
|
||||
|
||||
@Override
|
||||
public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
|
||||
builder.field(Fields.STATUS, getStatus().name().toLowerCase(Locale.ROOT));
|
||||
builder.field(Fields.NUMBER_OF_SHARDS, getNumberOfShards());
|
||||
builder.field(Fields.NUMBER_OF_REPLICAS, getNumberOfReplicas());
|
||||
builder.field(Fields.ACTIVE_PRIMARY_SHARDS, getActivePrimaryShards());
|
||||
builder.field(Fields.ACTIVE_SHARDS, getActiveShards());
|
||||
builder.field(Fields.RELOCATING_SHARDS, getRelocatingShards());
|
||||
builder.field(Fields.INITIALIZING_SHARDS, getInitializingShards());
|
||||
builder.field(Fields.UNASSIGNED_SHARDS, getUnassignedShards());
|
||||
public XContentBuilder toXContent(final XContentBuilder builder, final Params params) throws IOException {
|
||||
builder.field(STATUS, getStatus().name().toLowerCase(Locale.ROOT));
|
||||
builder.field(NUMBER_OF_SHARDS, getNumberOfShards());
|
||||
builder.field(NUMBER_OF_REPLICAS, getNumberOfReplicas());
|
||||
builder.field(ACTIVE_PRIMARY_SHARDS, getActivePrimaryShards());
|
||||
builder.field(ACTIVE_SHARDS, getActiveShards());
|
||||
builder.field(RELOCATING_SHARDS, getRelocatingShards());
|
||||
builder.field(INITIALIZING_SHARDS, getInitializingShards());
|
||||
builder.field(UNASSIGNED_SHARDS, getUnassignedShards());
|
||||
|
||||
if ("shards".equals(params.param("level", "indices"))) {
|
||||
builder.startObject(Fields.SHARDS);
|
||||
builder.startObject(SHARDS);
|
||||
|
||||
for (ClusterShardHealth shardHealth : shards.values()) {
|
||||
builder.startObject(Integer.toString(shardHealth.getId()));
|
||||
|
||||
builder.field(Fields.STATUS, shardHealth.getStatus().name().toLowerCase(Locale.ROOT));
|
||||
builder.field(Fields.PRIMARY_ACTIVE, shardHealth.isPrimaryActive());
|
||||
builder.field(Fields.ACTIVE_SHARDS, shardHealth.getActiveShards());
|
||||
builder.field(Fields.RELOCATING_SHARDS, shardHealth.getRelocatingShards());
|
||||
builder.field(Fields.INITIALIZING_SHARDS, shardHealth.getInitializingShards());
|
||||
builder.field(Fields.UNASSIGNED_SHARDS, shardHealth.getUnassignedShards());
|
||||
builder.field(STATUS, shardHealth.getStatus().name().toLowerCase(Locale.ROOT));
|
||||
builder.field(PRIMARY_ACTIVE, shardHealth.isPrimaryActive());
|
||||
builder.field(ACTIVE_SHARDS, shardHealth.getActiveShards());
|
||||
builder.field(RELOCATING_SHARDS, shardHealth.getRelocatingShards());
|
||||
builder.field(INITIALIZING_SHARDS, shardHealth.getInitializingShards());
|
||||
builder.field(UNASSIGNED_SHARDS, shardHealth.getUnassignedShards());
|
||||
|
||||
builder.endObject();
|
||||
}
|
||||
|
||||
@ -23,57 +23,69 @@ import org.elasticsearch.cluster.routing.IndexShardRoutingTable;
|
||||
import org.elasticsearch.cluster.routing.ShardRouting;
|
||||
import org.elasticsearch.common.io.stream.StreamInput;
|
||||
import org.elasticsearch.common.io.stream.StreamOutput;
|
||||
import org.elasticsearch.common.io.stream.Streamable;
|
||||
import org.elasticsearch.common.io.stream.Writeable;
|
||||
|
||||
import java.io.IOException;
|
||||
|
||||
public final class ClusterShardHealth implements Streamable {
|
||||
public final class ClusterShardHealth implements Writeable {
|
||||
|
||||
private int shardId;
|
||||
private final int shardId;
|
||||
private final ClusterHealthStatus status;
|
||||
private final int activeShards;
|
||||
private final int relocatingShards;
|
||||
private final int initializingShards;
|
||||
private final int unassignedShards;
|
||||
private final boolean primaryActive;
|
||||
|
||||
ClusterHealthStatus status = ClusterHealthStatus.RED;
|
||||
|
||||
private int activeShards = 0;
|
||||
|
||||
private int relocatingShards = 0;
|
||||
|
||||
private int initializingShards = 0;
|
||||
|
||||
private int unassignedShards = 0;
|
||||
|
||||
private boolean primaryActive = false;
|
||||
|
||||
private ClusterShardHealth() {
|
||||
|
||||
}
|
||||
|
||||
public ClusterShardHealth(int shardId, final IndexShardRoutingTable shardRoutingTable) {
|
||||
public ClusterShardHealth(final int shardId, final IndexShardRoutingTable shardRoutingTable) {
|
||||
this.shardId = shardId;
|
||||
int computeActiveShards = 0;
|
||||
int computeRelocatingShards = 0;
|
||||
int computeInitializingShards = 0;
|
||||
int computeUnassignedShards = 0;
|
||||
boolean computePrimaryActive = false;
|
||||
for (ShardRouting shardRouting : shardRoutingTable) {
|
||||
if (shardRouting.active()) {
|
||||
activeShards++;
|
||||
computeActiveShards++;
|
||||
if (shardRouting.relocating()) {
|
||||
// the shard is relocating, the one it is relocating to will be in initializing state, so we don't count it
|
||||
relocatingShards++;
|
||||
computeRelocatingShards++;
|
||||
}
|
||||
if (shardRouting.primary()) {
|
||||
primaryActive = true;
|
||||
computePrimaryActive = true;
|
||||
}
|
||||
} else if (shardRouting.initializing()) {
|
||||
initializingShards++;
|
||||
computeInitializingShards++;
|
||||
} else if (shardRouting.unassigned()) {
|
||||
unassignedShards++;
|
||||
computeUnassignedShards++;
|
||||
}
|
||||
}
|
||||
if (primaryActive) {
|
||||
if (activeShards == shardRoutingTable.size()) {
|
||||
status = ClusterHealthStatus.GREEN;
|
||||
ClusterHealthStatus computeStatus;
|
||||
if (computePrimaryActive) {
|
||||
if (computeActiveShards == shardRoutingTable.size()) {
|
||||
computeStatus = ClusterHealthStatus.GREEN;
|
||||
} else {
|
||||
status = ClusterHealthStatus.YELLOW;
|
||||
computeStatus = ClusterHealthStatus.YELLOW;
|
||||
}
|
||||
} else {
|
||||
status = ClusterHealthStatus.RED;
|
||||
computeStatus = ClusterHealthStatus.RED;
|
||||
}
|
||||
this.status = computeStatus;
|
||||
this.activeShards = computeActiveShards;
|
||||
this.relocatingShards = computeRelocatingShards;
|
||||
this.initializingShards = computeInitializingShards;
|
||||
this.unassignedShards = computeUnassignedShards;
|
||||
this.primaryActive = computePrimaryActive;
|
||||
}
|
||||
|
||||
public ClusterShardHealth(final StreamInput in) throws IOException {
|
||||
shardId = in.readVInt();
|
||||
status = ClusterHealthStatus.fromValue(in.readByte());
|
||||
activeShards = in.readVInt();
|
||||
relocatingShards = in.readVInt();
|
||||
initializingShards = in.readVInt();
|
||||
unassignedShards = in.readVInt();
|
||||
primaryActive = in.readBoolean();
|
||||
}
|
||||
|
||||
public int getId() {
|
||||
@ -104,25 +116,8 @@ public final class ClusterShardHealth implements Streamable {
|
||||
return unassignedShards;
|
||||
}
|
||||
|
||||
static ClusterShardHealth readClusterShardHealth(StreamInput in) throws IOException {
|
||||
ClusterShardHealth ret = new ClusterShardHealth();
|
||||
ret.readFrom(in);
|
||||
return ret;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void readFrom(StreamInput in) throws IOException {
|
||||
shardId = in.readVInt();
|
||||
status = ClusterHealthStatus.fromValue(in.readByte());
|
||||
activeShards = in.readVInt();
|
||||
relocatingShards = in.readVInt();
|
||||
initializingShards = in.readVInt();
|
||||
unassignedShards = in.readVInt();
|
||||
primaryActive = in.readBoolean();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void writeTo(StreamOutput out) throws IOException {
|
||||
public void writeTo(final StreamOutput out) throws IOException {
|
||||
out.writeVInt(shardId);
|
||||
out.writeByte(status.value());
|
||||
out.writeVInt(activeShards);
|
||||
|
||||
@ -27,7 +27,7 @@ import org.elasticsearch.cluster.routing.RoutingTable;
|
||||
import org.elasticsearch.cluster.routing.ShardRouting;
|
||||
import org.elasticsearch.common.io.stream.StreamInput;
|
||||
import org.elasticsearch.common.io.stream.StreamOutput;
|
||||
import org.elasticsearch.common.io.stream.Streamable;
|
||||
import org.elasticsearch.common.io.stream.Writeable;
|
||||
import org.elasticsearch.rest.RestStatus;
|
||||
|
||||
import java.io.IOException;
|
||||
@ -37,29 +37,18 @@ import java.util.Iterator;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
import static org.elasticsearch.cluster.health.ClusterIndexHealth.readClusterIndexHealth;
|
||||
public final class ClusterStateHealth implements Iterable<ClusterIndexHealth>, Writeable {
|
||||
|
||||
public final class ClusterStateHealth implements Iterable<ClusterIndexHealth>, Streamable {
|
||||
private int numberOfNodes = 0;
|
||||
private int numberOfDataNodes = 0;
|
||||
private int activeShards = 0;
|
||||
private int relocatingShards = 0;
|
||||
private int activePrimaryShards = 0;
|
||||
private int initializingShards = 0;
|
||||
private int unassignedShards = 0;
|
||||
private double activeShardsPercent = 100;
|
||||
private ClusterHealthStatus status = ClusterHealthStatus.RED;
|
||||
private Map<String, ClusterIndexHealth> indices = new HashMap<>();
|
||||
|
||||
public static ClusterStateHealth readClusterHealth(StreamInput in) throws IOException {
|
||||
ClusterStateHealth clusterStateHealth = new ClusterStateHealth();
|
||||
clusterStateHealth.readFrom(in);
|
||||
return clusterStateHealth;
|
||||
}
|
||||
|
||||
ClusterStateHealth() {
|
||||
// only intended for serialization
|
||||
}
|
||||
private final int numberOfNodes;
|
||||
private final int numberOfDataNodes;
|
||||
private final int activeShards;
|
||||
private final int relocatingShards;
|
||||
private final int activePrimaryShards;
|
||||
private final int initializingShards;
|
||||
private final int unassignedShards;
|
||||
private final double activeShardsPercent;
|
||||
private final ClusterHealthStatus status;
|
||||
private final Map<String, ClusterIndexHealth> indices = new HashMap<>();
|
||||
|
||||
/**
|
||||
* Creates a new <code>ClusterStateHealth</code> instance based on cluster meta data and its routing table as a convenience.
|
||||
@ -67,7 +56,7 @@ public final class ClusterStateHealth implements Iterable<ClusterIndexHealth>, S
|
||||
* @param clusterMetaData Current cluster meta data. Must not be null.
|
||||
* @param routingTables Current routing table. Must not be null.
|
||||
*/
|
||||
public ClusterStateHealth(MetaData clusterMetaData, RoutingTable routingTables) {
|
||||
public ClusterStateHealth(final MetaData clusterMetaData, final RoutingTable routingTables) {
|
||||
this(ClusterState.builder(ClusterName.DEFAULT).metaData(clusterMetaData).routingTable(routingTables).build());
|
||||
}
|
||||
|
||||
@ -76,7 +65,7 @@ public final class ClusterStateHealth implements Iterable<ClusterIndexHealth>, S
|
||||
*
|
||||
* @param clusterState The current cluster state. Must not be null.
|
||||
*/
|
||||
public ClusterStateHealth(ClusterState clusterState) {
|
||||
public ClusterStateHealth(final ClusterState clusterState) {
|
||||
this(clusterState, clusterState.metaData().getConcreteAllIndices());
|
||||
}
|
||||
|
||||
@ -86,7 +75,7 @@ public final class ClusterStateHealth implements Iterable<ClusterIndexHealth>, S
|
||||
* @param clusterState The current cluster state. Must not be null.
|
||||
* @param concreteIndices An array of index names to consider. Must not be null but may be empty.
|
||||
*/
|
||||
public ClusterStateHealth(ClusterState clusterState, String[] concreteIndices) {
|
||||
public ClusterStateHealth(final ClusterState clusterState, final String[] concreteIndices) {
|
||||
numberOfNodes = clusterState.nodes().getSize();
|
||||
numberOfDataNodes = clusterState.nodes().getDataNodes().size();
|
||||
|
||||
@ -102,27 +91,39 @@ public final class ClusterStateHealth implements Iterable<ClusterIndexHealth>, S
|
||||
indices.put(indexHealth.getIndex(), indexHealth);
|
||||
}
|
||||
|
||||
status = ClusterHealthStatus.GREEN;
|
||||
ClusterHealthStatus computeStatus = ClusterHealthStatus.GREEN;
|
||||
int computeActivePrimaryShards = 0;
|
||||
int computeActiveShards = 0;
|
||||
int computeRelocatingShards = 0;
|
||||
int computeInitializingShards = 0;
|
||||
int computeUnassignedShards = 0;
|
||||
|
||||
for (ClusterIndexHealth indexHealth : indices.values()) {
|
||||
activePrimaryShards += indexHealth.getActivePrimaryShards();
|
||||
activeShards += indexHealth.getActiveShards();
|
||||
relocatingShards += indexHealth.getRelocatingShards();
|
||||
initializingShards += indexHealth.getInitializingShards();
|
||||
unassignedShards += indexHealth.getUnassignedShards();
|
||||
computeActivePrimaryShards += indexHealth.getActivePrimaryShards();
|
||||
computeActiveShards += indexHealth.getActiveShards();
|
||||
computeRelocatingShards += indexHealth.getRelocatingShards();
|
||||
computeInitializingShards += indexHealth.getInitializingShards();
|
||||
computeUnassignedShards += indexHealth.getUnassignedShards();
|
||||
if (indexHealth.getStatus() == ClusterHealthStatus.RED) {
|
||||
status = ClusterHealthStatus.RED;
|
||||
} else if (indexHealth.getStatus() == ClusterHealthStatus.YELLOW && status != ClusterHealthStatus.RED) {
|
||||
status = ClusterHealthStatus.YELLOW;
|
||||
computeStatus = ClusterHealthStatus.RED;
|
||||
} else if (indexHealth.getStatus() == ClusterHealthStatus.YELLOW && computeStatus != ClusterHealthStatus.RED) {
|
||||
computeStatus = ClusterHealthStatus.YELLOW;
|
||||
}
|
||||
}
|
||||
|
||||
if (clusterState.blocks().hasGlobalBlock(RestStatus.SERVICE_UNAVAILABLE)) {
|
||||
status = ClusterHealthStatus.RED;
|
||||
computeStatus = ClusterHealthStatus.RED;
|
||||
}
|
||||
|
||||
this.status = computeStatus;
|
||||
this.activePrimaryShards = computeActivePrimaryShards;
|
||||
this.activeShards = computeActiveShards;
|
||||
this.relocatingShards = computeRelocatingShards;
|
||||
this.initializingShards = computeInitializingShards;
|
||||
this.unassignedShards = computeUnassignedShards;
|
||||
|
||||
// shortcut on green
|
||||
if (status.equals(ClusterHealthStatus.GREEN)) {
|
||||
if (computeStatus.equals(ClusterHealthStatus.GREEN)) {
|
||||
this.activeShardsPercent = 100;
|
||||
} else {
|
||||
List<ShardRouting> shardRoutings = clusterState.getRoutingTable().allShards();
|
||||
@ -136,6 +137,23 @@ public final class ClusterStateHealth implements Iterable<ClusterIndexHealth>, S
|
||||
}
|
||||
}
|
||||
|
||||
public ClusterStateHealth(final StreamInput in) throws IOException {
|
||||
activePrimaryShards = in.readVInt();
|
||||
activeShards = in.readVInt();
|
||||
relocatingShards = in.readVInt();
|
||||
initializingShards = in.readVInt();
|
||||
unassignedShards = in.readVInt();
|
||||
numberOfNodes = in.readVInt();
|
||||
numberOfDataNodes = in.readVInt();
|
||||
status = ClusterHealthStatus.fromValue(in.readByte());
|
||||
int size = in.readVInt();
|
||||
for (int i = 0; i < size; i++) {
|
||||
ClusterIndexHealth indexHealth = new ClusterIndexHealth(in);
|
||||
indices.put(indexHealth.getIndex(), indexHealth);
|
||||
}
|
||||
activeShardsPercent = in.readDouble();
|
||||
}
|
||||
|
||||
public int getActiveShards() {
|
||||
return activeShards;
|
||||
}
|
||||
@ -182,25 +200,7 @@ public final class ClusterStateHealth implements Iterable<ClusterIndexHealth>, S
|
||||
}
|
||||
|
||||
@Override
|
||||
public void readFrom(StreamInput in) throws IOException {
|
||||
activePrimaryShards = in.readVInt();
|
||||
activeShards = in.readVInt();
|
||||
relocatingShards = in.readVInt();
|
||||
initializingShards = in.readVInt();
|
||||
unassignedShards = in.readVInt();
|
||||
numberOfNodes = in.readVInt();
|
||||
numberOfDataNodes = in.readVInt();
|
||||
status = ClusterHealthStatus.fromValue(in.readByte());
|
||||
int size = in.readVInt();
|
||||
for (int i = 0; i < size; i++) {
|
||||
ClusterIndexHealth indexHealth = readClusterIndexHealth(in);
|
||||
indices.put(indexHealth.getIndex(), indexHealth);
|
||||
}
|
||||
activeShardsPercent = in.readDouble();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void writeTo(StreamOutput out) throws IOException {
|
||||
public void writeTo(final StreamOutput out) throws IOException {
|
||||
out.writeVInt(activePrimaryShards);
|
||||
out.writeVInt(activeShards);
|
||||
out.writeVInt(relocatingShards);
|
||||
|
||||
@ -170,7 +170,7 @@ public class ClusterStateHealthTests extends ESTestCase {
|
||||
BytesStreamOutput out = new BytesStreamOutput();
|
||||
clusterStateHealth.writeTo(out);
|
||||
StreamInput in = StreamInput.wrap(out.bytes());
|
||||
clusterStateHealth = ClusterStateHealth.readClusterHealth(in);
|
||||
clusterStateHealth = new ClusterStateHealth(in);
|
||||
}
|
||||
return clusterStateHealth;
|
||||
}
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user