You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@lucene.apache.org by ab...@apache.org on 2017/11/21 19:15:18 UTC
[2/2] lucene-solr:jira/solr-11285-sim: SOLR-11403: Improve support
for collection properties and create shard.
SOLR-11403: Improve support for collection properties and create shard.
Project: http://git-wip-us.apache.org/repos/asf/lucene-solr/repo
Commit: http://git-wip-us.apache.org/repos/asf/lucene-solr/commit/a9591e3d
Tree: http://git-wip-us.apache.org/repos/asf/lucene-solr/tree/a9591e3d
Diff: http://git-wip-us.apache.org/repos/asf/lucene-solr/diff/a9591e3d
Branch: refs/heads/jira/solr-11285-sim
Commit: a9591e3d35184e93b18a9ef89c0b8deb7208e6f9
Parents: 9bed50e
Author: Andrzej Bialecki <ab...@apache.org>
Authored: Tue Nov 21 20:14:46 2017 +0100
Committer: Andrzej Bialecki <ab...@apache.org>
Committed: Tue Nov 21 20:14:46 2017 +0100
----------------------------------------------------------------------
.../org/apache/solr/cloud/CreateShardCmd.java | 92 +++--
.../cloud/OverseerCollectionMessageHandler.java | 4 +-
.../solr/cloud/overseer/CollectionMutator.java | 8 +-
.../cloud/autoscaling/sim/SimCloudManager.java | 35 +-
.../sim/SimClusterStateProvider.java | 90 ++++-
.../sim/SimComputePlanActionTest.java | 381 -------------------
.../sim/SimExecutePlanActionTest.java | 222 -----------
.../autoscaling/sim/SimNodeStateProvider.java | 60 ++-
.../autoscaling/sim/TestComputePlanAction.java | 376 ++++++++++++++++++
.../autoscaling/sim/TestExecutePlanAction.java | 222 +++++++++++
.../cloud/autoscaling/sim/TestPolicyCloud.java | 368 ++++++++++++++++++
11 files changed, 1194 insertions(+), 664 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a9591e3d/solr/core/src/java/org/apache/solr/cloud/CreateShardCmd.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/CreateShardCmd.java b/solr/core/src/java/org/apache/solr/cloud/CreateShardCmd.java
index 403d99a..4ccf993 100644
--- a/solr/core/src/java/org/apache/solr/cloud/CreateShardCmd.java
+++ b/solr/core/src/java/org/apache/solr/cloud/CreateShardCmd.java
@@ -17,6 +17,7 @@
package org.apache.solr.cloud;
+import java.io.IOException;
import java.lang.invoke.MethodHandles;
import java.util.ArrayList;
import java.util.Collections;
@@ -28,6 +29,7 @@ import java.util.concurrent.TimeUnit;
import com.google.common.collect.ImmutableMap;
import org.apache.solr.client.solrj.cloud.autoscaling.Policy;
import org.apache.solr.client.solrj.cloud.autoscaling.PolicyHelper;
+import org.apache.solr.client.solrj.cloud.autoscaling.SolrCloudManager;
import org.apache.solr.cloud.OverseerCollectionMessageHandler.Cmd;
import org.apache.solr.common.SolrCloseableLatch;
import org.apache.solr.common.SolrException;
@@ -74,57 +76,20 @@ public class CreateShardCmd implements Cmd {
throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, "'collection' and 'shard' are required parameters");
DocCollection collection = clusterState.getCollection(collectionName);
- int numNrtReplicas = message.getInt(NRT_REPLICAS, message.getInt(REPLICATION_FACTOR, collection.getInt(NRT_REPLICAS, collection.getInt(REPLICATION_FACTOR, 1))));
- int numPullReplicas = message.getInt(PULL_REPLICAS, collection.getInt(PULL_REPLICAS, 0));
- int numTlogReplicas = message.getInt(TLOG_REPLICAS, collection.getInt(TLOG_REPLICAS, 0));
- int totalReplicas = numNrtReplicas + numPullReplicas + numTlogReplicas;
-
- if (numNrtReplicas + numTlogReplicas <= 0) {
- throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, NRT_REPLICAS + " + " + TLOG_REPLICAS + " must be greater than 0");
- }
-
- Object createNodeSetStr = message.get(OverseerCollectionMessageHandler.CREATE_NODE_SET);
ZkStateReader zkStateReader = ocmh.zkStateReader;
- boolean usePolicyFramework = CloudUtil.usePolicyFramework(collection, ocmh.cloudManager);
- List<ReplicaPosition> positions = null;
SolrCloseableLatch countDownLatch;
try {
- if (usePolicyFramework) {
- if (collection.getPolicyName() != null) message.getProperties().put(Policy.POLICY, collection.getPolicyName());
- positions = Assign.identifyNodes(ocmh.cloudManager,
- clusterState,
- Assign.getLiveOrLiveAndCreateNodeSetList(clusterState.getLiveNodes(), message, RANDOM),
- collectionName,
- message,
- Collections.singletonList(sliceName),
- numNrtReplicas,
- numTlogReplicas,
- numPullReplicas);
- } else {
- List<Assign.ReplicaCount> sortedNodeList = getNodesForNewReplicas(clusterState, collectionName, sliceName, totalReplicas,
- createNodeSetStr, ocmh.overseer.getSolrCloudManager());
- int i = 0;
- positions = new ArrayList<>();
- for (Map.Entry<Replica.Type, Integer> e : ImmutableMap.of(Replica.Type.NRT, numNrtReplicas,
- Replica.Type.TLOG, numTlogReplicas,
- Replica.Type.PULL, numPullReplicas
- ).entrySet()) {
- for (int j = 0; j < e.getValue(); j++) {
- positions.add(new ReplicaPosition(sliceName, j + 1, e.getKey(), sortedNodeList.get(i % sortedNodeList.size()).nodeName));
- i++;
- }
- }
- }
+ List<ReplicaPosition> positions = buildReplicaPositions(ocmh.cloudManager, clusterState, collectionName, message);
Overseer.getStateUpdateQueue(zkStateReader.getZkClient()).offer(Utils.toJSON(message));
// wait for a while until we see the shard
ocmh.waitForNewShard(collectionName, sliceName);
String async = message.getStr(ASYNC);
- countDownLatch = new SolrCloseableLatch(totalReplicas, ocmh);
+ countDownLatch = new SolrCloseableLatch(positions.size(), ocmh);
for (ReplicaPosition position : positions) {
String nodeName = position.node;
- String coreName = Assign.buildSolrCoreName(ocmh.overseer.getSolrCloudManager().getDistribStateManager(), collection, sliceName, position.type);
+ String coreName = Assign.buildSolrCoreName(ocmh.cloudManager.getDistribStateManager(), collection, sliceName, position.type);
log.info("Creating replica " + coreName + " as part of slice " + sliceName + " of collection " + collectionName
+ " on " + nodeName);
@@ -173,4 +138,51 @@ public class CreateShardCmd implements Cmd {
}
+ public static List<ReplicaPosition> buildReplicaPositions(SolrCloudManager cloudManager, ClusterState clusterState,
+ String collectionName, ZkNodeProps message) throws IOException, InterruptedException {
+ String sliceName = message.getStr(SHARD_ID_PROP);
+ DocCollection collection = clusterState.getCollection(collectionName);
+
+ int numNrtReplicas = message.getInt(NRT_REPLICAS, message.getInt(REPLICATION_FACTOR, collection.getInt(NRT_REPLICAS, collection.getInt(REPLICATION_FACTOR, 1))));
+ int numPullReplicas = message.getInt(PULL_REPLICAS, collection.getInt(PULL_REPLICAS, 0));
+ int numTlogReplicas = message.getInt(TLOG_REPLICAS, collection.getInt(TLOG_REPLICAS, 0));
+ int totalReplicas = numNrtReplicas + numPullReplicas + numTlogReplicas;
+
+ if (numNrtReplicas + numTlogReplicas <= 0) {
+ throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, NRT_REPLICAS + " + " + TLOG_REPLICAS + " must be greater than 0");
+ }
+
+ Object createNodeSetStr = message.get(OverseerCollectionMessageHandler.CREATE_NODE_SET);
+
+ boolean usePolicyFramework = CloudUtil.usePolicyFramework(collection, cloudManager);
+ List<ReplicaPosition> positions;
+ if (usePolicyFramework) {
+ if (collection.getPolicyName() != null) message.getProperties().put(Policy.POLICY, collection.getPolicyName());
+ positions = Assign.identifyNodes(cloudManager,
+ clusterState,
+ Assign.getLiveOrLiveAndCreateNodeSetList(clusterState.getLiveNodes(), message, RANDOM),
+ collection.getName(),
+ message,
+ Collections.singletonList(sliceName),
+ numNrtReplicas,
+ numTlogReplicas,
+ numPullReplicas);
+ } else {
+ List<Assign.ReplicaCount> sortedNodeList = getNodesForNewReplicas(clusterState, collection.getName(), sliceName, totalReplicas,
+ createNodeSetStr, cloudManager);
+ int i = 0;
+ positions = new ArrayList<>();
+ for (Map.Entry<Replica.Type, Integer> e : ImmutableMap.of(Replica.Type.NRT, numNrtReplicas,
+ Replica.Type.TLOG, numTlogReplicas,
+ Replica.Type.PULL, numPullReplicas
+ ).entrySet()) {
+ for (int j = 0; j < e.getValue(); j++) {
+ positions.add(new ReplicaPosition(sliceName, j + 1, e.getKey(), sortedNodeList.get(i % sortedNodeList.size()).nodeName));
+ i++;
+ }
+ }
+ }
+ return positions;
+ }
+
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a9591e3d/solr/core/src/java/org/apache/solr/cloud/OverseerCollectionMessageHandler.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/OverseerCollectionMessageHandler.java b/solr/core/src/java/org/apache/solr/cloud/OverseerCollectionMessageHandler.java
index 18431e5..848a71c 100644
--- a/solr/core/src/java/org/apache/solr/cloud/OverseerCollectionMessageHandler.java
+++ b/solr/core/src/java/org/apache/solr/cloud/OverseerCollectionMessageHandler.java
@@ -651,7 +651,7 @@ public class OverseerCollectionMessageHandler implements OverseerMessageHandler,
validateConfigOrThrowSolrException(configName);
boolean isLegacyCloud = Overseer.isLegacy(zkStateReader);
- createConfNode(overseer.getSolrCloudManager().getDistribStateManager(), configName, collectionName, isLegacyCloud);
+ createConfNode(cloudManager.getDistribStateManager(), configName, collectionName, isLegacyCloud);
reloadCollection(null, new ZkNodeProps(NAME, collectionName), results);
}
@@ -660,7 +660,7 @@ public class OverseerCollectionMessageHandler implements OverseerMessageHandler,
TimeOut timeout = new TimeOut(30, TimeUnit.SECONDS, timeSource);
boolean areChangesVisible = true;
while (!timeout.hasTimedOut()) {
- DocCollection collection = zkStateReader.getClusterState().getCollection(collectionName);
+ DocCollection collection = cloudManager.getClusterStateProvider().getClusterState().getCollection(collectionName);
areChangesVisible = true;
for (Map.Entry<String,Object> updateEntry : message.getProperties().entrySet()) {
String updateKey = updateEntry.getKey();
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a9591e3d/solr/core/src/java/org/apache/solr/cloud/overseer/CollectionMutator.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/overseer/CollectionMutator.java b/solr/core/src/java/org/apache/solr/cloud/overseer/CollectionMutator.java
index c14f180..951c752 100644
--- a/solr/core/src/java/org/apache/solr/cloud/overseer/CollectionMutator.java
+++ b/solr/core/src/java/org/apache/solr/cloud/overseer/CollectionMutator.java
@@ -43,12 +43,12 @@ import static org.apache.solr.common.params.CommonParams.NAME;
public class CollectionMutator {
private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
- protected final SolrCloudManager dataProvider;
+ protected final SolrCloudManager cloudManager;
protected final DistribStateManager stateManager;
- public CollectionMutator(SolrCloudManager dataProvider) {
- this.dataProvider = dataProvider;
- this.stateManager = dataProvider.getDistribStateManager();
+ public CollectionMutator(SolrCloudManager cloudManager) {
+ this.cloudManager = cloudManager;
+ this.stateManager = cloudManager.getDistribStateManager();
}
public ZkWriteCommand createShard(final ClusterState clusterState, ZkNodeProps message) {
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a9591e3d/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimCloudManager.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimCloudManager.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimCloudManager.java
index 3494fd2..85f7650 100644
--- a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimCloudManager.java
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimCloudManager.java
@@ -106,14 +106,16 @@ public class SimCloudManager implements SolrCloudManager {
public SimCloudManager(TimeSource timeSource) throws Exception {
this.stateManager = new SimDistribStateManager();
- this.timeSource = timeSource != null ? timeSource : TimeSource.NANO_TIME;
// init common paths
stateManager.makePath(ZkStateReader.CLUSTER_STATE);
stateManager.makePath(ZkStateReader.CLUSTER_PROPS);
stateManager.makePath(ZkStateReader.SOLR_AUTOSCALING_CONF_PATH);
stateManager.makePath(ZkStateReader.LIVE_NODES_ZKNODE);
+ stateManager.makePath(ZkStateReader.ROLES);
+
+ this.timeSource = timeSource != null ? timeSource : TimeSource.NANO_TIME;
this.clusterStateProvider = new SimClusterStateProvider(liveNodes, this);
- this.nodeStateProvider = new SimNodeStateProvider(liveNodes, this.clusterStateProvider, null);
+ this.nodeStateProvider = new SimNodeStateProvider(liveNodes, this.stateManager, this.clusterStateProvider, null);
this.queueFactory = new SimDistributedQueueFactory();
this.httpServer = new SimHttpServer();
this.simCloudManagerPool = ExecutorUtil.newMDCAwareCachedThreadPool(new DefaultSolrThreadFactory("simCloudManagerPool"));
@@ -156,10 +158,11 @@ public class SimCloudManager implements SolrCloudManager {
public static Map<String, Object> createNodeValues(String nodeName) {
Map<String, Object> values = new HashMap<>();
- String host, port, nodeId;
+ String host, nodeId;
+ int port;
if (nodeName == null) {
host = "127.0.0.1";
- port = "" + nodeIdPort++;
+ port = nodeIdPort++;
nodeId = host + ":" + port + "_solr";
values.put("ip_1", "127");
values.put("ip_2", "0");
@@ -175,7 +178,7 @@ public class SimCloudManager implements SolrCloudManager {
if (portCtx.length != 2) {
throw new RuntimeException("Invalid port_context in nodeName " + nodeName);
}
- port = portCtx[0];
+ port = Integer.parseInt(portCtx[0]);
nodeId = host + ":" + port + "_" + portCtx[1];
String[] ip = host.split("\\.");
if (ip.length == 4) {
@@ -193,6 +196,11 @@ public class SimCloudManager implements SolrCloudManager {
values.put(ImplicitSnitch.SYSLOADAVG, 1.0);
values.put(ImplicitSnitch.HEAPUSAGE, 123450000);
values.put("INDEX.sizeInBytes", 12345000);
+ values.put("sysprop.java.version", System.getProperty("java.version"));
+ values.put("sysprop.java.vendor", System.getProperty("java.vendor"));
+ // fake some metrics expected in tests
+ values.put("metrics:solr.node:ADMIN./admin/authorization.clientErrors:count", 0);
+ values.put("metrics:solr.jvm:buffers.direct.Count", 0);
return values;
}
@@ -325,6 +333,9 @@ public class SimCloudManager implements SolrCloudManager {
queryRequest.getContext().put("httpMethod", req.getMethod().toString());
SolrQueryResponse queryResponse = new SolrQueryResponse();
autoScalingHandler.handleRequest(queryRequest, queryResponse);
+ if (queryResponse.getException() != null) {
+ throw new IOException(queryResponse.getException());
+ }
SolrResponse rsp = new SolrResponseBase();
rsp.setResponse(queryResponse.getValues());
LOG.debug("-- response: " + rsp);
@@ -409,11 +420,21 @@ public class SimCloudManager implements SolrCloudManager {
results.add("overseer_collection_queue_size", 0);
results.add("success", "");
break;
+ case ADDROLE:
+ nodeStateProvider.simAddNodeValue(req.getParams().get("node"), "nodeRole", req.getParams().get("role"));
+ break;
+ case CREATESHARD:
+ try {
+ clusterStateProvider.simCreateShard(new ZkNodeProps(req.getParams().toNamedList().asMap(10)), results);
+ } catch (Exception e) {
+ throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, e);
+ }
+ break;
default:
- throw new UnsupportedOperationException("Unsupported collection admin action=" + action);
+ throw new UnsupportedOperationException("Unsupported collection admin action=" + action + " in request: " + req.getParams());
}
} else {
- throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, "action is a required param");
+ throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, "action is a required param in request: " + req.getParams());
}
return rsp;
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a9591e3d/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimClusterStateProvider.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimClusterStateProvider.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimClusterStateProvider.java
index 6c387ad..8dca4cc 100644
--- a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimClusterStateProvider.java
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimClusterStateProvider.java
@@ -41,6 +41,9 @@ import org.apache.solr.client.solrj.cloud.autoscaling.VersionedData;
import org.apache.solr.client.solrj.impl.ClusterStateProvider;
import org.apache.solr.cloud.Assign;
import org.apache.solr.cloud.CreateCollectionCmd;
+import org.apache.solr.cloud.CreateShardCmd;
+import org.apache.solr.cloud.overseer.CollectionMutator;
+import org.apache.solr.cloud.overseer.ZkWriteCommand;
import org.apache.solr.common.SolrException;
import org.apache.solr.common.cloud.ClusterState;
import org.apache.solr.common.cloud.DocCollection;
@@ -298,7 +301,17 @@ public class SimClusterStateProvider implements ClusterStateProvider {
}
// pick random one
Collections.shuffle(active);
- ReplicaInfo ri = active.get(0);
+ ReplicaInfo ri = null;
+ for (ReplicaInfo a : active) {
+ if (!a.getType().equals(Replica.Type.PULL)) {
+ ri = a;
+ break;
+ }
+ }
+ if (ri == null) {
+ LOG.warn("-- can't find any suitable replica type for " + dc.getName() + " / " + s.getName());
+ return;
+ }
ri.getVariables().put(ZkStateReader.LEADER_PROP, "true");
LOG.info("-- elected new leader for " + dc.getName() + " / " + s.getName() + ": " + ri);
} else {
@@ -334,6 +347,17 @@ public class SimClusterStateProvider implements ClusterStateProvider {
throw new RuntimeException(e);
}
});
+ Map<String, Object> colProps = collProperties.computeIfAbsent(collectionName, c -> new ConcurrentHashMap<>());
+ props.getProperties().entrySet().stream()
+ .filter(e -> !e.getKey().equals("action"))
+ .filter(e -> !e.getKey().equals("name"))
+ .forEach(e -> {
+ if (e.getKey().startsWith("collection.")) {
+ colProps.put(e.getKey().substring("collection.".length()), e.getValue());
+ } else {
+ colProps.put(e.getKey(), e.getValue());
+ }
+ });
cloudManager.submit(() -> {simRunLeaderElection(Collections.singleton(collectionName), true); return true;});
results.add("success", "");
}
@@ -425,6 +449,58 @@ public class SimClusterStateProvider implements ClusterStateProvider {
results.add("success", "");
}
+ public void simCreateShard(ZkNodeProps message, NamedList results) throws Exception {
+ if (message.getStr(CommonAdminParams.ASYNC) != null) {
+ results.add(CoreAdminParams.REQUESTID, message.getStr(CommonAdminParams.ASYNC));
+ }
+ String collectionName = message.getStr(COLLECTION_PROP);
+ String sliceName = message.getStr(SHARD_ID_PROP);
+ ClusterState clusterState = getClusterState();
+ lock.lock();
+ try {
+ ZkWriteCommand cmd = new CollectionMutator(cloudManager).createShard(clusterState, message);
+ if (cmd.noop) {
+ results.add("success", "no-op");
+ return;
+ }
+ // copy shard properties
+ DocCollection collection = cmd.collection;
+ Slice slice = collection.getSlice(sliceName);
+ Map<String, Object> props = sliceProperties.computeIfAbsent(collection.getName(), c -> new ConcurrentHashMap<>())
+ .computeIfAbsent(sliceName, s -> new ConcurrentHashMap<>());
+ props.clear();
+ slice.getProperties().entrySet().stream()
+ .filter(e -> !e.getKey().equals("range"))
+ .filter(e -> !e.getKey().equals("replicas"))
+ .forEach(e -> props.put(e.getKey(), e.getValue()));
+ // 2. create new replicas
+ List<ReplicaPosition> positions = CreateShardCmd.buildReplicaPositions(cloudManager, clusterState, collectionName, message);
+ AtomicInteger replicaNum = new AtomicInteger(1);
+ positions.forEach(pos -> {
+ Map<String, Object> replicaProps = new HashMap<>();
+ replicaProps.put(ZkStateReader.SHARD_ID_PROP, pos.shard);
+ replicaProps.put(ZkStateReader.NODE_NAME_PROP, pos.node);
+ replicaProps.put(ZkStateReader.REPLICA_TYPE, pos.type.toString());
+ String coreName = String.format(Locale.ROOT, "%s_%s_replica_%s%s", collectionName, pos.shard, pos.type.name().substring(0,1).toLowerCase(Locale.ROOT),
+ replicaNum.getAndIncrement());
+ try {
+ replicaProps.put(ZkStateReader.CORE_NAME_PROP, coreName);
+ ReplicaInfo ri = new ReplicaInfo("core_node" + Assign.incAndGetId(stateManager, collectionName, 0),
+ coreName, collectionName, pos.shard, pos.type, pos.node, replicaProps);
+ simAddReplica(pos.node, ri, false);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ });
+ Map<String, Object> colProps = collProperties.computeIfAbsent(collectionName, c -> new ConcurrentHashMap<>());
+
+ cloudManager.submit(() -> {simRunLeaderElection(Collections.singleton(collectionName), true); return true;});
+ results.add("success", "");
+ } finally {
+ lock.unlock();
+ }
+ }
+
public synchronized Map<String, Object> saveClusterProperties() throws Exception {
if (lastSavedProperties != null && lastSavedProperties.equals(clusterProperties)) {
return lastSavedProperties;
@@ -562,15 +638,23 @@ public class SimClusterStateProvider implements ClusterStateProvider {
.put(ri.getName(), r);
});
});
+
+ // add empty slices
+ sliceProperties.forEach((c, perSliceProps) -> {
+ perSliceProps.forEach((slice, props) -> {
+ collMap.computeIfAbsent(c, co -> new ConcurrentHashMap<>()).computeIfAbsent(slice, s -> new ConcurrentHashMap<>());
+ });
+ });
+
Map<String, DocCollection> res = new HashMap<>();
collMap.forEach((coll, shards) -> {
Map<String, Slice> slices = new HashMap<>();
shards.forEach((s, replicas) -> {
- Map<String, Object> sliceProps = sliceProperties.computeIfAbsent(coll, c -> new HashMap<>()).computeIfAbsent(s, sl -> new HashMap<>());
+ Map<String, Object> sliceProps = sliceProperties.computeIfAbsent(coll, c -> new ConcurrentHashMap<>()).computeIfAbsent(s, sl -> new ConcurrentHashMap<>());
Slice slice = new Slice(s, replicas, sliceProps);
slices.put(s, slice);
});
- Map<String, Object> collProps = collProperties.computeIfAbsent(coll, c -> new HashMap<>());
+ Map<String, Object> collProps = collProperties.computeIfAbsent(coll, c -> new ConcurrentHashMap<>());
DocCollection dc = new DocCollection(coll, slices, collProps, DocRouter.DEFAULT, 0, ZkStateReader.CLUSTER_STATE);
res.put(coll, dc);
});
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a9591e3d/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimComputePlanActionTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimComputePlanActionTest.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimComputePlanActionTest.java
deleted file mode 100644
index 606dee4..0000000
--- a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimComputePlanActionTest.java
+++ /dev/null
@@ -1,381 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.solr.cloud.autoscaling.sim;
-
-import java.io.IOException;
-import java.lang.invoke.MethodHandles;
-import java.util.Collection;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicReference;
-
-import org.apache.solr.SolrTestCaseJ4;
-import org.apache.solr.client.solrj.SolrClient;
-import org.apache.solr.client.solrj.SolrRequest;
-import org.apache.solr.client.solrj.SolrResponse;
-import org.apache.solr.client.solrj.cloud.autoscaling.TriggerEventType;
-import org.apache.solr.client.solrj.embedded.JettySolrRunner;
-import org.apache.solr.client.solrj.impl.CloudSolrClient;
-import org.apache.solr.client.solrj.impl.SolrClientNodeStateProvider;
-import org.apache.solr.client.solrj.request.CollectionAdminRequest;
-import org.apache.solr.cloud.SolrCloudTestCase;
-import org.apache.solr.cloud.autoscaling.ActionContext;
-import org.apache.solr.cloud.autoscaling.ComputePlanAction;
-import org.apache.solr.cloud.autoscaling.ScheduledTriggers;
-import org.apache.solr.cloud.autoscaling.TriggerAction;
-import org.apache.solr.cloud.autoscaling.TriggerEvent;
-import org.apache.solr.common.cloud.ClusterState;
-import org.apache.solr.common.cloud.DocCollection;
-import org.apache.solr.common.cloud.Replica;
-import org.apache.solr.common.cloud.ZkStateReader;
-import org.apache.solr.common.params.CollectionParams;
-import org.apache.solr.common.params.SolrParams;
-import org.apache.solr.common.util.NamedList;
-import org.apache.solr.common.util.TimeSource;
-import org.apache.solr.common.util.Utils;
-import org.apache.solr.util.LogLevel;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.BeforeClass;
-import org.junit.Test;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import static java.nio.charset.StandardCharsets.UTF_8;
-import static org.apache.solr.cloud.autoscaling.AutoScalingHandlerTest.createAutoScalingRequest;
-import static org.apache.solr.common.params.CollectionParams.CollectionAction.MOVEREPLICA;
-
-/**
- * Test for {@link ComputePlanAction}
- */
-@LogLevel("org.apache.solr.cloud.autoscaling=DEBUG;org.apache.solr.cloud.Overseer=DEBUG;org.apache.solr.cloud.overseer=DEBUG;org.apache.solr.client.solrj.impl.SolrClientDataProvider=DEBUG;")
-public class SimComputePlanActionTest extends SimSolrCloudTestCase {
- private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
-
- private static final AtomicBoolean fired = new AtomicBoolean(false);
- private static final int NODE_COUNT = 1;
- private static CountDownLatch triggerFiredLatch = new CountDownLatch(1);
- private static final AtomicReference<Map> actionContextPropsRef = new AtomicReference<>();
- private static final AtomicReference<TriggerEvent> eventRef = new AtomicReference<>();
-
- @BeforeClass
- public static void setupCluster() throws Exception {
- cluster = SimCloudManager.createCluster(1, TimeSource.get("simTime:50"));
- }
-
- @Before
- public void init() throws Exception {
-
- fired.set(false);
- triggerFiredLatch = new CountDownLatch(1);
- actionContextPropsRef.set(null);
-
- // remove everything from autoscaling.json in ZK
- cluster.getDistribStateManager().setData(ZkStateReader.SOLR_AUTOSCALING_CONF_PATH, "{}".getBytes(UTF_8), -1);
-
- if (cluster.getClusterStateProvider().getLiveNodes().size() > NODE_COUNT) {
- // stop some to get to original state
- int numJetties = cluster.getClusterStateProvider().getLiveNodes().size();
- for (int i = 0; i < numJetties - NODE_COUNT; i++) {
- String node = cluster.getSimClusterStateProvider().simGetRandomNode(random());
- cluster.getSimClusterStateProvider().simRemoveNode(node);
- }
- }
-
- try {
- cluster.request(CollectionAdminRequest.deleteCollection("testNodeLost"));
- } catch (Exception e) {
- // expected if testNodeLost hasn't run already
- }
- try {
- cluster.request(CollectionAdminRequest.deleteCollection("testNodeAdded"));
- } catch (Exception e) {
- // expected if testNodeAdded hasn't run already
- }
- try {
- cluster.request(CollectionAdminRequest.deleteCollection("testNodeWithMultipleReplicasLost"));
- } catch (Exception e) {
- // expected if testNodeWithMultipleReplicasLost hasn't run already
- }
-
- String setClusterPolicyCommand = "{" +
- " 'set-cluster-policy': [" +
- " {'cores':'<10', 'node':'#ANY'}," +
- " {'replica':'<2', 'shard': '#EACH', 'node': '#ANY'}," +
- " {'nodeRole':'overseer', 'replica':0}" +
- " ]" +
- "}";
- SolrRequest req = createAutoScalingRequest(SolrRequest.METHOD.POST, setClusterPolicyCommand);
- SolrResponse rsp = cluster.request(req);
- NamedList<Object> response = rsp.getResponse();
- assertEquals(response.get("result").toString(), "success");
-
- String setClusterPreferencesCommand = "{" +
- "'set-cluster-preferences': [" +
- "{'minimize': 'cores'}," +
- "{'maximize': 'freedisk','precision': 100}]" +
- "}";
- req = createAutoScalingRequest(SolrRequest.METHOD.POST, setClusterPreferencesCommand);
- rsp = cluster.request(req);
- response = rsp.getResponse();
- assertEquals(response.get("result").toString(), "success");
- cluster.getTimeSource().sleep(ScheduledTriggers.DEFAULT_COOLDOWN_PERIOD_MS);
- }
-
- @After
- public void printState() throws Exception {
- log.info("-------------_ FINAL STATE --------------");
- log.info("* Node values: " + Utils.toJSONString(cluster.getSimNodeStateProvider().simGetAllNodeValues()));
- log.info("* Live nodes: " + cluster.getClusterStateProvider().getLiveNodes());
- ClusterState state = cluster.getClusterStateProvider().getClusterState();
- for (String coll: cluster.getSimClusterStateProvider().simListCollections()) {
- log.info("* Collection " + coll + " state: " + state.getCollection(coll));
- }
-
- }
-
- @Test
- public void testNodeLost() throws Exception {
- // let's start a node so that we have at least two
- String node = cluster.simAddNode();
- AssertingTriggerAction.expectedNode = node;
-
- SolrClient solrClient = cluster.simGetSolrClient();
- String setTriggerCommand = "{" +
- "'set-trigger' : {" +
- "'name' : 'node_lost_trigger'," +
- "'event' : 'nodeLost'," +
- "'waitFor' : '7s'," +
- "'enabled' : true," +
- "'actions' : [{'name':'compute_plan', 'class' : 'solr.ComputePlanAction'}," +
- "{'name':'test','class':'" + SimComputePlanActionTest.AssertingTriggerAction.class.getName() + "'}]" +
- "}}";
- SolrRequest req = createAutoScalingRequest(SolrRequest.METHOD.POST, setTriggerCommand);
- NamedList<Object> response = solrClient.request(req);
- assertEquals(response.get("result").toString(), "success");
-
- CollectionAdminRequest.Create create = CollectionAdminRequest.createCollection("testNodeLost",
- "conf",1, 2);
- create.process(solrClient);
-
- waitForState("Timed out waiting for replicas of new collection to be active",
- "testNodeLost", clusterShape(1, 2));
-
- ClusterState clusterState = cluster.getClusterStateProvider().getClusterState();
- DocCollection collection = clusterState.getCollection("testNodeLost");
- List<Replica> replicas = collection.getReplicas(node);
- assertNotNull(replicas);
- assertFalse(replicas.isEmpty());
-
- // start another node because because when the other node goes away, the cluster policy requires only
- // 1 replica per node and none on the overseer
- String node2 = cluster.simAddNode();
- assertTrue(node2 + "is not live yet", cluster.getClusterStateProvider().getClusterState().liveNodesContain(node2) );
-
- // stop the original node
- cluster.simRemoveNode(node);
- log.info("Stopped_node : {}", node);
-
- assertTrue("Trigger was not fired even after 10 seconds", triggerFiredLatch.await(10, TimeUnit.SECONDS));
- assertTrue(fired.get());
- Map context = actionContextPropsRef.get();
- assertNotNull(context);
- List<SolrRequest> operations = (List<SolrRequest>) context.get("operations");
- assertNotNull("The operations computed by ComputePlanAction should not be null , " + eventRef.get(), operations);
- assertEquals("ComputePlanAction should have computed exactly 1 operation", 1, operations.size());
- SolrRequest solrRequest = operations.get(0);
- SolrParams params = solrRequest.getParams();
- assertEquals("Expected MOVEREPLICA action after adding node", MOVEREPLICA, CollectionParams.CollectionAction.get(params.get("action")));
- String replicaToBeMoved = params.get("replica");
- assertEquals("Unexpected node in computed operation", replicas.get(0).getName(), replicaToBeMoved);
-
- // shutdown the extra node that we had started
- cluster.simRemoveNode(node2);
- }
-
- public void testNodeWithMultipleReplicasLost() throws Exception {
- AssertingTriggerAction.expectedNode = null;
-
- // start 3 more nodes
- cluster.simAddNode();
- cluster.simAddNode();
- cluster.simAddNode();
-
- SolrClient solrClient = cluster.simGetSolrClient();
- String setTriggerCommand = "{" +
- "'set-trigger' : {" +
- "'name' : 'node_lost_trigger'," +
- "'event' : 'nodeLost'," +
- "'waitFor' : '1s'," +
- "'enabled' : true," +
- "'actions' : [{'name':'compute_plan', 'class' : 'solr.ComputePlanAction'}," +
- "{'name':'test','class':'" + SimComputePlanActionTest.AssertingTriggerAction.class.getName() + "'}]" +
- "}}";
- SolrRequest req = createAutoScalingRequest(SolrRequest.METHOD.POST, setTriggerCommand);
- NamedList<Object> response = solrClient.request(req);
- assertEquals(response.get("result").toString(), "success");
-
- CollectionAdminRequest.Create create = CollectionAdminRequest.createCollection("testNodeWithMultipleReplicasLost",
- "conf",2, 3);
-// create.setMaxShardsPerNode(2);
- create.process(solrClient);
-
- waitForState("Timed out waiting for replicas of new collection to be active",
- "testNodeWithMultipleReplicasLost", clusterShape(2, 3));
-
- ClusterState clusterState = cluster.getClusterStateProvider().getClusterState();
- DocCollection docCollection = clusterState.getCollection("testNodeWithMultipleReplicasLost");
-
- // lets find a node with at least 2 replicas
- String stoppedNodeName = null;
- List<Replica> replicasToBeMoved = null;
- for (String node : cluster.getClusterStateProvider().getLiveNodes()) {
- List<Replica> replicas = docCollection.getReplicas(node);
- if (replicas != null && replicas.size() == 2) {
- stoppedNodeName = node;
- replicasToBeMoved = replicas;
- cluster.simRemoveNode(node);
- break;
- }
- }
- assertNotNull(stoppedNodeName);
-
- assertTrue("Trigger was not fired even after 5 seconds", triggerFiredLatch.await(5, TimeUnit.SECONDS));
- assertTrue(fired.get());
-
- TriggerEvent triggerEvent = eventRef.get();
- assertNotNull(triggerEvent);
- assertEquals(TriggerEventType.NODELOST, triggerEvent.getEventType());
- // TODO assertEquals(stoppedNodeName, triggerEvent.getProperty(TriggerEvent.NODE_NAME));
-
- Map context = actionContextPropsRef.get();
- assertNotNull(context);
- List<SolrRequest> operations = (List<SolrRequest>) context.get("operations");
- assertNotNull("The operations computed by ComputePlanAction should not be null " + actionContextPropsRef.get(), operations);
- operations.forEach(solrRequest -> log.info(solrRequest.getParams().toString()));
- assertEquals("ComputePlanAction should have computed exactly 2 operation", 2, operations.size());
-
- for (SolrRequest solrRequest : operations) {
- SolrParams params = solrRequest.getParams();
- assertEquals("Expected MOVEREPLICA action after adding node", MOVEREPLICA, CollectionParams.CollectionAction.get(params.get("action")));
- String moved = params.get("replica");
- assertTrue(replicasToBeMoved.stream().anyMatch(replica -> replica.getName().equals(moved)));
- }
- }
-
- @Test
- public void testNodeAdded() throws Exception {
- AssertingTriggerAction.expectedNode = null;
- SolrClient solrClient = cluster.simGetSolrClient();
- String setTriggerCommand = "{" +
- "'set-trigger' : {" +
- "'name' : 'node_added_trigger'," +
- "'event' : 'nodeAdded'," +
- "'waitFor' : '1s'," +
- "'enabled' : true," +
- "'actions' : [{'name':'compute_plan', 'class' : 'solr.ComputePlanAction'}," +
- "{'name':'test','class':'" + SimComputePlanActionTest.AssertingTriggerAction.class.getName() + "'}]" +
- "}}";
- SolrRequest req = createAutoScalingRequest(SolrRequest.METHOD.POST, setTriggerCommand);
- NamedList<Object> response = solrClient.request(req);
- assertEquals(response.get("result").toString(), "success");
-
- // the default policy limits 1 replica per node, we need more right now
- String setClusterPolicyCommand = "{" +
- " 'set-cluster-policy': [" +
- " {'cores':'<10', 'node':'#ANY'}," +
- " {'replica':'<3', 'shard': '#EACH', 'node': '#ANY'}," +
- " {'nodeRole':'overseer', 'replica':0}" +
- " ]" +
- "}";
- req = createAutoScalingRequest(SolrRequest.METHOD.POST, setClusterPolicyCommand);
- response = solrClient.request(req);
- assertEquals(response.get("result").toString(), "success");
-
- CollectionAdminRequest.Create create = CollectionAdminRequest.createCollection("testNodeAdded",
- "conf",1, 2);
- create.process(solrClient);
-
- waitForState("Timed out waiting for replicas of new collection to be active",
- "testNodeAdded", (liveNodes, collectionState) -> collectionState.getReplicas().stream().allMatch(replica -> replica.isActive(liveNodes)));
-
- // reset to the original policy which has only 1 replica per shard per node
- setClusterPolicyCommand = "{" +
- " 'set-cluster-policy': [" +
- " {'cores':'<10', 'node':'#ANY'}," +
- " {'replica':'<2', 'shard': '#EACH', 'node': '#ANY'}," +
- " {'nodeRole':'overseer', 'replica':0}" +
- " ]" +
- "}";
- req = createAutoScalingRequest(SolrRequest.METHOD.POST, setClusterPolicyCommand);
- response = solrClient.request(req);
- assertEquals(response.get("result").toString(), "success");
-
- // start a node so that the 'violation' created by the previous policy update is fixed
- String newNode = cluster.simAddNode();
- assertTrue("Trigger was not fired even after 5 seconds", triggerFiredLatch.await(5, TimeUnit.SECONDS));
- assertTrue(fired.get());
- Map context = actionContextPropsRef.get();
- assertNotNull(context);
- log.info("Node values: " + Utils.toJSONString(cluster.getSimNodeStateProvider().simGetAllNodeValues()));
- log.info("Live nodes: " + cluster.getClusterStateProvider().getLiveNodes() + ", collection state: " + cluster.getClusterStateProvider().getClusterState().getCollection("testNodeAdded"));
- List<SolrRequest> operations = (List<SolrRequest>) context.get("operations");
- assertNotNull("The operations computed by ComputePlanAction should not be null" + context, operations);
- assertEquals("ComputePlanAction should have computed exactly 1 operation, but was: " + operations, 1, operations.size());
- SolrRequest request = operations.get(0);
- SolrParams params = request.getParams();
- assertEquals("Expected MOVEREPLICA action after adding node", MOVEREPLICA, CollectionParams.CollectionAction.get(params.get("action")));
- String nodeAdded = params.get("targetNode");
- assertEquals("Unexpected node in computed operation", newNode, nodeAdded);
- }
-
- public static class AssertingTriggerAction implements TriggerAction {
- static String expectedNode;
-
- @Override
- public String getName() {
- return null;
- }
-
- @Override
- public void process(TriggerEvent event, ActionContext context) {
- if (expectedNode != null) {
- Collection nodes = (Collection) event.getProperty(TriggerEvent.NODE_NAMES);
- if (nodes == null || !nodes.contains(expectedNode)) return;//this is not the event we are looking for
- }
- if (fired.compareAndSet(false, true)) {
- eventRef.set(event);
- actionContextPropsRef.set(context.getProperties());
- triggerFiredLatch.countDown();
- }
- }
-
- @Override
- public void close() throws IOException {
-
- }
-
- @Override
- public void init(Map<String, String> args) {
-
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a9591e3d/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimExecutePlanActionTest.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimExecutePlanActionTest.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimExecutePlanActionTest.java
deleted file mode 100644
index b5c78a2..0000000
--- a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimExecutePlanActionTest.java
+++ /dev/null
@@ -1,222 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.solr.cloud.autoscaling.sim;
-
-import java.lang.invoke.MethodHandles;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.stream.Collectors;
-
-import com.google.common.collect.Lists;
-import org.apache.solr.client.solrj.SolrClient;
-import org.apache.solr.client.solrj.SolrRequest;
-import org.apache.solr.client.solrj.cloud.autoscaling.TriggerEventType;
-import org.apache.solr.client.solrj.cloud.autoscaling.VersionedData;
-import org.apache.solr.client.solrj.request.CollectionAdminRequest;
-import org.apache.solr.cloud.autoscaling.ActionContext;
-import org.apache.solr.cloud.autoscaling.AutoScalingHandlerTest;
-import org.apache.solr.cloud.autoscaling.ExecutePlanAction;
-import org.apache.solr.cloud.autoscaling.NodeLostTrigger;
-import org.apache.solr.common.cloud.ClusterState;
-import org.apache.solr.common.cloud.DocCollection;
-import org.apache.solr.common.cloud.Replica;
-import org.apache.solr.common.cloud.ZkNodeProps;
-import org.apache.solr.common.cloud.ZkStateReader;
-import org.apache.solr.common.params.CollectionParams;
-import org.apache.solr.common.util.NamedList;
-import org.apache.solr.common.util.Utils;
-import org.apache.solr.util.LogLevel;
-import org.apache.solr.common.util.TimeSource;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.BeforeClass;
-import org.junit.Test;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import static org.apache.solr.common.cloud.ZkStateReader.SOLR_AUTOSCALING_CONF_PATH;
-
-/**
- * Test for {@link ExecutePlanAction}
- */
-@LogLevel("org.apache.solr.cloud.autoscaling=DEBUG")
-public class SimExecutePlanActionTest extends SimSolrCloudTestCase {
- private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
-
- private static final int NODE_COUNT = 2;
-
- @BeforeClass
- public static void setupCluster() throws Exception {
- cluster = SimCloudManager.createCluster(NODE_COUNT, TimeSource.get("simTime:50"));
- }
-
- @Before
- public void setUp() throws Exception {
- super.setUp();
- // clear any persisted auto scaling configuration
- cluster.getDistribStateManager().setData(SOLR_AUTOSCALING_CONF_PATH, Utils.toJSON(new ZkNodeProps()), -1);
-
- if (cluster.getClusterStateProvider().getLiveNodes().size() < NODE_COUNT) {
- // start some to get to original state
- int numJetties = cluster.getClusterStateProvider().getLiveNodes().size();
- for (int i = 0; i < NODE_COUNT - numJetties; i++) {
- cluster.simAddNode();
- }
- }
- }
-
- @After
- public void printState() throws Exception {
- log.info("-------------_ FINAL STATE --------------");
- log.info("* Node values: " + Utils.toJSONString(cluster.getSimNodeStateProvider().simGetAllNodeValues()));
- log.info("* Live nodes: " + cluster.getClusterStateProvider().getLiveNodes());
- ClusterState state = cluster.getClusterStateProvider().getClusterState();
- for (String coll: cluster.getSimClusterStateProvider().simListCollections()) {
- log.info("* Collection " + coll + " state: " + state.getCollection(coll));
- }
-
- }
-
- @Test
- public void testExecute() throws Exception {
- SolrClient solrClient = cluster.simGetSolrClient();
- String collectionName = "testExecute";
- CollectionAdminRequest.Create create = CollectionAdminRequest.createCollection(collectionName,
- "conf", 1, 2);
- create.setMaxShardsPerNode(1);
- create.process(solrClient);
-
- waitForState("Timed out waiting for replicas of new collection to be active",
- collectionName, clusterShape(1, 2));
-
- String sourceNodeName = cluster.getSimClusterStateProvider().simGetRandomNode(random());
- ClusterState clusterState = cluster.getClusterStateProvider().getClusterState();
- DocCollection docCollection = clusterState.getCollection(collectionName);
- List<Replica> replicas = docCollection.getReplicas(sourceNodeName);
- assertNotNull(replicas);
- assertFalse(replicas.isEmpty());
-
- List<String> otherNodes = cluster.getClusterStateProvider().getLiveNodes().stream()
- .filter(node -> !node.equals(sourceNodeName)).collect(Collectors.toList());
- assertFalse(otherNodes.isEmpty());
- String survivor = otherNodes.get(0);
-
- try (ExecutePlanAction action = new ExecutePlanAction()) {
- action.init(Collections.singletonMap("name", "execute_plan"));
-
- // used to signal if we found that ExecutePlanAction did in fact create the right znode before executing the operation
- AtomicBoolean znodeCreated = new AtomicBoolean(false);
-
- CollectionAdminRequest.AsyncCollectionAdminRequest moveReplica = new CollectionAdminRequest.MoveReplica(collectionName, replicas.get(0).getName(), survivor);
- CollectionAdminRequest.AsyncCollectionAdminRequest mockRequest = new CollectionAdminRequest.AsyncCollectionAdminRequest(CollectionParams.CollectionAction.OVERSEERSTATUS) {
- @Override
- public void setAsyncId(String asyncId) {
- super.setAsyncId(asyncId);
- String parentPath = ZkStateReader.SOLR_AUTOSCALING_TRIGGER_STATE_PATH + "/xyz/execute_plan";
- try {
- if (cluster.getDistribStateManager().hasData(parentPath)) {
- java.util.List<String> children = cluster.getDistribStateManager().listData(parentPath);
- if (!children.isEmpty()) {
- String child = children.get(0);
- VersionedData data = cluster.getDistribStateManager().getData(parentPath + "/" + child);
- Map m = (Map) Utils.fromJSON(data.getData());
- if (m.containsKey("requestid")) {
- znodeCreated.set(m.get("requestid").equals(asyncId));
- }
- }
- }
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
-
- }
- };
- List<CollectionAdminRequest.AsyncCollectionAdminRequest> operations = Lists.asList(moveReplica, new CollectionAdminRequest.AsyncCollectionAdminRequest[]{mockRequest});
- NodeLostTrigger.NodeLostEvent nodeLostEvent = new NodeLostTrigger.NodeLostEvent(TriggerEventType.NODELOST,
- "mock_trigger_name", Collections.singletonList(TimeSource.CURRENT_TIME.getTime()),
- Collections.singletonList(sourceNodeName));
- ActionContext actionContext = new ActionContext(cluster, null,
- new HashMap<>(Collections.singletonMap("operations", operations)));
- action.process(nodeLostEvent, actionContext);
-
-// assertTrue("ExecutePlanAction should have stored the requestid in ZK before executing the request", znodeCreated.get());
- List<NamedList<Object>> responses = (List<NamedList<Object>>) actionContext.getProperty("responses");
- assertNotNull(responses);
- assertEquals(2, responses.size());
- NamedList<Object> response = responses.get(0);
- assertNull(response.get("failure"));
- assertNotNull(response.get("success"));
- }
-
- waitForState("Timed out waiting for replicas of new collection to be active",
- collectionName, clusterShape(1, 2));
- }
-
- @Test
- public void testIntegration() throws Exception {
- SolrClient solrClient = cluster.simGetSolrClient();
-
- String setTriggerCommand = "{" +
- "'set-trigger' : {" +
- "'name' : 'node_lost_trigger'," +
- "'event' : 'nodeLost'," +
- "'waitFor' : '1s'," +
- "'enabled' : true," +
- "'actions' : [{'name':'compute_plan', 'class' : 'solr.ComputePlanAction'}," +
- "{'name':'execute_plan','class':'solr.ExecutePlanAction'}]" +
- "}}";
- SolrRequest req = AutoScalingHandlerTest.createAutoScalingRequest(SolrRequest.METHOD.POST, setTriggerCommand);
- NamedList<Object> response = solrClient.request(req);
- assertEquals(response.get("result").toString(), "success");
-
- String collectionName = "testIntegration";
- CollectionAdminRequest.Create create = CollectionAdminRequest.createCollection(collectionName,
- "conf", 1, 2);
- create.setMaxShardsPerNode(1);
- create.process(solrClient);
-
- waitForState("Timed out waiting for replicas of new collection to be active",
- collectionName, clusterShape(1, 2));
-
- String sourceNodeName = cluster.getSimClusterStateProvider().simGetRandomNode(random());
- ClusterState clusterState = cluster.getClusterStateProvider().getClusterState();
- DocCollection docCollection = clusterState.getCollection(collectionName);
- List<Replica> replicas = docCollection.getReplicas(sourceNodeName);
- assertNotNull(replicas);
- assertFalse(replicas.isEmpty());
-
- List<String> otherNodes = cluster.getClusterStateProvider().getLiveNodes().stream()
- .filter(node -> !node.equals(sourceNodeName)).collect(Collectors.toList());
- assertFalse(otherNodes.isEmpty());
- String survivor = otherNodes.get(0);
-
- cluster.simRemoveNode(sourceNodeName);
-
- waitForState("Timed out waiting for replicas of collection to be 2 again",
- collectionName, clusterShape(1, 2));
-
- clusterState = cluster.getClusterStateProvider().getClusterState();
- docCollection = clusterState.getCollection(collectionName);
- List<Replica> replicasOnSurvivor = docCollection.getReplicas(survivor);
- assertNotNull(replicasOnSurvivor);
- assertEquals(2, replicasOnSurvivor.size());
- }
-}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a9591e3d/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimNodeStateProvider.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimNodeStateProvider.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimNodeStateProvider.java
index 577f978..19e8905 100644
--- a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimNodeStateProvider.java
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimNodeStateProvider.java
@@ -6,14 +6,18 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;
+import org.apache.solr.client.solrj.cloud.autoscaling.BadVersionException;
import org.apache.solr.client.solrj.cloud.autoscaling.NodeStateProvider;
import org.apache.solr.client.solrj.cloud.autoscaling.ReplicaInfo;
+import org.apache.solr.common.cloud.ZkStateReader;
+import org.apache.solr.common.util.Utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -24,12 +28,16 @@ public class SimNodeStateProvider implements NodeStateProvider {
private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
private final Map<String, Map<String, Object>> nodeValues = new ConcurrentHashMap<>();
- private final SimClusterStateProvider stateProvider;
+ private final SimClusterStateProvider clusterStateProvider;
+ private final SimDistribStateManager stateManager;
private final Set<String> liveNodes;
- public SimNodeStateProvider(Set<String> liveNodes, SimClusterStateProvider stateProvider, Map<String, Map<String, Object>> nodeValues) {
+ public SimNodeStateProvider(Set<String> liveNodes, SimDistribStateManager stateManager,
+ SimClusterStateProvider clusterStateProvider,
+ Map<String, Map<String, Object>> nodeValues) {
this.liveNodes = liveNodes;
- this.stateProvider = stateProvider;
+ this.stateManager = stateManager;
+ this.clusterStateProvider = clusterStateProvider;
if (nodeValues != null) {
this.nodeValues.putAll(nodeValues);
}
@@ -46,20 +54,62 @@ public class SimNodeStateProvider implements NodeStateProvider {
public void simSetNodeValues(String node, Map<String, Object> values) {
nodeValues.put(node, new ConcurrentHashMap<>(values));
+ if (values.containsKey("nodeRole")) {
+ saveRoles();
+ }
}
public void simSetNodeValue(String node, String key, Object value) {
nodeValues.computeIfAbsent(node, n -> new ConcurrentHashMap<>()).put(key, value);
+ if (key.equals("nodeRole")) {
+ saveRoles();
+ }
+ }
+
+ public void simAddNodeValue(String node, String key, Object value) {
+ Map<String, Object> values = nodeValues.computeIfAbsent(node, n -> new ConcurrentHashMap<>());
+ Object existing = values.get(key);
+ if (existing == null) {
+ values.put(key, value);
+ } else if (existing instanceof Set) {
+ ((Set)existing).add(value);
+ } else {
+ Set<Object> vals = new HashSet<>();
+ vals.add(existing);
+ vals.add(value);
+ values.put(key, vals);
+ }
+ if (key.equals("nodeRole")) {
+ saveRoles();
+ }
}
public void simRemoveNodeValues(String node) {
- nodeValues.remove(node);
+ Map<String, Object> values = nodeValues.remove(node);
+ if (values != null && values.containsKey("nodeRole")) {
+ saveRoles();
+ }
}
public Map<String, Map<String, Object>> simGetAllNodeValues() {
return nodeValues;
}
+ private synchronized void saveRoles() {
+ final Map<String, Set<String>> roles = new HashMap<>();
+ nodeValues.forEach((n, values) -> {
+ String nodeRole = (String)values.get("nodeRole");
+ if (nodeRole != null) {
+ roles.computeIfAbsent(nodeRole, role -> new HashSet<>()).add(n);
+ }
+ });
+ try {
+ stateManager.setData(ZkStateReader.ROLES, Utils.toJSON(roles), -1);
+ } catch (Exception e) {
+ throw new RuntimeException("Unexpected exception saving roles " + roles, e);
+ }
+ }
+
// ---------- interface methods -------------
@Override
@@ -78,7 +128,7 @@ public class SimNodeStateProvider implements NodeStateProvider {
@Override
public Map<String, Map<String, List<ReplicaInfo>>> getReplicaInfo(String node, Collection<String> keys) {
- List<ReplicaInfo> replicas = stateProvider.simGetReplicaInfos(node);
+ List<ReplicaInfo> replicas = clusterStateProvider.simGetReplicaInfos(node);
if (replicas == null) {
return Collections.emptyMap();
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a9591e3d/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestComputePlanAction.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestComputePlanAction.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestComputePlanAction.java
new file mode 100644
index 0000000..e10dfc4
--- /dev/null
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestComputePlanAction.java
@@ -0,0 +1,376 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.solr.cloud.autoscaling.sim;
+
+import java.io.IOException;
+import java.lang.invoke.MethodHandles;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.solr.client.solrj.SolrClient;
+import org.apache.solr.client.solrj.SolrRequest;
+import org.apache.solr.client.solrj.SolrResponse;
+import org.apache.solr.client.solrj.cloud.autoscaling.TriggerEventType;
+import org.apache.solr.client.solrj.request.CollectionAdminRequest;
+import org.apache.solr.cloud.autoscaling.ActionContext;
+import org.apache.solr.cloud.autoscaling.ComputePlanAction;
+import org.apache.solr.cloud.autoscaling.ScheduledTriggers;
+import org.apache.solr.cloud.autoscaling.TriggerAction;
+import org.apache.solr.cloud.autoscaling.TriggerEvent;
+import org.apache.solr.common.cloud.ClusterState;
+import org.apache.solr.common.cloud.DocCollection;
+import org.apache.solr.common.cloud.Replica;
+import org.apache.solr.common.cloud.ZkStateReader;
+import org.apache.solr.common.params.CollectionParams;
+import org.apache.solr.common.params.SolrParams;
+import org.apache.solr.common.util.NamedList;
+import org.apache.solr.common.util.TimeSource;
+import org.apache.solr.common.util.Utils;
+import org.apache.solr.util.LogLevel;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.apache.solr.cloud.autoscaling.AutoScalingHandlerTest.createAutoScalingRequest;
+import static org.apache.solr.common.params.CollectionParams.CollectionAction.MOVEREPLICA;
+
+/**
+ * Test for {@link ComputePlanAction}
+ */
+@LogLevel("org.apache.solr.cloud.autoscaling=DEBUG;org.apache.solr.cloud.Overseer=DEBUG;org.apache.solr.cloud.overseer=DEBUG;org.apache.solr.client.solrj.impl.SolrClientDataProvider=DEBUG;")
+public class TestComputePlanAction extends SimSolrCloudTestCase {
+ private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ private static final AtomicBoolean fired = new AtomicBoolean(false);
+ private static final int NODE_COUNT = 1;
+ private static CountDownLatch triggerFiredLatch = new CountDownLatch(1);
+ private static final AtomicReference<Map> actionContextPropsRef = new AtomicReference<>();
+ private static final AtomicReference<TriggerEvent> eventRef = new AtomicReference<>();
+
+ @BeforeClass
+ public static void setupCluster() throws Exception {
+ cluster = SimCloudManager.createCluster(1, TimeSource.get("simTime:50"));
+ }
+
+ @Before
+ public void init() throws Exception {
+
+ fired.set(false);
+ triggerFiredLatch = new CountDownLatch(1);
+ actionContextPropsRef.set(null);
+
+ // remove everything from autoscaling.json in ZK
+ cluster.getDistribStateManager().setData(ZkStateReader.SOLR_AUTOSCALING_CONF_PATH, "{}".getBytes(UTF_8), -1);
+
+ if (cluster.getClusterStateProvider().getLiveNodes().size() > NODE_COUNT) {
+ // stop some to get to original state
+ int numJetties = cluster.getClusterStateProvider().getLiveNodes().size();
+ for (int i = 0; i < numJetties - NODE_COUNT; i++) {
+ String node = cluster.getSimClusterStateProvider().simGetRandomNode(random());
+ cluster.getSimClusterStateProvider().simRemoveNode(node);
+ }
+ }
+
+ try {
+ cluster.request(CollectionAdminRequest.deleteCollection("testNodeLost"));
+ } catch (Exception e) {
+ // expected if testNodeLost hasn't run already
+ }
+ try {
+ cluster.request(CollectionAdminRequest.deleteCollection("testNodeAdded"));
+ } catch (Exception e) {
+ // expected if testNodeAdded hasn't run already
+ }
+ try {
+ cluster.request(CollectionAdminRequest.deleteCollection("testNodeWithMultipleReplicasLost"));
+ } catch (Exception e) {
+ // expected if testNodeWithMultipleReplicasLost hasn't run already
+ }
+
+ String setClusterPolicyCommand = "{" +
+ " 'set-cluster-policy': [" +
+ " {'cores':'<10', 'node':'#ANY'}," +
+ " {'replica':'<2', 'shard': '#EACH', 'node': '#ANY'}," +
+ " {'nodeRole':'overseer', 'replica':0}" +
+ " ]" +
+ "}";
+ SolrRequest req = createAutoScalingRequest(SolrRequest.METHOD.POST, setClusterPolicyCommand);
+ SolrResponse rsp = cluster.request(req);
+ NamedList<Object> response = rsp.getResponse();
+ assertEquals(response.get("result").toString(), "success");
+
+ String setClusterPreferencesCommand = "{" +
+ "'set-cluster-preferences': [" +
+ "{'minimize': 'cores'}," +
+ "{'maximize': 'freedisk','precision': 100}]" +
+ "}";
+ req = createAutoScalingRequest(SolrRequest.METHOD.POST, setClusterPreferencesCommand);
+ rsp = cluster.request(req);
+ response = rsp.getResponse();
+ assertEquals(response.get("result").toString(), "success");
+ cluster.getTimeSource().sleep(ScheduledTriggers.DEFAULT_COOLDOWN_PERIOD_MS);
+ }
+
+ @After
+ public void printState() throws Exception {
+ log.info("-------------_ FINAL STATE --------------");
+ log.info("* Node values: " + Utils.toJSONString(cluster.getSimNodeStateProvider().simGetAllNodeValues()));
+ log.info("* Live nodes: " + cluster.getClusterStateProvider().getLiveNodes());
+ ClusterState state = cluster.getClusterStateProvider().getClusterState();
+ for (String coll: cluster.getSimClusterStateProvider().simListCollections()) {
+ log.info("* Collection " + coll + " state: " + state.getCollection(coll));
+ }
+
+ }
+
+ @Test
+ public void testNodeLost() throws Exception {
+ // let's start a node so that we have at least two
+ String node = cluster.simAddNode();
+ AssertingTriggerAction.expectedNode = node;
+
+ SolrClient solrClient = cluster.simGetSolrClient();
+ String setTriggerCommand = "{" +
+ "'set-trigger' : {" +
+ "'name' : 'node_lost_trigger'," +
+ "'event' : 'nodeLost'," +
+ "'waitFor' : '7s'," +
+ "'enabled' : true," +
+ "'actions' : [{'name':'compute_plan', 'class' : 'solr.ComputePlanAction'}," +
+ "{'name':'test','class':'" + TestComputePlanAction.AssertingTriggerAction.class.getName() + "'}]" +
+ "}}";
+ SolrRequest req = createAutoScalingRequest(SolrRequest.METHOD.POST, setTriggerCommand);
+ NamedList<Object> response = solrClient.request(req);
+ assertEquals(response.get("result").toString(), "success");
+
+ CollectionAdminRequest.Create create = CollectionAdminRequest.createCollection("testNodeLost",
+ "conf",1, 2);
+ create.process(solrClient);
+
+ waitForState("Timed out waiting for replicas of new collection to be active",
+ "testNodeLost", clusterShape(1, 2));
+
+ ClusterState clusterState = cluster.getClusterStateProvider().getClusterState();
+ DocCollection collection = clusterState.getCollection("testNodeLost");
+ List<Replica> replicas = collection.getReplicas(node);
+ assertNotNull(replicas);
+ assertFalse(replicas.isEmpty());
+
+ // start another node because because when the other node goes away, the cluster policy requires only
+ // 1 replica per node and none on the overseer
+ String node2 = cluster.simAddNode();
+ assertTrue(node2 + "is not live yet", cluster.getClusterStateProvider().getClusterState().liveNodesContain(node2) );
+
+ // stop the original node
+ cluster.simRemoveNode(node);
+ log.info("Stopped_node : {}", node);
+
+ assertTrue("Trigger was not fired even after 10 seconds", triggerFiredLatch.await(10, TimeUnit.SECONDS));
+ assertTrue(fired.get());
+ Map context = actionContextPropsRef.get();
+ assertNotNull(context);
+ List<SolrRequest> operations = (List<SolrRequest>) context.get("operations");
+ assertNotNull("The operations computed by ComputePlanAction should not be null , " + eventRef.get(), operations);
+ assertEquals("ComputePlanAction should have computed exactly 1 operation", 1, operations.size());
+ SolrRequest solrRequest = operations.get(0);
+ SolrParams params = solrRequest.getParams();
+ assertEquals("Expected MOVEREPLICA action after adding node", MOVEREPLICA, CollectionParams.CollectionAction.get(params.get("action")));
+ String replicaToBeMoved = params.get("replica");
+ assertEquals("Unexpected node in computed operation", replicas.get(0).getName(), replicaToBeMoved);
+
+ // shutdown the extra node that we had started
+ cluster.simRemoveNode(node2);
+ }
+
+ public void testNodeWithMultipleReplicasLost() throws Exception {
+ AssertingTriggerAction.expectedNode = null;
+
+ // start 3 more nodes
+ cluster.simAddNode();
+ cluster.simAddNode();
+ cluster.simAddNode();
+
+ SolrClient solrClient = cluster.simGetSolrClient();
+ String setTriggerCommand = "{" +
+ "'set-trigger' : {" +
+ "'name' : 'node_lost_trigger'," +
+ "'event' : 'nodeLost'," +
+ "'waitFor' : '1s'," +
+ "'enabled' : true," +
+ "'actions' : [{'name':'compute_plan', 'class' : 'solr.ComputePlanAction'}," +
+ "{'name':'test','class':'" + TestComputePlanAction.AssertingTriggerAction.class.getName() + "'}]" +
+ "}}";
+ SolrRequest req = createAutoScalingRequest(SolrRequest.METHOD.POST, setTriggerCommand);
+ NamedList<Object> response = solrClient.request(req);
+ assertEquals(response.get("result").toString(), "success");
+
+ CollectionAdminRequest.Create create = CollectionAdminRequest.createCollection("testNodeWithMultipleReplicasLost",
+ "conf",2, 3);
+// create.setMaxShardsPerNode(2);
+ create.process(solrClient);
+
+ waitForState("Timed out waiting for replicas of new collection to be active",
+ "testNodeWithMultipleReplicasLost", clusterShape(2, 3));
+
+ ClusterState clusterState = cluster.getClusterStateProvider().getClusterState();
+ DocCollection docCollection = clusterState.getCollection("testNodeWithMultipleReplicasLost");
+
+ // lets find a node with at least 2 replicas
+ String stoppedNodeName = null;
+ List<Replica> replicasToBeMoved = null;
+ for (String node : cluster.getClusterStateProvider().getLiveNodes()) {
+ List<Replica> replicas = docCollection.getReplicas(node);
+ if (replicas != null && replicas.size() == 2) {
+ stoppedNodeName = node;
+ replicasToBeMoved = replicas;
+ cluster.simRemoveNode(node);
+ break;
+ }
+ }
+ assertNotNull(stoppedNodeName);
+
+ assertTrue("Trigger was not fired even after 5 seconds", triggerFiredLatch.await(5, TimeUnit.SECONDS));
+ assertTrue(fired.get());
+
+ TriggerEvent triggerEvent = eventRef.get();
+ assertNotNull(triggerEvent);
+ assertEquals(TriggerEventType.NODELOST, triggerEvent.getEventType());
+ // TODO assertEquals(stoppedNodeName, triggerEvent.getProperty(TriggerEvent.NODE_NAME));
+
+ Map context = actionContextPropsRef.get();
+ assertNotNull(context);
+ List<SolrRequest> operations = (List<SolrRequest>) context.get("operations");
+ assertNotNull("The operations computed by ComputePlanAction should not be null " + actionContextPropsRef.get(), operations);
+ operations.forEach(solrRequest -> log.info(solrRequest.getParams().toString()));
+ assertEquals("ComputePlanAction should have computed exactly 2 operation", 2, operations.size());
+
+ for (SolrRequest solrRequest : operations) {
+ SolrParams params = solrRequest.getParams();
+ assertEquals("Expected MOVEREPLICA action after adding node", MOVEREPLICA, CollectionParams.CollectionAction.get(params.get("action")));
+ String moved = params.get("replica");
+ assertTrue(replicasToBeMoved.stream().anyMatch(replica -> replica.getName().equals(moved)));
+ }
+ }
+
+ @Test
+ public void testNodeAdded() throws Exception {
+ AssertingTriggerAction.expectedNode = null;
+ SolrClient solrClient = cluster.simGetSolrClient();
+ String setTriggerCommand = "{" +
+ "'set-trigger' : {" +
+ "'name' : 'node_added_trigger'," +
+ "'event' : 'nodeAdded'," +
+ "'waitFor' : '1s'," +
+ "'enabled' : true," +
+ "'actions' : [{'name':'compute_plan', 'class' : 'solr.ComputePlanAction'}," +
+ "{'name':'test','class':'" + TestComputePlanAction.AssertingTriggerAction.class.getName() + "'}]" +
+ "}}";
+ SolrRequest req = createAutoScalingRequest(SolrRequest.METHOD.POST, setTriggerCommand);
+ NamedList<Object> response = solrClient.request(req);
+ assertEquals(response.get("result").toString(), "success");
+
+ // the default policy limits 1 replica per node, we need more right now
+ String setClusterPolicyCommand = "{" +
+ " 'set-cluster-policy': [" +
+ " {'cores':'<10', 'node':'#ANY'}," +
+ " {'replica':'<3', 'shard': '#EACH', 'node': '#ANY'}," +
+ " {'nodeRole':'overseer', 'replica':0}" +
+ " ]" +
+ "}";
+ req = createAutoScalingRequest(SolrRequest.METHOD.POST, setClusterPolicyCommand);
+ response = solrClient.request(req);
+ assertEquals(response.get("result").toString(), "success");
+
+ CollectionAdminRequest.Create create = CollectionAdminRequest.createCollection("testNodeAdded",
+ "conf",1, 2);
+ create.process(solrClient);
+
+ waitForState("Timed out waiting for replicas of new collection to be active",
+ "testNodeAdded", (liveNodes, collectionState) -> collectionState.getReplicas().stream().allMatch(replica -> replica.isActive(liveNodes)));
+
+ // reset to the original policy which has only 1 replica per shard per node
+ setClusterPolicyCommand = "{" +
+ " 'set-cluster-policy': [" +
+ " {'cores':'<10', 'node':'#ANY'}," +
+ " {'replica':'<2', 'shard': '#EACH', 'node': '#ANY'}," +
+ " {'nodeRole':'overseer', 'replica':0}" +
+ " ]" +
+ "}";
+ req = createAutoScalingRequest(SolrRequest.METHOD.POST, setClusterPolicyCommand);
+ response = solrClient.request(req);
+ assertEquals(response.get("result").toString(), "success");
+
+ // start a node so that the 'violation' created by the previous policy update is fixed
+ String newNode = cluster.simAddNode();
+ assertTrue("Trigger was not fired even after 5 seconds", triggerFiredLatch.await(5, TimeUnit.SECONDS));
+ assertTrue(fired.get());
+ Map context = actionContextPropsRef.get();
+ assertNotNull(context);
+ log.info("Node values: " + Utils.toJSONString(cluster.getSimNodeStateProvider().simGetAllNodeValues()));
+ log.info("Live nodes: " + cluster.getClusterStateProvider().getLiveNodes() + ", collection state: " + cluster.getClusterStateProvider().getClusterState().getCollection("testNodeAdded"));
+ List<SolrRequest> operations = (List<SolrRequest>) context.get("operations");
+ assertNotNull("The operations computed by ComputePlanAction should not be null" + context, operations);
+ assertEquals("ComputePlanAction should have computed exactly 1 operation, but was: " + operations, 1, operations.size());
+ SolrRequest request = operations.get(0);
+ SolrParams params = request.getParams();
+ assertEquals("Expected MOVEREPLICA action after adding node", MOVEREPLICA, CollectionParams.CollectionAction.get(params.get("action")));
+ String nodeAdded = params.get("targetNode");
+ assertEquals("Unexpected node in computed operation", newNode, nodeAdded);
+ }
+
+ public static class AssertingTriggerAction implements TriggerAction {
+ static String expectedNode;
+
+ @Override
+ public String getName() {
+ return null;
+ }
+
+ @Override
+ public void process(TriggerEvent event, ActionContext context) {
+ if (expectedNode != null) {
+ Collection nodes = (Collection) event.getProperty(TriggerEvent.NODE_NAMES);
+ if (nodes == null || !nodes.contains(expectedNode)) return;//this is not the event we are looking for
+ }
+ if (fired.compareAndSet(false, true)) {
+ eventRef.set(event);
+ actionContextPropsRef.set(context.getProperties());
+ triggerFiredLatch.countDown();
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+
+ }
+
+ @Override
+ public void init(Map<String, String> args) {
+
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a9591e3d/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestExecutePlanAction.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestExecutePlanAction.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestExecutePlanAction.java
new file mode 100644
index 0000000..7045e6e
--- /dev/null
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestExecutePlanAction.java
@@ -0,0 +1,222 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.solr.cloud.autoscaling.sim;
+
+import java.lang.invoke.MethodHandles;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.Collectors;
+
+import com.google.common.collect.Lists;
+import org.apache.solr.client.solrj.SolrClient;
+import org.apache.solr.client.solrj.SolrRequest;
+import org.apache.solr.client.solrj.cloud.autoscaling.TriggerEventType;
+import org.apache.solr.client.solrj.cloud.autoscaling.VersionedData;
+import org.apache.solr.client.solrj.request.CollectionAdminRequest;
+import org.apache.solr.cloud.autoscaling.ActionContext;
+import org.apache.solr.cloud.autoscaling.AutoScalingHandlerTest;
+import org.apache.solr.cloud.autoscaling.ExecutePlanAction;
+import org.apache.solr.cloud.autoscaling.NodeLostTrigger;
+import org.apache.solr.common.cloud.ClusterState;
+import org.apache.solr.common.cloud.DocCollection;
+import org.apache.solr.common.cloud.Replica;
+import org.apache.solr.common.cloud.ZkNodeProps;
+import org.apache.solr.common.cloud.ZkStateReader;
+import org.apache.solr.common.params.CollectionParams;
+import org.apache.solr.common.util.NamedList;
+import org.apache.solr.common.util.Utils;
+import org.apache.solr.util.LogLevel;
+import org.apache.solr.common.util.TimeSource;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.apache.solr.common.cloud.ZkStateReader.SOLR_AUTOSCALING_CONF_PATH;
+
+/**
+ * Test for {@link ExecutePlanAction}
+ */
+@LogLevel("org.apache.solr.cloud.autoscaling=DEBUG")
+public class TestExecutePlanAction extends SimSolrCloudTestCase {
+ private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ private static final int NODE_COUNT = 2;
+
+ @BeforeClass
+ public static void setupCluster() throws Exception {
+ cluster = SimCloudManager.createCluster(NODE_COUNT, TimeSource.get("simTime:50"));
+ }
+
+ @Before
+ public void setUp() throws Exception {
+ super.setUp();
+ // clear any persisted auto scaling configuration
+ cluster.getDistribStateManager().setData(SOLR_AUTOSCALING_CONF_PATH, Utils.toJSON(new ZkNodeProps()), -1);
+
+ if (cluster.getClusterStateProvider().getLiveNodes().size() < NODE_COUNT) {
+ // start some to get to original state
+ int numJetties = cluster.getClusterStateProvider().getLiveNodes().size();
+ for (int i = 0; i < NODE_COUNT - numJetties; i++) {
+ cluster.simAddNode();
+ }
+ }
+ }
+
+ @After
+ public void printState() throws Exception {
+ log.info("-------------_ FINAL STATE --------------");
+ log.info("* Node values: " + Utils.toJSONString(cluster.getSimNodeStateProvider().simGetAllNodeValues()));
+ log.info("* Live nodes: " + cluster.getClusterStateProvider().getLiveNodes());
+ ClusterState state = cluster.getClusterStateProvider().getClusterState();
+ for (String coll: cluster.getSimClusterStateProvider().simListCollections()) {
+ log.info("* Collection " + coll + " state: " + state.getCollection(coll));
+ }
+
+ }
+
+ @Test
+ public void testExecute() throws Exception {
+ SolrClient solrClient = cluster.simGetSolrClient();
+ String collectionName = "testExecute";
+ CollectionAdminRequest.Create create = CollectionAdminRequest.createCollection(collectionName,
+ "conf", 1, 2);
+ create.setMaxShardsPerNode(1);
+ create.process(solrClient);
+
+ waitForState("Timed out waiting for replicas of new collection to be active",
+ collectionName, clusterShape(1, 2));
+
+ String sourceNodeName = cluster.getSimClusterStateProvider().simGetRandomNode(random());
+ ClusterState clusterState = cluster.getClusterStateProvider().getClusterState();
+ DocCollection docCollection = clusterState.getCollection(collectionName);
+ List<Replica> replicas = docCollection.getReplicas(sourceNodeName);
+ assertNotNull(replicas);
+ assertFalse(replicas.isEmpty());
+
+ List<String> otherNodes = cluster.getClusterStateProvider().getLiveNodes().stream()
+ .filter(node -> !node.equals(sourceNodeName)).collect(Collectors.toList());
+ assertFalse(otherNodes.isEmpty());
+ String survivor = otherNodes.get(0);
+
+ try (ExecutePlanAction action = new ExecutePlanAction()) {
+ action.init(Collections.singletonMap("name", "execute_plan"));
+
+ // used to signal if we found that ExecutePlanAction did in fact create the right znode before executing the operation
+ AtomicBoolean znodeCreated = new AtomicBoolean(false);
+
+ CollectionAdminRequest.AsyncCollectionAdminRequest moveReplica = new CollectionAdminRequest.MoveReplica(collectionName, replicas.get(0).getName(), survivor);
+ CollectionAdminRequest.AsyncCollectionAdminRequest mockRequest = new CollectionAdminRequest.AsyncCollectionAdminRequest(CollectionParams.CollectionAction.OVERSEERSTATUS) {
+ @Override
+ public void setAsyncId(String asyncId) {
+ super.setAsyncId(asyncId);
+ String parentPath = ZkStateReader.SOLR_AUTOSCALING_TRIGGER_STATE_PATH + "/xyz/execute_plan";
+ try {
+ if (cluster.getDistribStateManager().hasData(parentPath)) {
+ java.util.List<String> children = cluster.getDistribStateManager().listData(parentPath);
+ if (!children.isEmpty()) {
+ String child = children.get(0);
+ VersionedData data = cluster.getDistribStateManager().getData(parentPath + "/" + child);
+ Map m = (Map) Utils.fromJSON(data.getData());
+ if (m.containsKey("requestid")) {
+ znodeCreated.set(m.get("requestid").equals(asyncId));
+ }
+ }
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+
+ }
+ };
+ List<CollectionAdminRequest.AsyncCollectionAdminRequest> operations = Lists.asList(moveReplica, new CollectionAdminRequest.AsyncCollectionAdminRequest[]{mockRequest});
+ NodeLostTrigger.NodeLostEvent nodeLostEvent = new NodeLostTrigger.NodeLostEvent(TriggerEventType.NODELOST,
+ "mock_trigger_name", Collections.singletonList(TimeSource.CURRENT_TIME.getTime()),
+ Collections.singletonList(sourceNodeName));
+ ActionContext actionContext = new ActionContext(cluster, null,
+ new HashMap<>(Collections.singletonMap("operations", operations)));
+ action.process(nodeLostEvent, actionContext);
+
+// assertTrue("ExecutePlanAction should have stored the requestid in ZK before executing the request", znodeCreated.get());
+ List<NamedList<Object>> responses = (List<NamedList<Object>>) actionContext.getProperty("responses");
+ assertNotNull(responses);
+ assertEquals(2, responses.size());
+ NamedList<Object> response = responses.get(0);
+ assertNull(response.get("failure"));
+ assertNotNull(response.get("success"));
+ }
+
+ waitForState("Timed out waiting for replicas of new collection to be active",
+ collectionName, clusterShape(1, 2));
+ }
+
+ @Test
+ public void testIntegration() throws Exception {
+ SolrClient solrClient = cluster.simGetSolrClient();
+
+ String setTriggerCommand = "{" +
+ "'set-trigger' : {" +
+ "'name' : 'node_lost_trigger'," +
+ "'event' : 'nodeLost'," +
+ "'waitFor' : '1s'," +
+ "'enabled' : true," +
+ "'actions' : [{'name':'compute_plan', 'class' : 'solr.ComputePlanAction'}," +
+ "{'name':'execute_plan','class':'solr.ExecutePlanAction'}]" +
+ "}}";
+ SolrRequest req = AutoScalingHandlerTest.createAutoScalingRequest(SolrRequest.METHOD.POST, setTriggerCommand);
+ NamedList<Object> response = solrClient.request(req);
+ assertEquals(response.get("result").toString(), "success");
+
+ String collectionName = "testIntegration";
+ CollectionAdminRequest.Create create = CollectionAdminRequest.createCollection(collectionName,
+ "conf", 1, 2);
+ create.setMaxShardsPerNode(1);
+ create.process(solrClient);
+
+ waitForState("Timed out waiting for replicas of new collection to be active",
+ collectionName, clusterShape(1, 2));
+
+ String sourceNodeName = cluster.getSimClusterStateProvider().simGetRandomNode(random());
+ ClusterState clusterState = cluster.getClusterStateProvider().getClusterState();
+ DocCollection docCollection = clusterState.getCollection(collectionName);
+ List<Replica> replicas = docCollection.getReplicas(sourceNodeName);
+ assertNotNull(replicas);
+ assertFalse(replicas.isEmpty());
+
+ List<String> otherNodes = cluster.getClusterStateProvider().getLiveNodes().stream()
+ .filter(node -> !node.equals(sourceNodeName)).collect(Collectors.toList());
+ assertFalse(otherNodes.isEmpty());
+ String survivor = otherNodes.get(0);
+
+ cluster.simRemoveNode(sourceNodeName);
+
+ waitForState("Timed out waiting for replicas of collection to be 2 again",
+ collectionName, clusterShape(1, 2));
+
+ clusterState = cluster.getClusterStateProvider().getClusterState();
+ docCollection = clusterState.getCollection(collectionName);
+ List<Replica> replicasOnSurvivor = docCollection.getReplicas(survivor);
+ assertNotNull(replicasOnSurvivor);
+ assertEquals(2, replicasOnSurvivor.size());
+ }
+}