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/09/28 22:09:39 UTC
[23/24] lucene-solr:jira/solr-11285: Merge branch
'feature/autoscaling' into jira/solr-11285
Merge branch 'feature/autoscaling' into jira/solr-11285
Project: http://git-wip-us.apache.org/repos/asf/lucene-solr/repo
Commit: http://git-wip-us.apache.org/repos/asf/lucene-solr/commit/1f25f3ec
Tree: http://git-wip-us.apache.org/repos/asf/lucene-solr/tree/1f25f3ec
Diff: http://git-wip-us.apache.org/repos/asf/lucene-solr/diff/1f25f3ec
Branch: refs/heads/jira/solr-11285
Commit: 1f25f3ec293592f34b4c24acfd92dd58d97c6c31
Parents: 18b9936 fba51f3
Author: Andrzej Bialecki <ab...@apache.org>
Authored: Thu Sep 28 22:40:13 2017 +0200
Committer: Andrzej Bialecki <ab...@apache.org>
Committed: Thu Sep 28 22:40:13 2017 +0200
----------------------------------------------------------------------
lucene/CHANGES.txt | 3 +
.../lucene/analysis/synonym/SynonymMap.java | 3 +-
.../apache/lucene/search/ConjunctionScorer.java | 4 +-
.../apache/lucene/search/TestConjunctions.java | 56 +++-
solr/CHANGES.txt | 66 ++---
.../java/org/apache/solr/cloud/Overseer.java | 75 +----
.../solr/cloud/autoscaling/AutoScaling.java | 11 +-
.../cloud/autoscaling/AutoScalingHandler.java | 6 +-
.../cloud/autoscaling/ComputePlanAction.java | 50 ----
.../cloud/autoscaling/ExecutePlanAction.java | 110 +++++++-
.../cloud/autoscaling/NodeAddedTrigger.java | 10 +-
.../solr/cloud/autoscaling/NodeLostTrigger.java | 5 +
.../autoscaling/OverseerTriggerThread.java | 52 +++-
.../cloud/autoscaling/ScheduledTriggers.java | 84 +++++-
.../cloud/autoscaling/SearchRateTrigger.java | 275 -------------------
.../solr/handler/CdcrReplicatorManager.java | 8 +-
.../solr/handler/admin/MetricsHandler.java | 11 +-
.../org/apache/solr/response/DocsStreamer.java | 87 ++----
.../solr/response/RetrieveFieldsOptimizer.java | 130 +++++++++
.../apache/solr/search/SolrDocumentFetcher.java | 25 +-
.../facet/FacetFieldProcessorByHashDV.java | 11 +-
.../org/apache/solr/search/facet/HLLAgg.java | 1 +
.../org/apache/solr/search/facet/UniqueAgg.java | 1 +
.../solr/search/join/GraphPointsCollector.java | 1 +
.../apache/solr/cloud/CdcrBootstrapTest.java | 1 -
.../cloud/SharedFSAutoReplicaFailoverTest.java | 25 +-
.../cloud/TestCloudJSONFacetJoinDomain.java | 24 +-
.../AutoAddReplicasIntegrationTest.java | 2 +
.../autoscaling/ComputePlanActionTest.java | 57 +---
.../autoscaling/ExecutePlanActionTest.java | 42 ++-
.../cloud/autoscaling/NodeLostTriggerTest.java | 1 +
.../autoscaling/SearchRateTriggerTest.java | 168 -----------
.../admin/AutoscalingHistoryHandlerTest.java | 21 +-
.../solr/handler/admin/MetricsHandlerTest.java | 9 -
.../response/TestRetrieveFieldsOptimizer.java | 56 ++++
.../solr/search/facet/TestJsonFacets.java | 26 ++
.../conf/solrconfig.xml | 1 -
.../cloud/autoscaling/ClusterDataProvider.java | 1 -
.../solrj/cloud/autoscaling/NoneSuggester.java | 4 +-
.../solrj/cloud/autoscaling/PolicyHelper.java | 1 -
.../solrj/cloud/autoscaling/Preference.java | 8 +
.../solrj/cloud/autoscaling/ReplicaInfo.java | 24 --
.../autoscaling/SolrCloudDataProvider.java | 10 +-
.../solrj/impl/SolrClientCloudDataProvider.java | 12 +
.../impl/SolrClientClusterDataProvider.java | 2 +-
.../apache/solr/common/cloud/ZkStateReader.java | 3 +-
.../solr/common/params/AutoScalingParams.java | 7 -
.../solrj/cloud/autoscaling/TestPolicy.java | 38 ---
48 files changed, 762 insertions(+), 866 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/1f25f3ec/solr/core/src/java/org/apache/solr/cloud/Overseer.java
----------------------------------------------------------------------
diff --cc solr/core/src/java/org/apache/solr/cloud/Overseer.java
index 1ad8eef,f665857..baa5d25
--- 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,17 +19,14 @@@ 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.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.client.solrj.cloud.DistributedQueue;
+import org.apache.solr.client.solrj.cloud.autoscaling.ClusterDataProvider;
+import org.apache.solr.client.solrj.cloud.autoscaling.SolrCloudDataProvider;
- import org.apache.solr.cloud.autoscaling.AutoScaling;
- 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.cloud.overseer.CollectionMutator;
@@@ -53,11 -48,8 +45,9 @@@ import org.apache.solr.common.util.IOUt
import org.apache.solr.common.util.ObjectReleaseTracker;
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.request.LocalSolrQueryRequest;
- import org.apache.solr.response.SolrQueryResponse;
import org.apache.solr.update.UpdateShardHandler;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
@@@ -533,14 -523,8 +521,9 @@@ public class Overseer implements Closea
ccThread = new OverseerThread(ccTg, overseerCollectionConfigSetProcessor, "OverseerCollectionConfigSetProcessor-" + id);
ccThread.setDaemon(true);
- //TODO nocommit, autoscaling framework should start autoAddReplicas trigger automatically (implicitly)
- autoscalingTriggerCreator = new Thread(createAutoscalingTriggerIfNotExist(), "AutoscalingTriggerCreator");
- autoscalingTriggerCreator.setDaemon(true);
- autoscalingTriggerCreator.start();
-
ThreadGroup triggerThreadGroup = new ThreadGroup("Overseer autoscaling triggers");
- OverseerTriggerThread trigger = new OverseerTriggerThread(zkController, config);
+ OverseerTriggerThread trigger = new OverseerTriggerThread(zkController.getCoreContainer().getResourceLoader(),
- zkController.getSolrCloudDataProvider());
++ zkController.getSolrCloudDataProvider(), config);
triggerThread = new OverseerThread(triggerThreadGroup, trigger, "OverseerAutoScalingTriggerThread-" + id);
updaterThread.start();
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/1f25f3ec/solr/core/src/java/org/apache/solr/cloud/autoscaling/AutoScaling.java
----------------------------------------------------------------------
diff --cc solr/core/src/java/org/apache/solr/cloud/autoscaling/AutoScaling.java
index c314cd6,bd74c84..3842ac3
--- a/solr/core/src/java/org/apache/solr/cloud/autoscaling/AutoScaling.java
+++ b/solr/core/src/java/org/apache/solr/cloud/autoscaling/AutoScaling.java
@@@ -24,9 -24,10 +24,10 @@@ import java.util.Map
import com.google.common.base.Preconditions;
import org.apache.lucene.store.AlreadyClosedException;
+import org.apache.solr.client.solrj.cloud.autoscaling.SolrCloudDataProvider;
import org.apache.solr.client.solrj.cloud.autoscaling.TriggerEventType;
+ import org.apache.solr.common.util.Utils;
-import org.apache.solr.core.CoreContainer;
-import org.apache.solr.cloud.ZkController;
+import org.apache.solr.core.SolrResourceLoader;
public class AutoScaling {
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/1f25f3ec/solr/core/src/java/org/apache/solr/cloud/autoscaling/AutoScalingHandler.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/1f25f3ec/solr/core/src/java/org/apache/solr/cloud/autoscaling/ComputePlanAction.java
----------------------------------------------------------------------
diff --cc solr/core/src/java/org/apache/solr/cloud/autoscaling/ComputePlanAction.java
index 42cdf18,5d5e3eb..47468c4
--- a/solr/core/src/java/org/apache/solr/cloud/autoscaling/ComputePlanAction.java
+++ b/solr/core/src/java/org/apache/solr/cloud/autoscaling/ComputePlanAction.java
@@@ -20,22 -20,19 +20,15 @@@ package org.apache.solr.cloud.autoscali
import java.io.IOException;
import java.lang.invoke.MethodHandles;
import java.util.ArrayList;
--import java.util.HashMap;
- import java.util.HashSet;
import java.util.List;
import java.util.Map;
- import java.util.Set;
import org.apache.solr.client.solrj.SolrRequest;
import org.apache.solr.client.solrj.cloud.autoscaling.AutoScalingConfig;
+import org.apache.solr.client.solrj.cloud.autoscaling.ClusterDataProvider;
import org.apache.solr.client.solrj.cloud.autoscaling.Policy;
-import org.apache.solr.client.solrj.impl.CloudSolrClient;
-import org.apache.solr.client.solrj.impl.SolrClientDataProvider;
-import org.apache.solr.common.cloud.ZkStateReader;
+import org.apache.solr.client.solrj.cloud.autoscaling.SolrCloudDataProvider;
import org.apache.solr.common.params.CollectionParams;
- import org.apache.solr.client.solrj.cloud.autoscaling.NoneSuggester;
- import org.apache.solr.client.solrj.cloud.autoscaling.ReplicaInfo;
- import org.apache.solr.common.params.AutoScalingParams;
- import org.apache.solr.common.util.Pair;
-import org.apache.solr.core.CoreContainer;
-import org.apache.zookeeper.KeeperException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/1f25f3ec/solr/core/src/java/org/apache/solr/cloud/autoscaling/ExecutePlanAction.java
----------------------------------------------------------------------
diff --cc solr/core/src/java/org/apache/solr/cloud/autoscaling/ExecutePlanAction.java
index 0a536ed,8221ff8..a8bfd15
--- a/solr/core/src/java/org/apache/solr/cloud/autoscaling/ExecutePlanAction.java
+++ b/solr/core/src/java/org/apache/solr/cloud/autoscaling/ExecutePlanAction.java
@@@ -17,15 -17,30 +17,28 @@@
package org.apache.solr.cloud.autoscaling;
+ import java.io.IOException;
import java.lang.invoke.MethodHandles;
import java.util.ArrayList;
+ import java.util.Collections;
import java.util.List;
+ import java.util.concurrent.TimeUnit;
+ import java.util.concurrent.TimeoutException;
+ import org.apache.commons.lang3.exception.ExceptionUtils;
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.impl.CloudSolrClient;
-import org.apache.solr.client.solrj.impl.HttpSolrClient;
++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.SolrCloudDataProvider;
+ import org.apache.solr.client.solrj.request.CollectionAdminRequest;
+ import org.apache.solr.client.solrj.response.RequestStatusState;
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.util.NamedList;
+ import org.apache.solr.common.util.Utils;
-import org.apache.solr.core.CoreContainer;
+ import org.apache.zookeeper.CreateMode;
-import org.apache.zookeeper.KeeperException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@@ -49,21 -72,106 +65,113 @@@ public class ExecutePlanAction extends
for (SolrRequest operation : operations) {
log.info("Executing operation: {}", operation.getParams());
try {
- SolrResponse response = null;
- if (operation instanceof CollectionAdminRequest.AsyncCollectionAdminRequest) {
- CollectionAdminRequest.AsyncCollectionAdminRequest req = (CollectionAdminRequest.AsyncCollectionAdminRequest) operation;
- String asyncId = event.getSource() + '/' + event.getId() + '/' + counter;
- String znode = saveAsyncId(event, context, asyncId);
- log.debug("Saved requestId: {} in znode: {}", asyncId, znode);
- asyncId = req.processAsync(asyncId, cloudSolrClient);
- CollectionAdminRequest.RequestStatusResponse statusResponse = waitForTaskToFinish(cloudSolrClient, asyncId,
- DEFAULT_TASK_TIMEOUT_SECONDS, TimeUnit.SECONDS);
- if (statusResponse != null) {
- RequestStatusState state = statusResponse.getRequestStatus();
- if (state == RequestStatusState.COMPLETED || state == RequestStatusState.FAILED || state == RequestStatusState.NOT_FOUND) {
- try {
- zkClient.delete(znode, -1, true);
- } catch (KeeperException e) {
- log.warn("Unexpected exception while trying to delete znode: " + znode, e);
- }
- }
- response = statusResponse;
- }
- } else {
- response = operation.process(cloudSolrClient);
- }
+ SolrResponse response = dataProvider.request(operation);
++ // nocommit
++//=======
++// try (CloudSolrClient cloudSolrClient = new CloudSolrClient.Builder()
++// .withZkHost(container.getZkController().getZkServerAddress())
++// .withHttpClient(container.getUpdateShardHandler().getHttpClient())
++// .build()) {
++// int counter = 0;
++// for (SolrRequest operation : operations) {
++// log.info("Executing operation: {}", operation.getParams());
++// try {
++// SolrResponse response = null;
++// if (operation instanceof CollectionAdminRequest.AsyncCollectionAdminRequest) {
++// CollectionAdminRequest.AsyncCollectionAdminRequest req = (CollectionAdminRequest.AsyncCollectionAdminRequest) operation;
++// String asyncId = event.getSource() + '/' + event.getId() + '/' + counter;
++// String znode = saveAsyncId(event, context, asyncId);
++// log.debug("Saved requestId: {} in znode: {}", asyncId, znode);
++// asyncId = req.processAsync(asyncId, cloudSolrClient);
++// CollectionAdminRequest.RequestStatusResponse statusResponse = waitForTaskToFinish(cloudSolrClient, asyncId,
++// DEFAULT_TASK_TIMEOUT_SECONDS, TimeUnit.SECONDS);
++// if (statusResponse != null) {
++// RequestStatusState state = statusResponse.getRequestStatus();
++// if (state == RequestStatusState.COMPLETED || state == RequestStatusState.FAILED || state == RequestStatusState.NOT_FOUND) {
++// try {
++// zkClient.delete(znode, -1, true);
++// } catch (KeeperException e) {
++// log.warn("Unexpected exception while trying to delete znode: " + znode, e);
++// }
++// }
++// response = statusResponse;
++// }
++// } else {
++// response = operation.process(cloudSolrClient);
++// }
++//>>>>>>> feature/autoscaling
+ NamedList<Object> result = response.getResponse();
context.getProperties().compute("responses", (s, o) -> {
List<NamedList<Object>> responses = (List<NamedList<Object>>) o;
if (responses == null) responses = new ArrayList<>(operations.size());
- responses.add(response.getResponse());
+ responses.add(result);
return responses;
});
- } catch (SolrServerException | HttpSolrClient.RemoteSolrException e) {
++ } catch (IOException e) {
++ throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, "Unable to talk to ZooKeeper", e);
++// } catch (InterruptedException e) {
++// Thread.currentThread().interrupt();
++// throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, "ExecutePlanAction was interrupted", e);
+ } catch (Exception e) {
throw new SolrException(SolrException.ErrorCode.SERVER_ERROR,
"Unexpected exception executing operation: " + operation.getParams(), e);
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, "ExecutePlanAction was interrupted", e);
- } catch (KeeperException e) {
- throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, "Unable to talk to ZooKeeper", e);
}
+
- counter++;
++// counter++;
}
- } catch (IOException e) {
+ } catch (Exception e) {
throw new SolrException(SolrException.ErrorCode.SERVER_ERROR,
"Unexpected IOException while processing event: " + event, e);
}
}
+
- static CollectionAdminRequest.RequestStatusResponse waitForTaskToFinish(CloudSolrClient cloudSolrClient, String requestId, long duration, TimeUnit timeUnit) throws SolrServerException, IOException, InterruptedException {
++
++ static CollectionAdminRequest.RequestStatusResponse waitForTaskToFinish(SolrCloudDataProvider dataProvider, String requestId, long duration, TimeUnit timeUnit) throws IOException, InterruptedException {
+ long timeoutSeconds = timeUnit.toSeconds(duration);
+ RequestStatusState state = RequestStatusState.NOT_FOUND;
+ CollectionAdminRequest.RequestStatusResponse statusResponse = null;
+ for (int i = 0; i < timeoutSeconds; i++) {
+ try {
- statusResponse = CollectionAdminRequest.requestStatus(requestId).process(cloudSolrClient);
++ statusResponse = (CollectionAdminRequest.RequestStatusResponse)dataProvider.request(CollectionAdminRequest.requestStatus(requestId));
+ state = statusResponse.getRequestStatus();
+ if (state == RequestStatusState.COMPLETED || state == RequestStatusState.FAILED) {
+ log.info("Task with requestId={} finished with state={} in {}s", requestId, state, i * 5);
- CollectionAdminRequest.deleteAsyncId(requestId).process(cloudSolrClient);
++ dataProvider.request(CollectionAdminRequest.deleteAsyncId(requestId));
+ return statusResponse;
+ } else if (state == RequestStatusState.NOT_FOUND) {
+ // the request for this id was never actually submitted! no harm done, just bail out
+ log.warn("Task with requestId={} was not found on overseer", requestId);
- CollectionAdminRequest.deleteAsyncId(requestId).process(cloudSolrClient);
++ dataProvider.request(CollectionAdminRequest.deleteAsyncId(requestId));
+ return statusResponse;
+ }
+ } catch (Exception e) {
+ Throwable rootCause = ExceptionUtils.getRootCause(e);
+ if (rootCause instanceof IllegalStateException && rootCause.getMessage().contains("Connection pool shut down")) {
+ throw e;
+ }
+ if (rootCause instanceof TimeoutException && rootCause.getMessage().contains("Could not connect to ZooKeeper")) {
+ throw e;
+ }
+ log.error("Unexpected Exception while querying status of requestId=" + requestId, e);
+ }
+ if (i > 0 && i % 5 == 0) {
+ log.debug("Task with requestId={} still not complete after {}s. Last state={}", requestId, i * 5, state);
+ }
+ TimeUnit.SECONDS.sleep(5);
+ }
+ log.debug("Task with requestId={} did not complete within 5 minutes. Last state={}", requestId, state);
+ return statusResponse;
+ }
+
+ /**
+ * Saves the given asyncId in ZK as a persistent sequential node.
+ *
+ * @return the path of the newly created node in ZooKeeper
+ */
- private String saveAsyncId(TriggerEvent event, ActionContext context, String asyncId) throws InterruptedException, KeeperException {
++ private String saveAsyncId(DistribStateManager stateManager, TriggerEvent event, ActionContext context, String asyncId) throws InterruptedException, AlreadyExistsException, IOException {
+ String parentPath = ZkStateReader.SOLR_AUTOSCALING_TRIGGER_STATE_PATH + "/" + context.getSource().getName() + "/" + getName();
- CoreContainer container = context.getCoreContainer();
- SolrZkClient zkClient = container.getZkController().getZkClient();
- try {
- zkClient.makePath(parentPath, new byte[0], CreateMode.PERSISTENT, true);
- } catch (KeeperException.NodeExistsException e) {
- // ignore
- }
- return zkClient.create(parentPath + "/" + PREFIX, Utils.toJSON(Collections.singletonMap("requestid", asyncId)), CreateMode.PERSISTENT_SEQUENTIAL, true);
++ stateManager.makePath(parentPath);
++ return stateManager.createData(parentPath + "/" + PREFIX, Utils.toJSON(Collections.singletonMap("requestid", asyncId)), CreateMode.PERSISTENT_SEQUENTIAL);
+ }
+
}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/1f25f3ec/solr/core/src/java/org/apache/solr/cloud/autoscaling/NodeAddedTrigger.java
----------------------------------------------------------------------
diff --cc solr/core/src/java/org/apache/solr/cloud/autoscaling/NodeAddedTrigger.java
index 438e383,bbfd098..120414d
--- a/solr/core/src/java/org/apache/solr/cloud/autoscaling/NodeAddedTrigger.java
+++ b/solr/core/src/java/org/apache/solr/cloud/autoscaling/NodeAddedTrigger.java
@@@ -63,10 -96,16 +63,16 @@@ public class NodeAddedTrigger extends T
@Override
public void init() {
- super.init();
+ List<Map<String, String>> o = (List<Map<String, String>>) properties.get("actions");
+ if (o != null && !o.isEmpty()) {
+ for (int i = 0; i < o.size(); i++) {
+ Map<String, String> map = o.get(i);
+ actions.get(i).init(map);
+ }
+ }
// pick up added nodes for which marker paths were created
try {
- List<String> added = zkClient.getChildren(ZkStateReader.SOLR_AUTOSCALING_NODE_ADDED_PATH, null, true);
+ List<String> added = stateManager.listData(ZkStateReader.SOLR_AUTOSCALING_NODE_ADDED_PATH);
added.forEach(n -> {
// don't add nodes that have since gone away
if (lastLiveNodes.contains(n)) {
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/1f25f3ec/solr/core/src/java/org/apache/solr/cloud/autoscaling/NodeLostTrigger.java
----------------------------------------------------------------------
diff --cc solr/core/src/java/org/apache/solr/cloud/autoscaling/NodeLostTrigger.java
index 2174d41,cbcbb3e..c51ecdd
--- a/solr/core/src/java/org/apache/solr/cloud/autoscaling/NodeLostTrigger.java
+++ b/solr/core/src/java/org/apache/solr/cloud/autoscaling/NodeLostTrigger.java
@@@ -24,17 -26,21 +26,20 @@@ 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.Objects;
+import java.util.NoSuchElementException;
import java.util.Set;
import java.util.concurrent.TimeUnit;
+ import java.util.concurrent.atomic.AtomicReference;
-import org.apache.lucene.util.IOUtils;
+import org.apache.solr.client.solrj.cloud.autoscaling.SolrCloudDataProvider;
import org.apache.solr.client.solrj.cloud.autoscaling.TriggerEventType;
+ import org.apache.solr.cloud.ZkController;
import org.apache.solr.common.SolrException;
import org.apache.solr.common.cloud.ZkStateReader;
-import org.apache.solr.core.CoreContainer;
+import org.apache.solr.core.SolrResourceLoader;
import org.apache.solr.util.TimeSource;
-import org.apache.zookeeper.KeeperException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/1f25f3ec/solr/core/src/java/org/apache/solr/cloud/autoscaling/OverseerTriggerThread.java
----------------------------------------------------------------------
diff --cc solr/core/src/java/org/apache/solr/cloud/autoscaling/OverseerTriggerThread.java
index dbe2ea4,367c450..8cf3707
--- a/solr/core/src/java/org/apache/solr/cloud/autoscaling/OverseerTriggerThread.java
+++ b/solr/core/src/java/org/apache/solr/cloud/autoscaling/OverseerTriggerThread.java
@@@ -32,12 -30,16 +32,15 @@@ import java.util.concurrent.locks.Reent
import org.apache.lucene.store.AlreadyClosedException;
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.DistribStateManager;
+import org.apache.solr.client.solrj.cloud.autoscaling.SolrCloudDataProvider;
import org.apache.solr.client.solrj.cloud.autoscaling.TriggerEventType;
-import org.apache.solr.cloud.ZkController;
-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.cloud.ZooKeeperException;
import org.apache.solr.common.util.IOUtils;
+ import org.apache.solr.common.util.Utils;
+ import org.apache.solr.core.CloudConfig;
-import org.apache.zookeeper.KeeperException;
+import org.apache.solr.core.SolrResourceLoader;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.slf4j.Logger;
@@@ -51,8 -55,14 +56,10 @@@ public class OverseerTriggerThread impl
private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
- private final ZkController zkController;
+ private final SolrCloudDataProvider dataProvider;
+ private final CloudConfig cloudConfig;
+
- private final ZkStateReader zkStateReader;
-
- private final SolrZkClient zkClient;
-
private final ScheduledTriggers scheduledTriggers;
private final AutoScaling.TriggerFactory triggerFactory;
@@@ -72,10 -82,13 +79,11 @@@
private AutoScalingConfig autoScalingConfig;
- public OverseerTriggerThread(SolrResourceLoader loader, SolrCloudDataProvider dataProvider) {
- public OverseerTriggerThread(ZkController zkController, CloudConfig cloudConfig) {
- this.zkController = zkController;
++ public OverseerTriggerThread(SolrResourceLoader loader, SolrCloudDataProvider dataProvider, CloudConfig cloudConfig) {
+ this.dataProvider = dataProvider;
+ this.cloudConfig = cloudConfig;
- zkStateReader = zkController.getZkStateReader();
- zkClient = zkController.getZkClient();
- scheduledTriggers = new ScheduledTriggers(zkController);
- triggerFactory = new AutoScaling.TriggerFactory(zkController.getCoreContainer(), zkController);
+ scheduledTriggers = new ScheduledTriggers(loader, dataProvider);
+ triggerFactory = new AutoScaling.TriggerFactoryImpl(loader, dataProvider);
}
@Override
@@@ -97,10 -110,36 +105,33 @@@
public void run() {
int lastZnodeVersion = znodeVersion;
+ // we automatically add a trigger for auto add replicas if it does not exists already
+ while (!isClosed) {
+ try {
- AutoScalingConfig autoScalingConfig = zkStateReader.getAutoScalingConfig();
++ AutoScalingConfig autoScalingConfig = dataProvider.getDistribStateManager().getAutoScalingConfig();
+ AutoScalingConfig withAutoAddReplicasTrigger = withAutoAddReplicasTrigger(autoScalingConfig);
+ if (withAutoAddReplicasTrigger.equals(autoScalingConfig)) break;
+ log.debug("Adding .autoAddReplicas trigger");
- zkClient.setData(SOLR_AUTOSCALING_CONF_PATH, Utils.toJSON(withAutoAddReplicasTrigger), withAutoAddReplicasTrigger.getZkVersion(), true);
++ dataProvider.getDistribStateManager().setData(SOLR_AUTOSCALING_CONF_PATH, Utils.toJSON(withAutoAddReplicasTrigger), withAutoAddReplicasTrigger.getZkVersion());
+ break;
- } catch (KeeperException.BadVersionException bve) {
++ } catch (BadVersionException bve) {
+ // somebody else has changed the configuration so we must retry
+ } catch (InterruptedException e) {
+ // Restore the interrupted status
+ Thread.currentThread().interrupt();
+ log.warn("Interrupted", e);
+ break;
- } catch (KeeperException e) {
++ } catch (IOException e) {
+ log.error("A ZK error has occurred", e);
+ }
+ }
+
+ if (isClosed || Thread.currentThread().isInterrupted()) return;
+
try {
refreshAutoScalingConf(new AutoScalingWatcher());
- } catch (KeeperException.ConnectionLossException | KeeperException.SessionExpiredException e) {
+ } catch (ConnectException e) {
log.warn("ZooKeeper watch triggered for autoscaling conf, but Solr cannot talk to ZK: [{}]", e.getMessage());
- } catch (KeeperException e) {
- log.error("A ZK error has occurred", e);
- throw new ZooKeeperException(SolrException.ErrorCode.SERVER_ERROR, "A ZK error has occurred", e);
} catch (InterruptedException e) {
// Restore the interrupted status
Thread.currentThread().interrupt();
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/1f25f3ec/solr/core/src/java/org/apache/solr/cloud/autoscaling/ScheduledTriggers.java
----------------------------------------------------------------------
diff --cc solr/core/src/java/org/apache/solr/cloud/autoscaling/ScheduledTriggers.java
index c1cea01,7a96552..d98ea45
--- a/solr/core/src/java/org/apache/solr/cloud/autoscaling/ScheduledTriggers.java
+++ b/solr/core/src/java/org/apache/solr/cloud/autoscaling/ScheduledTriggers.java
@@@ -39,20 -40,27 +40,27 @@@ import java.util.concurrent.atomic.Atom
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Collectors;
+ import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.lucene.store.AlreadyClosedException;
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.SolrCloudDataProvider;
import org.apache.solr.client.solrj.cloud.autoscaling.TriggerEventProcessorStage;
++import org.apache.solr.client.solrj.cloud.autoscaling.VersionedData;
+ import org.apache.solr.client.solrj.impl.CloudSolrClient;
+ import org.apache.solr.client.solrj.request.CollectionAdminRequest.RequestStatusResponse;
+ import org.apache.solr.client.solrj.response.RequestStatusState;
import org.apache.solr.cloud.ActionThrottle;
-import org.apache.solr.cloud.Overseer;
-import org.apache.solr.cloud.ZkController;
+import org.apache.solr.cloud.Stats;
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.util.ExecutorUtil;
import org.apache.solr.common.util.IOUtils;
+ import org.apache.solr.common.util.Utils;
-import org.apache.solr.core.CoreContainer;
+import org.apache.solr.core.SolrResourceLoader;
import org.apache.solr.util.DefaultSolrThreadFactory;
-import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.Op;
+ import org.apache.zookeeper.OpResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@@ -157,6 -159,12 +167,12 @@@ public class ScheduledTriggers implemen
scheduledTriggers.replace(newTrigger.getName(), scheduledTrigger);
}
newTrigger.setProcessor(event -> {
- if (coreContainer.isShutDown()) {
++ if (dataProvider.isClosed()) {
+ String msg = String.format(Locale.ROOT, "Ignoring autoscaling event %s because Solr has been shutdown.", event.toString());
+ log.warn(msg);
+ listeners.fireListeners(event.getSource(), event, TriggerEventProcessorStage.ABORTED, msg);
+ return false;
+ }
ScheduledTrigger scheduledSource = scheduledTriggers.get(event.getSource());
if (scheduledSource == null) {
String msg = String.format(Locale.ROOT, "Ignoring autoscaling event %s because the source trigger: %s doesn't exist.", event.toString(), event.getSource());
@@@ -190,7 -199,12 +207,12 @@@
// let the action executor thread wait instead of the trigger thread so we use the throttle here
actionThrottle.minimumWaitBetweenActions();
actionThrottle.markAttemptingAction();
+
+ // in future, we could wait for pending tasks in a different thread and re-enqueue
+ // this event so that we continue processing other events and not block this action executor
+ waitForPendingTasks(newTrigger, actions);
+
- ActionContext actionContext = new ActionContext(coreContainer, newTrigger, new HashMap<>());
+ ActionContext actionContext = new ActionContext(dataProvider, newTrigger, new HashMap<>());
for (TriggerAction action : actions) {
List<String> beforeActions = (List<String>)actionContext.getProperties().computeIfAbsent(TriggerEventProcessorStage.BEFORE_ACTION.toString(), k -> new ArrayList<String>());
beforeActions.add(action.getName());
@@@ -236,6 -250,69 +258,65 @@@
scheduledTrigger.scheduledFuture = scheduledThreadPoolExecutor.scheduleWithFixedDelay(scheduledTrigger, 0, DEFAULT_SCHEDULED_TRIGGER_DELAY_SECONDS, TimeUnit.SECONDS);
}
+ private void waitForPendingTasks(AutoScaling.Trigger newTrigger, List<TriggerAction> actions) throws AlreadyClosedException {
- try (CloudSolrClient cloudSolrClient = new CloudSolrClient.Builder()
- .withZkHost(coreContainer.getZkController().getZkServerAddress())
- .withHttpClient(coreContainer.getUpdateShardHandler().getHttpClient())
- .build()) {
-
- SolrZkClient zkClient = coreContainer.getZkController().getZkClient();
++ DistribStateManager stateManager = dataProvider.getDistribStateManager();
++ try {
+
+ for (TriggerAction action : actions) {
+ if (action instanceof ExecutePlanAction) {
+ String parentPath = ZkStateReader.SOLR_AUTOSCALING_TRIGGER_STATE_PATH + "/" + newTrigger.getName() + "/" + action.getName();
- if (!zkClient.exists(parentPath, true)) {
++ if (!stateManager.hasData(parentPath)) {
+ break;
+ }
- List<String> children = zkClient.getChildren(parentPath, null, true);
++ List<String> children = stateManager.listData(parentPath);
+ if (children != null) {
+ for (String child : children) {
+ String path = parentPath + '/' + child;
- byte[] data = zkClient.getData(path, null, null, true);
++ VersionedData data = stateManager.getData(path, null);
+ if (data != null) {
- Map map = (Map) Utils.fromJSON(data);
++ Map map = (Map) Utils.fromJSON(data.getData());
+ String requestid = (String) map.get("requestid");
+ try {
+ log.debug("Found pending task with requestid={}", requestid);
- RequestStatusResponse statusResponse = waitForTaskToFinish(cloudSolrClient, requestid,
++ RequestStatusResponse statusResponse = waitForTaskToFinish(dataProvider, requestid,
+ ExecutePlanAction.DEFAULT_TASK_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+ if (statusResponse != null) {
+ RequestStatusState state = statusResponse.getRequestStatus();
+ if (state == RequestStatusState.COMPLETED || state == RequestStatusState.FAILED || state == RequestStatusState.NOT_FOUND) {
- zkClient.delete(path, -1, true);
++ stateManager.removeData(path, -1);
+ }
+ }
+ } catch (Exception e) {
- if (coreContainer.isShutDown()) {
++ if (dataProvider.isClosed()) {
+ throw e; // propagate the abort to the caller
+ }
+ Throwable rootCause = ExceptionUtils.getRootCause(e);
+ if (rootCause instanceof IllegalStateException && rootCause.getMessage().contains("Connection pool shut down")) {
+ throw e;
+ }
+ if (rootCause instanceof TimeoutException && rootCause.getMessage().contains("Could not connect to ZooKeeper")) {
+ throw e;
+ }
+ log.error("Unexpected exception while waiting for pending task with requestid: " + requestid + " to finish", e);
+ }
+ }
+ }
+ }
+ }
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, "Thread interrupted", e);
+ } catch (Exception e) {
- if (coreContainer.isShutDown()) {
++ if (dataProvider.isClosed()) {
+ throw new AlreadyClosedException("The Solr instance has been shutdown");
+ }
+ // we catch but don't rethrow because a failure to wait for pending tasks
+ // should not keep the actions from executing
+ log.error("Unexpected exception while waiting for pending tasks to finish", e);
+ }
+ }
+
/**
* Removes and stops the trigger with the given name. Also cleans up any leftover
* state / events in ZK.
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/1f25f3ec/solr/core/src/test/org/apache/solr/cloud/autoscaling/ComputePlanActionTest.java
----------------------------------------------------------------------
diff --cc solr/core/src/test/org/apache/solr/cloud/autoscaling/ComputePlanActionTest.java
index 93a13b0,d9b88b5..94b4094
--- a/solr/core/src/test/org/apache/solr/cloud/autoscaling/ComputePlanActionTest.java
+++ b/solr/core/src/test/org/apache/solr/cloud/autoscaling/ComputePlanActionTest.java
@@@ -63,7 -61,7 +61,7 @@@ public class ComputePlanActionTest exte
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<>();
++ private static final AtomicReference<TriggerEvent> eventRef = new AtomicReference<>();o
@BeforeClass
public static void setupCluster() throws Exception {
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/1f25f3ec/solr/core/src/test/org/apache/solr/cloud/autoscaling/ExecutePlanActionTest.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/1f25f3ec/solr/core/src/test/org/apache/solr/cloud/autoscaling/NodeLostTriggerTest.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/1f25f3ec/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/ClusterDataProvider.java
----------------------------------------------------------------------
diff --cc solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/ClusterDataProvider.java
index 1ffb00b,58972af..2753c3d
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/ClusterDataProvider.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/ClusterDataProvider.java
@@@ -23,15 -23,8 +23,14 @@@ import java.util.Collection
import java.util.List;
import java.util.Map;
+import org.apache.solr.common.cloud.ClusterState;
- import org.apache.zookeeper.Watcher;
+
+/**
+ * This interface abstracts the access to cluster data.
+ */
public interface ClusterDataProvider extends Closeable {
- /**Get the value of each tag for a given node
+ /**
+ * Get the value of each tag for a given node
*
* @param node node name
* @param tags tag names
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/1f25f3ec/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/PolicyHelper.java
----------------------------------------------------------------------
diff --cc solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/PolicyHelper.java
index 5dd8a33,bc2a5d7..3ed12a6
--- 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
@@@ -18,9 -18,8 +18,8 @@@
package org.apache.solr.client.solrj.cloud.autoscaling;
+import java.io.IOException;
import java.util.ArrayList;
--import java.util.Collection;
import java.util.EnumMap;
import java.util.List;
import java.util.Map;
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/1f25f3ec/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/SolrCloudDataProvider.java
----------------------------------------------------------------------
diff --cc solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/SolrCloudDataProvider.java
index 995de9f,0000000..2c30ecc
mode 100644,000000..100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/SolrCloudDataProvider.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/cloud/autoscaling/SolrCloudDataProvider.java
@@@ -1,63 -1,0 +1,61 @@@
+/*
+ * 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.client.solrj.cloud.autoscaling;
+
+import java.io.Closeable;
+import java.io.IOException;
- import java.util.List;
+import java.util.Map;
- import java.util.NoSuchElementException;
+
+import org.apache.solr.client.solrj.SolrRequest;
+import org.apache.solr.client.solrj.SolrResponse;
+import org.apache.solr.client.solrj.cloud.DistributedQueue;
- import org.apache.zookeeper.CreateMode;
- import org.apache.zookeeper.Op;
- import org.apache.zookeeper.OpResult;
- import org.apache.zookeeper.Watcher;
+
+/**
+ * This interface abstracts the access to a SolrCloud cluster, including interactions with Zookeeper, Solr
+ * and generic HTTP calls.
+ */
+public interface SolrCloudDataProvider extends Closeable {
+
+ ClusterDataProvider getClusterDataProvider();
+
+ DistribStateManager getDistribStateManager();
+
+ interface DistributedQueueFactory {
+ DistributedQueue makeQueue(String path) throws IOException;
+ void removeQueue(String path) throws IOException;
+ }
+
+ DistributedQueueFactory getDistributedQueueFactory();
+
+ // Solr-like methods
+
+ SolrResponse request(SolrRequest req) throws IOException;
+
+ byte[] httpRequest(String url, SolrRequest.METHOD method, Map<String, String> headers, String payload, int timeout, boolean followRedirects) throws IOException;
+
+ // distributed queue implementation
+
+ @Override
+ default void close() {
+
+ }
++
++ default boolean isClosed() {
++ return false;
++ }
+}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/1f25f3ec/solr/solrj/src/java/org/apache/solr/client/solrj/impl/SolrClientCloudDataProvider.java
----------------------------------------------------------------------
diff --cc solr/solrj/src/java/org/apache/solr/client/solrj/impl/SolrClientCloudDataProvider.java
index 246ef88,0000000..59d4d97
mode 100644,000000..100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/SolrClientCloudDataProvider.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/SolrClientCloudDataProvider.java
@@@ -1,156 -1,0 +1,168 @@@
+/*
+ * 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.client.solrj.impl;
+
+
+import java.io.IOException;
+import java.lang.invoke.MethodHandles;
+import java.util.List;
+import java.util.Map;
+import java.util.NoSuchElementException;
+
+import org.apache.http.HttpEntity;
+import org.apache.http.HttpResponse;
+import org.apache.http.client.HttpClient;
+import org.apache.http.client.config.RequestConfig;
+import org.apache.http.client.methods.HttpDelete;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.client.methods.HttpPut;
+import org.apache.http.client.methods.HttpRequestBase;
+import org.apache.http.client.protocol.HttpClientContext;
+import org.apache.http.entity.StringEntity;
+import org.apache.http.util.EntityUtils;
+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.autoscaling.ClusterDataProvider;
+import org.apache.solr.client.solrj.cloud.autoscaling.DistribStateManager;
+import org.apache.solr.client.solrj.cloud.autoscaling.SolrCloudDataProvider;
+import org.apache.solr.common.cloud.SolrZkClient;
+import org.apache.solr.common.cloud.ZkStateReader;
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.Op;
+import org.apache.zookeeper.OpResult;
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.data.Stat;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Class that implements {@link SolrCloudDataProvider} using a SolrClient
+ */
+public class SolrClientCloudDataProvider implements SolrCloudDataProvider {
+ private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ private final CloudSolrClient solrClient;
+ private final ZkDistribStateManager stateManager;
+ private final DistributedQueueFactory queueFactory;
+ private final ZkStateReader zkStateReader;
+ private final SolrZkClient zkClient;
++ private volatile boolean isClosed;
+
+ public SolrClientCloudDataProvider(DistributedQueueFactory queueFactory, CloudSolrClient solrClient) {
+ this.queueFactory = queueFactory;
+ this.solrClient = solrClient;
+ this.zkStateReader = solrClient.getZkStateReader();
+ this.zkClient = zkStateReader.getZkClient();
+ this.stateManager = new ZkDistribStateManager(zkClient);
++ this.isClosed = false;
++ }
++
++ @Override
++ public void close() {
++ isClosed = true;
++ }
++
++ @Override
++ public boolean isClosed() {
++ return isClosed;
+ }
+
+ @Override
+ public ClusterDataProvider getClusterDataProvider() {
+ return new SolrClientClusterDataProvider(solrClient);
+ }
+
+ @Override
+ public DistribStateManager getDistribStateManager() {
+ return stateManager;
+ }
+
+ @Override
+ public SolrResponse request(SolrRequest req) throws IOException {
+ try {
+ return req.process(solrClient);
+ } catch (SolrServerException e) {
+ throw new IOException(e);
+ }
+ }
+
+ private static final byte[] EMPTY = new byte[0];
+
+ @Override
+ public byte[] httpRequest(String url, SolrRequest.METHOD method, Map<String, String> headers, String payload, int timeout, boolean followRedirects) throws IOException {
+ HttpClient client = solrClient.getHttpClient();
+ final HttpRequestBase req;
+ HttpEntity entity = null;
+ if (payload != null) {
+ entity = new StringEntity(payload, "UTF-8");
+ }
+ switch (method) {
+ case GET:
+ req = new HttpGet(url);
+ break;
+ case POST:
+ req = new HttpPost(url);
+ if (entity != null) {
+ ((HttpPost)req).setEntity(entity);
+ }
+ break;
+ case PUT:
+ req = new HttpPut(url);
+ if (entity != null) {
+ ((HttpPut)req).setEntity(entity);
+ }
+ break;
+ case DELETE:
+ req = new HttpDelete(url);
+ break;
+ default:
+ throw new IOException("Unsupported method " + method);
+ }
+ if (headers != null) {
+ headers.forEach((k, v) -> req.addHeader(k, v));
+ }
+ RequestConfig.Builder requestConfigBuilder = HttpClientUtil.createDefaultRequestConfigBuilder();
+ if (timeout > 0) {
+ requestConfigBuilder.setSocketTimeout(timeout);
+ requestConfigBuilder.setConnectTimeout(timeout);
+ }
+ requestConfigBuilder.setRedirectsEnabled(followRedirects);
+ req.setConfig(requestConfigBuilder.build());
+ HttpClientContext httpClientRequestContext = HttpClientUtil.createNewHttpClientRequestContext();
+ HttpResponse rsp = client.execute(req, httpClientRequestContext);
+ int statusCode = rsp.getStatusLine().getStatusCode();
+ if (statusCode != 200) {
+ throw new IOException("Error sending request to " + url + ", HTTP response: " + rsp.toString());
+ }
+ HttpEntity responseEntity = rsp.getEntity();
+ if (responseEntity != null && responseEntity.getContent() != null) {
+ return EntityUtils.toByteArray(responseEntity);
+ } else {
+ return EMPTY;
+ }
+ }
+
+ @Override
+ public DistributedQueueFactory getDistributedQueueFactory() {
+ return queueFactory;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/1f25f3ec/solr/solrj/src/java/org/apache/solr/client/solrj/impl/SolrClientClusterDataProvider.java
----------------------------------------------------------------------
diff --cc solr/solrj/src/java/org/apache/solr/client/solrj/impl/SolrClientClusterDataProvider.java
index fd0722a,0000000..2cd24ae
mode 100644,000000..100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/SolrClientClusterDataProvider.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/SolrClientClusterDataProvider.java
@@@ -1,249 -1,0 +1,249 @@@
+/*
+ * 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.client.solrj.impl;
+
+import java.io.IOException;
+import java.lang.invoke.MethodHandles;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.solr.client.solrj.SolrRequest;
+import org.apache.solr.client.solrj.SolrServerException;
+import org.apache.solr.client.solrj.cloud.autoscaling.ClusterDataProvider;
+import org.apache.solr.client.solrj.cloud.autoscaling.ReplicaInfo;
+import org.apache.solr.client.solrj.request.GenericSolrRequest;
+import org.apache.solr.client.solrj.response.SimpleSolrResponse;
+import org.apache.solr.common.MapWriter;
+import org.apache.solr.common.SolrException;
+import org.apache.solr.common.cloud.ClusterState;
+import org.apache.solr.common.cloud.DocCollection;
+import org.apache.solr.common.cloud.ZkStateReader;
+import org.apache.solr.common.cloud.rule.ImplicitSnitch;
+import org.apache.solr.common.cloud.rule.RemoteCallback;
+import org.apache.solr.common.cloud.rule.SnitchContext;
+import org.apache.solr.common.params.CommonParams;
+import org.apache.solr.common.params.ModifiableSolrParams;
+import org.apache.solr.common.params.SolrParams;
+import org.apache.solr.common.util.NamedList;
+import org.apache.solr.common.util.StrUtils;
+import org.apache.solr.common.util.Utils;
+import org.apache.zookeeper.KeeperException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.apache.solr.client.solrj.cloud.autoscaling.Clause.METRICS_PREFIX;
+
+/**
+ *
+ */
+public class SolrClientClusterDataProvider implements ClusterDataProvider, MapWriter {
+ private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ private final CloudSolrClient solrClient;
+ private final ZkStateReader zkStateReader;
+ private final Map<String, Map<String, Map<String, List<ReplicaInfo>>>> data = new HashMap<>();
+ private Map<String, Object> snitchSession = new HashMap<>();
+ private Map<String, Map> nodeVsTags = new HashMap<>();
+
+ public SolrClientClusterDataProvider(CloudSolrClient solrClient) {
+ this.solrClient = solrClient;
+ this.zkStateReader = solrClient.getZkStateReader();
+ ClusterState clusterState = zkStateReader.getClusterState();
+ Map<String, ClusterState.CollectionRef> all = clusterState.getCollectionStates();
+ all.forEach((collName, ref) -> {
+ DocCollection coll = ref.get();
+ if (coll == null) return;
+ coll.forEachReplica((shard, replica) -> {
+ Map<String, Map<String, List<ReplicaInfo>>> nodeData = data.computeIfAbsent(replica.getNodeName(), k -> new HashMap<>());
+ Map<String, List<ReplicaInfo>> collData = nodeData.computeIfAbsent(collName, k -> new HashMap<>());
+ List<ReplicaInfo> replicas = collData.computeIfAbsent(shard, k -> new ArrayList<>());
- replicas.add(new ReplicaInfo(replica.getName(), collName, shard, replica.getType(), replica.getProperties()));
++ replicas.add(new ReplicaInfo(replica.getName(), collName, shard, replica.getType(), new HashMap<>()));
+ });
+ });
+ }
+
+ public Collection<String> getLiveNodes() {
+ return zkStateReader.getClusterState().getLiveNodes();
+ }
+
+ @Override
+ public String getPolicyNameByCollection(String coll) {
+ ClusterState.CollectionRef state = solrClient.getClusterStateProvider().getState(coll);
+ return state == null || state.get() == null ? null : (String) state.get().getProperties().get("policy");
+ }
+
+ @Override
+ public Map<String, Object> getNodeValues(String node, Collection<String> tags) {
+ AutoScalingSnitch snitch = new AutoScalingSnitch();
+ ClientSnitchCtx ctx = new ClientSnitchCtx(null, node, snitchSession, solrClient);
+ snitch.getTags(node, new HashSet<>(tags), ctx);
+ nodeVsTags.put(node, ctx.getTags());
+ return ctx.getTags();
+ }
+
+ @Override
+ public Map<String, Map<String, List<ReplicaInfo>>> getReplicaInfo(String node, Collection<String> keys) {
+ return data.computeIfAbsent(node, s -> Collections.emptyMap());//todo fill other details
+ }
+
+ @Override
+ public Map<String, Object> getClusterProperties() {
+ return zkStateReader.getClusterProperties();
+ }
+
+ @Override
+ public ClusterState getClusterState() {
+ return zkStateReader.getClusterState();
+ }
+
+ public void writeMap(EntryWriter ew) throws IOException {
+ ew.put("clusterState", zkStateReader.getClusterState());
+ ew.put("replicaInfo", Utils.getDeepCopy(data, 5));
+ ew.put("nodeValues", nodeVsTags);
+ }
+
+ static class ClientSnitchCtx
+ extends SnitchContext {
+ private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ ZkClientClusterStateProvider zkClientClusterStateProvider;
+ CloudSolrClient solrClient;
+
+ public ClientSnitchCtx(SnitchInfo perSnitch,
+ String node, Map<String, Object> session,
+ CloudSolrClient solrClient) {
+ super(perSnitch, node, session);
+ this.solrClient = solrClient;
+ this.zkClientClusterStateProvider = (ZkClientClusterStateProvider) solrClient.getClusterStateProvider();
+ }
+
+
+ @Override
+ public Map getZkJson(String path) throws KeeperException, InterruptedException {
+ return Utils.getJson(zkClientClusterStateProvider.getZkStateReader().getZkClient(), path, true);
+ }
+
+ public void invokeRemote(String node, ModifiableSolrParams params, String klas, RemoteCallback callback) {
+
+ }
+
+ public SimpleSolrResponse invoke(String solrNode, String path, SolrParams params)
+ throws IOException, SolrServerException {
+ String url = zkClientClusterStateProvider.getZkStateReader().getBaseUrlForNodeName(solrNode);
+
+ GenericSolrRequest request = new GenericSolrRequest(SolrRequest.METHOD.GET, path, params);
+ try (HttpSolrClient client = new HttpSolrClient.Builder()
+ .withHttpClient(solrClient.getHttpClient())
+ .withBaseSolrUrl(url)
+ .withResponseParser(new BinaryResponseParser())
+ .build()) {
+ NamedList<Object> rsp = client.request(request);
+ request.response.nl = rsp;
+ return request.response;
+ }
+ }
+
+ }
+
+ //uses metrics API to get node information
+ static class AutoScalingSnitch extends ImplicitSnitch {
+ @Override
+ protected void getRemoteInfo(String solrNode, Set<String> requestedTags, SnitchContext ctx) {
+ ClientSnitchCtx snitchContext = (ClientSnitchCtx) ctx;
+ Map<String, String> metricsKeyVsTag = new HashMap<>();
+ for (String tag : requestedTags) {
+ if (tag.startsWith(SYSPROP)) {
+ metricsKeyVsTag.put("solr.jvm:system.properties:" + tag.substring(SYSPROP.length()), tag);
+ } else if (tag.startsWith(METRICS_PREFIX)) {
+ metricsKeyVsTag.put(tag.substring(METRICS_PREFIX.length()), tag);
+ }
+ }
+ if (!metricsKeyVsTag.isEmpty()) {
+ ModifiableSolrParams params = new ModifiableSolrParams();
+ params.add("key", metricsKeyVsTag.keySet().toArray(new String[metricsKeyVsTag.size()]));
+ try {
+ SimpleSolrResponse rsp = snitchContext.invoke(solrNode, CommonParams.METRICS_PATH, params);
+ metricsKeyVsTag.forEach((key, tag) -> {
+ Object v = Utils.getObjectByPath(rsp.nl, true, Arrays.asList("metrics", key));
+ if (v != null) snitchContext.getTags().put(tag, v);
+ });
+ } catch (Exception e) {
+ log.warn("could not get tags from node " + solrNode, e);
+ }
+ }
+
+ Set<String> groups = new HashSet<>();
+ List<String> prefixes = new ArrayList<>();
+ if (requestedTags.contains(DISK)) {
+ groups.add("solr.node");
+ prefixes.add("CONTAINER.fs.usableSpace");
+ }
+ if (requestedTags.contains(CORES)) {
+ groups.add("solr.core");
+ prefixes.add("CORE.coreName");
+ }
+ if (requestedTags.contains(SYSLOADAVG)) {
+ groups.add("solr.jvm");
+ prefixes.add("os.systemLoadAverage");
+ }
+ if (requestedTags.contains(HEAPUSAGE)) {
+ groups.add("solr.jvm");
+ prefixes.add("memory.heap.usage");
+ }
+ if (groups.isEmpty() || prefixes.isEmpty()) return;
+
+ ModifiableSolrParams params = new ModifiableSolrParams();
+ params.add("group", StrUtils.join(groups, ','));
+ params.add("prefix", StrUtils.join(prefixes, ','));
+
+ try {
+ SimpleSolrResponse rsp = snitchContext.invoke(solrNode, CommonParams.METRICS_PATH, params);
+ Map m = rsp.nl.asMap(4);
+ if (requestedTags.contains(DISK)) {
+ Number n = (Number) Utils.getObjectByPath(m, true, "metrics/solr.node/CONTAINER.fs.usableSpace");
+ if (n != null) ctx.getTags().put(DISK, n.doubleValue() / 1024.0d / 1024.0d / 1024.0d);
+ }
+ if (requestedTags.contains(CORES)) {
+ int count = 0;
+ Map cores = (Map) m.get("metrics");
+ for (Object o : cores.keySet()) {
+ if (o.toString().startsWith("solr.core.")) count++;
+ }
+ ctx.getTags().put(CORES, count);
+ }
+ if (requestedTags.contains(SYSLOADAVG)) {
+ Number n = (Number) Utils.getObjectByPath(m, true, "metrics/solr.jvm/os.systemLoadAverage");
+ if (n != null) ctx.getTags().put(SYSLOADAVG, n.doubleValue() * 100.0d);
+ }
+ if (requestedTags.contains(HEAPUSAGE)) {
+ Number n = (Number) Utils.getObjectByPath(m, true, "metrics/solr.jvm/memory.heap.usage");
+ if (n != null) ctx.getTags().put(HEAPUSAGE, n.doubleValue() * 100.0d);
+ }
+ } catch (Exception e) {
+ throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, "", e);
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/lucene-solr/blob/1f25f3ec/solr/solrj/src/test/org/apache/solr/client/solrj/cloud/autoscaling/TestPolicy.java
----------------------------------------------------------------------
diff --cc solr/solrj/src/test/org/apache/solr/client/solrj/cloud/autoscaling/TestPolicy.java
index 5516902,96e2257..22afc51
--- a/solr/solrj/src/test/org/apache/solr/client/solrj/cloud/autoscaling/TestPolicy.java
+++ b/solr/solrj/src/test/org/apache/solr/client/solrj/cloud/autoscaling/TestPolicy.java
@@@ -1226,42 -1172,10 +1200,36 @@@ public class TestPolicy extends SolrTes
}
@Override
- public Collection<String> getNodes() {
+ public Collection<String> getLiveNodes() {
return Arrays.asList( "127.0.0.1:50097_solr", "127.0.0.1:50096_solr");
}
-
- @Override
- public ClusterState getClusterState() {
- throw new UnsupportedOperationException("getClusterState");
- }
-
};
+ SolrCloudDataProvider dataProvider = new SolrCloudDataProvider() {
+ @Override
+ public ClusterDataProvider getClusterDataProvider() {
+ return provider;
+ }
+
+ @Override
+ public DistribStateManager getDistribStateManager() {
+ return null;
+ }
+
+ @Override
+ public DistributedQueueFactory getDistributedQueueFactory() {
+ return null;
+ }
+
+ @Override
+ public SolrResponse request(SolrRequest req) throws IOException {
+ return null;
+ }
+
+ @Override
+ public byte[] httpRequest(String url, SolrRequest.METHOD method, Map<String, String> headers, String payload, int timeout, boolean followRedirects) throws IOException {
+ return new byte[0];
+ }
+ };
List<ReplicaPosition> locations = PolicyHelper.getReplicaLocations(
"newColl", new AutoScalingConfig((Map<String, Object>)Utils.fromJSONString(autoScaleJson)),
dataProvider, Collections.singletonMap("newColl", "c1"), Arrays.asList("shard1", "shard2"), 1, 0, 0, null);
@@@ -1315,42 -1229,10 +1283,36 @@@
}
@Override
- public Collection<String> getNodes() {
+ public Collection<String> getLiveNodes() {
return Arrays.asList("node1", "node2", "node3", "node4");
}
-
- @Override
- public ClusterState getClusterState() {
- throw new UnsupportedOperationException("getClusterState");
- }
-
};
+ SolrCloudDataProvider dataProvider = new SolrCloudDataProvider() {
+ @Override
+ public ClusterDataProvider getClusterDataProvider() {
+ return provider;
+ }
+
+ @Override
+ public DistribStateManager getDistribStateManager() {
+ return null;
+ }
+
+ @Override
+ public DistributedQueueFactory getDistributedQueueFactory() {
+ return null;
+ }
+
+ @Override
+ public SolrResponse request(SolrRequest req) throws IOException {
+ return null;
+ }
+
+ @Override
+ public byte[] httpRequest(String url, SolrRequest.METHOD method, Map<String, String> headers, String payload, int timeout, boolean followRedirects) throws IOException {
+ return new byte[0];
+ }
+ };
List<ReplicaPosition> locations = PolicyHelper.getReplicaLocations(
"newColl", new AutoScalingConfig((Map<String, Object>) Utils.fromJSONString(autoScaleJson)),
dataProvider, Collections.singletonMap("newColl", "policy1"), Arrays.asList("shard1", "shard2"), 3,0,0, null);