You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@solr.apache.org by no...@apache.org on 2022/09/13 12:23:42 UTC
[solr] branch main updated: SOLR-15715: Dedicated query coordinator nodes in the solr cluster (#996)
This is an automated email from the ASF dual-hosted git repository.
noble pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/solr.git
The following commit(s) were added to refs/heads/main by this push:
new d029eb14aa8 SOLR-15715: Dedicated query coordinator nodes in the solr cluster (#996)
d029eb14aa8 is described below
commit d029eb14aa8fbb592938f67bd1397bd8fe805168
Author: Noble Paul <no...@users.noreply.github.com>
AuthorDate: Tue Sep 13 22:23:36 2022 +1000
SOLR-15715: Dedicated query coordinator nodes in the solr cluster (#996)
---
solr/CHANGES.txt | 2 +
.../apache/solr/api/CoordinatorV2HttpSolrCall.java | 58 ++++++
.../apache/solr/cloud/api/collections/Assign.java | 1 +
.../impl/PlacementPluginAssignStrategy.java | 52 +++++
.../src/java/org/apache/solr/core/NodeRoles.java | 12 ++
.../solr/handler/component/HttpShardHandler.java | 2 +-
.../solr/request/DelegatedSolrQueryRequest.java | 167 +++++++++++++++
.../solr/servlet/CoordinatorHttpSolrCall.java | 223 +++++++++++++++++++++
.../apache/solr/servlet/SolrDispatchFilter.java | 38 +++-
.../apache/solr/search/TestCoordinatorRole.java | 85 ++++++++
.../modules/deployment-guide/pages/node-roles.adoc | 25 +++
.../org/apache/solr/client/solrj/SolrRequest.java | 11 +
.../solr/client/solrj/impl/CloudSolrClient.java | 15 ++
13 files changed, 683 insertions(+), 8 deletions(-)
diff --git a/solr/CHANGES.txt b/solr/CHANGES.txt
index 9521d4f8fea..ca6c5a18c03 100644
--- a/solr/CHANGES.txt
+++ b/solr/CHANGES.txt
@@ -49,6 +49,8 @@ New Features
* SOLR-15007: Add ability to roll up core level metrics to be node level metrics for a RequestHandler via configuration. (Justin Sweeney, David Smiley)
+* SOLR-15715: Dedicated query coordinator nodes in the solr cluster (noble, Hitesh Khamesra, Ishan Chattopadhyaya)
+
Improvements
---------------------
* SOLR-15986: CommitUpdateCommand and SplitIndexCommand can write user commit metadata. (Bruno Roustant)
diff --git a/solr/core/src/java/org/apache/solr/api/CoordinatorV2HttpSolrCall.java b/solr/core/src/java/org/apache/solr/api/CoordinatorV2HttpSolrCall.java
new file mode 100644
index 00000000000..f29d2407022
--- /dev/null
+++ b/solr/core/src/java/org/apache/solr/api/CoordinatorV2HttpSolrCall.java
@@ -0,0 +1,58 @@
+/*
+ * 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.api;
+
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+import org.apache.solr.core.CoreContainer;
+import org.apache.solr.core.SolrCore;
+import org.apache.solr.servlet.CoordinatorHttpSolrCall;
+import org.apache.solr.servlet.SolrDispatchFilter;
+
+public class CoordinatorV2HttpSolrCall extends V2HttpCall {
+ private String collectionName;
+ CoordinatorHttpSolrCall.Factory factory;
+
+ public CoordinatorV2HttpSolrCall(
+ CoordinatorHttpSolrCall.Factory factory,
+ SolrDispatchFilter solrDispatchFilter,
+ CoreContainer cc,
+ HttpServletRequest request,
+ HttpServletResponse response,
+ boolean retry) {
+ super(solrDispatchFilter, cc, request, response, retry);
+ this.factory = factory;
+ }
+
+ @Override
+ protected SolrCore getCoreByCollection(String collectionName, boolean isPreferLeader) {
+ this.collectionName = collectionName;
+ SolrCore core = super.getCoreByCollection(collectionName, isPreferLeader);
+ if (core != null) return core;
+ if (!path.endsWith("/select")) return null;
+ return CoordinatorHttpSolrCall.getCore(factory, this, collectionName, isPreferLeader);
+ }
+
+ @Override
+ protected void init() throws Exception {
+ super.init();
+ if (action == SolrDispatchFilter.Action.PROCESS && core != null) {
+ solrReq = CoordinatorHttpSolrCall.wrappedReq(solrReq, collectionName, this);
+ }
+ }
+}
diff --git a/solr/core/src/java/org/apache/solr/cloud/api/collections/Assign.java b/solr/core/src/java/org/apache/solr/cloud/api/collections/Assign.java
index 27300b1143d..e8c01a2829b 100644
--- a/solr/core/src/java/org/apache/solr/cloud/api/collections/Assign.java
+++ b/solr/core/src/java/org/apache/solr/cloud/api/collections/Assign.java
@@ -62,6 +62,7 @@ import org.slf4j.LoggerFactory;
public class Assign {
private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+ public static final String SYSTEM_COLL_PREFIX = ".sys.";
public static String getCounterNodePath(String collection) {
return ZkStateReader.COLLECTIONS_ZKNODE + "/" + collection + "/counter";
diff --git a/solr/core/src/java/org/apache/solr/cluster/placement/impl/PlacementPluginAssignStrategy.java b/solr/core/src/java/org/apache/solr/cluster/placement/impl/PlacementPluginAssignStrategy.java
index df310cdc355..4c2bec84ba5 100644
--- a/solr/core/src/java/org/apache/solr/cluster/placement/impl/PlacementPluginAssignStrategy.java
+++ b/solr/core/src/java/org/apache/solr/cluster/placement/impl/PlacementPluginAssignStrategy.java
@@ -19,10 +19,12 @@ package org.apache.solr.cluster.placement.impl;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.HashSet;
import java.util.List;
import java.util.Set;
import org.apache.solr.client.solrj.cloud.SolrCloudManager;
import org.apache.solr.cloud.api.collections.Assign;
+import org.apache.solr.cluster.Node;
import org.apache.solr.cluster.placement.DeleteCollectionRequest;
import org.apache.solr.cluster.placement.DeleteReplicasRequest;
import org.apache.solr.cluster.placement.PlacementContext;
@@ -48,6 +50,14 @@ public class PlacementPluginAssignStrategy implements Assign.AssignStrategy {
throws Assign.AssignmentException, IOException, InterruptedException {
PlacementContext placementContext = new SimplePlacementContextImpl(solrCloudManager);
+ if (assignRequests.size() == 1
+ && assignRequests.get(0).collectionName.startsWith(Assign.SYSTEM_COLL_PREFIX)) {
+ // this is a system collection
+ Assign.AssignRequest assignRequest = assignRequests.get(0);
+ if (assignRequest.nodes != null && !assignRequest.nodes.isEmpty()) {
+ return computeSystemCollectionPositions(placementContext, assignRequest);
+ }
+ }
List<PlacementRequest> placementRequests = new ArrayList<>(assignRequests.size());
for (Assign.AssignRequest assignRequest : assignRequests) {
@@ -106,4 +116,46 @@ public class PlacementPluginAssignStrategy implements Assign.AssignStrategy {
throw new Assign.AssignmentException(pe);
}
}
+
+ /** Very minimal placement logic for System collections */
+ private static List<ReplicaPosition> computeSystemCollectionPositions(
+ PlacementContext placementContext, Assign.AssignRequest assignRequest) throws IOException {
+ Set<Node> nodes = SimpleClusterAbstractionsImpl.NodeImpl.getNodes(assignRequest.nodes);
+ for (Node n : nodes) {
+ if (!placementContext.getCluster().getLiveNodes().contains(n)) {
+ throw new Assign.AssignmentException(
+ "Bad assign request: specified node is not a live node ("
+ + n.getName()
+ + ") for collection "
+ + assignRequest.collectionName);
+ }
+ }
+ PlacementRequestImpl request =
+ new PlacementRequestImpl(
+ placementContext.getCluster().getCollection(assignRequest.collectionName),
+ new HashSet<>(assignRequest.shardNames),
+ nodes,
+ assignRequest.numNrtReplicas,
+ assignRequest.numTlogReplicas,
+ assignRequest.numPullReplicas);
+ final List<ReplicaPosition> replicaPositions = new ArrayList<>();
+ ArrayList<Node> nodeList = new ArrayList<>(request.getTargetNodes());
+ for (String shard : request.getShardNames()) {
+ int replicaNumOfShard = 0;
+ for (org.apache.solr.cluster.Replica.ReplicaType replicaType :
+ org.apache.solr.cluster.Replica.ReplicaType.values()) {
+ for (int i = 0; i < request.getCountReplicasToCreate(replicaType); i++) {
+ Node assignedNode = nodeList.get(replicaNumOfShard++ % nodeList.size());
+ replicaPositions.add(
+ new ReplicaPosition(
+ request.getCollection().getName(),
+ shard,
+ i,
+ SimpleClusterAbstractionsImpl.ReplicaImpl.toCloudReplicaType(replicaType),
+ assignedNode.getName()));
+ }
+ }
+ }
+ return replicaPositions;
+ }
}
diff --git a/solr/core/src/java/org/apache/solr/core/NodeRoles.java b/solr/core/src/java/org/apache/solr/core/NodeRoles.java
index b36358f37dc..6bb1e1633f1 100644
--- a/solr/core/src/java/org/apache/solr/core/NodeRoles.java
+++ b/solr/core/src/java/org/apache/solr/core/NodeRoles.java
@@ -102,6 +102,18 @@ public class NodeRoles {
public String modeWhenRoleIsAbsent() {
return MODE_DISALLOWED;
}
+ },
+
+ COORDINATOR("coordinator") {
+ @Override
+ public String modeWhenRoleIsAbsent() {
+ return MODE_OFF;
+ }
+
+ @Override
+ public Set<String> supportedModes() {
+ return Set.of(MODE_ON, MODE_OFF);
+ }
};
public final String roleName;
diff --git a/solr/core/src/java/org/apache/solr/handler/component/HttpShardHandler.java b/solr/core/src/java/org/apache/solr/handler/component/HttpShardHandler.java
index dd1d7ac04b2..1cd131560a2 100644
--- a/solr/core/src/java/org/apache/solr/handler/component/HttpShardHandler.java
+++ b/solr/core/src/java/org/apache/solr/handler/component/HttpShardHandler.java
@@ -274,7 +274,7 @@ public class HttpShardHandler extends ShardHandler {
final String shards = params.get(ShardParams.SHARDS);
CoreDescriptor coreDescriptor = req.getCore().getCoreDescriptor();
- CloudDescriptor cloudDescriptor = coreDescriptor.getCloudDescriptor();
+ CloudDescriptor cloudDescriptor = req.getCloudDescriptor();
ZkController zkController = req.getCoreContainer().getZkController();
final ReplicaListTransformer replicaListTransformer =
diff --git a/solr/core/src/java/org/apache/solr/request/DelegatedSolrQueryRequest.java b/solr/core/src/java/org/apache/solr/request/DelegatedSolrQueryRequest.java
new file mode 100644
index 00000000000..70028099ef1
--- /dev/null
+++ b/solr/core/src/java/org/apache/solr/request/DelegatedSolrQueryRequest.java
@@ -0,0 +1,167 @@
+/*
+ * 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.request;
+
+import io.opentracing.Span;
+import io.opentracing.Tracer;
+import java.security.Principal;
+import java.util.List;
+import java.util.Map;
+import org.apache.solr.cloud.CloudDescriptor;
+import org.apache.solr.common.params.SolrParams;
+import org.apache.solr.common.util.CommandOperation;
+import org.apache.solr.common.util.ContentStream;
+import org.apache.solr.core.CoreContainer;
+import org.apache.solr.core.SolrCore;
+import org.apache.solr.schema.IndexSchema;
+import org.apache.solr.search.SolrIndexSearcher;
+import org.apache.solr.servlet.HttpSolrCall;
+import org.apache.solr.util.RTimerTree;
+
+public class DelegatedSolrQueryRequest implements SolrQueryRequest {
+ private final SolrQueryRequest delegate;
+
+ public DelegatedSolrQueryRequest(SolrQueryRequest delegate) {
+ this.delegate = delegate;
+ }
+
+ @Override
+ public SolrParams getParams() {
+ return delegate.getParams();
+ }
+
+ @Override
+ public void setParams(SolrParams params) {
+ delegate.setParams(params);
+ }
+
+ @Override
+ public Iterable<ContentStream> getContentStreams() {
+ return delegate.getContentStreams();
+ }
+
+ @Override
+ public SolrParams getOriginalParams() {
+ return delegate.getOriginalParams();
+ }
+
+ @Override
+ public Map<Object, Object> getContext() {
+ return delegate.getContext();
+ }
+
+ @Override
+ public void close() {
+ delegate.close();
+ }
+
+ @Override
+ public long getStartTime() {
+ return delegate.getStartTime();
+ }
+
+ @Override
+ public RTimerTree getRequestTimer() {
+ return delegate.getRequestTimer();
+ }
+
+ @Override
+ public SolrIndexSearcher getSearcher() {
+ return delegate.getSearcher();
+ }
+
+ @Override
+ public SolrCore getCore() {
+ return delegate.getCore();
+ }
+
+ @Override
+ public IndexSchema getSchema() {
+ return delegate.getSchema();
+ }
+
+ @Override
+ public void updateSchemaToLatest() {
+ delegate.updateSchemaToLatest();
+ }
+
+ @Override
+ public String getParamString() {
+ return delegate.getParamString();
+ }
+
+ @Override
+ public Map<String, Object> getJSON() {
+ return delegate.getJSON();
+ }
+
+ @Override
+ public void setJSON(Map<String, Object> json) {
+ delegate.setJSON(json);
+ }
+
+ @Override
+ public Principal getUserPrincipal() {
+ return delegate.getUserPrincipal();
+ }
+
+ @Override
+ public String getPath() {
+ return delegate.getPath();
+ }
+
+ @Override
+ public Map<String, String> getPathTemplateValues() {
+ return delegate.getPathTemplateValues();
+ }
+
+ @Override
+ public List<CommandOperation> getCommands(boolean validateInput) {
+ return delegate.getCommands(validateInput);
+ }
+
+ @Override
+ public String getHttpMethod() {
+ return delegate.getHttpMethod();
+ }
+
+ @Override
+ public HttpSolrCall getHttpSolrCall() {
+ return delegate.getHttpSolrCall();
+ }
+
+ @Override
+ public Tracer getTracer() {
+ return delegate.getTracer();
+ }
+
+ @Override
+ public Span getSpan() {
+ return delegate.getSpan();
+ }
+
+ @Override
+ public CoreContainer getCoreContainer() {
+ return delegate.getCoreContainer();
+ }
+
+ @Override
+ public CloudDescriptor getCloudDescriptor() {
+ return delegate.getCloudDescriptor();
+ }
+}
diff --git a/solr/core/src/java/org/apache/solr/servlet/CoordinatorHttpSolrCall.java b/solr/core/src/java/org/apache/solr/servlet/CoordinatorHttpSolrCall.java
new file mode 100644
index 00000000000..06d7f94cdd5
--- /dev/null
+++ b/solr/core/src/java/org/apache/solr/servlet/CoordinatorHttpSolrCall.java
@@ -0,0 +1,223 @@
+/*
+ * 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.servlet;
+
+import java.lang.invoke.MethodHandles;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.ConcurrentHashMap;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+import org.apache.solr.api.CoordinatorV2HttpSolrCall;
+import org.apache.solr.client.solrj.request.CollectionAdminRequest;
+import org.apache.solr.cloud.CloudDescriptor;
+import org.apache.solr.cloud.api.collections.Assign;
+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.Replica;
+import org.apache.solr.common.cloud.ZkStateReader;
+import org.apache.solr.common.params.SolrParams;
+import org.apache.solr.common.util.Utils;
+import org.apache.solr.core.CoreContainer;
+import org.apache.solr.core.CoreDescriptor;
+import org.apache.solr.core.SolrCore;
+import org.apache.solr.request.DelegatedSolrQueryRequest;
+import org.apache.solr.request.LocalSolrQueryRequest;
+import org.apache.solr.request.SolrQueryRequest;
+import org.apache.solr.response.SolrQueryResponse;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class CoordinatorHttpSolrCall extends HttpSolrCall {
+ public static final String SYNTHETIC_COLL_PREFIX =
+ Assign.SYSTEM_COLL_PREFIX + "COORDINATOR-COLL-";
+ private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+ private String collectionName;
+ private final Factory factory;
+
+ public CoordinatorHttpSolrCall(
+ Factory factory,
+ SolrDispatchFilter solrDispatchFilter,
+ CoreContainer cores,
+ HttpServletRequest request,
+ HttpServletResponse response,
+ boolean retry) {
+ super(solrDispatchFilter, cores, request, response, retry);
+ this.factory = factory;
+ }
+
+ @Override
+ protected SolrCore getCoreByCollection(String collectionName, boolean isPreferLeader) {
+ this.collectionName = collectionName;
+ SolrCore core = super.getCoreByCollection(collectionName, isPreferLeader);
+ if (core != null) return core;
+ if (!path.endsWith("/select")) return null;
+ return getCore(factory, this, collectionName, isPreferLeader);
+ }
+
+ public static SolrCore getCore(
+ Factory factory, HttpSolrCall solrCall, String collectionName, boolean isPreferLeader) {
+ String sytheticCoreName = factory.collectionVsCoreNameMapping.get(collectionName);
+ if (sytheticCoreName != null) {
+ return solrCall.cores.getCore(sytheticCoreName);
+ } else {
+ ZkStateReader zkStateReader = solrCall.cores.getZkController().getZkStateReader();
+ ClusterState clusterState = zkStateReader.getClusterState();
+ DocCollection coll = clusterState.getCollectionOrNull(collectionName, true);
+ if (coll != null) {
+ String confName = coll.getConfigName();
+ String syntheticCollectionName = SYNTHETIC_COLL_PREFIX + confName;
+
+ DocCollection syntheticColl = clusterState.getCollectionOrNull(syntheticCollectionName);
+ if (syntheticColl == null) {
+ // no such collection. let's create one
+ if (log.isInfoEnabled()) {
+ log.info(
+ "synthetic collection: {} does not exist, creating.. ", syntheticCollectionName);
+ }
+ createColl(syntheticCollectionName, solrCall.cores, confName);
+ }
+ SolrCore core = solrCall.getCoreByCollection(syntheticCollectionName, isPreferLeader);
+ if (core != null) {
+ factory.collectionVsCoreNameMapping.put(collectionName, core.getName());
+ solrCall.cores.getZkController().getZkStateReader().registerCore(collectionName);
+ if (log.isDebugEnabled()) {
+ log.debug("coordinator node, returns synthetic core: {}", core.getName());
+ }
+ } else {
+ // this node does not have a replica. add one
+ if (log.isInfoEnabled()) {
+ log.info(
+ "this node does not have a replica of the synthetic collection: {} , adding replica ",
+ syntheticCollectionName);
+ }
+
+ addReplica(syntheticCollectionName, solrCall.cores);
+ core = solrCall.getCoreByCollection(syntheticCollectionName, isPreferLeader);
+ }
+ return core;
+ }
+ return null;
+ }
+ }
+
+ private static void addReplica(String syntheticCollectionName, CoreContainer cores) {
+ SolrQueryResponse rsp = new SolrQueryResponse();
+ try {
+ cores
+ .getCollectionsHandler()
+ .handleRequestBody(
+ new LocalSolrQueryRequest(
+ null,
+ CollectionAdminRequest.addReplicaToShard(syntheticCollectionName, "shard1")
+ .setCreateNodeSet(cores.getZkController().getNodeName())
+ .getParams()),
+ rsp);
+ if (rsp.getValues().get("success") == null) {
+ throw new SolrException(
+ SolrException.ErrorCode.SERVER_ERROR,
+ "Could not auto-create collection: " + Utils.toJSONString(rsp.getValues()));
+ }
+ } catch (SolrException e) {
+ throw e;
+
+ } catch (Exception e) {
+ throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, e);
+ }
+ }
+
+ private static void createColl(
+ String syntheticCollectionName, CoreContainer cores, String confName) {
+ SolrQueryResponse rsp = new SolrQueryResponse();
+ try {
+ SolrParams params =
+ CollectionAdminRequest.createCollection(syntheticCollectionName, confName, 1, 1)
+ .setCreateNodeSet(cores.getZkController().getNodeName())
+ .getParams();
+ if (log.isInfoEnabled()) {
+ log.info("sending collection admin command : {}", Utils.toJSONString(params));
+ }
+ cores.getCollectionsHandler().handleRequestBody(new LocalSolrQueryRequest(null, params), rsp);
+ if (rsp.getValues().get("success") == null) {
+ throw new SolrException(
+ SolrException.ErrorCode.SERVER_ERROR,
+ "Could not create :"
+ + syntheticCollectionName
+ + " collection: "
+ + Utils.toJSONString(rsp.getValues()));
+ }
+ } catch (SolrException e) {
+ throw e;
+
+ } catch (Exception e) {
+ throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, e);
+ }
+ }
+
+ @Override
+ protected void init() throws Exception {
+ super.init();
+ if (action == SolrDispatchFilter.Action.PROCESS && core != null) {
+ solrReq = wrappedReq(solrReq, collectionName, this);
+ }
+ }
+
+ public static SolrQueryRequest wrappedReq(
+ SolrQueryRequest delegate, String collectionName, HttpSolrCall httpSolrCall) {
+ Properties p = new Properties();
+ p.put(CoreDescriptor.CORE_COLLECTION, collectionName);
+ p.put(CloudDescriptor.REPLICA_TYPE, Replica.Type.PULL.toString());
+ p.put(CoreDescriptor.CORE_SHARD, "_");
+
+ CloudDescriptor cloudDescriptor =
+ new CloudDescriptor(
+ delegate.getCore().getCoreDescriptor(), delegate.getCore().getName(), p);
+ return new DelegatedSolrQueryRequest(delegate) {
+ @Override
+ public HttpSolrCall getHttpSolrCall() {
+ return httpSolrCall;
+ }
+
+ @Override
+ public CloudDescriptor getCloudDescriptor() {
+ return cloudDescriptor;
+ }
+ };
+ }
+
+ // The factory that creates an instance of HttpSolrCall
+ public static class Factory implements SolrDispatchFilter.HttpSolrCallFactory {
+ private final Map<String, String> collectionVsCoreNameMapping = new ConcurrentHashMap<>();
+
+ @Override
+ public HttpSolrCall createInstance(
+ SolrDispatchFilter filter,
+ String path,
+ CoreContainer cores,
+ HttpServletRequest request,
+ HttpServletResponse response,
+ boolean retry) {
+ if ((path.startsWith("/____v2/") || path.equals("/____v2"))) {
+ return new CoordinatorV2HttpSolrCall(this, filter, cores, request, response, retry);
+ } else {
+ return new CoordinatorHttpSolrCall(this, filter, cores, request, response, retry);
+ }
+ }
+ }
+}
diff --git a/solr/core/src/java/org/apache/solr/servlet/SolrDispatchFilter.java b/solr/core/src/java/org/apache/solr/servlet/SolrDispatchFilter.java
index ea53b5ac06c..bc13aba4a15 100644
--- a/solr/core/src/java/org/apache/solr/servlet/SolrDispatchFilter.java
+++ b/solr/core/src/java/org/apache/solr/servlet/SolrDispatchFilter.java
@@ -48,6 +48,7 @@ import org.apache.solr.common.SolrException.ErrorCode;
import org.apache.solr.common.util.ExecutorUtil;
import org.apache.solr.common.util.SuppressForbidden;
import org.apache.solr.core.CoreContainer;
+import org.apache.solr.core.NodeRoles;
import org.apache.solr.core.SolrCore;
import org.apache.solr.logging.MDCLoggingContext;
import org.apache.solr.logging.MDCSnapshot;
@@ -84,6 +85,8 @@ public class SolrDispatchFilter extends BaseSolrFilter implements PathExcluder {
protected String abortErrorMessage = null;
+ private HttpSolrCallFactory solrCallFactory;
+
@Override
public void setExcludePatterns(List<Pattern> excludePatterns) {
this.excludePatterns = excludePatterns;
@@ -91,7 +94,7 @@ public class SolrDispatchFilter extends BaseSolrFilter implements PathExcluder {
private List<Pattern> excludePatterns;
- private final boolean isV2Enabled = !"true".equals(System.getProperty("disable.v2.api", "false"));
+ public final boolean isV2Enabled = !"true".equals(System.getProperty("disable.v2.api", "false"));
public HttpClient getHttpClient() {
try {
@@ -137,7 +140,15 @@ public class SolrDispatchFilter extends BaseSolrFilter implements PathExcluder {
public void init(FilterConfig config) throws ServletException {
try {
coreService = CoreContainerProvider.serviceForContext(config.getServletContext());
-
+ boolean isCoordinator =
+ NodeRoles.MODE_ON.equals(
+ coreService
+ .getService()
+ .getCoreContainer()
+ .nodeRoles
+ .getRoleMode(NodeRoles.Role.COORDINATOR));
+ solrCallFactory =
+ isCoordinator ? new CoordinatorHttpSolrCall.Factory() : new HttpSolrCallFactory() {};
if (log.isTraceEnabled()) {
log.trace("SolrDispatchFilter.init(): {}", this.getClass().getClassLoader());
}
@@ -276,11 +287,7 @@ public class SolrDispatchFilter extends BaseSolrFilter implements PathExcluder {
} catch (UnavailableException e) {
throw new SolrException(ErrorCode.SERVER_ERROR, "Core Container Unavailable");
}
- if (isV2Enabled && (path.startsWith("/____v2/") || path.equals("/____v2"))) {
- return new V2HttpCall(this, cores, request, response, false);
- } else {
- return new HttpSolrCall(this, cores, request, response, retry);
- }
+ return solrCallFactory.createInstance(this, path, cores, request, response, retry);
}
// TODO: make this a servlet filter
@@ -394,4 +401,21 @@ public class SolrDispatchFilter extends BaseSolrFilter implements PathExcluder {
void replaceRateLimitManager(RateLimitManager rateLimitManager) {
coreService.getService().setRateLimitManager(rateLimitManager);
}
+
+ /** internal API */
+ public interface HttpSolrCallFactory {
+ default HttpSolrCall createInstance(
+ SolrDispatchFilter filter,
+ String path,
+ CoreContainer cores,
+ HttpServletRequest request,
+ HttpServletResponse response,
+ boolean retry) {
+ if (filter.isV2Enabled && (path.startsWith("/____v2/") || path.equals("/____v2"))) {
+ return new V2HttpCall(filter, cores, request, response, retry);
+ } else {
+ return new HttpSolrCall(filter, cores, request, response, retry);
+ }
+ }
+ }
}
diff --git a/solr/core/src/test/org/apache/solr/search/TestCoordinatorRole.java b/solr/core/src/test/org/apache/solr/search/TestCoordinatorRole.java
new file mode 100644
index 00000000000..282b06d4640
--- /dev/null
+++ b/solr/core/src/test/org/apache/solr/search/TestCoordinatorRole.java
@@ -0,0 +1,85 @@
+/*
+ * 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.search;
+
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import org.apache.solr.client.solrj.SolrQuery;
+import org.apache.solr.client.solrj.embedded.JettySolrRunner;
+import org.apache.solr.client.solrj.impl.CloudSolrClient;
+import org.apache.solr.client.solrj.request.CollectionAdminRequest;
+import org.apache.solr.client.solrj.request.QueryRequest;
+import org.apache.solr.client.solrj.request.UpdateRequest;
+import org.apache.solr.client.solrj.response.QueryResponse;
+import org.apache.solr.cloud.SolrCloudTestCase;
+import org.apache.solr.common.SolrInputDocument;
+import org.apache.solr.common.cloud.DocCollection;
+import org.apache.solr.core.NodeRoles;
+import org.apache.solr.servlet.CoordinatorHttpSolrCall;
+import org.junit.BeforeClass;
+
+public class TestCoordinatorRole extends SolrCloudTestCase {
+
+ @BeforeClass
+ public static void setupCluster() throws Exception {
+ configureCluster(4).addConfig("conf", configset("cloud-minimal")).configure();
+ }
+
+ public void testSimple() throws Exception {
+ CloudSolrClient client = cluster.getSolrClient();
+ String COLLECTION_NAME = "test_coll";
+ String SYNTHETIC_COLLECTION = CoordinatorHttpSolrCall.SYNTHETIC_COLL_PREFIX + "conf";
+ CollectionAdminRequest.createCollection(COLLECTION_NAME, "conf", 2, 2)
+ .process(cluster.getSolrClient());
+ cluster.waitForActiveCollection(COLLECTION_NAME, 2, 4);
+ UpdateRequest ur = new UpdateRequest();
+ for (int i = 0; i < 10; i++) {
+ SolrInputDocument doc2 = new SolrInputDocument();
+ doc2.addField("id", "" + i);
+ ur.add(doc2);
+ }
+
+ ur.commit(client, COLLECTION_NAME);
+ QueryResponse rsp = client.query(COLLECTION_NAME, new SolrQuery("*:*"));
+ assertEquals(10, rsp.getResults().getNumFound());
+
+ System.setProperty(NodeRoles.NODE_ROLES_PROP, "coordinator:on");
+ JettySolrRunner coordinatorJetty = null;
+ try {
+ coordinatorJetty = cluster.startJettySolrRunner();
+ } finally {
+ System.clearProperty(NodeRoles.NODE_ROLES_PROP);
+ }
+ QueryResponse rslt =
+ new QueryRequest(new SolrQuery("*:*"))
+ .setPreferredNodes(List.of(coordinatorJetty.getNodeName()))
+ .process(client, COLLECTION_NAME);
+
+ assertEquals(10, rslt.getResults().size());
+
+ DocCollection collection =
+ cluster.getSolrClient().getClusterStateProvider().getCollection(SYNTHETIC_COLLECTION);
+ assertNotNull(collection);
+
+ Set<String> expectedNodes = new HashSet<>();
+ expectedNodes.add(coordinatorJetty.getNodeName());
+ collection.forEachReplica((s, replica) -> expectedNodes.remove(replica.getNodeName()));
+ assertTrue(expectedNodes.isEmpty());
+ }
+}
diff --git a/solr/solr-ref-guide/modules/deployment-guide/pages/node-roles.adoc b/solr/solr-ref-guide/modules/deployment-guide/pages/node-roles.adoc
index 17bfcece5b3..a6fff6c814e 100644
--- a/solr/solr-ref-guide/modules/deployment-guide/pages/node-roles.adoc
+++ b/solr/solr-ref-guide/modules/deployment-guide/pages/node-roles.adoc
@@ -58,6 +58,9 @@ If a node has been started with no `solr.node.roles` parameter, it will be assum
|`overseer`
|allowed, preferred, disallowed
+
+|`coordinator`
+|on, off
|===
=== `data` role
@@ -66,15 +69,37 @@ A node with this role (in mode "on") can host shards and replicas for collection
=== `overseer` role
A node with this role can perform duties of an overseer node (unless mode is `disallowed`). When one or more nodes have the overseer role in `preferred` mode, the overseer leader will be elected from one of these nodes. In case no node is designated as a preferred overseer or no such node is live, the overseer leader will be elected from one of the nodes that have the overseer role in `allowed` mode. If all nodes that are designated with overseer role (allowed or preferred) are down, the [...]
+=== `coordinator` role
+
+A node with this role can act as if it has replicas of all collections in the cluster when a query is performed. The workflow is as follows
+
+If the cluster has collections with very large no:of shards, performing distributed requests in your _`data node`_ will lead to
+
+* large heap utilization
+* frequent GC pauses
+
+In such cases, a few dedicated nodes can be started with a *`coordinator`* role and queries can be sent to that node and avoid intermittent and unpredictable load in data nodes. The coordinator node is stateless and does not host any data. So, we can create and destroy coordinator nodes without any data loass or down time.
+
+==== The work-flow in a `coordinator` node
+
+1. A request for *`coll-A`* that uses configset *`configset-A`* comes to coordinator node
+2. It checks if there is a core that uses the configset *`configset-A`* is present. If yes, that core acts as a replica of *`coll-A`* and performs a distributed request to all shards of *`coll-A`* and sends back a response
+3. if there is no such core, it checks if there is a synthetic collection *`.sys.COORDINATOR-COLL-configset-A`* and a replica for that collection is present locally. If not the collection and replica is created on the fly and it goes to *`step 1`*
+
+
+
== Example usage
Sometimes, when the nodes in a cluster are under heavy querying or indexing load, the overseer leader node might be unable to perform collection management duties efficiently. It might be reasonable to have dedicated nodes to act as the overseer. Such an effect can be achieved as follows:
* Most nodes (data nodes) in the cluster start with `-Dsolr.node.roles=data:on,overseer:allowed` (or with no parameter, since the default value for `solr.node.roles` is the same).
* One or more nodes (dedicated overseer nodes) can start with `-Dsolr.node.roles=overseer:preferred` (or `-Dsolr.node.roles=overseer:preferred,data:off`)
+* One or more dedicated coordinator nodes can start with `-Dsolr.node.roles=coordinator:on,data:off`
In this arrangement, such dedicated nodes can be provisioned on hardware with lesser resources like CPU, memory or disk space than other data nodes (since these are stateless nodes) and yet the cluster will behave optimally. In case the dedicated overseer nodes go down for some reason, the overseer leader will be elected from one of the data nodes (since they have overseer in "allowed" mode), and once one of the dedicated overseer nodes are back up again, it will be re-elected for the ov [...]
+Dedicated *`coordinator`* nodes can be provisioned with enough memory but very little storage. They can also be started and stopped based on demand as they are stateless
+
== Roles API
=== GET /api/cluster/node-roles/supported
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/SolrRequest.java b/solr/solrj/src/java/org/apache/solr/client/solrj/SolrRequest.java
index 83c0d3ccb15..269c74ff04e 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/SolrRequest.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/SolrRequest.java
@@ -22,6 +22,7 @@ import java.security.Principal;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
@@ -75,6 +76,7 @@ public abstract class SolrRequest<T extends SolrResponse> implements Serializabl
private METHOD method = METHOD.GET;
private String path = null;
private Map<String, String> headers;
+ private List<String> preferredNodes;
private ResponseParser responseParser;
private StreamingResponseCallback callback;
@@ -98,6 +100,15 @@ public abstract class SolrRequest<T extends SolrResponse> implements Serializabl
return this;
}
+ public SolrRequest<T> setPreferredNodes(List<String> nodes) {
+ this.preferredNodes = nodes;
+ return this;
+ }
+
+ public List<String> getPreferredNodes() {
+ return this.preferredNodes;
+ }
+
private String basicAuthUser, basicAuthPwd;
private String basePath;
diff --git a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/CloudSolrClient.java b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/CloudSolrClient.java
index 23901407774..359bde54b08 100644
--- a/solr/solrj/src/java/org/apache/solr/client/solrj/impl/CloudSolrClient.java
+++ b/solr/solrj/src/java/org/apache/solr/client/solrj/impl/CloudSolrClient.java
@@ -1092,6 +1092,21 @@ public abstract class CloudSolrClient extends SolrClient {
+ inputCollections);
}
+ List<String> preferredNodes = request.getPreferredNodes();
+ if (preferredNodes != null && !preferredNodes.isEmpty()) {
+ String joinedInputCollections = StrUtils.join(inputCollections, ',');
+ List<String> urlList = new ArrayList<>(preferredNodes.size());
+ for (String nodeName : preferredNodes) {
+ urlList.add(
+ Utils.getBaseUrlForNodeName(nodeName, urlScheme) + "/" + joinedInputCollections);
+ }
+ if (!urlList.isEmpty()) {
+ LBSolrClient.Req req = new LBSolrClient.Req(request, urlList);
+ LBSolrClient.Rsp rsp = getLbClient().request(req);
+ return rsp.getResponse();
+ }
+ }
+
// TODO: not a big deal because of the caching, but we could avoid looking
// at every shard when getting leaders if we tweaked some things