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

Jump to bottom

Add support for merging custom meta data in tribe node #21552

Merged
merged 4 commits into from Nov 21, 2016
Merged

Add support for merging custom meta data in tribe node #21552

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
5211fcb
Add support for merging custom meta data in tribe node
areek Nov 14, 2016
cec9960
Simplify updating tribe state
areek Nov 17, 2016
16d6500
Add tests for merging multiple custom metadata types in tribe node
areek Nov 17, 2016
ae4c137
cleanup merging custom md logic in tribe service
areek Nov 18, 2016
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
31 changes: 31 additions & 0 deletions core/src/main/java/org/elasticsearch/cluster/ClusterChangedEvent.java
View file
Open in desktop
Expand Up @@ -20,17 +20,21 @@
package org.elasticsearch.cluster;

import com.carrotsearch.hppc.cursors.ObjectCursor;
import com.carrotsearch.hppc.cursors.ObjectObjectCursor;
import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.cluster.metadata.IndexGraveyard;
import org.elasticsearch.cluster.metadata.MetaData;
import org.elasticsearch.cluster.node.DiscoveryNodes;
import org.elasticsearch.common.collect.ImmutableOpenMap;
import org.elasticsearch.gateway.GatewayService;
import org.elasticsearch.index.Index;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Objects;
import java.util.Set;
import java.util.stream.Collectors;

/**
Expand Down Expand Up @@ -143,6 +147,33 @@ public boolean metaDataChanged() {
return state.metaData() != previousState.metaData();
}

/**
* Returns a set of custom meta data types when any custom metadata for the cluster has changed
* between the previous cluster state and the new cluster state. custom meta data types are
* returned iff they have been added, updated or removed between the previous and the current state
*/
public Set<String> changedCustomMetaDataSet() {
Set<String> result = new HashSet<>();
ImmutableOpenMap<String, MetaData.Custom> currentCustoms = state.metaData().customs();
ImmutableOpenMap<String, MetaData.Custom> previousCustoms = previousState.metaData().customs();
if (currentCustoms.equals(previousCustoms) == false) {
for (ObjectObjectCursor<String, MetaData.Custom> currentCustomMetaData : currentCustoms) {
// new custom md added or existing custom md changed
if (previousCustoms.containsKey(currentCustomMetaData.key) == false
|| currentCustomMetaData.value.equals(previousCustoms.get(currentCustomMetaData.key)) == false) {
result.add(currentCustomMetaData.key);
}
}
// existing custom md deleted
for (ObjectObjectCursor<String, MetaData.Custom> previousCustomMetaData : previousCustoms) {
if (currentCustoms.containsKey(previousCustomMetaData.key) == false) {
result.add(previousCustomMetaData.key);
}
}
}
return result;
}

/**
* Returns <code>true</code> iff the {@link IndexMetaData} for a given index
* has changed between the previous cluster state and the new cluster state.
Expand Down
129 changes: 106 additions & 23 deletions core/src/main/java/org/elasticsearch/tribe/TribeService.java
View file
Open in desktop
Expand Up @@ -58,7 +58,6 @@
import org.elasticsearch.env.NodeEnvironment;
import org.elasticsearch.gateway.GatewayService;
import org.elasticsearch.node.Node;
import org.elasticsearch.plugins.Plugin;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.transport.TransportSettings;

Expand All @@ -72,6 +71,7 @@
import java.util.Set;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.function.Function;
import java.util.stream.Collectors;

import static java.util.Collections.unmodifiableMap;

Expand Down Expand Up @@ -134,6 +134,29 @@ public static Settings processSettings(Settings settings) {
return sb.build();
}

/**
* Interface to allow merging {@link org.elasticsearch.cluster.metadata.MetaData.Custom} in tribe node
* When multiple Mergable Custom metadata of the same type is found (from underlying clusters), the
* Custom metadata will be merged using {@link #merge(MetaData.Custom)} and the result will be stored
* in the tribe cluster state
*
* @param <T> type of custom meta data
*/
interface MergableCustomMetaData<T extends MetaData.Custom> {

/**
* Merges this custom metadata with other, returning either this or <code>other</code> custom metadata
Copy link
Contributor

Choose a reason for hiding this comment

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

nit: can we document that the neither original customs should be changed by this method?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

added it to the docs

* for tribe cluster state. This method should not mutate either <code>this</code> or the
* <code>other</code> custom metadata.
*
* @param other custom meta data
* @return the same instance or <code>other</code> custom metadata based on implementation
* if both the instances are considered equal, implementations should return this
* instance to avoid redundant cluster state changes.
*/
T merge(T other);
}

