Refactoring only: Move direct dispatch decision to SearchCluster

4 files changed, 54 insertions, 50 deletions

diff --git a/container-search/src/main/java/com/yahoo/prelude/cluster/ClusterSearcher.java b/container-search/src/main/java/com/yahoo/prelude/cluster/ClusterSearcher.java
index bc740cd646f..25031f7376f 100644
--- a/container-search/src/main/java/com/yahoo/prelude/cluster/ClusterSearcher.java
+++ b/container-search/src/main/java/com/yahoo/prelude/cluster/ClusterSearcher.java
@@ -113,7 +113,7 @@ public class ClusterSearcher extends Searcher {
         QrSearchersConfig.Searchcluster searchClusterConfig = getSearchClusterConfigFromClusterName(qrsConfig, clusterModelName);
         documentTypes = new LinkedHashSet<>();
         failoverToRemote = clusterConfig.failoverToRemote();
-        Dispatcher dispatcher = new Dispatcher(dispatchConfig, fs4ResourcePool);
+        Dispatcher dispatcher = new Dispatcher(dispatchConfig, fs4ResourcePool, clusterInfoConfig.nodeCount());
 
         String eventName = clusterModelName + ".cache_hit_ratio";
         cacheHitRatio = new Value(eventName, manager, new Value.Parameters().setNameExtension(false)
@@ -128,8 +128,6 @@ public class ClusterSearcher extends Searcher {
                 .com().yahoo().prelude().fastsearch().FastSearcher().docsum()
                 .defaultclass());
         
-        int containerClusterSize = clusterInfoConfig.nodeCount();
-
         for (DocumentdbInfoConfig.Documentdb docDb : documentDbConfig.documentdb()) {
             String docTypeName = docDb.name();
             documentTypes.add(docTypeName);
@@ -151,7 +149,7 @@ public class ClusterSearcher extends Searcher {
                 Backend b = createBackend(searchClusterConfig.dispatcher(dispatcherIndex));
                 FastSearcher searcher = searchDispatch(searchClusterIndex, fs4ResourcePool, 
                                                        searchClusterConfig, cacheParams, emulationConfig, docSumParams,
-                                                       documentDbConfig, b, dispatcher, dispatcherIndex, containerClusterSize);
+                                                       documentDbConfig, b, dispatcher, dispatcherIndex);
                 try {
                     searcher.setLocalDispatching( ! isRemote(searchClusterConfig.dispatcher(dispatcherIndex).host()));
                 } catch (UnknownHostException e) {
@@ -210,14 +208,13 @@ public class ClusterSearcher extends Searcher {
                                                DocumentdbInfoConfig documentdbInfoConfig,
                                                Backend backend,
                                                Dispatcher dispatcher,
-                                               int dispatcherIndex,
-                                               int containerClusterSize) {
+                                               int dispatcherIndex) {
         ClusterParams clusterParams = makeClusterParams(searchclusterIndex,
                                                         searchClusterConfig,
                                                         emulConfig, 
                                                         dispatcherIndex);
         return new FastSearcher(backend, fs4ResourcePool, dispatcher, docSumParams, clusterParams, cacheParams, 
-                                documentdbInfoConfig, containerClusterSize);
+                                documentdbInfoConfig);
     }
 
     private static VdsStreamingSearcher vdsCluster(int searchclusterIndex,
diff --git a/container-search/src/main/java/com/yahoo/prelude/fastsearch/FastSearcher.java b/container-search/src/main/java/com/yahoo/prelude/fastsearch/FastSearcher.java
index 644199520e7..e9624f0cbdf 100644
--- a/container-search/src/main/java/com/yahoo/prelude/fastsearch/FastSearcher.java
+++ b/container-search/src/main/java/com/yahoo/prelude/fastsearch/FastSearcher.java
@@ -71,7 +71,6 @@ public class FastSearcher extends VespaBackEndSearcher {
     private final FS4ResourcePool fs4ResourcePool;
     
     private final String selfHostname;
-    private final int containerClusterSize;
     
     /**
      * Creates a Fastsearcher.
@@ -88,18 +87,15 @@ public class FastSearcher extends VespaBackEndSearcher {
      * @param clusterParams the cluster number, and other cluster backend parameters
      * @param cacheParams   the size, lifetime, and controller of our cache
      * @param documentdbInfoConfig document database parameters
-     * @param containerClusterSize the size of the cluster this is part of
      */
     public FastSearcher(Backend dispatchBackend, FS4ResourcePool fs4ResourcePool,
                         Dispatcher dispatcher, SummaryParameters docSumParams, ClusterParams clusterParams,
-                        CacheParams cacheParams, DocumentdbInfoConfig documentdbInfoConfig,
-                        int containerClusterSize) {
+                        CacheParams cacheParams, DocumentdbInfoConfig documentdbInfoConfig) {
         init(docSumParams, clusterParams, cacheParams, documentdbInfoConfig);
         this.dispatchBackend = dispatchBackend;
         this.fs4ResourcePool = fs4ResourcePool;
         this.dispatcher = dispatcher;
         this.selfHostname = HostName.getLocalhost();
-        this.containerClusterSize = containerClusterSize;
     }
 
     private static SimpleDateFormat isoDateFormat;
@@ -226,39 +222,17 @@ public class FastSearcher extends VespaBackEndSearcher {
     private Backend chooseBackend(Query query) {
         // TODO 2016-08-16: Turn this on by default (by changing the 'false' below to 'true')
         if ( ! query.properties().getBoolean(dispatchDirect, false)) return dispatchBackend;
-        
-        // A search node in the search cluster in question is configured on the same host as the currently running container.
-        // It has all the data <==> No other nodes in the search cluster have the same group id as this node.
-        //         That local search node responds.
-        // The search cluster to be searched has at least as many nodes as the container cluster we're running in.
-        ImmutableCollection<SearchCluster.Node> localSearchNodes = dispatcher.searchCluster().nodesByHost().get(selfHostname);
-        // Only use direct dispatch if we have exactly 1 search node on the same machine:
-        if (localSearchNodes.size() != 1) return dispatchBackend;
-
-        SearchCluster.Node localSearchNode = localSearchNodes.iterator().next();
-        SearchCluster.Group localSearchGroup = dispatcher.searchCluster().groups().get(localSearchNode.group());
-
-        // Only use direct dispatch if the local search node has the entire corpus
-        if (localSearchGroup.nodes().size() != 1) return dispatchBackend;
-
-        // Only use direct dispatch if this container cluster has at least as many nodes as the search cluster
-        // to avoid load skew/preserve fanout in the case where a subset of the search nodes are also containers.
-        // This disregards the case where the search and container clusters are partially overlapping. 
-        // Such configurations produce skewed load in any case.
-        if (containerClusterSize < dispatcher.searchCluster().size()) return dispatchBackend;
-
-        // Only use direct dispatch if the upstream ClusterSearcher chose the local dispatch
-        // (otherwise, we may be in this method due to a failover situation)
-        if ( ! dispatchBackend.getHost().equals(selfHostname)) return dispatchBackend;
 
-        // Only use direct dispatch if the local grouop has sufficient coverage
-        if ( ! localSearchGroup.hasSufficientCoverage()) return dispatchBackend;
+        // Don't use direct dispatch if the upstream ClusterSearcher did not chose the local dispatch
+        // as that probably means that we are in a failover situation
+        if ( ! dispatchBackend.getHost().equals(selfHostname)) return dispatchBackend;
 
-        // Only use direct dispatch if the local search node is up
-        if ( ! localSearchNode.isWorking()) return dispatchBackend;
+        Optional<SearchCluster.Node> directDispatchRecipient = dispatcher.searchCluster().dispatchDirectlyFrom(selfHostname);
+        if ( ! directDispatchRecipient.isPresent()) return dispatchBackend;
 
-        query.trace(false, 2, "Dispatching directly to ", localSearchNode);
-        return fs4ResourcePool.getBackend(localSearchNode.hostname(), localSearchNode.fs4port());
+        query.trace(false, 2, "Dispatching directly to ", directDispatchRecipient.get());
+        return fs4ResourcePool.getBackend(directDispatchRecipient.get().hostname(), 
+                                          directDispatchRecipient.get().fs4port());
     }
 
     /**
diff --git a/container-search/src/main/java/com/yahoo/search/dispatch/Dispatcher.java b/container-search/src/main/java/com/yahoo/search/dispatch/Dispatcher.java
index ca6445cff44..6245a0ee442 100644
--- a/container-search/src/main/java/com/yahoo/search/dispatch/Dispatcher.java
+++ b/container-search/src/main/java/com/yahoo/search/dispatch/Dispatcher.java
@@ -3,7 +3,6 @@ package com.yahoo.search.dispatch;
 
 import com.google.common.annotations.Beta;
 import com.google.common.collect.ImmutableMap;
-import com.google.inject.Inject;
 import com.yahoo.collections.ListMap;
 import com.yahoo.component.AbstractComponent;
 import com.yahoo.compress.CompressionType;
@@ -51,10 +50,9 @@ public class Dispatcher extends AbstractComponent {
 
     private final Compressor compressor = new Compressor();
 
-    @Inject
-    public Dispatcher(DispatchConfig dispatchConfig, FS4ResourcePool fs4ResourcePool) {
+    public Dispatcher(DispatchConfig dispatchConfig, FS4ResourcePool fs4ResourcePool, int containerClusterSize) {
         this.client = new RpcClient();
-        this.searchCluster = new SearchCluster(dispatchConfig, fs4ResourcePool);
+        this.searchCluster = new SearchCluster(dispatchConfig, fs4ResourcePool, containerClusterSize);
 
         // Create node rpc connections, indexed by the legacy "partid", which allows us to bridge
         // between fs4 calls (for search) and rpc calls (for summary fetch)
diff --git a/container-search/src/main/java/com/yahoo/search/dispatch/SearchCluster.java b/container-search/src/main/java/com/yahoo/search/dispatch/SearchCluster.java
index bd47c0525ab..db987fde58c 100644
--- a/container-search/src/main/java/com/yahoo/search/dispatch/SearchCluster.java
+++ b/container-search/src/main/java/com/yahoo/search/dispatch/SearchCluster.java
@@ -1,6 +1,7 @@
 package com.yahoo.search.dispatch;
 
 import com.google.common.annotations.Beta;
+import com.google.common.collect.ImmutableCollection;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableMultimap;
@@ -19,6 +20,7 @@ import com.yahoo.prelude.fastsearch.FS4ResourcePool;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executor;
@@ -47,18 +49,20 @@ public class SearchCluster implements NodeManager<SearchCluster.Node> {
     private final ImmutableMap<Integer, Group> groups;
     private final ImmutableMultimap<String, Node> nodesByHost;
     private final ClusterMonitor<Node> clusterMonitor;
+    private final int containerClusterSize;
 
     // Only needed until query requests are moved to rpc
     private final FS4ResourcePool fs4ResourcePool;
 
-    public SearchCluster(DispatchConfig dispatchConfig, FS4ResourcePool fs4ResourcePool) {
-        this(dispatchConfig.minActivedocsPercentage(), toNodes(dispatchConfig), fs4ResourcePool);
+    public SearchCluster(DispatchConfig dispatchConfig, FS4ResourcePool fs4ResourcePool, int containerClusterSize) {
+        this(dispatchConfig.minActivedocsPercentage(), toNodes(dispatchConfig), fs4ResourcePool, containerClusterSize);
     }
     
-    public SearchCluster(double minActivedocsCoverage, List<Node> nodes, FS4ResourcePool fs4ResourcePool) {
+    public SearchCluster(double minActivedocsCoverage, List<Node> nodes, FS4ResourcePool fs4ResourcePool, int containerClusterSize) {
         this.minActivedocsCoveragePercentage = minActivedocsCoverage;
-        size = nodes.size();
+        this.size = nodes.size();
         this.fs4ResourcePool = fs4ResourcePool;
+        this.containerClusterSize = containerClusterSize;
         
         // Create groups
         ImmutableMap.Builder<Integer, Group> groupsBuilder = new ImmutableMap.Builder<>();
@@ -98,6 +102,37 @@ public class SearchCluster implements NodeManager<SearchCluster.Node> {
      */
     public ImmutableMultimap<String, Node> nodesByHost() { return nodesByHost; }
 
+    /** Whether direct dispatch (bypassing fdispatch) should be used when dispatching queries from the given hostname */
+    public Optional<Node> dispatchDirectlyFrom(String selfHostname) {
+        // A search node in the search cluster in question is configured on the same host as the currently running container.
+        // It has all the data <==> No other nodes in the search cluster have the same group id as this node.
+        //         That local search node responds.
+        // The search cluster to be searched has at least as many nodes as the container cluster we're running in.
+        ImmutableCollection<Node> localSearchNodes = nodesByHost().get(selfHostname);
+        // Only use direct dispatch if we have exactly 1 search node on the same machine:
+        if (localSearchNodes.size() != 1) return Optional.empty();
+
+        SearchCluster.Node localSearchNode = localSearchNodes.iterator().next();
+        SearchCluster.Group localSearchGroup = groups().get(localSearchNode.group());
+
+        // Only use direct dispatch if the local search node has the entire corpus
+        if (localSearchGroup.nodes().size() != 1) return Optional.empty();
+
+        // Only use direct dispatch if this container cluster has at least as many nodes as the search cluster
+        // to avoid load skew/preserve fanout in the case where a subset of the search nodes are also containers.
+        // This disregards the case where the search and container clusters are partially overlapping. 
+        // Such configurations produce skewed load in any case.
+        if (containerClusterSize < size()) return Optional.empty();
+
+        // Only use direct dispatch if the local group has sufficient coverage
+        if ( ! localSearchGroup.hasSufficientCoverage()) return Optional.empty();
+
+        // Only use direct dispatch if the local search node is up
+        if ( ! localSearchNode.isWorking()) return Optional.empty();
+
+        return Optional.of(localSearchNode);
+    }
+
     /** Used by the cluster monitor to manage node status */
     @Override
     public void working(Node node) { node.setWorking(true); }