SOLR-15694, SOLR-15715: Node roles and dedicated query coordinator nodes

Co-authored-by: Noble Paul <[email protected]>

solr/contrib/analytics/src/java/org/apache/solr/handler/component/AnalyticsComponent.java

@@ -109,7 +109,7 @@ public int distributedProcess(ResponseBuilder rb) throws IOException {
 
     // Send out a request to each shard and merge the responses into our AnalyticsRequestManager
     reqManager.shardStream.sendRequests(rb.req.getCore().getCoreDescriptor().getCollectionName(),
-        rb.req.getCore().getCoreContainer().getZkController().getZkServerAddress());
+        rb.req.getCoreContainer().getZkController().getZkServerAddress());
 
     reqManager.sendShards = false;
 

solr/contrib/ltr/src/java/org/apache/solr/ltr/response/transform/LTRFeatureLoggerTransformerFactory.java

@@ -217,7 +217,7 @@ public void setContext(ResultContext context) {
       }
       leafContexts = searcher.getTopReaderContext().leaves();
       if (threadManager != null) {
-        threadManager.setExecutor(context.getRequest().getCore().getCoreContainer().getUpdateShardHandler().getUpdateExecutor());
+        threadManager.setExecutor(context.getRequest().getCoreContainer().getUpdateShardHandler().getUpdateExecutor());
       }
 
       rerankingQueriesFromContext = SolrQueryRequestContextUtils.getScoringQueries(req);

solr/contrib/ltr/src/java/org/apache/solr/ltr/search/LTRQParserPlugin.java

@@ -153,7 +153,7 @@ public LTRQParser(String qstr, SolrParams localParams, SolrParams params,
     @Override
     public Query parse() throws SyntaxError {
       if (threadManager != null) {
-        threadManager.setExecutor(req.getCore().getCoreContainer().getUpdateShardHandler().getUpdateExecutor());
+        threadManager.setExecutor(req.getCoreContainer().getUpdateShardHandler().getUpdateExecutor());
       }
       // ReRanking Model
       final String[] modelNames = localParams.getParams(LTRQParserPlugin.MODEL);

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);
+    }
+  }
+}

solr/core/src/java/org/apache/solr/cloud/api/collections/Assign.java

@@ -23,6 +23,7 @@
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.LinkedHashMap;
 import java.util.LinkedHashSet;
 import java.util.List;
@@ -55,6 +56,8 @@
 import org.apache.solr.common.params.CollectionAdminParams;
 import org.apache.solr.common.util.StrUtils;
 import org.apache.solr.common.util.Utils;
+import org.apache.solr.core.NodeRoles;
+import org.apache.solr.handler.ClusterAPI;
 import org.apache.solr.util.NumberUtils;
 import org.apache.zookeeper.CreateMode;
 import org.apache.zookeeper.KeeperException;
@@ -233,7 +236,12 @@ private static boolean existCoreName(String coreName, Slice slice) {
     return false;
   }
 
-  public static List<String> getLiveOrLiveAndCreateNodeSetList(final Set<String> liveNodes, final ZkNodeProps message, final Random random) {
+  public static List<String> getLiveOrLiveAndCreateNodeSetList(
+      final Set<String> liveNodes,
+      final ZkNodeProps message,
+      final Random random,
+      DistribStateManager zk) {
+
     List<String> nodeList;
     final String createNodeSetStr = message.getStr(CREATE_NODE_SET);
     final List<String> createNodeList = (createNodeSetStr == null) ? null :
@@ -248,13 +256,28 @@ public static List<String> getLiveOrLiveAndCreateNodeSetList(final Set<String> l
         Collections.shuffle(nodeList, random);
       }
     } else {
-      nodeList = new ArrayList<>(liveNodes);
+      nodeList = new ArrayList<>(filterNonDataNodes(zk, liveNodes));
       Collections.shuffle(nodeList, random);
     }
 
     return nodeList;
   }
