Don't direct dispatch if the containerCluster.size<searchCluster.size

7 files changed, 72 insertions, 31 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 288dc229999..8b224dd6d51 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
@@ -1,6 +1,7 @@
 // Copyright 2016 Yahoo Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
 package com.yahoo.prelude.cluster;
 
+import com.yahoo.cloud.config.ClusterInfoConfig;
 import com.yahoo.collections.Tuple2;
 import com.yahoo.component.ComponentId;
 import com.yahoo.component.chain.Chain;
@@ -100,6 +101,7 @@ public class ClusterSearcher extends Searcher {
                            LegacyEmulationConfig emulationConfig,
                            QrMonitorConfig monitorConfig,
                            DispatchConfig dispatchConfig,
+                           ClusterInfoConfig clusterInfoConfig,
                            Statistics manager,
                            FS4ResourcePool fs4ResourcePool,
                            VipStatus vipStatus) {
@@ -126,6 +128,8 @@ public class ClusterSearcher extends Searcher {
         SummaryParameters docSumParams = new SummaryParameters(qrsConfig
                 .com().yahoo().prelude().fastsearch().FastSearcher().docsum()
                 .defaultclass());
+        
+        int containerClusterSize = clusterInfoConfig.nodeCount();
 
         for (DocumentdbInfoConfig.Documentdb docDb : documentDbConfig.documentdb()) {
             String docTypeName = docDb.name();
@@ -144,13 +148,13 @@ public class ClusterSearcher extends Searcher {
             addBackendSearcher(searcher);
             gotExpectedBackend = true;
         } else {
-            for (int i = 0; i < searchClusterConfig.dispatcher().size(); i++) {
-                Backend b = createBackend(searchClusterConfig.dispatcher(i));
+            for (int dispatcherIndex = 0; dispatcherIndex < searchClusterConfig.dispatcher().size(); dispatcherIndex++) {
+                Backend b = createBackend(searchClusterConfig.dispatcher(dispatcherIndex));
                 FastSearcher searcher = searchDispatch(searchClusterIndex, fs4ResourcePool, 
                                                        searchClusterConfig, cacheParams, emulationConfig, docSumParams,
-                                                       documentDbConfig, b, dispatcher, i);
+                                                       documentDbConfig, b, dispatcher, dispatcherIndex, containerClusterSize);
                 try {
-                    searcher.setLocalDispatching( ! isRemote(searchClusterConfig.dispatcher(i).host()));
+                    searcher.setLocalDispatching( ! isRemote(searchClusterConfig.dispatcher(dispatcherIndex).host()));
                 } catch (UnknownHostException e) {
                     throw new RuntimeException(e);
                 }
@@ -217,11 +221,14 @@ public class ClusterSearcher extends Searcher {
                                                DocumentdbInfoConfig documentdbInfoConfig,
                                                Backend backend,
                                                Dispatcher dispatcher,
-                                               int i) {
+                                               int dispatcherIndex,
+                                               int containerClusterSize) {
         ClusterParams clusterParams = makeClusterParams(searchclusterIndex,
                                                         searchClusterConfig,
-                                                        emulConfig, i);
-        return new FastSearcher(backend, fs4ResourcePool, dispatcher, docSumParams, clusterParams, cacheParams, documentdbInfoConfig);
+                                                        emulConfig, 
+                                                        dispatcherIndex);
+        return new FastSearcher(backend, fs4ResourcePool, dispatcher, docSumParams, clusterParams, cacheParams, 
+                                documentdbInfoConfig, containerClusterSize);
     }
 
     private static VdsStreamingSearcher vdsCluster(int searchclusterIndex,
diff --git a/container-search/src/main/java/com/yahoo/prelude/cluster/dispatchprototype/DispatchClusterSearcher.java b/container-search/src/main/java/com/yahoo/prelude/cluster/dispatchprototype/DispatchClusterSearcher.java
index a05d222abde..dbcee3234ea 100644
--- a/container-search/src/main/java/com/yahoo/prelude/cluster/dispatchprototype/DispatchClusterSearcher.java
+++ b/container-search/src/main/java/com/yahoo/prelude/cluster/dispatchprototype/DispatchClusterSearcher.java
@@ -2,6 +2,7 @@
 package com.yahoo.prelude.cluster.dispatchprototype;
 
 import com.google.common.annotations.Beta;
+import com.yahoo.cloud.config.ClusterInfoConfig;
 import com.yahoo.component.ComponentId;
 import com.yahoo.component.chain.dependencies.After;
 import com.yahoo.component.provider.ComponentRegistry;
@@ -46,6 +47,7 @@ public class DispatchClusterSearcher extends Searcher {
             final LegacyEmulationConfig emulationConfig,
             final QrMonitorConfig monitorConfig,
             final DispatchConfig dispatchConfig,
+            final ClusterInfoConfig clusterInfoConfig,
             final Statistics manager,
             final FS4ResourcePool listeners,
             final ComponentRegistry<ClusterSearcher> otherClusterSearchers,
@@ -62,12 +64,13 @@ public class DispatchClusterSearcher extends Searcher {
                 emulationConfig,
                 monitorConfig,
                 dispatchConfig,
+                clusterInfoConfig,
                 manager,
                 listeners,
                 vipStatus);
 
-        //Prevent the ClusterSearcher(s) implicitly set up by the model from warning that it can't contact
-        //the c++ TLD when we disable it in the system test.
+        // Prevent the ClusterSearcher(s) implicitly set up by the model from warning that it can't contact
+        // the c++ TLD when we disable it in the system test.
         otherClusterSearchers.allComponents().stream()
                 .forEach(ClusterSearcher::deconstruct);
     }
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 bf004b47b82..61835492e52 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
@@ -1,7 +1,6 @@
 // Copyright 2016 Yahoo Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
 package com.yahoo.prelude.fastsearch;
 
-import java.util.List;
 import java.util.Optional;
 
 import com.google.common.collect.ImmutableCollection;
@@ -78,7 +77,8 @@ public class FastSearcher extends VespaBackEndSearcher {
     private final FS4ResourcePool fs4ResourcePool;
     
     private final String selfHostname;
-
+    private final int containerClusterSize;
+    
     /**
      * Creates a Fastsearcher.
      *
@@ -94,15 +94,18 @@ 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) {
+                        CacheParams cacheParams, DocumentdbInfoConfig documentdbInfoConfig,
+                        int containerClusterSize) {
         init(docSumParams, clusterParams, cacheParams, documentdbInfoConfig);
         this.dispatchBackend = dispatchBackend;
         this.fs4ResourcePool = fs4ResourcePool;
         this.dispatcher = dispatcher;
         this.selfHostname = LinuxInetAddress.getLocalHost().getHostName();
+        this.containerClusterSize = containerClusterSize;
     }
 
     /** Clears the packet cache if the received timestamp is older than our timestamp */
@@ -266,17 +269,21 @@ public class FastSearcher extends VespaBackEndSearcher {
         // 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.cluster().nodesByHost().get(selfHostname);
+        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();
         // Only use direct dispatch if the local search node has the entire corpus
-        if (dispatcher.cluster().groups().get(localSearchNode.group()).nodes().size() != 1) return dispatchBackend;
+        if (dispatcher.searchCluster().groups().get(localSearchNode.group()).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;
 
         // TODO: Only use direct dispatch if the local search node is up
-        // TODO: Only use direct dispatch if the search cluster has at least as many nodes as this container cluster
-        // (to avoid load skew/preserve fanout in the case where a subset of the search nodes are also containers)
 
         query.trace(false, 2, "Dispatching directly to ", localSearchNode);
         return fs4ResourcePool.getBackend(localSearchNode.hostname(), localSearchNode.port());
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 e9cc7b5eafd..e1b9f717d61 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
@@ -41,7 +41,7 @@ public class Dispatcher extends AbstractComponent {
     private final Client client;
 
     /** A model of the search cluster this dispatches to */
-    private final SearchCluster cluster;
+    private final SearchCluster searchCluster;
     
     /** Connections to the search nodes this talks to, indexed by node id ("partid") */
     private final ImmutableMap<Integer, Client.NodeConnection> nodeConnections;
@@ -51,7 +51,7 @@ public class Dispatcher extends AbstractComponent {
     @Inject
     public Dispatcher(DispatchConfig dispatchConfig) {
         this.client = new RpcClient();
-        this.cluster = new SearchCluster(dispatchConfig);
+        this.searchCluster = new SearchCluster(dispatchConfig);
 
         // 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)
@@ -64,13 +64,13 @@ public class Dispatcher extends AbstractComponent {
 
     /** For testing */
     public Dispatcher(Map<Integer, Client.NodeConnection> nodeConnections, Client client) {
-        this.cluster = null;
+        this.searchCluster = null;
         this.nodeConnections = ImmutableMap.copyOf(nodeConnections);
         this.client = client;
     }
     
     /** Returns the search cluster this dispatches to */
-    public SearchCluster cluster() { return cluster; }
+    public SearchCluster searchCluster() { return searchCluster; }
 
     /** Fills the given summary class by sending RPC requests to the right search nodes */
     public void fill(Result result, String summaryClass, CompressionType compression) {
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 374d36bb161..4f1b1ebfec0 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
@@ -16,6 +16,7 @@ import java.util.stream.Collectors;
  */
 public class SearchCluster {
 
+    private final int size;
     private final ImmutableMap<Integer, Group> groups;
     private final ImmutableMultimap<String, Node> nodesByHost;
 
@@ -24,6 +25,8 @@ public class SearchCluster {
     }
     
     public SearchCluster(List<Node> nodes) {
+        size = nodes.size();
+        
         // Create groups
         ImmutableMap.Builder<Integer, Group> groupsBuilder = new ImmutableMap.Builder<>();
         for (Map.Entry<Integer, List<Node>> group : nodes.stream().collect(Collectors.groupingBy(Node::group)).entrySet())
@@ -44,6 +47,9 @@ public class SearchCluster {
         return nodesBuilder.build();
     }
     
+    /** Returns the number of nodes in this cluster (across all groups) */
+    public int size() { return size; }
+    
     /** Returns the groups of this cluster as an immutable map indexed by group id */
     public ImmutableMap<Integer, Group> groups() { return groups; }
 
diff --git a/container-search/src/test/java/com/yahoo/prelude/cluster/ClusterSearcherTestCase.java b/container-search/src/test/java/com/yahoo/prelude/cluster/ClusterSearcherTestCase.java
index 7ff2b5ffb45..c3d11e1c929 100644
--- a/container-search/src/test/java/com/yahoo/prelude/cluster/ClusterSearcherTestCase.java
+++ b/container-search/src/test/java/com/yahoo/prelude/cluster/ClusterSearcherTestCase.java
@@ -1,6 +1,7 @@
 // Copyright 2016 Yahoo Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
 package com.yahoo.prelude.cluster;
 
+import com.yahoo.cloud.config.ClusterInfoConfig;
 import com.yahoo.component.ComponentId;
 import com.yahoo.container.QrSearchersConfig;
 import com.yahoo.container.search.Fs4Config;
@@ -523,14 +524,23 @@ public class ClusterSearcherTestCase extends junit.framework.TestCase {
                 documentdb(new DocumentdbInfoConfig.Documentdb.Builder().name("type1")));
         LegacyEmulationConfig emulationCfg = new LegacyEmulationConfig(new LegacyEmulationConfig.Builder());
         QrMonitorConfig monitorCfg = new QrMonitorConfig(new QrMonitorConfig.Builder());
-        Statistics statMgr = Statistics.nullImplementation;
+        Statistics statistics = Statistics.nullImplementation;
         Fs4Config fs4Cfg = new Fs4Config(new Fs4Config.Builder());
-        FS4ResourcePool listeners = new FS4ResourcePool(fs4Cfg);
-        ClusterSearcher searcher = new ClusterSearcher(id,
-                qrsCfg, clusterCfg, documentDbCfg, emulationCfg, monitorCfg, new DispatchConfig(new DispatchConfig.Builder()), statMgr, listeners, new VipStatus());
+        FS4ResourcePool fs4ResourcePool = new FS4ResourcePool(fs4Cfg);
+        ClusterSearcher searcher = new ClusterSearcher(id, qrsCfg, clusterCfg, documentDbCfg, emulationCfg, monitorCfg, 
+                                                       new DispatchConfig(new DispatchConfig.Builder()), 
+                                                       createClusterInfoConfig(),
+                                                       statistics, fs4ResourcePool, new VipStatus());
         return searcher;
     }
 
+    private static ClusterInfoConfig createClusterInfoConfig() {
+        ClusterInfoConfig.Builder clusterInfoConfigBuilder = new ClusterInfoConfig.Builder();
+        clusterInfoConfigBuilder.clusterId("containerCluster1");
+        clusterInfoConfigBuilder.nodeCount(1);
+        return new ClusterInfoConfig(clusterInfoConfigBuilder);
+    }
+    
     private static class QueryTimeoutFixture {
         ClusterSearcher searcher;
         Execution exec;
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 927faee1e5c..e4cbff8def9 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
@@ -65,7 +65,8 @@ public class FastSearcherTestCase {
                                                      new SummaryParameters(null),
                                                      new ClusterParams("testhittype"),
                                                      new CacheParams(100, 1e64),
-                                                     documentdbInfoConfig);
+                                                     documentdbInfoConfig,
+                                                     1);
 
         MockFSChannel.setEmptyDocsums(false);
 
@@ -86,7 +87,8 @@ public class FastSearcherTestCase {
                                                      new SummaryParameters(null),
                                                      new ClusterParams("testhittype"),
                                                      new CacheParams(100, 1e64),
-                                                     documentdbInfoConfig);
+                                                     documentdbInfoConfig,
+                                                     1);
 
         String query = "?junkparam=ignored";
         Result result = doSearch(fastSearcher,new Query(query), 0, 10);
@@ -109,7 +111,8 @@ public class FastSearcherTestCase {
                                                      new SummaryParameters(null),
                                                      new ClusterParams("testhittype"),
                                                      new CacheParams(100, 1e64), 
-                                                     documentdbConfigWithOneDb);
+                                                     documentdbConfigWithOneDb,
+                                                     1);
 
         Query query = new Query("?query=foo&model.restrict=testDb");
         query.prepare();
@@ -298,7 +301,9 @@ public class FastSearcherTestCase {
                                 new MockDispatcher(), 
                                 new SummaryParameters(null),
                                 new ClusterParams("testhittype"), 
-                                new CacheParams(100, 1e64), config);
+                                new CacheParams(100, 1e64), 
+                                config,
+                                1);
     }
 
     @Ignore
@@ -311,7 +316,8 @@ public class FastSearcherTestCase {
                                                      new SummaryParameters(null),
                                                      new ClusterParams("testhittype"),
                                                      new CacheParams(100, 1e64),
-                                                     documentdbInfoConfig);
+                                                     documentdbInfoConfig,
+                                                     1);
 
         CacheControl c = fastSearcher.getCacheControl();
 
@@ -354,7 +360,8 @@ public class FastSearcherTestCase {
                                                      new SummaryParameters(null),
                                                      new ClusterParams("testhittype"),
                                                      new CacheParams(100, 1e64),
-                                                     documentdbInfoConfig);
+                                                     documentdbInfoConfig,
+                                                     1);
 
         Result result = doSearch(fastSearcher,new Query("?query=ignored"), 0, 2);
         result = doSearch(fastSearcher,new Query("?query=ignored"), 1, 1);
@@ -390,7 +397,8 @@ public class FastSearcherTestCase {
                                                      new SummaryParameters(null),
                                                      new ClusterParams("testhittype"),
                                                      new CacheParams(0, 0.0d),
-                                                     documentdbInfoConfig);
+                                                     documentdbInfoConfig,
+                                                     1);
         server.dispatch.packetData = BackendTestCase.PONG;
         Chain<Searcher> chain = new Chain<>(fastSearcher);
         Execution e = new Execution(chain, Execution.Context.createContextStub());