Refactoring only: Move direct dispatch decision to SearchCluster

7 files changed, 71 insertions, 75 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); }
diff --git a/container-search/src/test/java/com/yahoo/prelude/fastsearch/test/FastSearcherTestCase.java b/container-search/src/test/java/com/yahoo/prelude/fastsearch/test/FastSearcherTestCase.java
index b052622ccd3..8dd56ad15a1 100644
--- a/container-search/src/test/java/com/yahoo/prelude/fastsearch/test/FastSearcherTestCase.java
+++ b/container-search/src/test/java/com/yahoo/prelude/fastsearch/test/FastSearcherTestCase.java
@@ -63,8 +63,7 @@ public class FastSearcherTestCase {
                                                      new SummaryParameters(null),
                                                      new ClusterParams("testhittype"),
                                                      new CacheParams(100, 1e64),
-                                                     documentdbInfoConfig,
-                                                     1);
+                                                     documentdbInfoConfig);
 
         MockFSChannel.setEmptyDocsums(false);
 
@@ -85,8 +84,7 @@ public class FastSearcherTestCase {
                                                      new SummaryParameters(null),
                                                      new ClusterParams("testhittype"),
                                                      new CacheParams(100, 1e64),
-                                                     documentdbInfoConfig,
-                                                     1);
+                                                     documentdbInfoConfig);
 
         String query = "?junkparam=ignored";
         Result result = doSearch(fastSearcher,new Query(query), 0, 10);
@@ -109,8 +107,7 @@ public class FastSearcherTestCase {
                                                      new SummaryParameters(null),
                                                      new ClusterParams("testhittype"),
                                                      new CacheParams(100, 1e64), 
-                                                     documentdbConfigWithOneDb,
-                                                     1);
+                                                     documentdbConfigWithOneDb);
 
         Query query = new Query("?query=foo&model.restrict=testDb");
         query.prepare();
@@ -298,8 +295,7 @@ public class FastSearcherTestCase {
                                 new SummaryParameters(null),
                                 new ClusterParams("testhittype"), 
                                 new CacheParams(100, 1e64), 
-                                config,
-                                1);
+                                config);
     }
 
     @Ignore
@@ -312,8 +308,7 @@ public class FastSearcherTestCase {
                                                      new SummaryParameters(null),
                                                      new ClusterParams("testhittype"),
                                                      new CacheParams(100, 1e64),
-                                                     documentdbInfoConfig,
-                                                     1);
+                                                     documentdbInfoConfig);
 
         CacheControl c = fastSearcher.getCacheControl();
 
@@ -356,8 +351,7 @@ public class FastSearcherTestCase {
                                                      new SummaryParameters(null),
                                                      new ClusterParams("testhittype"),
                                                      new CacheParams(100, 1e64),
-                                                     documentdbInfoConfig,
-                                                     1);
+                                                     documentdbInfoConfig);
 
         Result result = doSearch(fastSearcher,new Query("?query=ignored"), 0, 2);
         result = doSearch(fastSearcher,new Query("?query=ignored"), 1, 1);
@@ -392,8 +386,7 @@ public class FastSearcherTestCase {
                                                      new SummaryParameters(null),
                                                      new ClusterParams("testhittype"),
                                                      new CacheParams(0, 0.0d),
-                                                     documentdbInfoConfig,
-                                                     1);
+                                                     documentdbInfoConfig);
         server.dispatch.packetData = BackendTestCase.PONG;
         Chain<Searcher> chain = new Chain<>(fastSearcher);
         Execution e = new Execution(chain, Execution.Context.createContextStub());
diff --git a/container-search/src/test/java/com/yahoo/prelude/fastsearch/test/FastSearcherTester.java b/container-search/src/test/java/com/yahoo/prelude/fastsearch/test/FastSearcherTester.java
index e4d7e5fa076..b6735acbe28 100644
--- a/container-search/src/test/java/com/yahoo/prelude/fastsearch/test/FastSearcherTester.java
+++ b/container-search/src/test/java/com/yahoo/prelude/fastsearch/test/FastSearcherTester.java
@@ -32,25 +32,24 @@ class FastSearcherTester {
     private final FastSearcher fastSearcher;
     private final MockDispatcher mockDispatcher;
 
-    public FastSearcherTester(int containerNodeCount, SearchCluster.Node searchNode) {
-        this(containerNodeCount, Collections.singletonList(searchNode));
+    public FastSearcherTester(int containerClusterSize, SearchCluster.Node searchNode) {
+        this(containerClusterSize, Collections.singletonList(searchNode));
     }
 
-    public FastSearcherTester(int containerNodeCount, String... hostAndPortAndGroupStrings) {
-        this(containerNodeCount, toNodes(hostAndPortAndGroupStrings));
+    public FastSearcherTester(int containerClusterSize, String... hostAndPortAndGroupStrings) {
+        this(containerClusterSize, toNodes(hostAndPortAndGroupStrings));
     }
 
-    public FastSearcherTester(int containerNodeCount, List<SearchCluster.Node> searchNodes) {
+    public FastSearcherTester(int containerClusterSize, List<SearchCluster.Node> searchNodes) {
         mockFS4ResourcePool = new MockFS4ResourcePool();
-        mockDispatcher = new MockDispatcher(searchNodes, mockFS4ResourcePool);
+        mockDispatcher = new MockDispatcher(searchNodes, mockFS4ResourcePool, containerClusterSize);
         fastSearcher = new FastSearcher(new MockBackend(selfHostname, MockFSChannel::new),
                                         mockFS4ResourcePool,
                                         mockDispatcher,
                                         new SummaryParameters(null),
                                         new ClusterParams("testhittype"),
                                         new CacheParams(100, 1e64),
-                                        new DocumentdbInfoConfig(new DocumentdbInfoConfig.Builder()),
-                                        containerNodeCount);
+                                        new DocumentdbInfoConfig(new DocumentdbInfoConfig.Builder()));
     }
 
     private static List<SearchCluster.Node> toNodes(String... hostAndPortAndGroupStrings) {
diff --git a/container-search/src/test/java/com/yahoo/prelude/fastsearch/test/MockDispatcher.java b/container-search/src/test/java/com/yahoo/prelude/fastsearch/test/MockDispatcher.java
index b4a631ff474..9f2abe505d6 100644
--- a/container-search/src/test/java/com/yahoo/prelude/fastsearch/test/MockDispatcher.java
+++ b/container-search/src/test/java/com/yahoo/prelude/fastsearch/test/MockDispatcher.java
@@ -11,11 +11,11 @@ import java.util.List;
 class MockDispatcher extends Dispatcher {
 
     public MockDispatcher(List<SearchCluster.Node> nodes) {
-        super(toDispatchConfig(nodes), new FS4ResourcePool(1));
+        super(toDispatchConfig(nodes), new FS4ResourcePool(1), 1);
     }
 
-    public MockDispatcher(List<SearchCluster.Node> nodes, FS4ResourcePool fs4ResourcePool) {
-        super(toDispatchConfig(nodes), fs4ResourcePool);
+    public MockDispatcher(List<SearchCluster.Node> nodes, FS4ResourcePool fs4ResourcePool, int containerClusterSize) {
+        super(toDispatchConfig(nodes), fs4ResourcePool, containerClusterSize);
     }
 
     private static DispatchConfig toDispatchConfig(List<SearchCluster.Node> nodes) {