honeymoose
/
OpenSearch
mirror of https://github.com/honeymoose/OpenSearch.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Tribe nodes should apply cluster state updates in batches 

This commit applies the general mechanism for applying cluster state
updates in batches to tribe nodes.

Relates #14899, relates #14725
This commit is contained in:
Jason Tedor 2015-11-24 20:39:26 -05:00
parent 6f2c36dcb7
commit 53b3cd83a5
1 changed files with 157 additions and 118 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

67
core/src/main/java/org/elasticsearch/tribe/TribeService.java
View File

@ -20,14 +20,14 @@
package org.elasticsearch.tribe;
import com.carrotsearch.hppc.cursors.ObjectObjectCursor;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.action.support.master.TransportMasterNodeReadAction;
import org.elasticsearch.cluster.ClusterChangedEvent;
import org.elasticsearch.cluster.ClusterService;
import org.elasticsearch.cluster.ClusterState;
import org.elasticsearch.cluster.ClusterStateListener;
import org.elasticsearch.cluster.ClusterStateUpdateTask;
import org.elasticsearch.cluster.ClusterStateTaskConfig;
import org.elasticsearch.cluster.ClusterStateTaskExecutor;
import org.elasticsearch.cluster.block.ClusterBlock;
import org.elasticsearch.cluster.block.ClusterBlockLevel;
import org.elasticsearch.cluster.block.ClusterBlocks;
@ -37,6 +37,7 @@ import org.elasticsearch.cluster.node.DiscoveryNode;
import org.elasticsearch.cluster.node.DiscoveryNodes;
import org.elasticsearch.cluster.routing.IndexRoutingTable;
import org.elasticsearch.cluster.routing.RoutingTable;
import org.elasticsearch.common.Priority;
import org.elasticsearch.common.Strings;
import org.elasticsearch.common.component.AbstractLifecycleComponent;
import org.elasticsearch.common.inject.Inject;
@ -208,26 +209,61 @@ public class TribeService extends AbstractLifecycleComponent<TribeService> {
}
}
class TribeClusterStateListener implements ClusterStateListener {
class TribeClusterStateListener implements ClusterStateListener {
private final String tribeName;
private final TribeNodeClusterStateTaskExecutor executor;
TribeClusterStateListener(Node tribeNode) {
this.tribeName = tribeNode.settings().get(TRIBE_NAME);
String tribeName = tribeNode.settings().get(TRIBE_NAME);
this.tribeName = tribeName;
executor = new TribeNodeClusterStateTaskExecutor(tribeName);
}
@Override
public void clusterChanged(final ClusterChangedEvent event) {
logger.debug("[{}] received cluster event, [{}]", tribeName, event.source());
clusterService.submitStateUpdateTask("cluster event from " + tribeName + ", " + event.source(), new ClusterStateUpdateTask() {
clusterService.submitStateUpdateTask(
"cluster event from " + tribeName + ", " + event.source(),
event,
ClusterStateTaskConfig.build(Priority.NORMAL),
executor,
(source, t) -> logger.warn("failed to process [{}]", t, source));
}
}
class TribeNodeClusterStateTaskExecutor implements ClusterStateTaskExecutor<ClusterChangedEvent> {
private final String tribeName;
TribeNodeClusterStateTaskExecutor(String tribeName) {
this.tribeName = tribeName;
}
@Override
public boolean runOnlyOnMaster() {
return false;
}
@Override
public ClusterState execute(ClusterState currentState) throws Exception {
ClusterState tribeState = event.state();
public BatchResult<ClusterChangedEvent> execute(ClusterState currentState, List<ClusterChangedEvent> tasks) throws Exception {
ClusterState accumulator = ClusterState.builder(currentState).build();
BatchResult.Builder<ClusterChangedEvent> builder = BatchResult.builder();
try {
// we only need to apply the latest cluster state update
accumulator = applyUpdate(accumulator, tasks.get(tasks.size() - 1));
builder.successes(tasks);
} catch (Throwable t) {
builder.failures(tasks, t);
}
return builder.build(accumulator);
}
private ClusterState applyUpdate(ClusterState currentState, ClusterChangedEvent task) {
boolean clusterStateChanged = false;
ClusterState tribeState = task.state();
DiscoveryNodes.Builder nodes = DiscoveryNodes.builder(currentState.nodes());
// -- merge nodes
// go over existing nodes, and see if they need to be removed
@ -235,6 +271,7 @@ public class TribeService extends AbstractLifecycleComponent<TribeService> {
String markedTribeName = discoNode.attributes().get(TRIBE_NAME);
if (markedTribeName != null && markedTribeName.equals(tribeName)) {
if (tribeState.nodes().get(discoNode.id()) == null) {
clusterStateChanged = true;
logger.info("[{}] removing node [{}]", tribeName, discoNode);
nodes.remove(discoNode.id());
}
@ -250,6 +287,7 @@ public class TribeService extends AbstractLifecycleComponent<TribeService> {
}
tribeAttr.put(TRIBE_NAME, tribeName);
DiscoveryNode discoNode = new DiscoveryNode(tribe.name(), tribe.id(), tribe.getHostName(), tribe.getHostAddress(), tribe.address(), unmodifiableMap(tribeAttr), tribe.version());
clusterStateChanged = true;
logger.info("[{}] adding node [{}]", tribeName, discoNode);
nodes.put(discoNode);
}
@ -264,6 +302,7 @@ public class TribeService extends AbstractLifecycleComponent<TribeService> {
String markedTribeName = index.getSettings().get(TRIBE_NAME);
if (markedTribeName != null && markedTribeName.equals(tribeName)) {
IndexMetaData tribeIndex = tribeState.metaData().index(index.getIndex());
clusterStateChanged = true;
if (tribeIndex == null || tribeIndex.getState() == IndexMetaData.State.CLOSE) {
logger.info("[{}] removing index [{}]", tribeName, index.getIndex());
removeIndex(blocks, metaData, routingTable, index);
@ -287,6 +326,7 @@ public class TribeService extends AbstractLifecycleComponent<TribeService> {
if (indexMetaData == null) {
if (!droppedIndices.contains(tribeIndex.getIndex())) {
// a new index, add it, and add the tribe name as a setting
clusterStateChanged = true;
logger.info("[{}] adding index [{}]", tribeName, tribeIndex.getIndex());
addNewIndex(tribeState, blocks, metaData, routingTable, tribeIndex);
}
@ -298,6 +338,7 @@ public class TribeService extends AbstractLifecycleComponent<TribeService> {
// we chose any tribe, carry on
} else if (ON_CONFLICT_DROP.equals(onConflict)) {
// drop the indices, there is a conflict
clusterStateChanged = true;
logger.info("[{}] dropping index [{}] due to conflict with [{}]", tribeName, tribeIndex.getIndex(), existingFromTribe);
removeIndex(blocks, metaData, routingTable, tribeIndex);
droppedIndices.add(tribeIndex.getIndex());
@ -306,6 +347,7 @@ public class TribeService extends AbstractLifecycleComponent<TribeService> {
String preferredTribeName = onConflict.substring(ON_CONFLICT_PREFER.length());
if (tribeName.equals(preferredTribeName)) {
// the new one is hte preferred one, replace...
clusterStateChanged = true;
logger.info("[{}] adding index [{}], preferred over [{}]", tribeName, tribeIndex.getIndex(), existingFromTribe);
removeIndex(blocks, metaData, routingTable, tribeIndex);
addNewIndex(tribeState, blocks, metaData, routingTable, tribeIndex);
@ -315,8 +357,12 @@ public class TribeService extends AbstractLifecycleComponent<TribeService> {
}
}
if (!clusterStateChanged) {
return currentState;
} else {
return ClusterState.builder(currentState).incrementVersion().blocks(blocks).nodes(nodes).metaData(metaData).routingTable(routingTable.build()).build();
}
}
private void removeIndex(ClusterBlocks.Builder blocks, MetaData.Builder metaData, RoutingTable.Builder routingTable, IndexMetaData index) {
metaData.remove(index.getIndex());
@ -338,12 +384,5 @@ public class TribeService extends AbstractLifecycleComponent<TribeService> {
blocks.addIndexBlock(tribeIndex.getIndex(), IndexMetaData.INDEX_WRITE_BLOCK);
}
}
@Override
public void onFailure(String source, Throwable t) {
logger.warn("failed to process [{}]", t, source);
}
});
}
}
}