// internal settings only
public static final Setting<String> TRIBE_NAME_SETTING = Setting.simpleString("tribe.name", Property.NodeScope);
private final ClusterService clusterService;
Expand Down Expand Up @@ -270,7 +293,7 @@ protected void doStart() {
public void startNodes() {
for (Node node : nodes) {
try {
node.injector().getInstance(ClusterService.class).add(new TribeClusterStateListener(node));
getClusterService(node).add(new TribeClusterStateListener(node));
node.start();
} catch (Exception e) {
// calling close is safe for non started nodes, we can just iterate over all
Expand Down Expand Up @@ -348,23 +371,19 @@ public String describeTasks(List<ClusterChangedEvent> tasks) {

@Override
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 (Exception e) {
builder.failures(tasks, e);
}

return builder.build(accumulator);
ClusterState.Builder newState = ClusterState.builder(currentState).incrementVersion();
boolean clusterStateChanged = updateNodes(currentState, tasks, newState);
clusterStateChanged |= updateIndicesAndMetaData(currentState, tasks, newState);
builder.successes(tasks);
return builder.build(clusterStateChanged ? newState.build() : currentState);
}

private ClusterState applyUpdate(ClusterState currentState, ClusterChangedEvent task) {
private boolean updateNodes(ClusterState currentState, List<ClusterChangedEvent> tasks, ClusterState.Builder newState) {
boolean clusterStateChanged = false;
ClusterState tribeState = task.state();
// we only need to apply the latest cluster state update
ClusterChangedEvent latestTask = tasks.get(tasks.size() - 1);
ClusterState tribeState = latestTask.state();
DiscoveryNodes.Builder nodes = DiscoveryNodes.builder(currentState.nodes());
// -- merge nodes
// go over existing nodes, and see if they need to be removed
Expand All @@ -385,16 +404,25 @@ private ClusterState applyUpdate(ClusterState currentState, ClusterChangedEvent
Map<String, String> tribeAttr = new HashMap<>(tribe.getAttributes());
tribeAttr.put(TRIBE_NAME_SETTING.getKey(), tribeName);
DiscoveryNode discoNode = new DiscoveryNode(tribe.getName(), tribe.getId(), tribe.getEphemeralId(),
tribe.getHostName(), tribe.getHostAddress(), tribe.getAddress(), unmodifiableMap(tribeAttr), tribe.getRoles(),
tribe.getVersion());
tribe.getHostName(), tribe.getHostAddress(), tribe.getAddress(), unmodifiableMap(tribeAttr), tribe.getRoles(),
tribe.getVersion());
clusterStateChanged = true;
logger.info("[{}] adding node [{}]", tribeName, discoNode);
nodes.remove(tribe.getId()); // remove any existing node with the same id but different ephemeral id
nodes.add(discoNode);
}
}
if (clusterStateChanged) {
newState.nodes(nodes);
}
return clusterStateChanged;
}

// -- merge metadata
private boolean updateIndicesAndMetaData(ClusterState currentState, List<ClusterChangedEvent> tasks, ClusterState.Builder newState) {
// we only need to apply the latest cluster state update
ClusterChangedEvent latestTask = tasks.get(tasks.size() - 1);
ClusterState tribeState = latestTask.state();
boolean clusterStateChanged = false;
ClusterBlocks.Builder blocks = ClusterBlocks.builder().blocks(currentState.blocks());
MetaData.Builder metaData = MetaData.builder(currentState.metaData());
RoutingTable.Builder routingTable = RoutingTable.builder(currentState.routingTable());
Expand Down Expand Up @@ -462,13 +490,49 @@ private ClusterState applyUpdate(ClusterState currentState, ClusterChangedEvent
}
}
}
clusterStateChanged |= updateCustoms(currentState, tasks, metaData);
if (clusterStateChanged) {
newState.blocks(blocks);
newState.metaData(metaData);
newState.routingTable(routingTable.build());
}
return clusterStateChanged;
}

if (!clusterStateChanged) {
return currentState;
} else {
return ClusterState.builder(currentState).incrementVersion().blocks(blocks).nodes(nodes).metaData(metaData)
.routingTable(routingTable.build()).build();
private boolean updateCustoms(ClusterState currentState, List<ClusterChangedEvent> tasks, MetaData.Builder metaData) {
boolean clusterStateChanged = false;
Set<String> changedCustomMetaDataTypeSet = tasks.stream()
.map(ClusterChangedEvent::changedCustomMetaDataSet)
.flatMap(Collection::stream)
.collect(Collectors.toSet());
final List<Node> tribeClientNodes = TribeService.this.nodes;
Map<String, MetaData.Custom> mergedCustomMetaDataMap = mergeChangedCustomMetaData(changedCustomMetaDataTypeSet,
customMetaDataType -> tribeClientNodes.stream()
Copy link
Contributor

Choose a reason for hiding this comment

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

👍

.map(TribeService::getClusterService).map(ClusterService::state)
.map(ClusterState::metaData)
.map(clusterMetaData -> ((MetaData.Custom) clusterMetaData.custom(customMetaDataType)))
.filter(custom1 -> custom1 != null && custom1 instanceof MergableCustomMetaData)
.map(custom2 -> (MergableCustomMetaData) custom2)
.collect(Collectors.toList())
);
for (String changedCustomMetaDataType : changedCustomMetaDataTypeSet) {
MetaData.Custom mergedCustomMetaData = mergedCustomMetaDataMap.get(changedCustomMetaDataType);
if (mergedCustomMetaData == null) {
// we ignore merging custom md which doesn't implement MergableCustomMetaData interface
if (currentState.metaData().custom(changedCustomMetaDataType) instanceof MergableCustomMetaData) {
Copy link
Contributor

Choose a reason for hiding this comment

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

I'm wondering if we really need this check - when is it relevant?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

when a non-mergable custom md changes, it is in changedCustomMetaDataTypeSet but is ignored by mergeChangedCustomMetaData (returns a null for the custom type). So we can delete non-mergable custom md from the tribe cluster, while trying to merge mergable custom md. This check prevents deleting these non-mergable custom md.

// custom md has been removed
clusterStateChanged = true;
logger.info("[{}] removing custom meta data type [{}]", tribeName, changedCustomMetaDataType);
metaData.removeCustom(changedCustomMetaDataType);
}
} else {
// custom md has been changed
clusterStateChanged = true;
logger.info("[{}] updating custom meta data type [{}] data [{}]", tribeName, changedCustomMetaDataType, mergedCustomMetaData);
metaData.putCustom(changedCustomMetaDataType, mergedCustomMetaData);
}
}
return clusterStateChanged;
}

private void removeIndex(ClusterBlocks.Builder blocks, MetaData.Builder metaData, RoutingTable.Builder routingTable,
Expand All @@ -494,4 +558,23 @@ private void addNewIndex(ClusterState tribeState, ClusterBlocks.Builder blocks,
}
}
}

private static ClusterService getClusterService(Node node) {
return node.injector().getInstance(ClusterService.class);
}

// pkg-private for testing
static Map<String, MetaData.Custom> mergeChangedCustomMetaData(Set<String> changedCustomMetaDataTypeSet,
Function<String, List<MergableCustomMetaData>> customMetaDataByTribeNode) {

Map<String, MetaData.Custom> changedCustomMetaDataMap = new HashMap<>(changedCustomMetaDataTypeSet.size());
for (String customMetaDataType : changedCustomMetaDataTypeSet) {
customMetaDataByTribeNode.apply(customMetaDataType).stream()
.reduce((mergableCustomMD, mergableCustomMD2) ->
((MergableCustomMetaData) mergableCustomMD.merge((MetaData.Custom) mergableCustomMD2)))
.ifPresent(mergedCustomMetaData ->
changedCustomMetaDataMap.put(customMetaDataType, ((MetaData.Custom) mergedCustomMetaData)));
}
return changedCustomMetaDataMap;
}
}