-
+  public static Collection<String> filterNonDataNodes(
+      DistribStateManager zk, Collection<String> liveNodes) {
+    try {
+      List<String> noData = ClusterAPI.getNodesByRole(NodeRoles.Role.DATA, NodeRoles.MODE_OFF, zk);
+      if (noData.isEmpty()) {
+        return liveNodes;
+      } else {
+        liveNodes = new HashSet<>(liveNodes);
+        liveNodes.removeAll(noData);
+        return liveNodes;
+      }
+    } catch (Exception e) {
+      throw new SolrException(
+          SolrException.ErrorCode.SERVER_ERROR, "Error fetching roles from Zookeeper", e);
+    }
+  }
   /**
    * <b>Note:</b> where possible, the {@link #usePolicyFramework(DocCollection, SolrCloudManager)} method should
    * be used instead of this method

solr/core/src/java/org/apache/solr/cloud/api/collections/CategoryRoutedAlias.java

@@ -123,7 +123,7 @@ public RoutedAliasTypes getRoutedAliasType() {
   @Override
   public void validateRouteValue(AddUpdateCommand cmd) throws SolrException {
     if (this.aliases == null) {
-      updateParsedCollectionAliases(cmd.getReq().getCore().getCoreContainer().getZkController().zkStateReader, false);
+      updateParsedCollectionAliases(cmd.getReq().getCoreContainer().getZkController().zkStateReader, false);
     }
 
     Object fieldValue = cmd.getSolrInputDocument().getFieldValue(getRouteField());

solr/core/src/java/org/apache/solr/cloud/api/collections/CreateCollectionCmd.java

@@ -459,7 +459,12 @@ public static List<ReplicaPosition> buildReplicaPositions(SolrCloudManager cloud
     // but (for now) require that each core goes on a distinct node.
 
     List<ReplicaPosition> replicaPositions;
-    List<String> nodeList = Assign.getLiveOrLiveAndCreateNodeSetList(clusterState.getLiveNodes(), message, OverseerCollectionMessageHandler.RANDOM);
+    List<String> nodeList =
+        Assign.getLiveOrLiveAndCreateNodeSetList(
+            clusterState.getLiveNodes(),
+            message,
+            OverseerCollectionMessageHandler.RANDOM,
+            cloudManager.getDistribStateManager());
     if (nodeList.isEmpty()) {
       log.warn("It is unusual to create a collection ({}) without cores.", collectionName);
 

solr/core/src/java/org/apache/solr/cloud/api/collections/RestoreCmd.java

@@ -189,8 +189,12 @@ private RestoreContext(ZkNodeProps message, OverseerCollectionMessageHandler ocm
       this.backupCollectionState = this.backupManager.readCollectionState(this.backupCollection);
 
       this.shardHandler = ocmh.shardHandlerFactory.getShardHandler();
-      this.nodeList = Assign.getLiveOrLiveAndCreateNodeSetList(
-              zkStateReader.getClusterState().getLiveNodes(), message, OverseerCollectionMessageHandler.RANDOM);
+      this.nodeList =
+          Assign.getLiveOrLiveAndCreateNodeSetList(
+              zkStateReader.getClusterState().getLiveNodes(),
+              message,
+              OverseerCollectionMessageHandler.RANDOM,
+              container.getZkController().getSolrCloudManager().getDistribStateManager());
     }
 
     @Override

solr/core/src/java/org/apache/solr/cloud/api/collections/RoutedAlias.java

@@ -324,7 +324,7 @@ private String createAllRequiredCollections(AddUpdateCommand cmd, CandidateColle
 
     SolrQueryRequest req = cmd.getReq();
     SolrCore core = req.getCore();
-    CoreContainer coreContainer = core.getCoreContainer();
+    CoreContainer coreContainer = req.getCoreContainer();
     do {
       switch (targetCollectionDesc.getCreationType()) {
         case NONE:

solr/core/src/java/org/apache/solr/cloud/api/collections/TimeRoutedAlias.java

@@ -373,7 +373,7 @@ public void validateRouteValue(AddUpdateCommand cmd) throws SolrException {
     } catch (DateTimeParseException e) {
       startTime = DateMathParser.parseMath(new Date(), start).toInstant();
       SolrCore core = cmd.getReq().getCore();
-      ZkStateReader zkStateReader = core.getCoreContainer().getZkController().zkStateReader;
+      ZkStateReader zkStateReader = cmd.getReq().getCoreContainer().getZkController().zkStateReader;
       Aliases aliases = zkStateReader.getAliases();
       Map<String, String> props = new HashMap<>(aliases.getCollectionAliasProperties(aliasName));
       start = DateTimeFormatter.ISO_INSTANT.format(startTime);

solr/core/src/java/org/apache/solr/core/CoreContainer.java

@@ -249,6 +249,7 @@ public CoreLoadFailure(CoreDescriptor cd, Exception loadFailure) {
   protected volatile AutoscalingHistoryHandler autoscalingHistoryHandler;
 
   private volatile SolrClientCache solrClientCache;
+  public final NodeRoles nodeRoles = new NodeRoles(System.getProperty(NodeRoles.NODE_ROLES_PROP));
 
   private final ObjectCache objectCache = new ObjectCache();
 

solr/core/src/java/org/apache/solr/core/NodeRoles.java

@@ -0,0 +1,152 @@
+/*
+ * 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.core;
+
+import java.util.Collections;
+import java.util.EnumMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import com.google.common.collect.ImmutableSet;
+import org.apache.solr.common.SolrException;
+import org.apache.solr.common.StringUtils;
+import org.apache.solr.common.cloud.ZkStateReader;
+import org.apache.solr.common.util.StrUtils;
+
+public class NodeRoles {
+  public static final String NODE_ROLES_PROP = "solr.node.roles";
+
+  /** Roles to be assumed on nodes that don't have roles specified for them at startup */
+  public static final String DEFAULT_ROLES_STRING = "data:on,overseer:allowed";
+
+  // Map of roles to mode that are applicable for this node.
+  private Map<Role, String> nodeRoles;
+
+  public NodeRoles(String rolesString) {
+    Map<Role, String> roles = new EnumMap<>(Role.class);
+    if (StringUtils.isEmpty(rolesString)) {
+      rolesString = DEFAULT_ROLES_STRING;
+    }
+    List<String> rolesList = StrUtils.splitSmart(rolesString, ',');
+    for (String s : rolesList) {
+      List<String> roleMode = StrUtils.splitSmart(s, ':');
+      Role r = Role.getRole(roleMode.get(0));
+      String m = roleMode.get(1);
+      if (r.supportedModes().contains(m)) {
+        roles.put(r, m);
+      } else {
+        throw new SolrException(
+            SolrException.ErrorCode.SERVER_ERROR,
+            "Unknown role mode '" + roleMode.get(1) + "' for role '" + r + "'");
+      }
+    }
+    for (Role r : Role.values()) {
+      if (!roles.containsKey(r)) {
+        roles.put(r, r.modeWhenRoleIsAbsent());
+      }
+    }
+    nodeRoles = Collections.unmodifiableMap(roles);
+  }
+
+  public Map<Role, String> getRoles() {
+    return nodeRoles;
+  }
+
+  public String getRoleMode(Role role) {
+    return nodeRoles.get(role);
+  }
+
+  public boolean isOverseerAllowedOrPreferred() {
+    String roleMode = nodeRoles.get(Role.OVERSEER);
+    return MODE_ALLOWED.equals(roleMode) || MODE_PREFERRED.equals(roleMode);
+  }
+
+  public static final String MODE_ON = "on";
+  public static final String MODE_OFF = "off";
+  public static final String MODE_ALLOWED = "allowed";
+  public static final String MODE_PREFERRED = "preferred";
+  public static final String MODE_DISALLOWED = "disallowed";
+
+  public enum Role {
+    DATA("data") {
+      @Override
+      public Set<String> supportedModes() {
+        return ImmutableSet.of(MODE_ON, MODE_OFF);
+      }
+
+      @Override
+      public String modeWhenRoleIsAbsent() {
+        return MODE_OFF;
+      }
+    },
+    OVERSEER("overseer") {
+      @Override
+      public Set<String> supportedModes() {
+        return ImmutableSet.of(MODE_ALLOWED, MODE_PREFERRED, MODE_DISALLOWED);
+      }
+
+      @Override
+      public String modeWhenRoleIsAbsent() {
+        return MODE_DISALLOWED;
+      }
+    },
+
+    COORDINATOR("coordinator") {
+      @Override
+      public String modeWhenRoleIsAbsent() {
+        return MODE_OFF;
+      }
+
+      @Override
+      public Set<String> supportedModes() {
+        return ImmutableSet.of(MODE_ON, MODE_OFF);
+      }
+    };
+
+    public final String roleName;
+
+    Role(String name) {
+      this.roleName = name;
+    }
+
+    public static Role getRole(String value) {
+      for (Role role : Role.values()) {
+        if (value.equals(role.roleName)) return role;
+      }
+      throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, "Unknown role: " + value);
+    }
+
+    public abstract Set<String> supportedModes();
+
+    /** Default mode for a role in nodes where this role is not specified. */
+    public abstract String modeWhenRoleIsAbsent();
+
+    @Override
+    public String toString() {
+      return roleName;
+    }
+  }
+
+  public static String getZNodeForRole(Role role) {
+    return ZkStateReader.NODE_ROLES + "/" + role.roleName;
+  }
+
+  public static String getZNodeForRoleMode(Role role, String mode) {
+    return ZkStateReader.NODE_ROLES + "/" + role.roleName + "/" + mode;
+  }
+}