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

Put CCR tasks on (data && remote cluster clients) (#54146) 

Today we assign CCR persistent tasks to nodes with the data role. It
could be that the data node is not capable of connecting to remote
clusters, in which case the task will fail since it can not connect to
the remote cluster with the leader shard. Instead, we need to assign
such tasks to nodes that are capable of connecting to remote
clusters. This commit addresses this by enabling such persistent tasks
to only be assigned to nodes that have the data role, and also have the
remote cluster client role.
This commit is contained in:
Jason Tedor 2020-03-26 23:38:44 -04:00
parent 5f007b7cb1
commit c547fabb2b
No known key found for this signature in database
GPG Key ID: FA89F05560F16BC5
2 changed files with 124 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/ShardFollowTasksExecutor.java
View File

@ -31,6 +31,7 @@ import org.elasticsearch.cluster.ClusterState;
import org.elasticsearch.cluster.metadata.AliasMetaData;
import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.cluster.metadata.MappingMetaData;
import org.elasticsearch.cluster.node.DiscoveryNode;
import org.elasticsearch.cluster.routing.IndexRoutingTable;
import org.elasticsearch.cluster.service.ClusterService;
import org.elasticsearch.common.CheckedConsumer;
@ -54,6 +55,7 @@ import org.elasticsearch.index.translog.Translog;
import org.elasticsearch.persistent.AllocatedPersistentTask;
import org.elasticsearch.persistent.PersistentTaskState;
import org.elasticsearch.persistent.PersistentTasksCustomMetaData;
import org.elasticsearch.persistent.PersistentTasksCustomMetaData.Assignment;
import org.elasticsearch.persistent.PersistentTasksExecutor;
import org.elasticsearch.tasks.TaskId;
import org.elasticsearch.threadpool.Scheduler;
@ -76,6 +78,7 @@ import java.util.function.BiConsumer;
import java.util.function.Consumer;
import java.util.function.LongConsumer;
import java.util.function.LongSupplier;
import java.util.function.Predicate;
import java.util.function.Supplier;
import static org.elasticsearch.xpack.ccr.CcrLicenseChecker.wrapClient;
@ -115,6 +118,21 @@ public class ShardFollowTasksExecutor extends PersistentTasksExecutor<ShardFollo
}
}
private static final Assignment NO_ASSIGNMENT = new Assignment(null, "no nodes found with data and remote cluster client roles");
@Override
public Assignment getAssignment(final ShardFollowTask params, final ClusterState clusterState) {
final DiscoveryNode node = selectLeastLoadedNode(
clusterState,
((Predicate<DiscoveryNode>) DiscoveryNode::isDataNode).and(DiscoveryNode::isRemoteClusterClient)
);
if (node == null) {
return NO_ASSIGNMENT;
} else {
return new Assignment(node.getId(), "node is the least loaded data node and remote cluster client");
}
}
@Override
protected AllocatedPersistentTask createTask(long id, String type, String action, TaskId parentTaskId,
PersistentTasksCustomMetaData.PersistentTask<ShardFollowTask> taskInProgress,

106
x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/action/ShardFollowTasksExecutorAssignmentTests.java Normal file
View File

@ -0,0 +1,106 @@
/*
* 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.ccr.action;
import org.elasticsearch.Version;
import org.elasticsearch.client.Client;
import org.elasticsearch.cluster.ClusterName;
import org.elasticsearch.cluster.ClusterState;
import org.elasticsearch.cluster.node.DiscoveryNode;
import org.elasticsearch.cluster.node.DiscoveryNodeRole;
import org.elasticsearch.cluster.node.DiscoveryNodes;
import org.elasticsearch.cluster.service.ClusterService;
import org.elasticsearch.common.UUIDs;
import org.elasticsearch.common.settings.ClusterSettings;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.settings.SettingsModule;
import org.elasticsearch.persistent.PersistentTasksCustomMetaData.Assignment;
import org.elasticsearch.test.ESTestCase;
import org.elasticsearch.threadpool.ThreadPool;
import org.elasticsearch.xpack.ccr.CcrSettings;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
import java.util.function.BiConsumer;
import java.util.function.Supplier;
import static org.hamcrest.Matchers.equalTo;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
public class ShardFollowTasksExecutorAssignmentTests extends ESTestCase {
public void testAssignmentToNodeWithDataAndRemoteClusterClientRoles() {
runAssignmentTest(
new HashSet<>(Arrays.asList(DiscoveryNodeRole.DATA_ROLE, DiscoveryNodeRole.REMOTE_CLUSTER_CLIENT_ROLE)),
randomIntBetween(0, 8),
() -> new HashSet<>(randomSubsetOf(new HashSet<>(Arrays.asList(DiscoveryNodeRole.DATA_ROLE, DiscoveryNodeRole.INGEST_ROLE)))),
(theSpecial, assignment) -> {
assertTrue(assignment.isAssigned());
assertThat(assignment.getExecutorNode(), equalTo(theSpecial.getId()));
}
);
}
public void testDataRoleWithoutRemoteClusterServiceRole() {
runNoAssignmentTest(Collections.singleton(DiscoveryNodeRole.DATA_ROLE));
}
public void testRemoteClusterClientRoleWithoutDataRole() {
runNoAssignmentTest(Collections.singleton(DiscoveryNodeRole.REMOTE_CLUSTER_CLIENT_ROLE));
}
private void runNoAssignmentTest(final Set<DiscoveryNodeRole> roles) {
runAssignmentTest(
roles,
0,
Collections::emptySet,
(theSpecial, assignment) -> {
assertFalse(assignment.isAssigned());
assertThat(assignment.getExplanation(), equalTo("no nodes found with data and remote cluster client roles"));
}
);
}
private void runAssignmentTest(
final Set<DiscoveryNodeRole> theSpecialRoles,
final int numberOfOtherNodes,
final Supplier<Set<DiscoveryNodeRole>> otherNodesRolesSupplier,
final BiConsumer<DiscoveryNode, Assignment> consumer
) {
final ClusterService clusterService = mock(ClusterService.class);
when(clusterService.getClusterSettings())
.thenReturn(new ClusterSettings(Settings.EMPTY, Collections.singleton(CcrSettings.CCR_WAIT_FOR_METADATA_TIMEOUT)));
final SettingsModule settingsModule = mock(SettingsModule.class);
when(settingsModule.getSettings()).thenReturn(Settings.EMPTY);
final ShardFollowTasksExecutor executor =
new ShardFollowTasksExecutor(mock(Client.class), mock(ThreadPool.class), clusterService, settingsModule);
final ClusterState.Builder clusterStateBuilder = ClusterState.builder(new ClusterName("test"));
final DiscoveryNodes.Builder nodesBuilder = DiscoveryNodes.builder();
final DiscoveryNode theSpecial = newNode(theSpecialRoles);
nodesBuilder.add(theSpecial);
for (int i = 0; i < numberOfOtherNodes; i++) {
nodesBuilder.add(newNode(otherNodesRolesSupplier.get()));
}
clusterStateBuilder.nodes(nodesBuilder);
final Assignment assignment = executor.getAssignment(mock(ShardFollowTask.class), clusterStateBuilder.build());
consumer.accept(theSpecial, assignment);
}
private static DiscoveryNode newNode(final Set<DiscoveryNodeRole> roles) {
return new DiscoveryNode(
"node_" + UUIDs.randomBase64UUID(random()),
buildNewFakeTransportAddress(),
Collections.emptyMap(),
roles,
Version.CURRENT
);
}
}