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/10/14 18:57:14 UTC
[5/5] lucene-solr:master: SOLR-11285: Support simulations at scale in
the autoscaling framework, part 1 (refactoring).
SOLR-11285: Support simulations at scale in the autoscaling framework,
part 1 (refactoring).
Project: http://git-wip-us.apache.org/repos/asf/lucene-solr/repo
Commit: http://git-wip-us.apache.org/repos/asf/lucene-solr/commit/8dd489ed
Tree: http://git-wip-us.apache.org/repos/asf/lucene-solr/tree/8dd489ed
Diff: http://git-wip-us.apache.org/repos/asf/lucene-solr/diff/8dd489ed
Branch: refs/heads/master
Commit: 8dd489edcee009e40f44f98c79e69434bd6333ea
Parents: 9543e85
Author: Andrzej Bialecki <ab...@apache.org>
Authored: Sat Oct 14 18:15:40 2017 +0200
Committer: Andrzej Bialecki <ab...@apache.org>
Committed: Sat Oct 14 18:15:40 2017 +0200
----------------------------------------------------------------------
solr/CHANGES.txt | 6 +
.../src/java/org/apache/solr/api/ApiBag.java | 4 +
.../org/apache/solr/cloud/AddReplicaCmd.java | 14 +-
.../src/java/org/apache/solr/cloud/Assign.java | 95 +++---
.../apache/solr/cloud/CreateCollectionCmd.java | 2 +-
.../org/apache/solr/cloud/CreateShardCmd.java | 14 +-
.../org/apache/solr/cloud/DeleteShardCmd.java | 2 +-
.../org/apache/solr/cloud/ElectionContext.java | 5 +-
.../java/org/apache/solr/cloud/MigrateCmd.java | 2 +-
.../org/apache/solr/cloud/MoveReplicaCmd.java | 2 +-
.../java/org/apache/solr/cloud/Overseer.java | 188 +++---------
.../OverseerCollectionConfigSetProcessor.java | 10 +-
.../cloud/OverseerCollectionMessageHandler.java | 15 +-
.../solr/cloud/OverseerNodePrioritizer.java | 3 +-
.../apache/solr/cloud/OverseerStatusCmd.java | 8 +-
.../solr/cloud/OverseerTaskProcessor.java | 4 +-
.../apache/solr/cloud/OverseerTaskQueue.java | 4 +-
.../org/apache/solr/cloud/RecoveryStrategy.java | 7 +-
.../java/org/apache/solr/cloud/RestoreCmd.java | 2 +-
.../org/apache/solr/cloud/SplitShardCmd.java | 4 +-
.../src/java/org/apache/solr/cloud/Stats.java | 147 +++++++++
.../org/apache/solr/cloud/ZkController.java | 44 ++-
.../apache/solr/cloud/ZkDistributedQueue.java | 45 ++-
.../solr/cloud/ZkDistributedQueueFactory.java | 43 +++
.../solr/cloud/autoscaling/ActionContext.java | 12 +-
.../autoscaling/AutoAddReplicasPlanAction.java | 19 +-
.../solr/cloud/autoscaling/AutoScaling.java | 50 ++--
.../cloud/autoscaling/AutoScalingHandler.java | 11 +-
.../cloud/autoscaling/ComputePlanAction.java | 68 ++---
.../cloud/autoscaling/ExecutePlanAction.java | 73 +++--
.../cloud/autoscaling/HttpTriggerListener.java | 38 +--
.../cloud/autoscaling/NodeAddedTrigger.java | 136 +--------
.../solr/cloud/autoscaling/NodeLostTrigger.java | 143 +--------
.../autoscaling/OverseerTriggerThread.java | 69 ++---
.../cloud/autoscaling/ScheduledTriggers.java | 111 +++----
.../cloud/autoscaling/SystemLogListener.java | 15 +-
.../solr/cloud/autoscaling/TriggerBase.java | 159 +++++++++-
.../cloud/autoscaling/TriggerEventQueue.java | 31 +-
.../solr/cloud/autoscaling/TriggerListener.java | 4 +-
.../cloud/autoscaling/TriggerListenerBase.java | 8 +-
.../cloud/overseer/ClusterStateMutator.java | 10 +-
.../solr/cloud/overseer/CollectionMutator.java | 10 +-
.../apache/solr/cloud/overseer/NodeMutator.java | 3 +
.../solr/cloud/overseer/ReplicaMutator.java | 38 ++-
.../solr/cloud/overseer/SliceMutator.java | 17 +-
.../solr/cloud/overseer/ZkStateWriter.java | 5 +-
.../org/apache/solr/core/CoreContainer.java | 4 +
.../java/org/apache/solr/core/ZkContainer.java | 2 +
.../processor/DistributedUpdateProcessor.java | 2 +-
.../test/org/apache/solr/cloud/AssignTest.java | 19 +-
.../solr/cloud/DeleteInactiveReplicaTest.java | 5 +-
.../org/apache/solr/cloud/DeleteShardTest.java | 5 +-
.../apache/solr/cloud/DistributedQueueTest.java | 8 +-
.../org/apache/solr/cloud/ForceLeaderTest.java | 6 +-
.../apache/solr/cloud/MultiThreadedOCPTest.java | 6 +-
...verseerCollectionConfigSetProcessorTest.java | 71 ++++-
.../org/apache/solr/cloud/OverseerTest.java | 37 ++-
.../AutoAddReplicasPlanActionTest.java | 2 +-
.../autoscaling/ComputePlanActionTest.java | 18 +-
.../autoscaling/ExecutePlanActionTest.java | 15 +-
.../cloud/autoscaling/NodeAddedTriggerTest.java | 21 +-
.../cloud/autoscaling/NodeLostTriggerTest.java | 21 +-
.../solr/cloud/autoscaling/TestPolicyCloud.java | 17 +-
.../autoscaling/TriggerIntegrationTest.java | 6 +-
.../cloud/overseer/TestClusterStateMutator.java | 16 +-
.../solr/cloud/overseer/ZkStateReaderTest.java | 8 +-
.../solr/cloud/overseer/ZkStateWriterTest.java | 11 +-
.../client/solrj/cloud/DistributedQueue.java | 58 ++++
.../solrj/cloud/DistributedQueueFactory.java | 28 ++
.../autoscaling/AlreadyExistsException.java | 35 +++
.../cloud/autoscaling/AutoScalingConfig.java | 8 +-
.../cloud/autoscaling/BadVersionException.java | 40 +++
.../cloud/autoscaling/ClusterDataProvider.java | 52 ----
.../autoscaling/DelegatingCloudManager.java | 66 +++++
.../DelegatingClusterStateProvider.java | 113 +++++++
.../DelegatingNodeStateProvider.java | 43 +++
.../cloud/autoscaling/DistribStateManager.java | 69 +++++
.../cloud/autoscaling/MoveReplicaSuggester.java | 8 +-
.../cloud/autoscaling/NodeStateProvider.java | 42 +++
.../client/solrj/cloud/autoscaling/Policy.java | 35 ++-
.../solrj/cloud/autoscaling/PolicyHelper.java | 36 +--
.../solrj/cloud/autoscaling/ReplicaCount.java | 2 +-
.../solrj/cloud/autoscaling/ReplicaInfo.java | 69 ++++-
.../client/solrj/cloud/autoscaling/Row.java | 15 +-
.../cloud/autoscaling/SolrCloudManager.java | 60 ++++
.../solrj/cloud/autoscaling/VersionedData.java | 50 ++++
.../solr/client/solrj/cloud/package-info.java | 23 ++
.../solr/client/solrj/impl/CloudSolrClient.java | 6 +-
.../client/solrj/impl/ClusterStateProvider.java | 33 ++-
.../solrj/impl/HttpClusterStateProvider.java | 38 ++-
.../solrj/impl/SolrClientCloudManager.java | 166 +++++++++++
.../solrj/impl/SolrClientDataProvider.java | 250 ----------------
.../solrj/impl/SolrClientNodeStateProvider.java | 230 ++++++++++++++
.../impl/ZkClientClusterStateProvider.java | 38 ++-
.../solrj/impl/ZkDistribStateManager.java | 166 +++++++++++
.../solrj/request/CollectionAdminRequest.java | 4 +
.../solrj/cloud/autoscaling/TestPolicy.java | 296 +++++++++++--------
.../solrj/impl/CloudSolrClientCacheTest.java | 22 +-
98 files changed, 2662 insertions(+), 1439 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/8dd489ed/solr/CHANGES.txt
----------------------------------------------------------------------
diff --git a/solr/CHANGES.txt b/solr/CHANGES.txt
index 82786e0..2de3fee 100644
--- a/solr/CHANGES.txt
+++ b/solr/CHANGES.txt
@@ -52,6 +52,12 @@ Bug Fixes
* SOLR-11447: ZkStateWriter should process commands in atomic. (Cao Manh Dat, shalin)
+Optimizations
+----------------------
+* SOLR-11285: Refactor autoscaling framework to avoid direct references to Zookeeper and Solr
+ to make testing and large scale simulations easier. (ab)
+
+
================== 7.1.0 ==================
Consult the LUCENE_CHANGES.txt file for additional, low level, changes in this release.
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/8dd489ed/solr/core/src/java/org/apache/solr/api/ApiBag.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/api/ApiBag.java b/solr/core/src/java/org/apache/solr/api/ApiBag.java
index 55730b8..2caaeb9 100644
--- a/solr/core/src/java/org/apache/solr/api/ApiBag.java
+++ b/solr/core/src/java/org/apache/solr/api/ApiBag.java
@@ -327,6 +327,10 @@ public class ApiBag {
public List<Map> getErrs() {
return errs;
}
+
+ public String toString() {
+ return super.toString() + ", errors: " + getErrs() + ", ";
+ }
}
public static class LazyLoadedApi extends Api {
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/8dd489ed/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 62f01c5..d4331f9 100644
--- a/solr/core/src/java/org/apache/solr/cloud/AddReplicaCmd.java
+++ b/solr/core/src/java/org/apache/solr/cloud/AddReplicaCmd.java
@@ -18,6 +18,7 @@
package org.apache.solr.cloud;
+import java.io.IOException;
import java.lang.invoke.MethodHandles;
import java.util.Collection;
import java.util.Collections;
@@ -42,7 +43,6 @@ import org.apache.solr.common.params.ShardParams;
import org.apache.solr.common.util.NamedList;
import org.apache.solr.common.util.Utils;
import org.apache.solr.handler.component.ShardHandler;
-import org.apache.zookeeper.KeeperException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -70,7 +70,7 @@ public class AddReplicaCmd implements OverseerCollectionMessageHandler.Cmd {
}
ZkNodeProps addReplica(ClusterState clusterState, ZkNodeProps message, NamedList results, Runnable onComplete)
- throws KeeperException, InterruptedException {
+ throws IOException, InterruptedException {
log.debug("addReplica() : {}", Utils.toJSONString(message));
String collection = message.getStr(COLLECTION_PROP);
String node = message.getStr(CoreAdminParams.NODE);
@@ -119,7 +119,7 @@ public class AddReplicaCmd implements OverseerCollectionMessageHandler.Cmd {
}
} else {
node = getNodesForNewReplicas(clusterState, collection, shard, 1, node,
- ocmh.overseer.getZkController().getCoreContainer()).get(0).nodeName;// TODO: use replica type in this logic too
+ ocmh.overseer.getSolrCloudManager(), ocmh.overseer.getCoreContainer()).get(0).nodeName;// TODO: use replica type in this logic too
}
}
log.info("Node Identified {} for creating new replica", node);
@@ -128,7 +128,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.zkStateReader.getZkClient(), coll, shard, replicaType);
+ coreName = Assign.buildCoreName(ocmh.overseer.getSolrCloudManager().getDistribStateManager(), coll, shard, replicaType);
} else if (!skipCreateReplicaInClusterState) {
//Validate that the core name is unique in that collection
for (Slice slice : coll.getSlices()) {
@@ -158,7 +158,11 @@ public class AddReplicaCmd implements OverseerCollectionMessageHandler.Cmd {
if (coreNodeName != null) {
props = props.plus(ZkStateReader.CORE_NODE_NAME_PROP, coreNodeName);
}
- Overseer.getStateUpdateQueue(zkStateReader.getZkClient()).offer(Utils.toJSON(props));
+ try {
+ Overseer.getStateUpdateQueue(zkStateReader.getZkClient()).offer(Utils.toJSON(props));
+ } catch (Exception e) {
+ throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, "Exception updating Overseer state queue", e);
+ }
}
params.set(CoreAdminParams.CORE_NODE_NAME,
ocmh.waitToSeeReplicasInState(collection, Collections.singletonList(coreName)).get(coreName).getName());
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/8dd489ed/solr/core/src/java/org/apache/solr/cloud/Assign.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/Assign.java b/solr/core/src/java/org/apache/solr/cloud/Assign.java
index f22312f..36663e4 100644
--- a/solr/core/src/java/org/apache/solr/cloud/Assign.java
+++ b/solr/core/src/java/org/apache/solr/cloud/Assign.java
@@ -31,10 +31,12 @@ import java.util.Set;
import java.util.stream.Collectors;
import com.google.common.collect.ImmutableMap;
+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.impl.CloudSolrClient;
-import org.apache.solr.client.solrj.impl.SolrClientDataProvider;
-import org.apache.solr.client.solrj.impl.ZkClientClusterStateProvider;
+import org.apache.solr.client.solrj.cloud.autoscaling.SolrCloudManager;
+import org.apache.solr.client.solrj.cloud.autoscaling.VersionedData;
import org.apache.solr.client.solrj.cloud.autoscaling.AutoScalingConfig;
import org.apache.solr.cloud.rule.ReplicaAssigner;
import org.apache.solr.cloud.rule.Rule;
@@ -44,9 +46,7 @@ import org.apache.solr.common.cloud.DocCollection;
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.SolrZkClient;
import org.apache.solr.common.cloud.ZkNodeProps;
-import org.apache.solr.common.cloud.ZkStateReader;
import org.apache.solr.common.util.StrUtils;
import org.apache.solr.common.util.Utils;
import org.apache.solr.core.CoreContainer;
@@ -55,7 +55,6 @@ import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.apache.zookeeper.data.Stat;
import static org.apache.solr.client.solrj.cloud.autoscaling.Policy.POLICY;
import static org.apache.solr.cloud.OverseerCollectionMessageHandler.CREATE_NODE_SET;
@@ -68,43 +67,47 @@ import static org.apache.solr.common.cloud.ZkStateReader.CORE_NAME_PROP;
public class Assign {
private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
- public static int incAndGetId(SolrZkClient zkClient, String collection, int defaultValue) {
+ public static int incAndGetId(DistribStateManager stateManager, String collection, int defaultValue) {
String path = "/collections/"+collection;
try {
- if (!zkClient.exists(path, true)) {
+ if (!stateManager.hasData(path)) {
try {
- zkClient.makePath(path, true);
- } catch (KeeperException.NodeExistsException e) {
+ stateManager.makePath(path);
+ } catch (AlreadyExistsException e) {
// it's okay if another beats us creating the node
}
}
path += "/counter";
- if (!zkClient.exists(path, true)) {
+ if (!stateManager.hasData(path)) {
try {
- zkClient.create(path, NumberUtils.intToBytes(defaultValue), CreateMode.PERSISTENT, true);
- } catch (KeeperException.NodeExistsException e) {
+ stateManager.createData(path, NumberUtils.intToBytes(defaultValue), CreateMode.PERSISTENT);
+ } catch (AlreadyExistsException e) {
// it's okay if another beats us creating the node
}
}
} catch (InterruptedException e) {
Thread.interrupted();
throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, "Error creating counter node in Zookeeper for collection:" + collection, e);
- } catch (KeeperException e) {
+ } catch (IOException | KeeperException e) {
throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, "Error creating counter node in Zookeeper for collection:" + collection, e);
}
while (true) {
- Stat stat = new Stat();
try {
- byte[] data = zkClient.getData(path, null, stat, true);
- int currentId = NumberUtils.bytesToInt(data);
- data = NumberUtils.intToBytes(++currentId);
- zkClient.setData(path, data, stat.getVersion(), true);
- return currentId;
- } catch (KeeperException e) {
- if (e.code() != KeeperException.Code.BADVERSION) {
- throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, "Error inc and get counter from Zookeeper for collection:"+collection, e);
+ int version = 0;
+ int currentId = 0;
+ VersionedData data = stateManager.getData(path, null);
+ if (data != null) {
+ currentId = NumberUtils.bytesToInt(data.getData());
+ version = data.getVersion();
}
+ byte[] bytes = NumberUtils.intToBytes(++currentId);
+ stateManager.setData(path, bytes, version);
+ return currentId;
+ } catch (BadVersionException e) {
+ continue;
+ } catch (IOException | KeeperException e) {
+ throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, "Error inc and get counter from Zookeeper for collection:"+collection, e);
} catch (InterruptedException e) {
Thread.interrupted();
throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, "Error inc and get counter from Zookeeper for collection:" + collection, e);
@@ -112,14 +115,14 @@ public class Assign {
}
}
- public static String assignNode(SolrZkClient client, DocCollection collection) {
+ public static String assignNode(DistribStateManager stateManager, DocCollection collection) {
// for backward compatibility;
int defaultValue = defaultCounterValue(collection, false);
- String coreNodeName = "core_node" + incAndGetId(client, collection.getName(), defaultValue);
+ String coreNodeName = "core_node" + incAndGetId(stateManager, collection.getName(), defaultValue);
while (collection.getReplica(coreNodeName) != null) {
// there is wee chance that, the new coreNodeName id not totally unique,
// but this will be guaranteed unique for new collections
- coreNodeName = "core_node" + incAndGetId(client, collection.getName(), defaultValue);
+ coreNodeName = "core_node" + incAndGetId(stateManager, collection.getName(), defaultValue);
}
return coreNodeName;
}
@@ -185,20 +188,20 @@ public class Assign {
return defaultValue * 20;
}
- public static String buildCoreName(SolrZkClient zkClient, DocCollection collection, String shard, Replica.Type type, boolean newCollection) {
+ public static String buildCoreName(DistribStateManager stateManager, DocCollection collection, String shard, Replica.Type type, boolean newCollection) {
Slice slice = collection.getSlice(shard);
int defaultValue = defaultCounterValue(collection, newCollection);
- int replicaNum = incAndGetId(zkClient, collection.getName(), defaultValue);
+ int replicaNum = incAndGetId(stateManager, collection.getName(), defaultValue);
String coreName = buildCoreName(collection.getName(), shard, type, replicaNum);
while (existCoreName(coreName, slice)) {
- replicaNum = incAndGetId(zkClient, collection.getName(), defaultValue);
+ replicaNum = incAndGetId(stateManager, collection.getName(), defaultValue);
coreName = buildCoreName(collection.getName(), shard, type, replicaNum);
}
return coreName;
}
- public static String buildCoreName(SolrZkClient zkClient, DocCollection collection, String shard, Replica.Type type) {
- return buildCoreName(zkClient, collection, shard, type, false);
+ public static String buildCoreName(DistribStateManager stateManager, DocCollection collection, String shard, Replica.Type type) {
+ return buildCoreName(stateManager, collection, shard, type, false);
}
private static boolean existCoreName(String coreName, Slice slice) {
@@ -243,10 +246,10 @@ public class Assign {
List<String> shardNames,
int numNrtReplicas,
int numTlogReplicas,
- int numPullReplicas) throws KeeperException, InterruptedException {
+ int numPullReplicas) throws IOException, InterruptedException {
List<Map> rulesMap = (List) message.get("rule");
String policyName = message.getStr(POLICY);
- AutoScalingConfig autoScalingConfig = ocmh.zkStateReader.getAutoScalingConfig();
+ AutoScalingConfig autoScalingConfig = ocmh.overseer.getSolrCloudManager().getDistribStateManager().getAutoScalingConfig();
if (rulesMap == null && policyName == null && autoScalingConfig.getPolicy().getClusterPolicy().isEmpty()) {
log.debug("Identify nodes using default");
@@ -295,7 +298,7 @@ public class Assign {
PolicyHelper.SESSION_REF.set(ocmh.policySessionRef);
try {
return getPositionsUsingPolicy(collectionName,
- shardNames, numNrtReplicas, numTlogReplicas, numPullReplicas, policyName, ocmh.zkStateReader, nodeList);
+ shardNames, numNrtReplicas, numTlogReplicas, numPullReplicas, policyName, ocmh.overseer.getSolrCloudManager(), nodeList);
} finally {
PolicyHelper.SESSION_REF.remove();
}
@@ -324,7 +327,7 @@ public class Assign {
// could be created on live nodes given maxShardsPerNode, Replication factor (if from createShard) etc.
public static List<ReplicaCount> getNodesForNewReplicas(ClusterState clusterState, String collectionName,
String shard, int nrtReplicas,
- Object createNodeSet, CoreContainer cc) throws KeeperException, InterruptedException {
+ Object createNodeSet, SolrCloudManager cloudManager, CoreContainer cc) throws IOException, InterruptedException {
log.debug("getNodesForNewReplicas() shard: {} , replicas : {} , createNodeSet {}", shard, nrtReplicas, createNodeSet );
DocCollection coll = clusterState.getCollection(collectionName);
Integer maxShardsPerNode = coll.getMaxShardsPerNode();
@@ -356,13 +359,14 @@ public class Assign {
List l = (List) coll.get(DocCollection.RULE);
List<ReplicaPosition> replicaPositions = null;
if (l != null) {
+ // TODO: make it so that this method doesn't require access to CC
replicaPositions = getNodesViaRules(clusterState, shard, nrtReplicas, cc, coll, createNodeList, l);
}
String policyName = coll.getStr(POLICY);
- AutoScalingConfig autoScalingConfig = cc.getZkController().zkStateReader.getAutoScalingConfig();
+ AutoScalingConfig autoScalingConfig = cloudManager.getDistribStateManager().getAutoScalingConfig();
if (policyName != null || !autoScalingConfig.getPolicy().getClusterPolicy().isEmpty()) {
replicaPositions = Assign.getPositionsUsingPolicy(collectionName, Collections.singletonList(shard), nrtReplicas, 0, 0,
- policyName, cc.getZkController().zkStateReader, createNodeList);
+ policyName, cloudManager, createNodeList);
}
if(replicaPositions != null){
@@ -383,21 +387,17 @@ public class Assign {
int nrtReplicas,
int tlogReplicas,
int pullReplicas,
- String policyName, ZkStateReader zkStateReader,
- List<String> nodesList) throws KeeperException, InterruptedException {
+ String policyName, SolrCloudManager cloudManager,
+ List<String> nodesList) throws IOException, InterruptedException {
log.debug("shardnames {} NRT {} TLOG {} PULL {} , policy {}, nodeList {}", shardNames, nrtReplicas, tlogReplicas, pullReplicas, policyName, nodesList);
- SolrClientDataProvider clientDataProvider = null;
List<ReplicaPosition> replicaPositions = null;
- AutoScalingConfig autoScalingConfig = zkStateReader.getAutoScalingConfig();
- try (CloudSolrClient csc = new CloudSolrClient.Builder()
- .withClusterStateProvider(new ZkClientClusterStateProvider(zkStateReader))
- .build()) {
- clientDataProvider = new SolrClientDataProvider(csc);
+ AutoScalingConfig autoScalingConfig = cloudManager.getDistribStateManager().getAutoScalingConfig();
+ try {
Map<String, String> kvMap = Collections.singletonMap(collName, policyName);
replicaPositions = PolicyHelper.getReplicaLocations(
collName,
autoScalingConfig,
- clientDataProvider,
+ cloudManager,
kvMap,
shardNames,
nrtReplicas,
@@ -405,11 +405,10 @@ public class Assign {
pullReplicas,
nodesList);
return replicaPositions;
- } catch (IOException e) {
+ } catch (Exception e) {
throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, "Error closing CloudSolrClient",e);
} finally {
if (log.isTraceEnabled()) {
- if (clientDataProvider != null) log.trace("CLUSTER_DATA_PROVIDER: " + Utils.toJSONString(clientDataProvider));
if (replicaPositions != null)
log.trace("REPLICA_POSITIONS: " + Utils.toJSONString(Utils.getDeepCopy(replicaPositions, 7, true)));
log.trace("AUTOSCALING_CONF: " + Utils.toJSONString(autoScalingConfig));
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/8dd489ed/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 0b65915..93bc47a1 100644
--- a/solr/core/src/java/org/apache/solr/cloud/CreateCollectionCmd.java
+++ b/solr/core/src/java/org/apache/solr/cloud/CreateCollectionCmd.java
@@ -225,7 +225,7 @@ public class CreateCollectionCmd implements Cmd {
Map<String,ShardRequest> coresToCreate = new LinkedHashMap<>();
for (ReplicaPosition replicaPosition : replicaPositions) {
String nodeName = replicaPosition.node;
- String coreName = Assign.buildCoreName(ocmh.zkStateReader.getZkClient(), zkStateReader.getClusterState().getCollection(collectionName),
+ String coreName = Assign.buildCoreName(ocmh.overseer.getSolrCloudManager().getDistribStateManager(), zkStateReader.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));
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/8dd489ed/solr/core/src/java/org/apache/solr/cloud/CreateShardCmd.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/CreateShardCmd.java b/solr/core/src/java/org/apache/solr/cloud/CreateShardCmd.java
index 21258e4..90cecd7 100644
--- a/solr/core/src/java/org/apache/solr/cloud/CreateShardCmd.java
+++ b/solr/core/src/java/org/apache/solr/cloud/CreateShardCmd.java
@@ -17,6 +17,7 @@
package org.apache.solr.cloud;
+import java.io.IOException;
import java.lang.invoke.MethodHandles;
import java.util.ArrayList;
import java.util.Collections;
@@ -27,6 +28,7 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import com.google.common.collect.ImmutableMap;
+import org.apache.solr.client.solrj.cloud.autoscaling.AutoScalingConfig;
import org.apache.solr.client.solrj.cloud.autoscaling.Policy;
import org.apache.solr.client.solrj.cloud.autoscaling.PolicyHelper;
import org.apache.solr.cloud.OverseerCollectionMessageHandler.Cmd;
@@ -41,7 +43,6 @@ import org.apache.solr.common.params.CoreAdminParams;
import org.apache.solr.common.util.NamedList;
import org.apache.solr.common.util.SimpleOrderedMap;
import org.apache.solr.common.util.Utils;
-import org.apache.zookeeper.KeeperException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -52,7 +53,6 @@ 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.SHARD_ID_PROP;
-import static org.apache.solr.common.cloud.ZkStateReader.SOLR_AUTOSCALING_CONF_PATH;
import static org.apache.solr.common.cloud.ZkStateReader.TLOG_REPLICAS;
import static org.apache.solr.common.params.CommonAdminParams.ASYNC;
@@ -103,7 +103,7 @@ public class CreateShardCmd implements Cmd {
numPullReplicas);
} else {
List<Assign.ReplicaCount> sortedNodeList = getNodesForNewReplicas(clusterState, collectionName, sliceName, totalReplicas,
- createNodeSetStr, ocmh.overseer.getZkController().getCoreContainer());
+ createNodeSetStr, ocmh.overseer.getSolrCloudManager(), ocmh.overseer.getCoreContainer());
int i = 0;
positions = new ArrayList<>();
for (Map.Entry<Replica.Type, Integer> e : ImmutableMap.of(Replica.Type.NRT, numNrtReplicas,
@@ -124,7 +124,7 @@ public class CreateShardCmd implements Cmd {
countDownLatch = new CountDownLatch(totalReplicas);
for (ReplicaPosition position : positions) {
String nodeName = position.node;
- String coreName = Assign.buildCoreName(ocmh.zkStateReader.getZkClient(), collection, sliceName, position.type);
+ String coreName = Assign.buildCoreName(ocmh.overseer.getSolrCloudManager().getDistribStateManager(), collection, sliceName, position.type);
log.info("Creating replica " + coreName + " as part of slice " + sliceName + " of collection " + collectionName
+ " on " + nodeName);
@@ -173,8 +173,8 @@ public class CreateShardCmd implements Cmd {
}
static boolean usePolicyFramework(DocCollection collection, OverseerCollectionMessageHandler ocmh)
- throws KeeperException, InterruptedException {
- Map autoScalingJson = Utils.getJson(ocmh.zkStateReader.getZkClient(), SOLR_AUTOSCALING_CONF_PATH, true);
- return autoScalingJson.get(Policy.CLUSTER_POLICY) != null || collection.getPolicyName() != null;
+ throws IOException, InterruptedException {
+ AutoScalingConfig autoScalingConfig = ocmh.overseer.getSolrCloudManager().getDistribStateManager().getAutoScalingConfig();
+ return !autoScalingConfig.getPolicy().getClusterPolicy().isEmpty() || collection.getPolicyName() != null;
}
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/8dd489ed/solr/core/src/java/org/apache/solr/cloud/DeleteShardCmd.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/DeleteShardCmd.java b/solr/core/src/java/org/apache/solr/cloud/DeleteShardCmd.java
index fd868bc..f13fed5 100644
--- a/solr/core/src/java/org/apache/solr/cloud/DeleteShardCmd.java
+++ b/solr/core/src/java/org/apache/solr/cloud/DeleteShardCmd.java
@@ -26,7 +26,7 @@ import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
-import org.apache.solr.common.cloud.DistributedQueue;
+import org.apache.solr.client.solrj.cloud.DistributedQueue;
import org.apache.solr.cloud.OverseerCollectionMessageHandler.Cmd;
import org.apache.solr.cloud.overseer.OverseerAction;
import org.apache.solr.common.SolrException;
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/8dd489ed/solr/core/src/java/org/apache/solr/cloud/ElectionContext.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/ElectionContext.java b/solr/core/src/java/org/apache/solr/cloud/ElectionContext.java
index ba88e75..7169ea8 100644
--- a/solr/core/src/java/org/apache/solr/cloud/ElectionContext.java
+++ b/solr/core/src/java/org/apache/solr/cloud/ElectionContext.java
@@ -233,7 +233,8 @@ class ShardLeaderElectionContextBase extends ElectionContext {
ZkStateReader.SHARD_ID_PROP, shardId,
ZkStateReader.COLLECTION_PROP, collection,
ZkStateReader.BASE_URL_PROP, leaderProps.get(ZkStateReader.BASE_URL_PROP),
- ZkStateReader.CORE_NAME_PROP, leaderProps.get(ZkStateReader.CORE_NAME_PROP));
+ ZkStateReader.CORE_NAME_PROP, leaderProps.get(ZkStateReader.CORE_NAME_PROP),
+ ZkStateReader.STATE_PROP, Replica.State.ACTIVE.toString());
Overseer.getStateUpdateQueue(zkClient).offer(Utils.toJSON(m));
}
}
@@ -508,7 +509,7 @@ final class ShardLeaderElectionContext extends ShardLeaderElectionContextBase {
}
}
- public void publishActiveIfRegisteredAndNotActive(SolrCore core) throws KeeperException, InterruptedException {
+ public void publishActiveIfRegisteredAndNotActive(SolrCore core) throws Exception {
if (core.getCoreDescriptor().getCloudDescriptor().hasRegistered()) {
ZkStateReader zkStateReader = zkController.getZkStateReader();
zkStateReader.forceUpdateCollection(collection);
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/8dd489ed/solr/core/src/java/org/apache/solr/cloud/MigrateCmd.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/MigrateCmd.java b/solr/core/src/java/org/apache/solr/cloud/MigrateCmd.java
index f70c836..cacccb0 100644
--- a/solr/core/src/java/org/apache/solr/cloud/MigrateCmd.java
+++ b/solr/core/src/java/org/apache/solr/cloud/MigrateCmd.java
@@ -257,7 +257,7 @@ public class MigrateCmd implements OverseerCollectionMessageHandler.Cmd {
log.info("Creating a replica of temporary collection: {} on the target leader node: {}",
tempSourceCollectionName, targetLeader.getNodeName());
- String tempCollectionReplica2 = Assign.buildCoreName(ocmh.zkStateReader.getZkClient(),
+ String tempCollectionReplica2 = Assign.buildCoreName(ocmh.overseer.getSolrCloudManager().getDistribStateManager(),
zkStateReader.getClusterState().getCollection(tempSourceCollectionName), tempSourceSlice.getName(), Replica.Type.NRT);
props = new HashMap<>();
props.put(Overseer.QUEUE_OPERATION, ADDREPLICA.toLower());
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/8dd489ed/solr/core/src/java/org/apache/solr/cloud/MoveReplicaCmd.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/MoveReplicaCmd.java b/solr/core/src/java/org/apache/solr/cloud/MoveReplicaCmd.java
index 04bf054..f94353e 100644
--- a/solr/core/src/java/org/apache/solr/cloud/MoveReplicaCmd.java
+++ b/solr/core/src/java/org/apache/solr/cloud/MoveReplicaCmd.java
@@ -184,7 +184,7 @@ public class MoveReplicaCmd implements Cmd{
private void moveNormalReplica(ClusterState clusterState, NamedList results, String targetNode, String async,
DocCollection coll, Replica replica, Slice slice, int timeout) throws Exception {
- String newCoreName = Assign.buildCoreName(ocmh.zkStateReader.getZkClient(), coll, slice.getName(), replica.getType());
+ String newCoreName = Assign.buildCoreName(ocmh.overseer.getSolrCloudManager().getDistribStateManager(), coll, slice.getName(), replica.getType());
ZkNodeProps addReplicasProps = new ZkNodeProps(
COLLECTION_PROP, coll.getName(),
SHARD_ID_PROP, slice.getName(),
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/8dd489ed/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 fbf901c..b32158d 100644
--- a/solr/core/src/java/org/apache/solr/cloud/Overseer.java
+++ b/solr/core/src/java/org/apache/solr/cloud/Overseer.java
@@ -19,19 +19,16 @@ package org.apache.solr.cloud;
import java.io.Closeable;
import java.io.IOException;
import java.lang.invoke.MethodHandles;
-import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
-import java.util.Locale;
import java.util.Map;
import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.atomic.AtomicInteger;
import com.codahale.metrics.Timer;
-import org.apache.solr.client.solrj.SolrResponse;
+import org.apache.solr.client.solrj.cloud.autoscaling.SolrCloudManager;
+import org.apache.solr.client.solrj.impl.ClusterStateProvider;
import org.apache.solr.cloud.autoscaling.OverseerTriggerThread;
import org.apache.solr.cloud.overseer.ClusterStateMutator;
import org.apache.solr.cloud.overseer.CollectionMutator;
@@ -51,6 +48,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;
@@ -115,11 +113,11 @@ public class Overseer implements Closeable {
}
public Stats getStateUpdateQueueStats() {
- return stateUpdateQueue.getStats();
+ return stateUpdateQueue.getZkStats();
}
public Stats getWorkQueueStats() {
- return workQueue.getStats();
+ return workQueue.getZkStats();
}
@Override
@@ -164,7 +162,7 @@ public class Overseer implements Closeable {
byte[] data = fallbackQueue.peek();
while (fallbackQueueSize > 0 && data != null) {
final ZkNodeProps message = ZkNodeProps.load(data);
- log.debug("processMessage: fallbackQueueSize: {}, message = {}", fallbackQueue.getStats().getQueueLength(), message);
+ log.debug("processMessage: fallbackQueueSize: {}, message = {}", fallbackQueue.getZkStats().getQueueLength(), message);
// force flush to ZK after each message because there is no fallback if workQueue items
// are removed from workQueue but fail to be written to ZK
try {
@@ -217,7 +215,7 @@ public class Overseer implements Closeable {
for (Pair<String, byte[]> head : queue) {
byte[] data = head.second();
final ZkNodeProps message = ZkNodeProps.load(data);
- log.debug("processMessage: queueSize: {}, message = {} current state version: {}", stateUpdateQueue.getStats().getQueueLength(), message, clusterState.getZkClusterStateVersion());
+ log.debug("processMessage: queueSize: {}, message = {} current state version: {}", stateUpdateQueue.getZkStats().getQueueLength(), message, clusterState.getZkClusterStateVersion());
processedNodes.add(head.first());
fallbackQueueSize = processedNodes.size();
@@ -279,7 +277,7 @@ public class Overseer implements Closeable {
// ZooKeeper in which case another Overseer should take over
// TODO: if ordering for the message is not important, we could
// track retries and put it back on the end of the queue
- log.error("Overseer could not process the current clusterstate state update message, skipping the message.", e);
+ log.error("Overseer could not process the current clusterstate state update message, skipping the message: " + message, e);
stats.error(operation);
} finally {
timerContext.stop();
@@ -338,19 +336,19 @@ public class Overseer implements Closeable {
if (collectionAction != null) {
switch (collectionAction) {
case CREATE:
- return Collections.singletonList(new ClusterStateMutator(getZkStateReader()).createCollection(clusterState, message));
+ return Collections.singletonList(new ClusterStateMutator(getSolrCloudManager()).createCollection(clusterState, message));
case DELETE:
- return Collections.singletonList(new ClusterStateMutator(getZkStateReader()).deleteCollection(clusterState, message));
+ return Collections.singletonList(new ClusterStateMutator(getSolrCloudManager()).deleteCollection(clusterState, message));
case CREATESHARD:
- return Collections.singletonList(new CollectionMutator(getZkStateReader()).createShard(clusterState, message));
+ return Collections.singletonList(new CollectionMutator(getSolrCloudManager()).createShard(clusterState, message));
case DELETESHARD:
- return Collections.singletonList(new CollectionMutator(getZkStateReader()).deleteShard(clusterState, message));
+ return Collections.singletonList(new CollectionMutator(getSolrCloudManager()).deleteShard(clusterState, message));
case ADDREPLICA:
- return Collections.singletonList(new SliceMutator(getZkStateReader()).addReplica(clusterState, message));
+ return Collections.singletonList(new SliceMutator(getSolrCloudManager()).addReplica(clusterState, message));
case ADDREPLICAPROP:
- return Collections.singletonList(new ReplicaMutator(getZkStateReader()).addReplicaProperty(clusterState, message));
+ return Collections.singletonList(new ReplicaMutator(getSolrCloudManager()).addReplicaProperty(clusterState, message));
case DELETEREPLICAPROP:
- return Collections.singletonList(new ReplicaMutator(getZkStateReader()).deleteReplicaProperty(clusterState, message));
+ return Collections.singletonList(new ReplicaMutator(getSolrCloudManager()).deleteReplicaProperty(clusterState, message));
case BALANCESHARDUNIQUE:
ExclusiveSliceProperty dProp = new ExclusiveSliceProperty(clusterState, message);
if (dProp.balanceProperty()) {
@@ -360,9 +358,9 @@ public class Overseer implements Closeable {
break;
case MODIFYCOLLECTION:
CollectionsHandler.verifyRuleParams(zkController.getCoreContainer() ,message.getProperties());
- return Collections.singletonList(new CollectionMutator(reader).modifyCollection(clusterState,message));
+ return Collections.singletonList(new CollectionMutator(getSolrCloudManager()).modifyCollection(clusterState,message));
case MIGRATESTATEFORMAT:
- return Collections.singletonList(new ClusterStateMutator(reader).migrateStateFormat(clusterState, message));
+ return Collections.singletonList(new ClusterStateMutator(getSolrCloudManager()).migrateStateFormat(clusterState, message));
default:
throw new RuntimeException("unknown operation:" + operation
+ " contents:" + message.getProperties());
@@ -374,17 +372,17 @@ public class Overseer implements Closeable {
}
switch (overseerAction) {
case STATE:
- return Collections.singletonList(new ReplicaMutator(getZkStateReader()).setState(clusterState, message));
+ return Collections.singletonList(new ReplicaMutator(getSolrCloudManager()).setState(clusterState, message));
case LEADER:
- return Collections.singletonList(new SliceMutator(getZkStateReader()).setShardLeader(clusterState, message));
+ return Collections.singletonList(new SliceMutator(getSolrCloudManager()).setShardLeader(clusterState, message));
case DELETECORE:
- return Collections.singletonList(new SliceMutator(getZkStateReader()).removeReplica(clusterState, message));
+ return Collections.singletonList(new SliceMutator(getSolrCloudManager()).removeReplica(clusterState, message));
case ADDROUTINGRULE:
- return Collections.singletonList(new SliceMutator(getZkStateReader()).addRoutingRule(clusterState, message));
+ return Collections.singletonList(new SliceMutator(getSolrCloudManager()).addRoutingRule(clusterState, message));
case REMOVEROUTINGRULE:
- return Collections.singletonList(new SliceMutator(getZkStateReader()).removeRoutingRule(clusterState, message));
+ return Collections.singletonList(new SliceMutator(getSolrCloudManager()).removeRoutingRule(clusterState, message));
case UPDATESHARDSTATE:
- return Collections.singletonList(new SliceMutator(getZkStateReader()).updateShardState(clusterState, message));
+ return Collections.singletonList(new SliceMutator(getSolrCloudManager()).updateShardState(clusterState, message));
case QUIT:
if (myId.equals(message.get(ID))) {
log.info("Quit command received {} {}", message, LeaderElector.getNodeName(myId));
@@ -527,7 +525,8 @@ public class Overseer implements Closeable {
ccThread.setDaemon(true);
ThreadGroup triggerThreadGroup = new ThreadGroup("Overseer autoscaling triggers");
- OverseerTriggerThread trigger = new OverseerTriggerThread(zkController, config);
+ OverseerTriggerThread trigger = new OverseerTriggerThread(zkController.getCoreContainer().getResourceLoader(),
+ zkController.getSolrCloudManager(), config);
triggerThread = new OverseerThread(triggerThreadGroup, trigger, "OverseerAutoScalingTriggerThread-" + id);
updaterThread.start();
@@ -543,7 +542,15 @@ public class Overseer implements Closeable {
ZkController getZkController(){
return zkController;
}
-
+
+ public CoreContainer getCoreContainer() {
+ return zkController.getCoreContainer();
+ }
+
+ public SolrCloudManager getSolrCloudManager() {
+ return zkController.getSolrCloudManager();
+ }
+
/**
* For tests.
*
@@ -624,7 +631,7 @@ public class Overseer implements Closeable {
* This method will create the /overseer znode in ZooKeeper if it does not exist already.
*
* @param zkClient the {@link SolrZkClient} to be used for reading/writing to the queue
- * @param zkStats a {@link Overseer.Stats} object which tracks statistics for all zookeeper operations performed by this queue
+ * @param zkStats a {@link Stats} object which tracks statistics for all zookeeper operations performed by this queue
* @return a {@link ZkDistributedQueue} object
*/
static ZkDistributedQueue getStateUpdateQueue(final SolrZkClient zkClient, Stats zkStats) {
@@ -644,7 +651,7 @@ public class Overseer implements Closeable {
* This method will create the /overseer znode in ZooKeeper if it does not exist already.
*
* @param zkClient the {@link SolrZkClient} to be used for reading/writing to the queue
- * @param zkStats a {@link Overseer.Stats} object which tracks statistics for all zookeeper operations performed by this queue
+ * @param zkStats a {@link Stats} object which tracks statistics for all zookeeper operations performed by this queue
* @return a {@link ZkDistributedQueue} object
*/
static ZkDistributedQueue getInternalWorkQueue(final SolrZkClient zkClient, Stats zkStats) {
@@ -772,126 +779,13 @@ public class Overseer implements Closeable {
return "true".equals(legacyProperty);
}
- public ZkStateReader getZkStateReader() {
- return reader;
- }
-
- /**
- * Used to hold statistics about overseer operations. It will be exposed
- * to the OverseerCollectionProcessor to return statistics.
- *
- * This is experimental API and subject to change.
- */
- public static class Stats {
- static final int MAX_STORED_FAILURES = 10;
-
- final Map<String, Stat> stats = new ConcurrentHashMap<>();
- private volatile int queueLength;
-
- public Map<String, Stat> getStats() {
- return stats;
- }
-
- public int getSuccessCount(String operation) {
- Stat stat = stats.get(operation.toLowerCase(Locale.ROOT));
- return stat == null ? 0 : stat.success.get();
- }
-
- public int getErrorCount(String operation) {
- Stat stat = stats.get(operation.toLowerCase(Locale.ROOT));
- return stat == null ? 0 : stat.errors.get();
- }
-
- public void success(String operation) {
- String op = operation.toLowerCase(Locale.ROOT);
- Stat stat = stats.get(op);
- if (stat == null) {
- stat = new Stat();
- stats.put(op, stat);
- }
- stat.success.incrementAndGet();
- }
-
- public void error(String operation) {
- String op = operation.toLowerCase(Locale.ROOT);
- Stat stat = stats.get(op);
- if (stat == null) {
- stat = new Stat();
- stats.put(op, stat);
- }
- stat.errors.incrementAndGet();
- }
-
- public Timer.Context time(String operation) {
- String op = operation.toLowerCase(Locale.ROOT);
- Stat stat = stats.get(op);
- if (stat == null) {
- stat = new Stat();
- stats.put(op, stat);
- }
- return stat.requestTime.time();
- }
-
- public void storeFailureDetails(String operation, ZkNodeProps request, SolrResponse resp) {
- String op = operation.toLowerCase(Locale.ROOT);
- Stat stat = stats.get(op);
- if (stat == null) {
- stat = new Stat();
- stats.put(op, stat);
- }
- LinkedList<FailedOp> failedOps = stat.failureDetails;
- synchronized (failedOps) {
- if (failedOps.size() >= MAX_STORED_FAILURES) {
- failedOps.removeFirst();
- }
- failedOps.addLast(new FailedOp(request, resp));
- }
- }
-
- public List<FailedOp> getFailureDetails(String operation) {
- Stat stat = stats.get(operation.toLowerCase(Locale.ROOT));
- if (stat == null || stat.failureDetails.isEmpty()) return null;
- LinkedList<FailedOp> failedOps = stat.failureDetails;
- synchronized (failedOps) {
- ArrayList<FailedOp> ret = new ArrayList<>(failedOps);
- return ret;
- }
- }
-
- public int getQueueLength() {
- return queueLength;
- }
-
- public void setQueueLength(int queueLength) {
- this.queueLength = queueLength;
- }
-
- public void clear() {
- stats.clear();
- }
+ public static boolean isLegacy(ClusterStateProvider clusterStateProvider) {
+ String legacyProperty = clusterStateProvider.getClusterProperty(ZkStateReader.LEGACY_CLOUD, "false");
+ return "true".equals(legacyProperty);
}
- public static class Stat {
- public final AtomicInteger success;
- public final AtomicInteger errors;
- public final Timer requestTime;
- public final LinkedList<FailedOp> failureDetails;
-
- public Stat() {
- this.success = new AtomicInteger();
- this.errors = new AtomicInteger();
- this.requestTime = new Timer();
- this.failureDetails = new LinkedList<>();
- }
+ public ZkStateReader getZkStateReader() {
+ return reader;
}
- public static class FailedOp {
- public final ZkNodeProps req;
- public final SolrResponse resp;
-
- public FailedOp(ZkNodeProps req, SolrResponse resp) {
- this.req = req;
- this.resp = resp;
- }
- }
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/8dd489ed/solr/core/src/java/org/apache/solr/cloud/OverseerCollectionConfigSetProcessor.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/OverseerCollectionConfigSetProcessor.java b/solr/core/src/java/org/apache/solr/cloud/OverseerCollectionConfigSetProcessor.java
index 8c7a056..570843a 100644
--- a/solr/core/src/java/org/apache/solr/cloud/OverseerCollectionConfigSetProcessor.java
+++ b/solr/core/src/java/org/apache/solr/cloud/OverseerCollectionConfigSetProcessor.java
@@ -34,9 +34,9 @@ import static org.apache.solr.cloud.OverseerConfigSetMessageHandler.CONFIGSETS_A
public class OverseerCollectionConfigSetProcessor extends OverseerTaskProcessor {
public OverseerCollectionConfigSetProcessor(ZkStateReader zkStateReader, String myId,
- final ShardHandler shardHandler,
- String adminPath, Overseer.Stats stats, Overseer overseer,
- OverseerNodePrioritizer overseerNodePrioritizer) {
+ final ShardHandler shardHandler,
+ String adminPath, Stats stats, Overseer overseer,
+ OverseerNodePrioritizer overseerNodePrioritizer) {
this(
zkStateReader,
myId,
@@ -55,7 +55,7 @@ public class OverseerCollectionConfigSetProcessor extends OverseerTaskProcessor
protected OverseerCollectionConfigSetProcessor(ZkStateReader zkStateReader, String myId,
final ShardHandlerFactory shardHandlerFactory,
String adminPath,
- Overseer.Stats stats,
+ Stats stats,
Overseer overseer,
OverseerNodePrioritizer overseerNodePrioritizer,
OverseerTaskQueue workQueue,
@@ -80,7 +80,7 @@ public class OverseerCollectionConfigSetProcessor extends OverseerTaskProcessor
String myId,
final ShardHandlerFactory shardHandlerFactory,
String adminPath,
- Overseer.Stats stats,
+ Stats stats,
Overseer overseer,
OverseerNodePrioritizer overseerNodePrioritizer) {
final OverseerCollectionMessageHandler collMessageHandler = new OverseerCollectionMessageHandler(
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/8dd489ed/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 509d17c..08d6d10 100644
--- a/solr/core/src/java/org/apache/solr/cloud/OverseerCollectionMessageHandler.java
+++ b/solr/core/src/java/org/apache/solr/cloud/OverseerCollectionMessageHandler.java
@@ -36,8 +36,8 @@ import com.google.common.collect.ImmutableMap;
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.PolicyHelper;
-import org.apache.solr.common.cloud.DistributedQueue;
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;
@@ -145,7 +145,7 @@ public class OverseerCollectionMessageHandler implements OverseerMessageHandler
String adminPath;
ZkStateReader zkStateReader;
String myId;
- Overseer.Stats stats;
+ Stats stats;
// Set that tracks collections that are currently being processed by a running task.
// This is used for handling mutual exclusion of the tasks.
@@ -172,7 +172,7 @@ public class OverseerCollectionMessageHandler implements OverseerMessageHandler
public OverseerCollectionMessageHandler(ZkStateReader zkStateReader, String myId,
final ShardHandlerFactory shardHandlerFactory,
String adminPath,
- Overseer.Stats stats,
+ Stats stats,
Overseer overseer,
OverseerNodePrioritizer overseerPrioritizer) {
this.zkStateReader = zkStateReader;
@@ -442,7 +442,7 @@ public class OverseerCollectionMessageHandler implements OverseerMessageHandler
}
}
- void deleteCoreNode(String collectionName, String replicaName, Replica replica, String core) throws KeeperException, InterruptedException {
+ void deleteCoreNode(String collectionName, String replicaName, Replica replica, String core) throws Exception {
ZkNodeProps m = new ZkNodeProps(
Overseer.QUEUE_OPERATION, OverseerAction.DELETECORE.toLower(),
ZkStateReader.CORE_NAME_PROP, core,
@@ -463,8 +463,7 @@ public class OverseerCollectionMessageHandler implements OverseerMessageHandler
}
//TODO should we not remove in the next release ?
- private void migrateStateFormat(ClusterState state, ZkNodeProps message, NamedList results)
- throws KeeperException, InterruptedException {
+ private void migrateStateFormat(ClusterState state, ZkNodeProps message, NamedList results) throws Exception {
final String collectionName = message.getStr(COLLECTION_PROP);
boolean firstLoop = true;
@@ -635,7 +634,7 @@ public class OverseerCollectionMessageHandler implements OverseerMessageHandler
private void modifyCollection(ClusterState clusterState, ZkNodeProps message, NamedList results)
- throws KeeperException, InterruptedException {
+ throws Exception {
final String collectionName = message.getStr(ZkStateReader.COLLECTION_PROP);
//the rest of the processing is based on writing cluster state properties
@@ -713,7 +712,7 @@ public class OverseerCollectionMessageHandler implements OverseerMessageHandler
}
ZkNodeProps addReplica(ClusterState clusterState, ZkNodeProps message, NamedList results, Runnable onComplete)
- throws KeeperException, InterruptedException {
+ throws Exception {
return ((AddReplicaCmd) commandMap.get(ADDREPLICA)).addReplica(clusterState, message, results, onComplete);
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/8dd489ed/solr/core/src/java/org/apache/solr/cloud/OverseerNodePrioritizer.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/OverseerNodePrioritizer.java b/solr/core/src/java/org/apache/solr/cloud/OverseerNodePrioritizer.java
index 7db1315..34ee041 100644
--- a/solr/core/src/java/org/apache/solr/cloud/OverseerNodePrioritizer.java
+++ b/solr/core/src/java/org/apache/solr/cloud/OverseerNodePrioritizer.java
@@ -31,7 +31,6 @@ import org.apache.solr.common.util.Utils;
import org.apache.solr.handler.component.ShardHandler;
import org.apache.solr.handler.component.ShardHandlerFactory;
import org.apache.solr.handler.component.ShardRequest;
-import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -56,7 +55,7 @@ public class OverseerNodePrioritizer {
this.shardHandlerFactory = shardHandlerFactory;
}
- public synchronized void prioritizeOverseerNodes(String overseerId) throws KeeperException, InterruptedException {
+ public synchronized void prioritizeOverseerNodes(String overseerId) throws Exception {
SolrZkClient zk = zkStateReader.getZkClient();
if(!zk.exists(ZkStateReader.ROLES,true))return;
Map m = (Map) Utils.fromJSON(zk.getData(ZkStateReader.ROLES, null, new Stat(), true));
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/8dd489ed/solr/core/src/java/org/apache/solr/cloud/OverseerStatusCmd.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/OverseerStatusCmd.java b/solr/core/src/java/org/apache/solr/cloud/OverseerStatusCmd.java
index a24deb1..aba4872 100644
--- a/solr/core/src/java/org/apache/solr/cloud/OverseerStatusCmd.java
+++ b/solr/core/src/java/org/apache/solr/cloud/OverseerStatusCmd.java
@@ -64,8 +64,8 @@ public class OverseerStatusCmd implements Cmd {
NamedList stateUpdateQueueStats = new NamedList();
NamedList workQueueStats = new NamedList();
NamedList collectionQueueStats = new NamedList();
- Overseer.Stats stats = ocmh.stats;
- for (Map.Entry<String, Overseer.Stat> entry : stats.getStats().entrySet()) {
+ Stats stats = ocmh.stats;
+ for (Map.Entry<String, Stats.Stat> entry : stats.getStats().entrySet()) {
String key = entry.getKey();
NamedList<Object> lst = new SimpleOrderedMap<>();
if (key.startsWith("collection_")) {
@@ -74,10 +74,10 @@ public class OverseerStatusCmd implements Cmd {
int errors = stats.getErrorCount(entry.getKey());
lst.add("requests", successes);
lst.add("errors", errors);
- List<Overseer.FailedOp> failureDetails = stats.getFailureDetails(key);
+ List<Stats.FailedOp> failureDetails = stats.getFailureDetails(key);
if (failureDetails != null) {
List<SimpleOrderedMap<Object>> failures = new ArrayList<>();
- for (Overseer.FailedOp failedOp : failureDetails) {
+ for (Stats.FailedOp failedOp : failureDetails) {
SimpleOrderedMap<Object> fail = new SimpleOrderedMap<>();
fail.add("request", failedOp.req.getProperties());
fail.add("response", failedOp.resp.getResponse());
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/8dd489ed/solr/core/src/java/org/apache/solr/cloud/OverseerTaskProcessor.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/OverseerTaskProcessor.java b/solr/core/src/java/org/apache/solr/cloud/OverseerTaskProcessor.java
index 4dffb21..d014fc4 100644
--- a/solr/core/src/java/org/apache/solr/cloud/OverseerTaskProcessor.java
+++ b/solr/core/src/java/org/apache/solr/cloud/OverseerTaskProcessor.java
@@ -91,7 +91,7 @@ public class OverseerTaskProcessor implements Runnable, Closeable {
private boolean isClosed;
- private Overseer.Stats stats;
+ private Stats stats;
// Set of tasks that have been picked up for processing but not cleaned up from zk work-queue.
// It may contain tasks that have completed execution, have been entered into the completed/failed map in zk but not
@@ -122,7 +122,7 @@ public class OverseerTaskProcessor implements Runnable, Closeable {
private OverseerNodePrioritizer prioritizer;
public OverseerTaskProcessor(ZkStateReader zkStateReader, String myId,
- Overseer.Stats stats,
+ Stats stats,
OverseerMessageHandlerSelector selector,
OverseerNodePrioritizer prioritizer,
OverseerTaskQueue workQueue,
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/8dd489ed/solr/core/src/java/org/apache/solr/cloud/OverseerTaskQueue.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/OverseerTaskQueue.java b/solr/core/src/java/org/apache/solr/cloud/OverseerTaskQueue.java
index b381c3b..4a6ea8a 100644
--- a/solr/core/src/java/org/apache/solr/cloud/OverseerTaskQueue.java
+++ b/solr/core/src/java/org/apache/solr/cloud/OverseerTaskQueue.java
@@ -45,10 +45,10 @@ public class OverseerTaskQueue extends ZkDistributedQueue {
private static final String RESPONSE_PREFIX = "qnr-" ;
public OverseerTaskQueue(SolrZkClient zookeeper, String dir) {
- this(zookeeper, dir, new Overseer.Stats());
+ this(zookeeper, dir, new Stats());
}
- public OverseerTaskQueue(SolrZkClient zookeeper, String dir, Overseer.Stats stats) {
+ public OverseerTaskQueue(SolrZkClient zookeeper, String dir, Stats stats) {
super(zookeeper, dir, stats);
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/8dd489ed/solr/core/src/java/org/apache/solr/cloud/RecoveryStrategy.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/RecoveryStrategy.java b/solr/core/src/java/org/apache/solr/cloud/RecoveryStrategy.java
index 73e6142..53e3743 100644
--- a/solr/core/src/java/org/apache/solr/cloud/RecoveryStrategy.java
+++ b/solr/core/src/java/org/apache/solr/cloud/RecoveryStrategy.java
@@ -64,7 +64,6 @@ import org.apache.solr.update.processor.DistributedUpdateProcessor;
import org.apache.solr.util.RefCounted;
import org.apache.solr.util.SolrPluginUtils;
import org.apache.solr.util.plugin.NamedListInitializedPlugin;
-import org.apache.zookeeper.KeeperException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -174,7 +173,7 @@ public class RecoveryStrategy implements Runnable, Closeable {
final private void recoveryFailed(final SolrCore core,
final ZkController zkController, final String baseUrl,
- final String shardZkNodeName, final CoreDescriptor cd) throws KeeperException, InterruptedException {
+ final String shardZkNodeName, final CoreDescriptor cd) throws Exception {
SolrException.log(LOG, "Recovery failed - I give up.");
try {
zkController.publish(cd, Replica.State.RECOVERY_FAILED);
@@ -298,7 +297,7 @@ public class RecoveryStrategy implements Runnable, Closeable {
}
}
- final public void doRecovery(SolrCore core) throws KeeperException, InterruptedException {
+ final public void doRecovery(SolrCore core) throws Exception {
if (core.getCoreDescriptor().getCloudDescriptor().requiresTransactionLog()) {
doSyncOrReplicateRecovery(core);
} else {
@@ -441,7 +440,7 @@ public class RecoveryStrategy implements Runnable, Closeable {
}
// TODO: perhaps make this grab a new core each time through the loop to handle core reloads?
- final public void doSyncOrReplicateRecovery(SolrCore core) throws KeeperException, InterruptedException {
+ final public void doSyncOrReplicateRecovery(SolrCore core) throws Exception {
boolean replayed = false;
boolean successfulRecovery = false;
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/8dd489ed/solr/core/src/java/org/apache/solr/cloud/RestoreCmd.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/RestoreCmd.java b/solr/core/src/java/org/apache/solr/cloud/RestoreCmd.java
index f2d3794..f25a3cf 100644
--- a/solr/core/src/java/org/apache/solr/cloud/RestoreCmd.java
+++ b/solr/core/src/java/org/apache/solr/cloud/RestoreCmd.java
@@ -33,7 +33,7 @@ import java.util.Optional;
import java.util.Properties;
import java.util.Set;
-import org.apache.solr.common.cloud.DistributedQueue;
+import org.apache.solr.client.solrj.cloud.DistributedQueue;
import org.apache.solr.client.solrj.cloud.autoscaling.PolicyHelper;
import org.apache.solr.cloud.overseer.OverseerAction;
import org.apache.solr.common.cloud.ReplicaPosition;
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/8dd489ed/solr/core/src/java/org/apache/solr/cloud/SplitShardCmd.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/SplitShardCmd.java b/solr/core/src/java/org/apache/solr/cloud/SplitShardCmd.java
index bde92d6..10cc7c8 100644
--- a/solr/core/src/java/org/apache/solr/cloud/SplitShardCmd.java
+++ b/solr/core/src/java/org/apache/solr/cloud/SplitShardCmd.java
@@ -27,7 +27,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
-import org.apache.solr.common.cloud.DistributedQueue;
+import org.apache.solr.client.solrj.cloud.DistributedQueue;
import org.apache.solr.client.solrj.cloud.autoscaling.PolicyHelper;
import org.apache.solr.client.solrj.request.CoreAdminRequest;
import org.apache.solr.cloud.OverseerCollectionMessageHandler.Cmd;
@@ -207,7 +207,7 @@ public class SplitShardCmd implements Cmd {
for (int i = 0; i < subRanges.size(); i++) {
String subSlice = slice + "_" + i;
subSlices.add(subSlice);
- String subShardName = Assign.buildCoreName(ocmh.zkStateReader.getZkClient(), collection, subSlice, Replica.Type.NRT);
+ String subShardName = Assign.buildCoreName(ocmh.overseer.getSolrCloudManager().getDistribStateManager(), collection, subSlice, Replica.Type.NRT);
subShardNames.add(subShardName);
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/8dd489ed/solr/core/src/java/org/apache/solr/cloud/Stats.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/Stats.java b/solr/core/src/java/org/apache/solr/cloud/Stats.java
new file mode 100644
index 0000000..36593f6
--- /dev/null
+++ b/solr/core/src/java/org/apache/solr/cloud/Stats.java
@@ -0,0 +1,147 @@
+/*
+ * 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;
+
+import java.util.ArrayList;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import com.codahale.metrics.Timer;
+import org.apache.solr.client.solrj.SolrResponse;
+import org.apache.solr.common.cloud.ZkNodeProps;
+
+/**
+ * Used to hold statistics about some SolrCloud operations.
+ *
+ * This is experimental API and subject to change.
+ */
+public class Stats {
+ static final int MAX_STORED_FAILURES = 10;
+
+ final Map<String, Stat> stats = new ConcurrentHashMap<>();
+ private volatile int queueLength;
+
+ public Map<String, Stat> getStats() {
+ return stats;
+ }
+
+ public int getSuccessCount(String operation) {
+ Stat stat = stats.get(operation.toLowerCase(Locale.ROOT));
+ return stat == null ? 0 : stat.success.get();
+ }
+
+ public int getErrorCount(String operation) {
+ Stat stat = stats.get(operation.toLowerCase(Locale.ROOT));
+ return stat == null ? 0 : stat.errors.get();
+ }
+
+ public void success(String operation) {
+ String op = operation.toLowerCase(Locale.ROOT);
+ Stat stat = stats.get(op);
+ if (stat == null) {
+ stat = new Stat();
+ stats.put(op, stat);
+ }
+ stat.success.incrementAndGet();
+ }
+
+ public void error(String operation) {
+ String op = operation.toLowerCase(Locale.ROOT);
+ Stat stat = stats.get(op);
+ if (stat == null) {
+ stat = new Stat();
+ stats.put(op, stat);
+ }
+ stat.errors.incrementAndGet();
+ }
+
+ public Timer.Context time(String operation) {
+ String op = operation.toLowerCase(Locale.ROOT);
+ Stat stat = stats.get(op);
+ if (stat == null) {
+ stat = new Stat();
+ stats.put(op, stat);
+ }
+ return stat.requestTime.time();
+ }
+
+ public void storeFailureDetails(String operation, ZkNodeProps request, SolrResponse resp) {
+ String op = operation.toLowerCase(Locale.ROOT);
+ Stat stat = stats.get(op);
+ if (stat == null) {
+ stat = new Stat();
+ stats.put(op, stat);
+ }
+ LinkedList<FailedOp> failedOps = stat.failureDetails;
+ synchronized (failedOps) {
+ if (failedOps.size() >= MAX_STORED_FAILURES) {
+ failedOps.removeFirst();
+ }
+ failedOps.addLast(new FailedOp(request, resp));
+ }
+ }
+
+ public List<FailedOp> getFailureDetails(String operation) {
+ Stat stat = stats.get(operation.toLowerCase(Locale.ROOT));
+ if (stat == null || stat.failureDetails.isEmpty()) return null;
+ LinkedList<FailedOp> failedOps = stat.failureDetails;
+ synchronized (failedOps) {
+ ArrayList<FailedOp> ret = new ArrayList<>(failedOps);
+ return ret;
+ }
+ }
+
+ public int getQueueLength() {
+ return queueLength;
+ }
+
+ public void setQueueLength(int queueLength) {
+ this.queueLength = queueLength;
+ }
+
+ public void clear() {
+ stats.clear();
+ }
+
+ public static class Stat {
+ public final AtomicInteger success;
+ public final AtomicInteger errors;
+ public final Timer requestTime;
+ public final LinkedList<FailedOp> failureDetails;
+
+ public Stat() {
+ this.success = new AtomicInteger();
+ this.errors = new AtomicInteger();
+ this.requestTime = new Timer();
+ this.failureDetails = new LinkedList<>();
+ }
+ }
+
+ public static class FailedOp {
+ public final ZkNodeProps req;
+ public final SolrResponse resp;
+
+ public FailedOp(ZkNodeProps req, SolrResponse resp) {
+ this.req = req;
+ this.resp = resp;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/8dd489ed/solr/core/src/java/org/apache/solr/cloud/ZkController.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/ZkController.java b/solr/core/src/java/org/apache/solr/cloud/ZkController.java
index e55c966..35171da 100644
--- a/solr/core/src/java/org/apache/solr/cloud/ZkController.java
+++ b/solr/core/src/java/org/apache/solr/cloud/ZkController.java
@@ -51,8 +51,12 @@ import java.util.concurrent.atomic.AtomicReference;
import com.google.common.base.Strings;
import org.apache.commons.lang.StringUtils;
+import org.apache.solr.client.solrj.cloud.autoscaling.SolrCloudManager;
+import org.apache.solr.client.solrj.impl.CloudSolrClient;
import org.apache.solr.client.solrj.impl.HttpSolrClient;
import org.apache.solr.client.solrj.impl.HttpSolrClient.Builder;
+import org.apache.solr.client.solrj.impl.SolrClientCloudManager;
+import org.apache.solr.client.solrj.impl.ZkClientClusterStateProvider;
import org.apache.solr.client.solrj.request.CoreAdminRequest.WaitForState;
import org.apache.solr.client.solrj.cloud.autoscaling.TriggerEventType;
import org.apache.solr.cloud.overseer.OverseerAction;
@@ -191,6 +195,8 @@ public class ZkController {
private final SolrZkClient zkClient;
private final ZkCmdExecutor cmdExecutor;
public final ZkStateReader zkStateReader;
+ private SolrCloudManager cloudManager;
+ private CloudSolrClient cloudSolrClient;
private final String zkServerAddress; // example: 127.0.0.1:54062/solr
@@ -435,7 +441,7 @@ public class ZkController {
});
init(registerOnReconnect);
-
+
assert ObjectReleaseTracker.track(this);
}
@@ -554,6 +560,12 @@ public class ZkController {
IOUtils.closeQuietly(overseerElector.getContext());
IOUtils.closeQuietly(overseer);
} finally {
+ if (cloudSolrClient != null) {
+ IOUtils.closeQuietly(cloudSolrClient);
+ }
+ if (cloudManager != null) {
+ IOUtils.closeQuietly(cloudManager);
+ }
try {
try {
zkStateReader.close();
@@ -588,6 +600,22 @@ public class ZkController {
return zkStateReader.getClusterState();
}
+ public SolrCloudManager getSolrCloudManager() {
+ if (cloudManager != null) {
+ return cloudManager;
+ }
+ synchronized(this) {
+ if (cloudManager != null) {
+ return cloudManager;
+ }
+ cloudSolrClient = new CloudSolrClient.Builder()
+ .withClusterStateProvider(new ZkClientClusterStateProvider(zkStateReader))
+ .build();
+ cloudManager = new SolrClientCloudManager(new ZkDistributedQueueFactory(zkClient), cloudSolrClient);
+ }
+ return cloudManager;
+ }
+
/**
* Returns config file data (in bytes)
*/
@@ -1290,18 +1318,18 @@ public class ZkController {
return baseURL;
}
- public void publish(final CoreDescriptor cd, final Replica.State state) throws KeeperException, InterruptedException {
+ public void publish(final CoreDescriptor cd, final Replica.State state) throws Exception {
publish(cd, state, true);
}
- public void publish(final CoreDescriptor cd, final Replica.State state, boolean updateLastState) throws KeeperException, InterruptedException {
+ public void publish(final CoreDescriptor cd, final Replica.State state, boolean updateLastState) throws Exception {
publish(cd, state, updateLastState, false);
}
/**
* Publish core state to overseer.
*/
- public void publish(final CoreDescriptor cd, final Replica.State state, boolean updateLastState, boolean forcePublish) throws KeeperException, InterruptedException {
+ public void publish(final CoreDescriptor cd, final Replica.State state, boolean updateLastState, boolean forcePublish) throws Exception {
if (!forcePublish) {
try (SolrCore core = cc.getCore(cd.getName())) {
if (core == null || core.isClosed()) {
@@ -1410,7 +1438,7 @@ public class ZkController {
return true;
}
- public void unregister(String coreName, CoreDescriptor cd) throws InterruptedException, KeeperException {
+ public void unregister(String coreName, CoreDescriptor cd) throws Exception {
final String coreNodeName = cd.getCloudDescriptor().getCoreNodeName();
final String collection = cd.getCloudDescriptor().getCollectionName();
@@ -1441,8 +1469,7 @@ public class ZkController {
overseerJobQueue.offer(Utils.toJSON(m));
}
- public void createCollection(String collection) throws KeeperException,
- InterruptedException {
+ public void createCollection(String collection) throws Exception {
ZkNodeProps m = new ZkNodeProps(Overseer.QUEUE_OPERATION,
CollectionParams.CollectionAction.CREATE.toLower(), ZkStateReader.NODE_NAME_PROP, getNodeName(),
ZkStateReader.COLLECTION_PROP, collection);
@@ -1570,6 +1597,9 @@ public class ZkController {
Thread.currentThread().interrupt();
log.error("", e);
throw new ZooKeeperException(SolrException.ErrorCode.SERVER_ERROR, "", e);
+ } catch (Exception e) {
+ log.error("", e);
+ throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, "", e);
}
doGetShardIdAndNodeNameProcess(cd);
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/8dd489ed/solr/core/src/java/org/apache/solr/cloud/ZkDistributedQueue.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/ZkDistributedQueue.java b/solr/core/src/java/org/apache/solr/cloud/ZkDistributedQueue.java
index c007dd4..fee7475 100644
--- a/solr/core/src/java/org/apache/solr/cloud/ZkDistributedQueue.java
+++ b/solr/core/src/java/org/apache/solr/cloud/ZkDistributedQueue.java
@@ -19,7 +19,10 @@ package org.apache.solr.cloud;
import java.lang.invoke.MethodHandles;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.NoSuchElementException;
import java.util.TreeSet;
import java.util.concurrent.TimeUnit;
@@ -31,9 +34,9 @@ import java.util.function.Predicate;
import com.codahale.metrics.Timer;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
+import org.apache.solr.client.solrj.cloud.DistributedQueue;
import org.apache.solr.common.SolrException;
import org.apache.solr.common.SolrException.ErrorCode;
-import org.apache.solr.common.cloud.DistributedQueue;
import org.apache.solr.common.cloud.SolrZkClient;
import org.apache.solr.common.cloud.ZkCmdExecutor;
import org.apache.solr.common.util.Pair;
@@ -47,7 +50,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
- * A distributed queue. Optimized for single-consumer,
+ * A ZK-based distributed queue. Optimized for single-consumer,
* multiple-producer: if there are multiple consumers on the same ZK queue,
* the results should be correct but inefficient
*/
@@ -71,7 +74,7 @@ public class ZkDistributedQueue implements DistributedQueue {
final SolrZkClient zookeeper;
- final Overseer.Stats stats;
+ final Stats stats;
/**
* A lock that guards all of the mutable state that follows.
@@ -104,14 +107,14 @@ public class ZkDistributedQueue implements DistributedQueue {
private final AtomicInteger offerPermits = new AtomicInteger(0);
public ZkDistributedQueue(SolrZkClient zookeeper, String dir) {
- this(zookeeper, dir, new Overseer.Stats());
+ this(zookeeper, dir, new Stats());
}
- public ZkDistributedQueue(SolrZkClient zookeeper, String dir, Overseer.Stats stats) {
+ public ZkDistributedQueue(SolrZkClient zookeeper, String dir, Stats stats) {
this(zookeeper, dir, stats, 0);
}
- public ZkDistributedQueue(SolrZkClient zookeeper, String dir, Overseer.Stats stats, int maxQueueSize) {
+ public ZkDistributedQueue(SolrZkClient zookeeper, String dir, Stats stats, int maxQueueSize) {
this.dir = dir;
ZkCmdExecutor cmdExecutor = new ZkCmdExecutor(zookeeper.getZkClientTimeout());
@@ -331,10 +334,35 @@ public class ZkDistributedQueue implements DistributedQueue {
}
}
- public Overseer.Stats getStats() {
+ public Stats getZkStats() {
return stats;
}
+ @Override
+ public Map<String, Object> getStats() {
+ if (stats == null) {
+ return Collections.emptyMap();
+ }
+ Map<String, Object> res = new HashMap<>();
+ res.put("queueLength", stats.getQueueLength());
+ final Map<String, Object> statsMap = new HashMap<>();
+ res.put("stats", statsMap);
+ stats.getStats().forEach((op, stat) -> {
+ final Map<String, Object> statMap = new HashMap<>();
+ statMap.put("success", stat.success.get());
+ statMap.put("errors", stat.errors.get());
+ final List<Map<String, Object>> failed = new ArrayList<>(stat.failureDetails.size());
+ statMap.put("failureDetails", failed);
+ stat.failureDetails.forEach(failedOp -> {
+ Map<String, Object> fo = new HashMap<>();
+ fo.put("req", failedOp.req);
+ fo.put("resp", failedOp.resp);
+ });
+ statsMap.put(op, statMap);
+ });
+ return res;
+ }
+
/**
* Returns the name if the first known child node, or {@code null} if the queue is empty.
* This is the only place {@link #knownChildren} is ever updated!
@@ -400,7 +428,8 @@ public class ZkDistributedQueue implements DistributedQueue {
/**
* Return the currently-known set of elements, using child names from memory. If no children are found, or no
* children pass {@code acceptFilter}, waits up to {@code waitMillis} for at least one child to become available.
- * Package-private to support {@link OverseerTaskQueue} specifically.
+ * <p>
+ * Package-private to support {@link OverseerTaskQueue} specifically.</p>
*/
@Override
public Collection<Pair<String, byte[]>> peekElements(int max, long waitMillis, Predicate<String> acceptFilter) throws KeeperException, InterruptedException {
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/8dd489ed/solr/core/src/java/org/apache/solr/cloud/ZkDistributedQueueFactory.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/ZkDistributedQueueFactory.java b/solr/core/src/java/org/apache/solr/cloud/ZkDistributedQueueFactory.java
new file mode 100644
index 0000000..1cf79e5
--- /dev/null
+++ b/solr/core/src/java/org/apache/solr/cloud/ZkDistributedQueueFactory.java
@@ -0,0 +1,43 @@
+/*
+ * 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;
+
+import java.io.IOException;
+
+import org.apache.solr.client.solrj.cloud.DistributedQueue;
+import org.apache.solr.client.solrj.cloud.DistributedQueueFactory;
+import org.apache.solr.common.cloud.SolrZkClient;
+
+/**
+ * Implementation of {@link DistributedQueueFactory} that uses ZooKeeper.
+ */
+public class ZkDistributedQueueFactory implements DistributedQueueFactory {
+ private final SolrZkClient zkClient;
+
+ public ZkDistributedQueueFactory(SolrZkClient zkClient) {
+ this.zkClient = zkClient;
+ }
+ @Override
+ public DistributedQueue makeQueue(String path) throws IOException {
+ return new ZkDistributedQueue(zkClient, path);
+ }
+
+ @Override
+ public void removeQueue(String path) throws IOException {
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/8dd489ed/solr/core/src/java/org/apache/solr/cloud/autoscaling/ActionContext.java
----------------------------------------------------------------------
diff --git a/solr/core/src/java/org/apache/solr/cloud/autoscaling/ActionContext.java b/solr/core/src/java/org/apache/solr/cloud/autoscaling/ActionContext.java
index 178a972..9348a22 100644
--- a/solr/core/src/java/org/apache/solr/cloud/autoscaling/ActionContext.java
+++ b/solr/core/src/java/org/apache/solr/cloud/autoscaling/ActionContext.java
@@ -20,8 +20,8 @@ package org.apache.solr.cloud.autoscaling;
import java.io.IOException;
import java.util.Map;
+import org.apache.solr.client.solrj.cloud.autoscaling.SolrCloudManager;
import org.apache.solr.common.MapWriter;
-import org.apache.solr.core.CoreContainer;
/**
* Provides additional context for the TriggerAction such as the trigger instance on
@@ -30,18 +30,18 @@ import org.apache.solr.core.CoreContainer;
*/
public class ActionContext implements MapWriter {
- private final CoreContainer coreContainer;
+ private final SolrCloudManager cloudManager;
private final AutoScaling.Trigger source;
private final Map<String, Object> properties;
- public ActionContext(CoreContainer coreContainer, AutoScaling.Trigger source, Map<String, Object> properties) {
- this.coreContainer = coreContainer;
+ public ActionContext(SolrCloudManager cloudManager, AutoScaling.Trigger source, Map<String, Object> properties) {
+ this.cloudManager = cloudManager;
this.source = source;
this.properties = properties;
}
- public CoreContainer getCoreContainer() {
- return coreContainer;
+ public SolrCloudManager getCloudManager() {
+ return cloudManager;
}
public AutoScaling.Trigger getSource() {