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/13 20:39:04 UTC
[1/2] lucene-solr:jira/solr-11285-sim: SOLR-11403: Improve the
simulation framework to support test cases.
Repository: lucene-solr
Updated Branches:
refs/heads/jira/solr-11285-sim 41a28dc64 -> a44d9cf65
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a44d9cf6/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimSolrCloudTestCase.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimSolrCloudTestCase.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimSolrCloudTestCase.java
new file mode 100644
index 0000000..af64ef7
--- /dev/null
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimSolrCloudTestCase.java
@@ -0,0 +1,180 @@
+package org.apache.solr.cloud.autoscaling.sim;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Predicate;
+
+import org.apache.solr.SolrTestCaseJ4;
+import org.apache.solr.common.cloud.ClusterState;
+import org.apache.solr.common.cloud.CollectionStatePredicate;
+import org.apache.solr.common.cloud.CollectionStateWatcher;
+import org.apache.solr.common.cloud.DocCollection;
+import org.apache.solr.common.cloud.Replica;
+import org.apache.solr.common.cloud.Slice;
+import org.apache.solr.common.cloud.ZkStateReader;
+import org.apache.solr.util.TimeOut;
+import org.apache.zookeeper.WatchedEvent;
+import org.apache.zookeeper.Watcher;
+import org.junit.AfterClass;
+import org.junit.Before;
+
+/**
+ *
+ */
+public class SimSolrCloudTestCase extends SolrTestCaseJ4 {
+
+ public static final int DEFAULT_TIMEOUT = 90;
+
+ /** The cluster. */
+ protected static SimCloudManager cluster;
+
+ protected static SimCloudManager configureCluster(int nodeCount) throws Exception {
+ return SimCloudManager.createCluster(nodeCount);
+ }
+
+ @AfterClass
+ public static void shutdownCluster() throws Exception {
+ if (cluster != null) {
+ cluster.close();
+ }
+ cluster = null;
+ }
+
+ @Before
+ public void checkClusterConfiguration() {
+ if (cluster == null)
+ throw new RuntimeException("MiniSolrCloudCluster not configured - have you called configureCluster().configure()?");
+ }
+
+ /* Cluster helper methods ************************************/
+
+ /**
+ * Get the collection state for a particular collection
+ */
+ protected DocCollection getCollectionState(String collectionName) throws IOException {
+ return cluster.getClusterStateProvider().getClusterState().getCollection(collectionName);
+ }
+
+ /**
+ * Wait for a particular collection state to appear in the cluster client's state reader
+ *
+ * This is a convenience method using the {@link #DEFAULT_TIMEOUT}
+ *
+ * @param message a message to report on failure
+ * @param collection the collection to watch
+ * @param predicate a predicate to match against the collection state
+ */
+ protected void waitForState(String message, String collection, CollectionStatePredicate predicate) {
+ AtomicReference<DocCollection> state = new AtomicReference<>();
+ AtomicReference<Set<String>> liveNodesLastSeen = new AtomicReference<>();
+ try {
+ waitForState(collection, DEFAULT_TIMEOUT, TimeUnit.SECONDS, (n, c) -> {
+ state.set(c);
+ liveNodesLastSeen.set(n);
+ return predicate.matches(n, c);
+ });
+ } catch (Exception e) {
+ fail(message + "\n" + e.getMessage() + "\nLive Nodes: " + liveNodesLastSeen.get() + "\nLast available state: " + state.get());
+ }
+ }
+
+ /**
+ * Block until a CollectionStatePredicate returns true, or the wait times out
+ *
+ * Note that the predicate may be called again even after it has returned true, so
+ * implementors should avoid changing state within the predicate call itself.
+ *
+ * @param collection the collection to watch
+ * @param wait how long to wait
+ * @param unit the units of the wait parameter
+ * @param predicate the predicate to call on state changes
+ * @throws InterruptedException on interrupt
+ * @throws TimeoutException on timeout
+ * @throws IOException on watcher register / unregister error
+ */
+ public void waitForState(final String collection, long wait, TimeUnit unit, CollectionStatePredicate predicate)
+ throws InterruptedException, TimeoutException, IOException {
+ TimeOut timeout = new TimeOut(wait, unit);
+ while (!timeout.hasTimedOut()) {
+ ClusterState state = cluster.getClusterStateProvider().getClusterState();
+ DocCollection coll = state.getCollectionOrNull(collection);
+ if (coll == null) { // already removed or does not exist
+ return;
+ }
+ if (predicate.matches(state.getLiveNodes(), coll)) {
+ return;
+ }
+ Thread.sleep(50);
+ }
+ throw new TimeoutException();
+ }
+
+ /**
+ * Return a {@link CollectionStatePredicate} that returns true if a collection has the expected
+ * number of shards and replicas
+ */
+ public static CollectionStatePredicate clusterShape(int expectedShards, int expectedReplicas) {
+ return (liveNodes, collectionState) -> {
+ if (collectionState == null)
+ return false;
+ if (collectionState.getSlices().size() != expectedShards)
+ return false;
+ for (Slice slice : collectionState) {
+ int activeReplicas = 0;
+ for (Replica replica : slice) {
+ if (replica.isActive(liveNodes))
+ activeReplicas++;
+ }
+ if (activeReplicas != expectedReplicas)
+ return false;
+ }
+ return true;
+ };
+ }
+
+ /**
+ * Get a (reproducibly) random shard from a {@link DocCollection}
+ */
+ protected static Slice getRandomShard(DocCollection collection) {
+ List<Slice> shards = new ArrayList<>(collection.getActiveSlices());
+ if (shards.size() == 0)
+ fail("Couldn't get random shard for collection as it has no shards!\n" + collection.toString());
+ Collections.shuffle(shards, random());
+ return shards.get(0);
+ }
+
+ /**
+ * Get a (reproducibly) random replica from a {@link Slice}
+ */
+ protected static Replica getRandomReplica(Slice slice) {
+ List<Replica> replicas = new ArrayList<>(slice.getReplicas());
+ if (replicas.size() == 0)
+ fail("Couldn't get random replica from shard as it has no replicas!\n" + slice.toString());
+ Collections.shuffle(replicas, random());
+ return replicas.get(0);
+ }
+
+ /**
+ * Get a (reproducibly) random replica from a {@link Slice} matching a predicate
+ */
+ protected static Replica getRandomReplica(Slice slice, Predicate<Replica> matchPredicate) {
+ List<Replica> replicas = new ArrayList<>(slice.getReplicas());
+ if (replicas.size() == 0)
+ fail("Couldn't get random replica from shard as it has no replicas!\n" + slice.toString());
+ Collections.shuffle(replicas, random());
+ for (Replica replica : replicas) {
+ if (matchPredicate.test(replica))
+ return replica;
+ }
+ fail("Couldn't get random replica that matched conditions\n" + slice.toString());
+ return null; // just to keep the compiler happy - fail will always throw an Exception
+ }
+}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a44d9cf6/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestSimClusterStateProvider.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestSimClusterStateProvider.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestSimClusterStateProvider.java
index 679c90e..7d13d1d 100644
--- a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestSimClusterStateProvider.java
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/TestSimClusterStateProvider.java
@@ -7,6 +7,7 @@ import com.google.common.collect.ImmutableSet;
import org.apache.solr.SolrTestCaseJ4;
import org.apache.solr.common.cloud.ClusterState;
import org.junit.Before;
+import org.junit.BeforeClass;
import org.junit.Test;
import static java.nio.charset.StandardCharsets.UTF_8;
@@ -14,9 +15,7 @@ import static java.nio.charset.StandardCharsets.UTF_8;
/**
*
*/
-public class TestSimClusterStateProvider extends SolrTestCaseJ4 {
-
- private SimClusterStateProvider dataProvider;
+public class TestSimClusterStateProvider extends SimSolrCloudTestCase {
private static final String COLL = "gettingstarted";
private static final Set<String> initialLiveNodes = ImmutableSet.of("192.168.1.108:7574_solr", "192.168.1.108:8983_solr");
@@ -63,8 +62,7 @@ public class TestSimClusterStateProvider extends SolrTestCaseJ4 {
public void setup() throws Exception {
ClusterState cs = ClusterState.load(1, coll1State.getBytes(UTF_8),
initialLiveNodes, "/collections/gettingstarted/state.json");
- dataProvider = new SimClusterStateProvider(ConcurrentHashMap.newKeySet());
- dataProvider.simSetClusterState(cs);
+ cluster = SimCloudManager.createCluster(cs);
}
@Test
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a44d9cf6/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/DistribStateManager.java
----------------------------------------------------------------------
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/DistribStateManager.java b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/DistribStateManager.java
index c6a348e..fb5790c 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/DistribStateManager.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/DistribStateManager.java
@@ -48,6 +48,8 @@ public interface DistribStateManager extends SolrCloseable {
void makePath(String path) throws AlreadyExistsException, IOException, KeeperException, InterruptedException;
+ void makePath(String path, byte[] data, CreateMode createMode, boolean failOnExists) throws AlreadyExistsException, IOException, KeeperException, InterruptedException;
+
/**
* Create data (leaf) node at specified path.
* @param path base path name of the node.
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a44d9cf6/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/PolicyHelper.java
----------------------------------------------------------------------
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/PolicyHelper.java b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/PolicyHelper.java
index 60535de..786863c 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/PolicyHelper.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/PolicyHelper.java
@@ -29,7 +29,6 @@ import java.util.concurrent.atomic.AtomicLong;
import org.apache.solr.client.solrj.SolrRequest;
import org.apache.solr.client.solrj.cloud.autoscaling.Suggester.Hint;
import org.apache.solr.client.solrj.impl.ClusterStateProvider;
-import org.apache.solr.client.solrj.impl.SolrClientCloudManager;
import org.apache.solr.common.MapWriter;
import org.apache.solr.common.SolrException;
import org.apache.solr.common.cloud.Replica;
@@ -110,7 +109,7 @@ public class PolicyHelper {
public static final int SESSION_EXPIRY = 180;//3 seconds
public static ThreadLocal<Long> REF_VERSION = new ThreadLocal<>();
- public static MapWriter getDiagnostics(Policy policy, SolrClientCloudManager cloudManager) {
+ public static MapWriter getDiagnostics(Policy policy, SolrCloudManager cloudManager) {
Policy.Session session = policy.createSession(cloudManager);
List<Row> sorted = session.getSorted();
List<Violation> violations = session.getViolations();
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a44d9cf6/solr/solrj/src/java/org/apache/solr/client/solrj/impl/CloudSolrClient.java
----------------------------------------------------------------------
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/CloudSolrClient.java b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/CloudSolrClient.java
index d746fe6..bced5c7 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/CloudSolrClient.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/CloudSolrClient.java
@@ -81,6 +81,7 @@ import org.apache.solr.common.util.NamedList;
import org.apache.solr.common.util.SimpleOrderedMap;
import org.apache.solr.common.util.SolrjNamedThreadFactory;
import org.apache.solr.common.util.StrUtils;
+import org.apache.solr.common.util.Utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
@@ -1026,13 +1027,13 @@ public class CloudSolrClient extends SolrClient {
if (!liveNodes.isEmpty()) {
List<String> liveNodesList = new ArrayList<>(liveNodes);
Collections.shuffle(liveNodesList, rand);
- theUrlList.add(ZkStateReader.getBaseUrlForNodeName(liveNodesList.get(0),
+ theUrlList.add(Utils.getBaseUrlForNodeName(liveNodesList.get(0),
(String) stateProvider.getClusterProperty(ZkStateReader.URL_SCHEME,"http")));
}
} else if (ADMIN_PATHS.contains(request.getPath())) {
for (String liveNode : liveNodes) {
- theUrlList.add(ZkStateReader.getBaseUrlForNodeName(liveNode,
+ theUrlList.add(Utils.getBaseUrlForNodeName(liveNode,
(String) stateProvider.getClusterProperty(ZkStateReader.URL_SCHEME,"http")));
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a44d9cf6/solr/solrj/src/java/org/apache/solr/client/solrj/impl/HttpClusterStateProvider.java
----------------------------------------------------------------------
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/HttpClusterStateProvider.java b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/HttpClusterStateProvider.java
index d617d16..b623157 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/HttpClusterStateProvider.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/HttpClusterStateProvider.java
@@ -39,6 +39,7 @@ import org.apache.solr.common.cloud.ZkStateReader;
import org.apache.solr.common.params.ModifiableSolrParams;
import org.apache.solr.common.util.NamedList;
import org.apache.solr.common.util.SimpleOrderedMap;
+import org.apache.solr.common.util.Utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -90,7 +91,7 @@ public class HttpClusterStateProvider implements ClusterStateProvider {
public CollectionRef getState(String collection) {
for (String nodeName: liveNodes) {
try (HttpSolrClient client = new HttpSolrClient.Builder().
- withBaseSolrUrl(ZkStateReader.getBaseUrlForNodeName(nodeName, urlScheme)).
+ withBaseSolrUrl(Utils.getBaseUrlForNodeName(nodeName, urlScheme)).
withHttpClient(httpClient).build()) {
ClusterState cs = fetchClusterState(client, collection, null);
return cs.getCollectionRef(collection);
@@ -102,7 +103,7 @@ public class HttpClusterStateProvider implements ClusterStateProvider {
return null;
}
log.warn("Attempt to fetch cluster state from " +
- ZkStateReader.getBaseUrlForNodeName(nodeName, urlScheme) + " failed.", e);
+ Utils.getBaseUrlForNodeName(nodeName, urlScheme) + " failed.", e);
}
}
throw new RuntimeException("Tried fetching cluster state using the node names we knew of, i.e. " + liveNodes +". However, "
@@ -160,7 +161,7 @@ public class HttpClusterStateProvider implements ClusterStateProvider {
if (TimeUnit.SECONDS.convert((System.nanoTime() - liveNodesTimestamp), TimeUnit.NANOSECONDS) > getCacheTimeout()) {
for (String nodeName: liveNodes) {
try (HttpSolrClient client = new HttpSolrClient.Builder().
- withBaseSolrUrl(ZkStateReader.getBaseUrlForNodeName(nodeName, urlScheme)).
+ withBaseSolrUrl(Utils.getBaseUrlForNodeName(nodeName, urlScheme)).
withHttpClient(httpClient).build()) {
Set<String> liveNodes = fetchLiveNodes(client);
this.liveNodes = (liveNodes);
@@ -168,7 +169,7 @@ public class HttpClusterStateProvider implements ClusterStateProvider {
return liveNodes;
} catch (Exception e) {
log.warn("Attempt to fetch live_nodes from " +
- ZkStateReader.getBaseUrlForNodeName(nodeName, urlScheme) + " failed.", e);
+ Utils.getBaseUrlForNodeName(nodeName, urlScheme) + " failed.", e);
}
}
throw new RuntimeException("Tried fetching live_nodes using all the node names we knew of, i.e. " + liveNodes +". However, "
@@ -209,7 +210,7 @@ public class HttpClusterStateProvider implements ClusterStateProvider {
TimeUnit.SECONDS.convert((System.nanoTime() - aliasesTimestamp), TimeUnit.NANOSECONDS) > getCacheTimeout()) {
for (String nodeName: liveNodes) {
try (HttpSolrClient client = new HttpSolrClient.Builder().
- withBaseSolrUrl(ZkStateReader.getBaseUrlForNodeName(nodeName, urlScheme)).
+ withBaseSolrUrl(Utils.getBaseUrlForNodeName(nodeName, urlScheme)).
withHttpClient(httpClient).build()) {
Map<String, List<String>> aliases = new CollectionAdminRequest.ListAliases().process(client).getAliasesAsLists();
@@ -226,7 +227,7 @@ public class HttpClusterStateProvider implements ClusterStateProvider {
return aliases;
}
log.warn("Attempt to fetch cluster state from " +
- ZkStateReader.getBaseUrlForNodeName(nodeName, urlScheme) + " failed.", e);
+ Utils.getBaseUrlForNodeName(nodeName, urlScheme) + " failed.", e);
}
}
@@ -244,13 +245,13 @@ public class HttpClusterStateProvider implements ClusterStateProvider {
public ClusterState getClusterState() throws IOException {
for (String nodeName: liveNodes) {
try (HttpSolrClient client = new HttpSolrClient.Builder().
- withBaseSolrUrl(ZkStateReader.getBaseUrlForNodeName(nodeName, urlScheme)).
+ withBaseSolrUrl(Utils.getBaseUrlForNodeName(nodeName, urlScheme)).
withHttpClient(httpClient).build()) {
ClusterState cs = fetchClusterState(client, null, null);
return cs;
} catch (SolrServerException | RemoteSolrException | IOException e) {
log.warn("Attempt to fetch cluster state from " +
- ZkStateReader.getBaseUrlForNodeName(nodeName, urlScheme) + " failed.", e);
+ Utils.getBaseUrlForNodeName(nodeName, urlScheme) + " failed.", e);
}
}
throw new RuntimeException("Tried fetching cluster state using the node names we knew of, i.e. " + liveNodes +". However, "
@@ -264,14 +265,14 @@ public class HttpClusterStateProvider implements ClusterStateProvider {
public Map<String, Object> getClusterProperties() {
for (String nodeName: liveNodes) {
try (HttpSolrClient client = new HttpSolrClient.Builder().
- withBaseSolrUrl(ZkStateReader.getBaseUrlForNodeName(nodeName, urlScheme)).
+ withBaseSolrUrl(Utils.getBaseUrlForNodeName(nodeName, urlScheme)).
withHttpClient(httpClient).build()) {
Map<String, Object> clusterProperties = new HashMap<>();
fetchClusterState(client, null, clusterProperties);
return clusterProperties;
} catch (SolrServerException | RemoteSolrException | IOException e) {
log.warn("Attempt to fetch cluster state from " +
- ZkStateReader.getBaseUrlForNodeName(nodeName, urlScheme) + " failed.", e);
+ Utils.getBaseUrlForNodeName(nodeName, urlScheme) + " failed.", e);
}
}
throw new RuntimeException("Tried fetching cluster state using the node names we knew of, i.e. " + liveNodes +". However, "
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a44d9cf6/solr/solrj/src/java/org/apache/solr/client/solrj/impl/ZkDistribStateManager.java
----------------------------------------------------------------------
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/ZkDistribStateManager.java b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/ZkDistribStateManager.java
index 6abc4ba..94c0a5a 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/ZkDistribStateManager.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/ZkDistribStateManager.java
@@ -96,6 +96,17 @@ public class ZkDistribStateManager implements DistribStateManager {
}
@Override
+ public void makePath(String path, byte[] data, CreateMode createMode, boolean failOnExists) throws AlreadyExistsException, IOException, KeeperException, InterruptedException {
+ try {
+ zkClient.makePath(path, data, createMode, null, failOnExists, true);
+ } catch (KeeperException.NodeExistsException e) {
+ throw new AlreadyExistsException(path);
+ } catch (InterruptedException e) {
+ throw e;
+ }
+ }
+
+ @Override
public String createData(String path, byte[] data, CreateMode mode) throws NoSuchElementException, AlreadyExistsException, IOException, KeeperException, InterruptedException {
try {
return zkClient.create(path, data, mode, true);
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a44d9cf6/solr/solrj/src/java/org/apache/solr/common/cloud/ZkStateReader.java
----------------------------------------------------------------------
diff --git a/solr/solrj/src/java/org/apache/solr/common/cloud/ZkStateReader.java b/solr/solrj/src/java/org/apache/solr/common/cloud/ZkStateReader.java
index b70d2f1..cd6425b 100644
--- a/solr/solrj/src/java/org/apache/solr/common/cloud/ZkStateReader.java
+++ b/solr/solrj/src/java/org/apache/solr/common/cloud/ZkStateReader.java
@@ -17,9 +17,7 @@
package org.apache.solr.common.cloud;
import java.io.Closeable;
-import java.io.UnsupportedEncodingException;
import java.lang.invoke.MethodHandles;
-import java.net.URLDecoder;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
@@ -998,21 +996,7 @@ public class ZkStateReader implements Closeable {
* @lucene.experimental
*/
public String getBaseUrlForNodeName(final String nodeName) {
- return getBaseUrlForNodeName(nodeName, getClusterProperty(URL_SCHEME, "http"));
- }
-
- public static String getBaseUrlForNodeName(final String nodeName, String urlScheme) {
- final int _offset = nodeName.indexOf("_");
- if (_offset < 0) {
- throw new IllegalArgumentException("nodeName does not contain expected '_' separator: " + nodeName);
- }
- final String hostAndPort = nodeName.substring(0,_offset);
- try {
- final String path = URLDecoder.decode(nodeName.substring(1+_offset), "UTF-8");
- return urlScheme + "://" + hostAndPort + (path.isEmpty() ? "" : ("/" + path));
- } catch (UnsupportedEncodingException e) {
- throw new IllegalStateException("JVM Does not seem to support UTF-8", e);
- }
+ return Utils.getBaseUrlForNodeName(nodeName, getClusterProperty(URL_SCHEME, "http"));
}
/** Watches a single collection's format2 state.json. */
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a44d9cf6/solr/solrj/src/java/org/apache/solr/common/util/Utils.java
----------------------------------------------------------------------
diff --git a/solr/solrj/src/java/org/apache/solr/common/util/Utils.java b/solr/solrj/src/java/org/apache/solr/common/util/Utils.java
index a7d8cf3..2e00424 100644
--- a/solr/solrj/src/java/org/apache/solr/common/util/Utils.java
+++ b/solr/solrj/src/java/org/apache/solr/common/util/Utils.java
@@ -21,7 +21,9 @@ import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.Reader;
import java.io.StringReader;
+import java.io.UnsupportedEncodingException;
import java.lang.invoke.MethodHandles;
+import java.net.URLDecoder;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
@@ -426,4 +428,18 @@ public class Utils {
}
}
}
+
+ public static String getBaseUrlForNodeName(final String nodeName, String urlScheme) {
+ final int _offset = nodeName.indexOf("_");
+ if (_offset < 0) {
+ throw new IllegalArgumentException("nodeName does not contain expected '_' separator: " + nodeName);
+ }
+ final String hostAndPort = nodeName.substring(0,_offset);
+ try {
+ final String path = URLDecoder.decode(nodeName.substring(1+_offset), "UTF-8");
+ return urlScheme + "://" + hostAndPort + (path.isEmpty() ? "" : ("/" + path));
+ } catch (UnsupportedEncodingException e) {
+ throw new IllegalStateException("JVM Does not seem to support UTF-8", e);
+ }
+ }
}
[2/2] lucene-solr:jira/solr-11285-sim: SOLR-11403: Improve the
simulation framework to support test cases.
Posted by ab...@apache.org.
SOLR-11403: Improve the simulation framework to support test cases.
Project: http://git-wip-us.apache.org/repos/asf/lucene-solr/repo
Commit: http://git-wip-us.apache.org/repos/asf/lucene-solr/commit/a44d9cf6
Tree: http://git-wip-us.apache.org/repos/asf/lucene-solr/tree/a44d9cf6
Diff: http://git-wip-us.apache.org/repos/asf/lucene-solr/diff/a44d9cf6
Branch: refs/heads/jira/solr-11285-sim
Commit: a44d9cf6554b1af2859cc455d59b49e82c277538
Parents: 41a28dc
Author: Andrzej Bialecki <ab...@apache.org>
Authored: Mon Nov 13 21:38:29 2017 +0100
Committer: Andrzej Bialecki <ab...@apache.org>
Committed: Mon Nov 13 21:38:29 2017 +0100
----------------------------------------------------------------------
.../org/apache/solr/cloud/AddReplicaCmd.java | 6 +-
.../solr/cloud/CloudConfigSetService.java | 2 +-
.../apache/solr/cloud/CreateCollectionCmd.java | 60 ++-
.../java/org/apache/solr/cloud/Overseer.java | 19 +-
.../cloud/OverseerCollectionMessageHandler.java | 24 +-
.../cloud/autoscaling/AutoScalingHandler.java | 66 ++--
.../org/apache/solr/core/CoreContainer.java | 4 +-
.../cloud/autoscaling/sim/SimCloudManager.java | 152 +++++++-
.../sim/SimClusterStateProvider.java | 191 ++++++++--
.../sim/SimComputePlanActionTest.java | 364 +++++++++++++++++++
.../autoscaling/sim/SimDistribStateManager.java | 41 ++-
.../autoscaling/sim/SimSolrCloudTestCase.java | 180 +++++++++
.../sim/TestSimClusterStateProvider.java | 8 +-
.../cloud/autoscaling/DistribStateManager.java | 2 +
.../solrj/cloud/autoscaling/PolicyHelper.java | 3 +-
.../solr/client/solrj/impl/CloudSolrClient.java | 5 +-
.../solrj/impl/HttpClusterStateProvider.java | 21 +-
.../solrj/impl/ZkDistribStateManager.java | 11 +
.../apache/solr/common/cloud/ZkStateReader.java | 18 +-
.../java/org/apache/solr/common/util/Utils.java | 16 +
20 files changed, 1014 insertions(+), 179 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a44d9cf6/solr/core/src/java/org/apache/solr/cloud/AddReplicaCmd.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/AddReplicaCmd.java b/solr/core/src/java/org/apache/solr/cloud/AddReplicaCmd.java
index e6e7ae7..5564bb3 100644
--- a/solr/core/src/java/org/apache/solr/cloud/AddReplicaCmd.java
+++ b/solr/core/src/java/org/apache/solr/cloud/AddReplicaCmd.java
@@ -69,7 +69,7 @@ public class AddReplicaCmd implements OverseerCollectionMessageHandler.Cmd {
@Override
public void call(ClusterState state, ZkNodeProps message, NamedList results) throws Exception {
- addReplica(ocmh.zkStateReader.getClusterState(), message, results, null);
+ addReplica(ocmh.cloudManager.getClusterStateProvider().getClusterState(), message, results, null);
}
ZkNodeProps addReplica(ClusterState clusterState, ZkNodeProps message, NamedList results, Runnable onComplete)
@@ -124,7 +124,7 @@ public class AddReplicaCmd implements OverseerCollectionMessageHandler.Cmd {
}
} else {
node = getNodesForNewReplicas(clusterState, collection, shard, 1, node,
- ocmh.overseer.getSolrCloudManager()).get(0).nodeName;// TODO: use replica type in this logic too
+ ocmh.cloudManager).get(0).nodeName;// TODO: use replica type in this logic too
}
}
log.info("Node Identified {} for creating new replica", node);
@@ -133,7 +133,7 @@ public class AddReplicaCmd implements OverseerCollectionMessageHandler.Cmd {
throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, "Node: " + node + " is not live");
}
if (coreName == null) {
- coreName = Assign.buildCoreName(ocmh.overseer.getSolrCloudManager().getDistribStateManager(), coll, shard, replicaType);
+ coreName = Assign.buildCoreName(ocmh.cloudManager.getDistribStateManager(), coll, shard, replicaType);
} else if (!skipCreateReplicaInClusterState) {
//Validate that the core name is unique in that collection
for (Slice slice : coll.getSlices()) {
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a44d9cf6/solr/core/src/java/org/apache/solr/cloud/CloudConfigSetService.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/CloudConfigSetService.java b/solr/core/src/java/org/apache/solr/cloud/CloudConfigSetService.java
index 6e0583f..3cdc903 100644
--- a/solr/core/src/java/org/apache/solr/cloud/CloudConfigSetService.java
+++ b/solr/core/src/java/org/apache/solr/cloud/CloudConfigSetService.java
@@ -42,7 +42,7 @@ public class CloudConfigSetService extends ConfigSetService {
try {
// for back compat with cores that can create collections without the collections API
if (!zkController.getZkClient().exists(ZkStateReader.COLLECTIONS_ZKNODE + "/" + cd.getCollectionName(), true)) {
- CreateCollectionCmd.createCollectionZkNode(zkController.getZkClient(), cd.getCollectionName(), cd.getCloudDescriptor().getParams());
+ CreateCollectionCmd.createCollectionZkNode(zkController.getSolrCloudManager().getDistribStateManager(), cd.getCollectionName(), cd.getCloudDescriptor().getParams());
}
} catch (KeeperException e) {
SolrException.log(log, null, e);
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a44d9cf6/solr/core/src/java/org/apache/solr/cloud/CreateCollectionCmd.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/CreateCollectionCmd.java b/solr/core/src/java/org/apache/solr/cloud/CreateCollectionCmd.java
index 3925a7e..5cadf6e 100644
--- a/solr/core/src/java/org/apache/solr/cloud/CreateCollectionCmd.java
+++ b/solr/core/src/java/org/apache/solr/cloud/CreateCollectionCmd.java
@@ -18,6 +18,7 @@
package org.apache.solr.cloud;
+import java.io.IOException;
import java.lang.invoke.MethodHandles;
import java.util.ArrayList;
import java.util.Collections;
@@ -26,11 +27,15 @@ import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import java.util.NoSuchElementException;
import java.util.Properties;
import java.util.concurrent.TimeUnit;
+import org.apache.solr.client.solrj.cloud.autoscaling.AlreadyExistsException;
+import org.apache.solr.client.solrj.cloud.autoscaling.DistribStateManager;
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.VersionedData;
import org.apache.solr.cloud.OverseerCollectionMessageHandler.Cmd;
import org.apache.solr.cloud.overseer.ClusterStateMutator;
import org.apache.solr.common.SolrException;
@@ -40,7 +45,6 @@ import org.apache.solr.common.cloud.DocRouter;
import org.apache.solr.common.cloud.ImplicitDocRouter;
import org.apache.solr.common.cloud.Replica;
import org.apache.solr.common.cloud.ReplicaPosition;
-import org.apache.solr.common.cloud.SolrZkClient;
import org.apache.solr.common.cloud.ZkConfigManager;
import org.apache.solr.common.cloud.ZkNodeProps;
import org.apache.solr.common.cloud.ZkStateReader;
@@ -81,11 +85,11 @@ import static org.apache.solr.common.util.StrUtils.formatString;
public class CreateCollectionCmd implements Cmd {
private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
private final OverseerCollectionMessageHandler ocmh;
- private SolrZkClient zkClient;
+ private final DistribStateManager stateManager;
public CreateCollectionCmd(OverseerCollectionMessageHandler ocmh) {
this.ocmh = ocmh;
- this.zkClient = ocmh.zkStateReader.getZkClient();
+ this.stateManager = ocmh.cloudManager.getDistribStateManager();
}
@Override
@@ -148,8 +152,8 @@ public class CreateCollectionCmd implements Cmd {
// add our new cores to existing nodes serving the least number of cores
// but (for now) require that each core goes on a distinct node.
- final List<String> nodeList = Assign.getLiveOrLiveAndCreateNodeSetList(clusterState.getLiveNodes(), message, RANDOM);
List<ReplicaPosition> replicaPositions;
+ final List<String> nodeList = Assign.getLiveOrLiveAndCreateNodeSetList(clusterState.getLiveNodes(), message, RANDOM);
if (nodeList.isEmpty()) {
log.warn("It is unusual to create a collection ("+collectionName+") without cores.");
@@ -189,7 +193,7 @@ public class CreateCollectionCmd implements Cmd {
ZkStateReader zkStateReader = ocmh.zkStateReader;
boolean isLegacyCloud = Overseer.isLegacy(zkStateReader);
- ocmh.createConfNode(configName, collectionName, isLegacyCloud);
+ ocmh.createConfNode(stateManager, configName, collectionName, isLegacyCloud);
Map<String,String> collectionParams = new HashMap<>();
Map<String,Object> collectionProps = message.getProperties();
@@ -199,7 +203,7 @@ public class CreateCollectionCmd implements Cmd {
}
}
- createCollectionZkNode(zkClient, collectionName, collectionParams);
+ createCollectionZkNode(stateManager, collectionName, collectionParams);
Overseer.getStateUpdateQueue(zkStateReader.getZkClient()).offer(Utils.toJSON(message));
@@ -208,7 +212,7 @@ public class CreateCollectionCmd implements Cmd {
boolean created = false;
while (! waitUntil.hasTimedOut()) {
Thread.sleep(100);
- created = zkStateReader.getClusterState().hasCollection(collectionName);
+ created = ocmh.cloudManager.getClusterStateProvider().getClusterState().hasCollection(collectionName);
if(created) break;
}
if (!created)
@@ -228,7 +232,8 @@ public class CreateCollectionCmd implements Cmd {
Map<String,ShardRequest> coresToCreate = new LinkedHashMap<>();
for (ReplicaPosition replicaPosition : replicaPositions) {
String nodeName = replicaPosition.node;
- String coreName = Assign.buildCoreName(ocmh.overseer.getSolrCloudManager().getDistribStateManager(), zkStateReader.getClusterState().getCollection(collectionName),
+ String coreName = Assign.buildCoreName(ocmh.cloudManager.getDistribStateManager(),
+ ocmh.cloudManager.getClusterStateProvider().getClusterState().getCollection(collectionName),
replicaPosition.shard, replicaPosition.type, true);
log.debug(formatString("Creating core {0} as part of shard {1} of collection {2} on {3}"
, coreName, replicaPosition.shard, collectionName, nodeName));
@@ -373,12 +378,12 @@ public class CreateCollectionCmd implements Cmd {
}
}
- public static void createCollectionZkNode(SolrZkClient zkClient, String collection, Map<String,String> params) {
+ public static void createCollectionZkNode(DistribStateManager stateManager, String collection, Map<String,String> params) {
log.debug("Check for collection zkNode:" + collection);
String collectionPath = ZkStateReader.COLLECTIONS_ZKNODE + "/" + collection;
try {
- if (!zkClient.exists(collectionPath, true)) {
+ if (!stateManager.hasData(collectionPath)) {
log.debug("Creating collection in ZooKeeper:" + collection);
try {
@@ -392,7 +397,7 @@ public class CreateCollectionCmd implements Cmd {
// if the config name wasn't passed in, use the default
if (!collectionProps.containsKey(ZkController.CONFIGNAME_PROP)) {
// users can create the collection node and conf link ahead of time, or this may return another option
- getConfName(zkClient, collection, collectionPath, collectionProps);
+ getConfName(stateManager, collection, collectionPath, collectionProps);
}
} else if (System.getProperty("bootstrap_confdir") != null) {
@@ -415,19 +420,21 @@ public class CreateCollectionCmd implements Cmd {
// the conf name should should be the collection name of this core
collectionProps.put(ZkController.CONFIGNAME_PROP, collection);
} else {
- getConfName(zkClient, collection, collectionPath, collectionProps);
+ getConfName(stateManager, collection, collectionPath, collectionProps);
}
collectionProps.remove(ZkStateReader.NUM_SHARDS_PROP); // we don't put numShards in the collections properties
ZkNodeProps zkProps = new ZkNodeProps(collectionProps);
- zkClient.makePath(collectionPath, Utils.toJSON(zkProps), CreateMode.PERSISTENT, null, true);
+ stateManager.makePath(collectionPath, Utils.toJSON(zkProps), CreateMode.PERSISTENT, false);
} catch (KeeperException e) {
// it's okay if the node already exists
if (e.code() != KeeperException.Code.NODEEXISTS) {
throw e;
}
+ } catch (AlreadyExistsException e) {
+ // it's okay if the node already exists
}
} else {
log.debug("Collection zkNode exists");
@@ -439,6 +446,8 @@ public class CreateCollectionCmd implements Cmd {
return;
}
throw new SolrException(ErrorCode.SERVER_ERROR, "Error creating collection node in Zookeeper", e);
+ } catch (IOException e) {
+ throw new SolrException(ErrorCode.SERVER_ERROR, "Error creating collection node in Zookeeper", e);
} catch (InterruptedException e) {
Thread.interrupted();
throw new SolrException(ErrorCode.SERVER_ERROR, "Error creating collection node in Zookeeper", e);
@@ -446,8 +455,8 @@ public class CreateCollectionCmd implements Cmd {
}
- private static void getConfName(SolrZkClient zkClient, String collection, String collectionPath, Map<String,Object> collectionProps) throws KeeperException,
- InterruptedException {
+ private static void getConfName(DistribStateManager stateManager, String collection, String collectionPath, Map<String,Object> collectionProps) throws IOException,
+ KeeperException, InterruptedException {
// check for configName
log.debug("Looking for collection configName");
if (collectionProps.containsKey("configName")) {
@@ -459,17 +468,17 @@ public class CreateCollectionCmd implements Cmd {
int retry = 1;
int retryLimt = 6;
for (; retry < retryLimt; retry++) {
- if (zkClient.exists(collectionPath, true)) {
- ZkNodeProps cProps = ZkNodeProps.load(zkClient.getData(collectionPath, null, null, true));
+ if (stateManager.hasData(collectionPath)) {
+ VersionedData data = stateManager.getData(collectionPath);
+ ZkNodeProps cProps = ZkNodeProps.load(data.getData());
if (cProps.containsKey(ZkController.CONFIGNAME_PROP)) {
break;
}
}
try {
- configNames = zkClient.getChildren(ZkConfigManager.CONFIGS_ZKNODE, null,
- true);
- } catch (NoNodeException e) {
+ configNames = stateManager.listData(ZkConfigManager.CONFIGS_ZKNODE);
+ } catch (NoSuchElementException | NoNodeException e) {
// just keep trying
}
@@ -505,15 +514,4 @@ public class CreateCollectionCmd implements Cmd {
"Could not find configName for collection " + collection + " found:" + configNames);
}
}
-
- public static boolean usePolicyFramework(ZkStateReader zkStateReader, ZkNodeProps message) {
- Map autoScalingJson = Collections.emptyMap();
- try {
- autoScalingJson = Utils.getJson(zkStateReader.getZkClient(), SOLR_AUTOSCALING_CONF_PATH, true);
- } catch (Exception e) {
- return false;
- }
- return autoScalingJson.get(Policy.CLUSTER_POLICY) != null || message.getStr(Policy.POLICY) != null;
- }
-
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a44d9cf6/solr/core/src/java/org/apache/solr/cloud/Overseer.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/Overseer.java b/solr/core/src/java/org/apache/solr/cloud/Overseer.java
index eed8d41..d1bb13a 100644
--- a/solr/core/src/java/org/apache/solr/cloud/Overseer.java
+++ b/solr/core/src/java/org/apache/solr/cloud/Overseer.java
@@ -49,6 +49,7 @@ import org.apache.solr.common.util.ObjectReleaseTracker;
import org.apache.solr.common.util.Pair;
import org.apache.solr.common.util.Utils;
import org.apache.solr.core.CloudConfig;
+import org.apache.solr.core.CoreContainer;
import org.apache.solr.handler.admin.CollectionsHandler;
import org.apache.solr.handler.component.ShardHandler;
import org.apache.solr.update.UpdateShardHandler;
@@ -79,7 +80,7 @@ public class Overseer implements SolrCloseable {
enum LeaderStatus {DONT_KNOW, NO, YES}
private class ClusterStateUpdater implements Runnable, Closeable {
-
+
private final ZkStateReader reader;
private final SolrZkClient zkClient;
private final String myId;
@@ -87,7 +88,7 @@ public class Overseer implements SolrCloseable {
private final ZkDistributedQueue stateUpdateQueue;
//TODO remove in 9.0, we do not push message into this queue anymore
//Internal queue where overseer stores events that have not yet been published into cloudstate
- //If Overseer dies while extracting the main queue a new overseer will start from this queue
+ //If Overseer dies while extracting the main queue a new overseer will start from this queue
private final ZkDistributedQueue workQueue;
// Internal map which holds the information about running tasks.
private final DistributedMap runningMap;
@@ -119,7 +120,7 @@ public class Overseer implements SolrCloseable {
public Stats getWorkQueueStats() {
return workQueue.getZkStats();
}
-
+
@Override
public void run() {
@@ -465,9 +466,9 @@ public class Overseer implements SolrCloseable {
public boolean isClosed() {
return this.isClosed;
}
-
+
}
-
+
private OverseerThread ccThread;
private OverseerThread updaterThread;
@@ -477,7 +478,7 @@ public class Overseer implements SolrCloseable {
private final ZkStateReader reader;
private final ShardHandler shardHandler;
-
+
private final UpdateShardHandler updateShardHandler;
private final String adminPath;
@@ -504,7 +505,7 @@ public class Overseer implements SolrCloseable {
this.stats = new Stats();
this.config = config;
}
-
+
public synchronized void start(String id) {
this.id = id;
closed = false;
@@ -543,6 +544,10 @@ public class Overseer implements SolrCloseable {
return zkController;
}
+ public CoreContainer getCoreContainer() {
+ return zkController.getCoreContainer();
+ }
+
public SolrCloudManager getSolrCloudManager() {
return zkController.getSolrCloudManager();
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a44d9cf6/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 f4e43b0..4d415a1 100644
--- a/solr/core/src/java/org/apache/solr/cloud/OverseerCollectionMessageHandler.java
+++ b/solr/core/src/java/org/apache/solr/cloud/OverseerCollectionMessageHandler.java
@@ -36,7 +36,11 @@ import org.apache.commons.lang.StringUtils;
import org.apache.solr.client.solrj.SolrResponse;
import org.apache.solr.client.solrj.SolrServerException;
import org.apache.solr.client.solrj.cloud.DistributedQueue;
+import org.apache.solr.client.solrj.cloud.autoscaling.AlreadyExistsException;
+import org.apache.solr.client.solrj.cloud.autoscaling.BadVersionException;
+import org.apache.solr.client.solrj.cloud.autoscaling.DistribStateManager;
import org.apache.solr.client.solrj.cloud.autoscaling.PolicyHelper;
+import org.apache.solr.client.solrj.cloud.autoscaling.SolrCloudManager;
import org.apache.solr.client.solrj.impl.HttpSolrClient;
import org.apache.solr.client.solrj.impl.HttpSolrClient.RemoteSolrException;
import org.apache.solr.client.solrj.request.AbstractUpdateRequest;
@@ -74,6 +78,7 @@ import org.apache.solr.handler.component.ShardResponse;
import org.apache.solr.util.DefaultSolrThreadFactory;
import org.apache.solr.util.RTimer;
import org.apache.solr.util.TimeOut;
+import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -144,6 +149,7 @@ public class OverseerCollectionMessageHandler implements OverseerMessageHandler,
ShardHandlerFactory shardHandlerFactory;
String adminPath;
ZkStateReader zkStateReader;
+ SolrCloudManager cloudManager;
String myId;
Stats stats;
@@ -183,6 +189,7 @@ public class OverseerCollectionMessageHandler implements OverseerMessageHandler,
this.myId = myId;
this.stats = stats;
this.overseer = overseer;
+ this.cloudManager = overseer.getSolrCloudManager();
this.isClosed = false;
commandMap = new ImmutableMap.Builder<CollectionAction, Cmd>()
.put(REPLACENODE, new ReplaceNodeCmd(this))
@@ -229,7 +236,7 @@ public class OverseerCollectionMessageHandler implements OverseerMessageHandler,
CollectionAction action = getCollectionAction(operation);
Cmd command = commandMap.get(action);
if (command != null) {
- command.call(zkStateReader.getClusterState(), message, results);
+ command.call(cloudManager.getClusterStateProvider().getClusterState(), message, results);
} else {
throw new SolrException(ErrorCode.BAD_REQUEST, "Unknown operation:"
+ operation);
@@ -642,7 +649,7 @@ public class OverseerCollectionMessageHandler implements OverseerMessageHandler,
validateConfigOrThrowSolrException(configName);
boolean isLegacyCloud = Overseer.isLegacy(zkStateReader);
- createConfNode(configName, collectionName, isLegacyCloud);
+ createConfNode(overseer.getSolrCloudManager().getDistribStateManager(), configName, collectionName, isLegacyCloud);
reloadCollection(null, new ZkNodeProps(NAME, collectionName), results);
}
@@ -746,8 +753,8 @@ public class OverseerCollectionMessageHandler implements OverseerMessageHandler,
}
- void validateConfigOrThrowSolrException(String configName) throws KeeperException, InterruptedException {
- boolean isValid = zkStateReader.getZkClient().exists(ZkConfigManager.CONFIGS_ZKNODE + "/" + configName, true);
+ void validateConfigOrThrowSolrException(String configName) throws IOException, KeeperException, InterruptedException {
+ boolean isValid = cloudManager.getDistribStateManager().hasData(ZkConfigManager.CONFIGS_ZKNODE + "/" + configName);
if(!isValid) {
throw new SolrException(ErrorCode.BAD_REQUEST, "Can not find the specified config set: " + configName);
}
@@ -757,16 +764,16 @@ public class OverseerCollectionMessageHandler implements OverseerMessageHandler,
* This doesn't validate the config (path) itself and is just responsible for creating the confNode.
* That check should be done before the config node is created.
*/
- void createConfNode(String configName, String coll, boolean isLegacyCloud) throws KeeperException, InterruptedException {
+ public static void createConfNode(DistribStateManager stateManager, String configName, String coll, boolean isLegacyCloud) throws IOException, AlreadyExistsException, BadVersionException, KeeperException, InterruptedException {
if (configName != null) {
String collDir = ZkStateReader.COLLECTIONS_ZKNODE + "/" + coll;
log.debug("creating collections conf node {} ", collDir);
byte[] data = Utils.toJSON(makeMap(ZkController.CONFIGNAME_PROP, configName));
- if (zkStateReader.getZkClient().exists(collDir, true)) {
- zkStateReader.getZkClient().setData(collDir, data, true);
+ if (stateManager.hasData(collDir)) {
+ stateManager.setData(collDir, data, -1);
} else {
- zkStateReader.getZkClient().makePath(collDir, data, true);
+ stateManager.makePath(collDir, data, CreateMode.PERSISTENT, false);
}
} else {
if(isLegacyCloud){
@@ -775,7 +782,6 @@ public class OverseerCollectionMessageHandler implements OverseerMessageHandler,
throw new SolrException(ErrorCode.BAD_REQUEST,"Unable to get config name");
}
}
-
}
private void collectionCmd(ZkNodeProps message, ModifiableSolrParams params,
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a44d9cf6/solr/core/src/java/org/apache/solr/cloud/autoscaling/AutoScalingHandler.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/autoscaling/AutoScalingHandler.java b/solr/core/src/java/org/apache/solr/cloud/autoscaling/AutoScalingHandler.java
index 0dc6011..d835f85 100644
--- a/solr/core/src/java/org/apache/solr/cloud/autoscaling/AutoScalingHandler.java
+++ b/solr/core/src/java/org/apache/solr/cloud/autoscaling/AutoScalingHandler.java
@@ -36,29 +36,23 @@ import java.util.stream.Stream;
import org.apache.solr.api.Api;
import org.apache.solr.api.ApiBag;
-import org.apache.solr.client.solrj.cloud.DistributedQueueFactory;
import org.apache.solr.client.solrj.cloud.autoscaling.AutoScalingConfig;
+import org.apache.solr.client.solrj.cloud.autoscaling.BadVersionException;
import org.apache.solr.client.solrj.cloud.autoscaling.Clause;
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.Preference;
-import org.apache.solr.client.solrj.cloud.autoscaling.Row;
import org.apache.solr.client.solrj.cloud.autoscaling.SolrCloudManager;
import org.apache.solr.client.solrj.cloud.autoscaling.TriggerEventProcessorStage;
import org.apache.solr.client.solrj.cloud.autoscaling.TriggerEventType;
-import org.apache.solr.client.solrj.impl.CloudSolrClient;
-import org.apache.solr.client.solrj.impl.SolrClientCloudManager;
-import org.apache.solr.cloud.ZkDistributedQueueFactory;
import org.apache.solr.common.MapWriter;
import org.apache.solr.common.SolrException;
-import org.apache.solr.common.cloud.SolrZkClient;
-import org.apache.solr.common.cloud.ZkStateReader;
import org.apache.solr.common.params.AutoScalingParams;
import org.apache.solr.common.params.CollectionAdminParams;
import org.apache.solr.common.util.CommandOperation;
import org.apache.solr.common.util.StrUtils;
import org.apache.solr.common.util.Utils;
-import org.apache.solr.core.CoreContainer;
+import org.apache.solr.core.SolrResourceLoader;
import org.apache.solr.handler.RequestHandlerBase;
import org.apache.solr.handler.RequestHandlerUtils;
import org.apache.solr.request.SolrQueryRequest;
@@ -83,15 +77,17 @@ import static org.apache.solr.common.params.CommonParams.JSON;
public class AutoScalingHandler extends RequestHandlerBase implements PermissionNameProvider {
public static final String HANDLER_PATH = "/admin/autoscaling";
private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
- protected final CoreContainer container;
+ protected final SolrCloudManager cloudManager;
+ protected final SolrResourceLoader loader;
private final List<Map<String, String>> DEFAULT_ACTIONS = new ArrayList<>(3);
private static Set<String> singletonCommands = Stream.of("set-cluster-preferences", "set-cluster-policy")
.collect(collectingAndThen(toSet(), Collections::unmodifiableSet));
private static final TimeSource timeSource = TimeSource.CURRENT_TIME;
- public AutoScalingHandler(CoreContainer container) {
- this.container = container;
+ public AutoScalingHandler(SolrCloudManager cloudManager, SolrResourceLoader loader) {
+ this.cloudManager = cloudManager;
+ this.loader = loader;
Map<String, String> map = new HashMap<>(2);
map.put(NAME, "compute_plan");
map.put(CLASS, "solr.ComputePlanAction");
@@ -119,7 +115,7 @@ public class AutoScalingHandler extends RequestHandlerBase implements Permission
throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, "Unknown path: " + path);
}
- AutoScalingConfig autoScalingConf = container.getZkController().zkStateReader.getAutoScalingConfig();
+ AutoScalingConfig autoScalingConf = cloudManager.getDistribStateManager().getAutoScalingConfig();
if (parts.size() == 2) {
autoScalingConf.writeMap(new MapWriter.EntryWriter() {
@@ -157,21 +153,14 @@ public class AutoScalingHandler extends RequestHandlerBase implements Permission
private void handleSuggestions(SolrQueryResponse rsp, AutoScalingConfig autoScalingConf) throws IOException {
- try (CloudSolrClient build = new CloudSolrClient.Builder()
- .withHttpClient(container.getUpdateShardHandler().getHttpClient())
- .withZkHost(container.getZkController().getZkServerAddress()).build()) {
- DistributedQueueFactory queueFactory = new ZkDistributedQueueFactory(container.getZkController().getZkClient());
- rsp.getValues().add("suggestions",
- PolicyHelper.getSuggestions(autoScalingConf, new SolrClientCloudManager(queueFactory, build)));
- }
-
-
+ rsp.getValues().add("suggestions",
+ PolicyHelper.getSuggestions(autoScalingConf, cloudManager));
}
public void processOps(SolrQueryRequest req, SolrQueryResponse rsp, List<CommandOperation> ops)
throws KeeperException, InterruptedException, IOException {
while (true) {
- AutoScalingConfig initialConfig = container.getZkController().zkStateReader.getAutoScalingConfig();
+ AutoScalingConfig initialConfig = cloudManager.getDistribStateManager().getAutoScalingConfig();
AutoScalingConfig currentConfig = initialConfig;
for (CommandOperation op : ops) {
switch (op.name) {
@@ -216,7 +205,7 @@ public class AutoScalingHandler extends RequestHandlerBase implements Permission
if (!currentConfig.equals(initialConfig)) {
// update in ZK
- if (zkSetAutoScalingConfig(container.getZkController().getZkStateReader(), currentConfig)) {
+ if (setAutoScalingConfig(currentConfig)) {
break;
} else {
// someone else updated the config, get the latest one and re-apply our ops
@@ -233,12 +222,7 @@ public class AutoScalingHandler extends RequestHandlerBase implements Permission
private void handleDiagnostics(SolrQueryResponse rsp, AutoScalingConfig autoScalingConf) throws IOException {
Policy policy = autoScalingConf.getPolicy();
- try (CloudSolrClient build = new CloudSolrClient.Builder()
- .withHttpClient(container.getUpdateShardHandler().getHttpClient())
- .withZkHost(container.getZkController().getZkServerAddress()).build()) {
- DistributedQueueFactory queueFactory = new ZkDistributedQueueFactory(container.getZkController().getZkClient());
- rsp.getValues().add("diagnostics", PolicyHelper.getDiagnostics(policy, new SolrClientCloudManager(queueFactory, build)));
- }
+ rsp.getValues().add("diagnostics", PolicyHelper.getDiagnostics(policy, cloudManager));
}
private AutoScalingConfig handleSetClusterPolicy(SolrQueryRequest req, SolrQueryResponse rsp, CommandOperation op,
@@ -291,7 +275,7 @@ public class AutoScalingHandler extends RequestHandlerBase implements Permission
return currentConfig;
}
- container.getZkController().getZkStateReader().getClusterState().forEachCollection(coll -> {
+ cloudManager.getClusterStateProvider().getClusterState().forEachCollection(coll -> {
if (policyName.equals(coll.getPolicyName()))
op.addError(StrUtils.formatString("policy : {0} is being used by collection {1}", policyName, coll.getName()));
});
@@ -459,7 +443,7 @@ public class AutoScalingHandler extends RequestHandlerBase implements Permission
// validate that we can load the listener class
// todo allow creation from blobstore
try {
- container.getResourceLoader().findClass(listenerClass, TriggerListener.class);
+ loader.findClass(listenerClass, TriggerListener.class);
} catch (Exception e) {
log.warn("error loading listener class ", e);
op.addError("Listener not found: " + listenerClass + ". error message:" + e.getMessage());
@@ -524,7 +508,7 @@ public class AutoScalingHandler extends RequestHandlerBase implements Permission
}
String klass = action.get(CLASS);
try {
- container.getResourceLoader().findClass(klass, TriggerAction.class);
+ loader.findClass(klass, TriggerAction.class);
} catch (Exception e) {
log.warn("Could not load class : ", e);
op.addError("Action not found: " + klass + " " + e.getMessage());
@@ -621,26 +605,22 @@ public class AutoScalingHandler extends RequestHandlerBase implements Permission
}
- private boolean zkSetAutoScalingConfig(ZkStateReader reader, AutoScalingConfig currentConfig) throws KeeperException, InterruptedException, IOException {
+ private boolean setAutoScalingConfig(AutoScalingConfig currentConfig) throws KeeperException, InterruptedException, IOException {
verifyAutoScalingConf(currentConfig);
try {
- reader.getZkClient().setData(SOLR_AUTOSCALING_CONF_PATH, Utils.toJSON(currentConfig), currentConfig.getZkVersion(), true);
- } catch (KeeperException.BadVersionException bve) {
+ cloudManager.getDistribStateManager().setData(SOLR_AUTOSCALING_CONF_PATH, Utils.toJSON(currentConfig), currentConfig.getZkVersion());
+ } catch (BadVersionException bve) {
// somebody else has changed the configuration so we must retry
return false;
}
+ log.debug("-- saved version " + currentConfig.getZkVersion() + ": " + currentConfig);
return true;
}
private void verifyAutoScalingConf(AutoScalingConfig autoScalingConf) throws IOException {
- try (CloudSolrClient build = new CloudSolrClient.Builder()
- .withHttpClient(container.getUpdateShardHandler().getHttpClient())
- .withZkHost(container.getZkController().getZkServerAddress()).build()) {
- DistributedQueueFactory queueFactory = new ZkDistributedQueueFactory(container.getZkController().getZkClient());
- Policy.Session session = autoScalingConf.getPolicy()
- .createSession(new SolrClientCloudManager(queueFactory, build));
- log.debug("Verified autoscaling configuration");
- }
+ Policy.Session session = autoScalingConf.getPolicy()
+ .createSession(cloudManager);
+ log.debug("Verified autoscaling configuration");
}
@Override
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a44d9cf6/solr/core/src/java/org/apache/solr/core/CoreContainer.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/core/CoreContainer.java b/solr/core/src/java/org/apache/solr/core/CoreContainer.java
index 5ed72d7..a761d19 100644
--- a/solr/core/src/java/org/apache/solr/core/CoreContainer.java
+++ b/solr/core/src/java/org/apache/solr/core/CoreContainer.java
@@ -540,7 +540,9 @@ public class CoreContainer {
// may want to add some configuration here in the future
metricsCollectorHandler.init(null);
- autoScalingHandler = createHandler(AutoScalingHandler.HANDLER_PATH, AutoScalingHandler.class.getName(), AutoScalingHandler.class);
+ autoScalingHandler = new AutoScalingHandler(getZkController().getSolrCloudManager(), loader);
+ containerHandlers.put(AutoScalingHandler.HANDLER_PATH, autoScalingHandler);
+ autoScalingHandler.initializeMetrics(metricManager, SolrInfoBean.Group.node.toString(), AutoScalingHandler.HANDLER_PATH);
containerHandlers.put(AUTHZ_PATH, securityConfHandler);
securityConfHandler.initializeMetrics(metricManager, SolrInfoBean.Group.node.toString(), AUTHZ_PATH);
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a44d9cf6/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 547fb37..37041c2 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
@@ -17,6 +17,7 @@
package org.apache.solr.cloud.autoscaling.sim;
+import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.lang.invoke.MethodHandles;
import java.util.ArrayList;
@@ -24,6 +25,7 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Random;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
@@ -32,30 +34,55 @@ import org.apache.solr.client.solrj.SolrRequest;
import org.apache.solr.client.solrj.SolrResponse;
import org.apache.solr.client.solrj.SolrServerException;
import org.apache.solr.client.solrj.cloud.DistributedQueueFactory;
+import org.apache.solr.client.solrj.cloud.autoscaling.AutoScalingConfig;
import org.apache.solr.client.solrj.cloud.autoscaling.DistribStateManager;
import org.apache.solr.client.solrj.cloud.autoscaling.NodeStateProvider;
+import org.apache.solr.client.solrj.cloud.autoscaling.Policy;
+import org.apache.solr.client.solrj.cloud.autoscaling.ReplicaInfo;
import org.apache.solr.client.solrj.cloud.autoscaling.SolrCloudManager;
import org.apache.solr.client.solrj.impl.ClusterStateProvider;
import org.apache.solr.client.solrj.request.CollectionAdminRequest;
+import org.apache.solr.client.solrj.request.RequestWriter;
import org.apache.solr.client.solrj.request.UpdateRequest;
import org.apache.solr.client.solrj.response.SolrResponseBase;
import org.apache.solr.client.solrj.response.UpdateResponse;
+import org.apache.solr.cloud.Assign;
import org.apache.solr.cloud.Overseer;
+import org.apache.solr.cloud.OverseerCollectionMessageHandler;
+import org.apache.solr.cloud.autoscaling.AutoScalingHandler;
import org.apache.solr.cloud.autoscaling.OverseerTriggerThread;
+import org.apache.solr.cloud.overseer.ClusterStateMutator;
import org.apache.solr.common.SolrException;
import org.apache.solr.common.SolrInputDocument;
+import org.apache.solr.common.cloud.ClusterState;
+import org.apache.solr.common.cloud.DocRouter;
+import org.apache.solr.common.cloud.ImplicitDocRouter;
+import org.apache.solr.common.cloud.ReplicaPosition;
+import org.apache.solr.common.cloud.ZkNodeProps;
+import org.apache.solr.common.cloud.ZkStateReader;
import org.apache.solr.common.cloud.rule.ImplicitSnitch;
import org.apache.solr.common.params.CollectionAdminParams;
import org.apache.solr.common.params.CollectionParams;
+import org.apache.solr.common.params.CommonParams;
import org.apache.solr.common.params.CoreAdminParams;
+import org.apache.solr.common.params.ModifiableSolrParams;
import org.apache.solr.common.params.SolrParams;
+import org.apache.solr.common.util.ContentStreamBase;
import org.apache.solr.common.util.IOUtils;
import org.apache.solr.common.util.NamedList;
import org.apache.solr.core.CloudConfig;
import org.apache.solr.core.SolrResourceLoader;
+import org.apache.solr.request.LocalSolrQueryRequest;
+import org.apache.solr.response.SolrQueryResponse;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import static org.apache.solr.cloud.OverseerCollectionMessageHandler.NUM_SLICES;
+import static org.apache.solr.common.cloud.ZkStateReader.NRT_REPLICAS;
+import static org.apache.solr.common.cloud.ZkStateReader.PULL_REPLICAS;
+import static org.apache.solr.common.cloud.ZkStateReader.REPLICATION_FACTOR;
+import static org.apache.solr.common.cloud.ZkStateReader.TLOG_REPLICAS;
+
/**
* Simulated {@link SolrCloudManager}.
*/
@@ -65,6 +92,7 @@ public class SimCloudManager implements SolrCloudManager {
private final SimDistribStateManager stateManager;
private final SimClusterStateProvider clusterStateProvider;
private final SimNodeStateProvider nodeStateProvider;
+ private final AutoScalingHandler autoScalingHandler;
private final Set<String> liveNodes = ConcurrentHashMap.newKeySet();
private final SimDistributedQueueFactory queueFactory;
private SolrClient solrClient;
@@ -76,12 +104,18 @@ public class SimCloudManager implements SolrCloudManager {
private static int nodeIdPort = 10000;
- public SimCloudManager() {
+ public SimCloudManager() throws Exception {
this.stateManager = new SimDistribStateManager();
- this.clusterStateProvider = new SimClusterStateProvider(liveNodes);
+ // 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);
+ this.clusterStateProvider = new SimClusterStateProvider(liveNodes, this);
this.nodeStateProvider = new SimNodeStateProvider(liveNodes, this.clusterStateProvider, null);
this.queueFactory = new SimDistributedQueueFactory();
this.httpServer = new SimHttpServer();
+ this.autoScalingHandler = new AutoScalingHandler(this, new SolrResourceLoader());
ThreadGroup triggerThreadGroup = new ThreadGroup("Simulated Overseer autoscaling triggers");
OverseerTriggerThread trigger = new OverseerTriggerThread(new SolrResourceLoader(), this,
new CloudConfig.CloudConfigBuilder("nonexistent", 0, "sim").build());
@@ -98,7 +132,7 @@ public class SimCloudManager implements SolrCloudManager {
public static SimCloudManager createCluster(int numNodes) throws Exception {
SimCloudManager cloudManager = new SimCloudManager();
for (int i = 1; i <= numNodes; i++) {
- Map<String, Object> values = createNodeValues();
+ Map<String, Object> values = createNodeValues(null);
String nodeId = (String)values.get(ImplicitSnitch.NODE);
cloudManager.getSimClusterStateProvider().simAddNode(nodeId);
cloudManager.getSimNodeStateProvider().simSetNodeValues(nodeId, values);
@@ -106,15 +140,46 @@ public class SimCloudManager implements SolrCloudManager {
return cloudManager;
}
- public static Map<String, Object> createNodeValues() {
- String host = "127.0.0.1";
- String port = "" + nodeIdPort++;
- String nodeId = host + ":" + port + "_solr";
+ public static SimCloudManager createCluster(ClusterState initialState) throws Exception {
+ SimCloudManager cloudManager = new SimCloudManager();
+ cloudManager.getSimClusterStateProvider().simSetClusterState(initialState);
+ for (String node : cloudManager.getClusterStateProvider().getLiveNodes()) {
+ cloudManager.getSimNodeStateProvider().simSetNodeValues(node, createNodeValues(node));
+ }
+ return cloudManager;
+ }
+
+ public static Map<String, Object> createNodeValues(String nodeName) {
Map<String, Object> values = new HashMap<>();
- values.put("ip_1", "127");
- values.put("ip_2", "0");
- values.put("ip_3", "0");
- values.put("ip_4", "1");
+ String host, port, nodeId;
+ if (nodeName == null) {
+ host = "127.0.0.1";
+ port = "" + nodeIdPort++;
+ nodeId = host + ":" + port + "_solr";
+ values.put("ip_1", "127");
+ values.put("ip_2", "0");
+ values.put("ip_3", "0");
+ values.put("ip_4", "1");
+ } else {
+ String[] hostPortCtx = nodeName.split(":");
+ if (hostPortCtx.length != 2) {
+ throw new RuntimeException("Invalid nodeName " + nodeName);
+ }
+ host = hostPortCtx[0];
+ String[] portCtx = hostPortCtx[1].split("_");
+ if (portCtx.length != 2) {
+ throw new RuntimeException("Invalid port_context in nodeName " + nodeName);
+ }
+ port = portCtx[0];
+ nodeId = host + ":" + port + "_" + portCtx[1];
+ String[] ip = host.split("\\.");
+ if (ip.length == 4) {
+ values.put("ip_1", ip[0]);
+ values.put("ip_2", ip[1]);
+ values.put("ip_3", ip[2]);
+ values.put("ip_4", ip[3]);
+ }
+ }
values.put(ImplicitSnitch.HOST, host);
values.put(ImplicitSnitch.PORT, port);
values.put(ImplicitSnitch.NODE, nodeId);
@@ -125,11 +190,12 @@ public class SimCloudManager implements SolrCloudManager {
return values;
}
- public void simAddNode() throws Exception {
- Map<String, Object> values = createNodeValues();
+ public String simAddNode() throws Exception {
+ Map<String, Object> values = createNodeValues(null);
String nodeId = (String)values.get(ImplicitSnitch.NODE);
clusterStateProvider.simAddNode(nodeId);
nodeStateProvider.simSetNodeValues(nodeId, values);
+ return nodeId;
}
public void simRemoveNode(String nodeId) throws Exception {
@@ -145,6 +211,25 @@ public class SimCloudManager implements SolrCloudManager {
return systemColl;
}
+ public SolrClient simGetSolrClient() {
+ if (solrClient != null) {
+ return solrClient;
+ } else {
+ return new SolrClient() {
+ @Override
+ public NamedList<Object> request(SolrRequest request, String collection) throws SolrServerException, IOException {
+ SolrResponse rsp = SimCloudManager.this.request(request);
+ return rsp.getResponse();
+ }
+
+ @Override
+ public void close() throws IOException {
+
+ }
+ };
+ }
+ }
+
// ---------- type-safe methods to obtain simulator components ----------
public SimClusterStateProvider getSimClusterStateProvider() {
@@ -194,12 +279,36 @@ public class SimCloudManager implements SolrCloudManager {
throw new IOException(e);
}
} else {
- return simHandleSolrRequest(req);
+ try {
+ return simHandleSolrRequest(req);
+ } catch (Exception e) {
+ throw new IOException(e);
+ }
}
}
- public SolrResponse simHandleSolrRequest(SolrRequest req) throws IOException {
- LOG.info("--- got SolrRequest: " + req);
+ public SolrResponse simHandleSolrRequest(SolrRequest req) throws IOException, InterruptedException {
+ LOG.debug("--- got SolrRequest: " + req.getMethod() + " " + req.getPath() +
+ (req.getParams() != null ? " " + req.getParams().toQueryString() : ""));
+ if (req.getPath() != null && req.getPath().startsWith("/admin/autoscaling") ||
+ req.getPath().startsWith("/cluster/autoscaling")) {
+ ModifiableSolrParams params = new ModifiableSolrParams(req.getParams());
+ params.set(CommonParams.PATH, req.getPath());
+ LocalSolrQueryRequest queryRequest = new LocalSolrQueryRequest(null, params);
+ RequestWriter.ContentWriter cw = req.getContentWriter("application/json");
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ cw.write(baos);
+ String payload = baos.toString("UTF-8");
+ LOG.debug("-- payload: " + payload);
+ queryRequest.setContentStreams(Collections.singletonList(new ContentStreamBase.StringStream(payload)));
+ queryRequest.getContext().put("httpMethod", req.getMethod().toString());
+ SolrQueryResponse queryResponse = new SolrQueryResponse();
+ autoScalingHandler.handleRequest(queryRequest, queryResponse);
+ SolrResponse rsp = new SolrResponseBase();
+ rsp.setResponse(queryResponse.getValues());
+ LOG.debug("-- response: " + rsp);
+ return rsp;
+ }
if (req instanceof UpdateRequest) {
// support only updates to the system collection
UpdateRequest ureq = (UpdateRequest)req;
@@ -214,11 +323,12 @@ public class SimCloudManager implements SolrCloudManager {
}
// support only a specific subset of collection admin ops
if (!(req instanceof CollectionAdminRequest)) {
- throw new UnsupportedOperationException("Only CollectionAdminRequest-s are supported: " + req.getClass().getName());
+ throw new UnsupportedOperationException("Only some CollectionAdminRequest-s are supported: " + req.getClass().getName());
}
SolrParams params = req.getParams();
String a = params.get(CoreAdminParams.ACTION);
SolrResponse rsp = new SolrResponseBase();
+ rsp.setResponse(new NamedList<>());
if (a != null) {
CollectionParams.CollectionAction action = CollectionParams.CollectionAction.get(a);
if (action == null) {
@@ -231,12 +341,20 @@ public class SimCloudManager implements SolrCloudManager {
case ADDREPLICAPROP:
break;
case CREATE:
+ try {
+ clusterStateProvider.simCreateCollection(new ZkNodeProps(req.getParams().toNamedList().asMap(10)));
+ rsp.getResponse().add("success", "");
+ } catch (Exception e) {
+ throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, e);
+ }
break;
case CLUSTERPROP:
break;
case CREATESHARD:
break;
case DELETE:
+ clusterStateProvider.simDeleteCollection(req.getParams().get(CommonParams.NAME));
+ rsp.getResponse().add("success", "");
break;
case DELETENODE:
break;
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a44d9cf6/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 18ecea1..215b763 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
@@ -26,35 +26,72 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
+import java.util.Locale;
import java.util.Map;
+import java.util.Random;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReentrantLock;
+import org.apache.solr.client.solrj.cloud.autoscaling.AutoScalingConfig;
+import org.apache.solr.client.solrj.cloud.autoscaling.DistribStateManager;
+import org.apache.solr.client.solrj.cloud.autoscaling.Policy;
import org.apache.solr.client.solrj.cloud.autoscaling.ReplicaInfo;
+import org.apache.solr.client.solrj.cloud.autoscaling.SolrCloudManager;
+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.OverseerCollectionMessageHandler;
+import org.apache.solr.cloud.overseer.ClusterStateMutator;
+import org.apache.solr.common.SolrException;
import org.apache.solr.common.SolrInputDocument;
import org.apache.solr.common.cloud.ClusterState;
import org.apache.solr.common.cloud.DocCollection;
import org.apache.solr.common.cloud.DocRouter;
+import org.apache.solr.common.cloud.ImplicitDocRouter;
import org.apache.solr.common.cloud.Replica;
+import org.apache.solr.common.cloud.ReplicaPosition;
import org.apache.solr.common.cloud.Slice;
+import org.apache.solr.common.cloud.ZkNodeProps;
import org.apache.solr.common.cloud.ZkStateReader;
+import org.apache.solr.common.params.CommonParams;
import org.apache.solr.common.util.ExecutorUtil;
+import org.apache.solr.common.util.Utils;
import org.apache.solr.util.DefaultSolrThreadFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import static org.apache.solr.cloud.OverseerCollectionMessageHandler.NUM_SLICES;
+import static org.apache.solr.common.cloud.ZkStateReader.NRT_REPLICAS;
+import static org.apache.solr.common.cloud.ZkStateReader.PULL_REPLICAS;
+import static org.apache.solr.common.cloud.ZkStateReader.REPLICATION_FACTOR;
+import static org.apache.solr.common.cloud.ZkStateReader.TLOG_REPLICAS;
+
/**
*
*/
public class SimClusterStateProvider implements ClusterStateProvider {
private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+ static final Random RANDOM;
+ static {
+ // We try to make things reproducible in the context of our tests by initializing the random instance
+ // based on the current seed
+ String seed = System.getProperty("tests.seed");
+ if (seed == null) {
+ RANDOM = new Random();
+ } else {
+ RANDOM = new Random(seed.hashCode());
+ }
+ }
+
private final Map<String, List<ReplicaInfo>> nodeReplicaMap = new ConcurrentHashMap<>();
private final Set<String> liveNodes;
+ private final DistribStateManager stateManager;
+ private final SolrCloudManager cloudManager;
private final Map<String, Object> clusterProperties = new ConcurrentHashMap<>();
private final Map<String, Map<String, Object>> collProperties = new ConcurrentHashMap<>();
@@ -64,23 +101,29 @@ public class SimClusterStateProvider implements ClusterStateProvider {
private final ReentrantLock lock = new ReentrantLock();
+ private ClusterState lastSavedState = null;
+ private Map<String, Object> lastSavedProperties = null;
+
/**
* Zero-arg constructor. The instance needs to be initialized using the <code>sim*</code> methods in order
* to ensure proper behavior, otherwise it will behave as a cluster with zero live nodes and zero replicas.
*/
- public SimClusterStateProvider(Set<String> liveNodes) {
+ public SimClusterStateProvider(Set<String> liveNodes, SolrCloudManager cloudManager) {
this.liveNodes = liveNodes;
+ this.cloudManager = cloudManager;
+ this.stateManager = cloudManager.getDistribStateManager();
simStateProviderPool = ExecutorUtil.newMDCAwareCachedThreadPool(new DefaultSolrThreadFactory("simClusterStateProviderPool"));
}
// ============== SIMULATOR SETUP METHODS ====================
- public void simSetClusterState(ClusterState initialState) {
+ public void simSetClusterState(ClusterState initialState) throws Exception {
lock.lock();
try {
collProperties.clear();
sliceProperties.clear();
nodeReplicaMap.clear();
+ liveNodes.clear();
liveNodes.addAll(initialState.getLiveNodes());
initialState.forEachCollection(dc -> {
collProperties.computeIfAbsent(dc.getName(), name -> new HashMap<>()).putAll(dc.getProperties());
@@ -95,11 +138,20 @@ public class SimClusterStateProvider implements ClusterStateProvider {
});
});
});
+ saveClusterState();
} finally {
lock.unlock();
}
}
+ public String simGetRandomNode(Random random) {
+ if (liveNodes.isEmpty()) {
+ return null;
+ }
+ List<String> nodes = new ArrayList<>(liveNodes);
+ return nodes.get(random.nextInt(nodes.size()));
+ }
+
// todo: maybe hook up DistribStateManager /live_nodes ?
// todo: maybe hook up DistribStateManager /clusterstate.json watchers?
public boolean simAddNode(String nodeId) throws Exception {
@@ -112,7 +164,7 @@ public class SimClusterStateProvider implements ClusterStateProvider {
// todo: maybe hook up DistribStateManager /live_nodes ?
// todo: maybe hook up DistribStateManager /clusterstate.json watchers?
- public boolean simRemoveNode(String nodeId) {
+ public boolean simRemoveNode(String nodeId) throws Exception {
lock.lock();
try {
// mark every replica on that node as down
@@ -123,7 +175,7 @@ public class SimClusterStateProvider implements ClusterStateProvider {
});
}
boolean res = liveNodes.remove(nodeId);
- simStateProviderPool.submit(() -> simRunLeaderElection());
+ simStateProviderPool.submit(() -> simRunLeaderElection(true));
return res;
} finally {
lock.unlock();
@@ -144,7 +196,7 @@ public class SimClusterStateProvider implements ClusterStateProvider {
try {
List<ReplicaInfo> replicas = nodeReplicaMap.computeIfAbsent(nodeId, n -> new ArrayList<>());
replicas.add(replicaInfo);
- simStateProviderPool.submit(() -> simRunLeaderElection());
+ simStateProviderPool.submit(() -> simRunLeaderElection(true));
} finally {
lock.unlock();
}
@@ -158,7 +210,7 @@ public class SimClusterStateProvider implements ClusterStateProvider {
for (int i = 0; i < replicas.size(); i++) {
if (coreNodeName.equals(replicas.get(i).getCore())) {
replicas.remove(i);
- simStateProviderPool.submit(() -> simRunLeaderElection());
+ simStateProviderPool.submit(() -> simRunLeaderElection(true));
return;
}
}
@@ -168,8 +220,29 @@ public class SimClusterStateProvider implements ClusterStateProvider {
}
}
- public synchronized void simRunLeaderElection() {
+ public synchronized ClusterState saveClusterState() throws IOException {
+ ClusterState currentState = getClusterState();
+ if (lastSavedState != null && lastSavedState.equals(currentState)) {
+ return lastSavedState;
+ }
+ byte[] data = Utils.toJSON(currentState);
+ try {
+ VersionedData oldData = stateManager.getData(ZkStateReader.CLUSTER_STATE);
+ int version = oldData != null ? oldData.getVersion() : -1;
+ stateManager.setData(ZkStateReader.CLUSTER_STATE, data, version);
+ } catch (Exception e) {
+ throw new IOException(e);
+ }
+ lastSavedState = currentState;
+ return currentState;
+ }
+
+ public synchronized void simRunLeaderElection(boolean saveClusterState) {
+
try {
+ if (saveClusterState) {
+ saveClusterState();
+ }
ClusterState state = getClusterState();
state.forEachCollection(dc -> {
dc.getSlices().forEach(s -> {
@@ -219,7 +292,51 @@ public class SimClusterStateProvider implements ClusterStateProvider {
}
}
- public void simDeleteCollection(String collection) throws Exception {
+ public void simCreateCollection(ZkNodeProps props) throws Exception {
+ AutoScalingConfig autoScalingConfig = stateManager.getAutoScalingConfig();
+ String collectionName = props.getStr(CommonParams.NAME);
+ Integer numSlices = props.getInt(OverseerCollectionMessageHandler.NUM_SLICES, null);
+ int numTlogReplicas = props.getInt(TLOG_REPLICAS, 0);
+ int numNrtReplicas = props.getInt(NRT_REPLICAS, props.getInt(REPLICATION_FACTOR, numTlogReplicas>0?0:1));
+ int numPullReplicas = props.getInt(PULL_REPLICAS, 0);
+ String policy = props.getStr(Policy.POLICY);
+ List<String> shardNames = new ArrayList<>();
+ String router = props.getStr("router.name", DocRouter.DEFAULT_NAME);
+ if(ImplicitDocRouter.NAME.equals(router)){
+ ClusterStateMutator.getShardNames(shardNames, props.getStr("shards", null));
+ numSlices = shardNames.size();
+ } else {
+ if (numSlices == null ) {
+ throw new SolrException(SolrException.ErrorCode.BAD_REQUEST, NUM_SLICES + " is a required param (when using CompositeId router).");
+ }
+ ClusterStateMutator.getShardNames(numSlices, shardNames);
+ }
+
+ boolean usePolicyFramework = !autoScalingConfig.isEmpty() && !autoScalingConfig.getPolicy().getClusterPolicy().isEmpty();
+ final List<String> nodeList = Assign.getLiveOrLiveAndCreateNodeSetList(liveNodes, props, RANDOM);
+ List<ReplicaPosition> replicaPositions = Assign.getPositionsUsingPolicy(collectionName, shardNames, numNrtReplicas,
+ numTlogReplicas, numPullReplicas, policy, cloudManager, nodeList);
+ AtomicInteger replicaNum = new AtomicInteger(1);
+ replicaPositions.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);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ });
+
+ }
+
+ public void simDeleteCollection(String collection) throws IOException {
lock.lock();
try {
collProperties.remove(collection);
@@ -232,64 +349,91 @@ public class SimClusterStateProvider implements ClusterStateProvider {
}
}
});
+ saveClusterState();
} finally {
lock.unlock();
}
}
- public void simDeleteAllCollections() {
+ public void simDeleteAllCollections() throws Exception {
lock.lock();
try {
nodeReplicaMap.clear();
collProperties.clear();
sliceProperties.clear();
+ saveClusterState();
} finally {
lock.unlock();
}
}
- // todo: maybe hook up DistribStateManager /clusterstate.json watchers?
- public void simSetClusterProperties(Map<String, Object> properties) {
- clusterProperties.clear();
- if (properties != null) {
- this.clusterProperties.putAll(properties);
+ public synchronized Map<String, Object> saveClusterProperties() throws Exception {
+ if (lastSavedProperties != null && lastSavedProperties.equals(clusterProperties)) {
+ return lastSavedProperties;
}
+ byte[] data = Utils.toJSON(clusterProperties);
+ VersionedData oldData = stateManager.getData(ZkStateReader.CLUSTER_PROPS);
+ int version = oldData != null ? oldData.getVersion() : -1;
+ stateManager.setData(ZkStateReader.CLUSTER_PROPS, data, version);
+ lastSavedProperties = (Map)Utils.fromJSON(data);
+ return lastSavedProperties;
}
- public void simSetClusterProperty(String key, Object value) {
- if (value != null) {
- clusterProperties.put(key, value);
- } else {
- clusterProperties.remove(key);
+ public void simSetClusterProperties(Map<String, Object> properties) throws Exception {
+ lock.lock();
+ try {
+ clusterProperties.clear();
+ if (properties != null) {
+ this.clusterProperties.putAll(properties);
+ }
+ saveClusterProperties();
+ } finally {
+ lock.unlock();
}
}
- // todo: maybe hook up DistribStateManager /clusterstate.json watchers?
- public void simSetCollectionProperties(String coll, Map<String, Object> properties) {
+ public void simSetClusterProperty(String key, Object value) throws Exception {
+ lock.lock();
+ try {
+ if (value != null) {
+ clusterProperties.put(key, value);
+ } else {
+ clusterProperties.remove(key);
+ }
+ saveClusterProperties();
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ public void simSetCollectionProperties(String coll, Map<String, Object> properties) throws Exception {
if (properties == null) {
collProperties.remove(coll);
+ saveClusterState();
} else {
lock.lock();
try {
Map<String, Object> props = collProperties.computeIfAbsent(coll, c -> new HashMap<>());
props.clear();
props.putAll(properties);
+ saveClusterState();
} finally {
lock.unlock();
}
}
}
- public void simSetCollectionProperty(String coll, String key, String value) {
+ public void simSetCollectionProperty(String coll, String key, String value) throws Exception {
Map<String, Object> props = collProperties.computeIfAbsent(coll, c -> new HashMap<>());
if (value == null) {
props.remove(key);
} else {
props.put(key, value);
}
+ saveClusterState();
}
- public void setSliceProperties(String coll, String slice, Map<String, Object> properties) {
+ public void setSliceProperties(String coll, String slice, Map<String, Object> properties) throws Exception {
Map<String, Object> sliceProps = sliceProperties.computeIfAbsent(coll, c -> new HashMap<>()).computeIfAbsent(slice, s -> new HashMap<>());
lock.lock();
try {
@@ -297,6 +441,7 @@ public class SimClusterStateProvider implements ClusterStateProvider {
if (properties != null) {
sliceProps.putAll(properties);
}
+ saveClusterState();
} finally {
lock.unlock();
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/a44d9cf6/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
new file mode 100644
index 0000000..0918029
--- /dev/null
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimComputePlanActionTest.java
@@ -0,0 +1,364 @@
+/*
+ * 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.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.Utils;
+import org.apache.solr.util.LogLevel;
+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(4);
+ }
+
+ @Before
+ public void setUp() throws Exception {
+ super.setUp();
+
+ 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");
+ }
+
+ @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);
+ 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", 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/a44d9cf6/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimDistribStateManager.java
----------------------------------------------------------------------
diff --git a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimDistribStateManager.java b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimDistribStateManager.java
index e3e10d1..614b1b9 100644
--- a/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimDistribStateManager.java
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/sim/SimDistribStateManager.java
@@ -72,8 +72,8 @@ public class SimDistribStateManager implements DistribStateManager {
private final Node parent;
private byte[] data = null;
private Map<String, Node> children = new ConcurrentHashMap<>();
- List<Watcher> dataWatches = new ArrayList<>();
- List<Watcher> childrenWatches = new ArrayList<>();
+ Set<Watcher> dataWatches = ConcurrentHashMap.newKeySet();
+ Set<Watcher> childrenWatches = ConcurrentHashMap.newKeySet();
Node(Node parent, String name, String path, CreateMode mode, String clientId) {
this.parent = parent;
@@ -109,10 +109,11 @@ public class SimDistribStateManager implements DistribStateManager {
this.data = null;
}
this.version++;
- for (Watcher w : dataWatches) {
+ Set<Watcher> currentWatchers = new HashSet<>(dataWatches);
+ dataWatches.clear();
+ for (Watcher w : currentWatchers) {
w.process(new WatchedEvent(Watcher.Event.EventType.NodeDataChanged, Watcher.Event.KeeperState.SyncConnected, path));
}
- dataWatches.clear();
} finally {
dataLock.unlock();
}
@@ -136,10 +137,11 @@ public class SimDistribStateManager implements DistribStateManager {
dataLock.lock();
try {
children.put(name, child);
- for (Watcher w : childrenWatches) {
+ Set<Watcher> currentWatchers = new HashSet<>(childrenWatches);
+ childrenWatches.clear();
+ for (Watcher w : currentWatchers) {
w.process(new WatchedEvent(Watcher.Event.EventType.NodeChildrenChanged, Watcher.Event.KeeperState.SyncConnected, path));
}
- childrenWatches.clear();
} finally {
dataLock.unlock();
}
@@ -154,10 +156,14 @@ public class SimDistribStateManager implements DistribStateManager {
throw new BadVersionException(version, path);
}
children.remove(name);
- for (Watcher w : childrenWatches) {
+ Set<Watcher> currentWatchers = new HashSet<>(childrenWatches);
+ childrenWatches.clear();
+ for (Watcher w : currentWatchers) {
w.process(new WatchedEvent(Watcher.Event.EventType.NodeChildrenChanged, Watcher.Event.KeeperState.SyncConnected, path));
}
- for (Watcher w : n.dataWatches) {
+ currentWatchers = new HashSet<>(n.dataWatches);
+ n.dataWatches.clear();
+ for (Watcher w : currentWatchers) {
w.process(new WatchedEvent(Watcher.Event.EventType.NodeDeleted, Watcher.Event.KeeperState.SyncConnected, n.path));
}
// TODO: not sure if it's correct to recurse and fire watches???
@@ -346,6 +352,24 @@ public class SimDistribStateManager implements DistribStateManager {
}
@Override
+ public void makePath(String path, byte[] data, CreateMode createMode, boolean failOnExists) throws AlreadyExistsException, IOException, KeeperException, InterruptedException {
+ multiLock.lock();
+ try {
+ if (failOnExists && hasData(path)) {
+ throw new AlreadyExistsException(path);
+ }
+ Node n = traverse(path, true, createMode);
+ try {
+ n.setData(data, -1);
+ } catch (BadVersionException e) {
+ throw new IOException("should not happen!", e);
+ }
+ } finally {
+ multiLock.unlock();
+ }
+ }
+
+ @Override
public String createData(String path, byte[] data, CreateMode mode) throws AlreadyExistsException, NoSuchElementException, IOException {
if ((CreateMode.EPHEMERAL == mode || CreateMode.PERSISTENT == mode) && hasData(path)) {
throw new AlreadyExistsException(path);
@@ -465,6 +489,7 @@ public class SimDistribStateManager implements DistribStateManager {
VersionedData data = getData(ZkStateReader.SOLR_AUTOSCALING_CONF_PATH, watcher);
if (data != null && data.getData() != null && data.getData().length > 0) {
map = (Map<String, Object>) Utils.fromJSON(data.getData());
+ version = data.getVersion();
}
} catch (NoSuchElementException e) {
// ignore