diff options
8 files changed, 55 insertions, 57 deletions
diff --git a/container-search/src/main/java/com/yahoo/search/cluster/ClusterMonitor.java b/container-search/src/main/java/com/yahoo/search/cluster/ClusterMonitor.java index d871a256a62..74a007d35d3 100644 --- a/container-search/src/main/java/com/yahoo/search/cluster/ClusterMonitor.java +++ b/container-search/src/main/java/com/yahoo/search/cluster/ClusterMonitor.java @@ -110,7 +110,7 @@ public class ClusterMonitor<T> { public void ping(Executor executor) { for (Iterator<BaseNodeMonitor<T>> i = nodeMonitorIterator(); i.hasNext() && !closed.get(); ) { BaseNodeMonitor<T> monitor= i.next(); - nodeManager.ping(monitor.getNode(), executor); // Cause call to failed or responded + nodeManager.ping(this, monitor.getNode(), executor); // Cause call to failed or responded } if (closed.get()) return; // Do nothing to change state if close has started. nodeManager.pingIterationCompleted(); diff --git a/container-search/src/main/java/com/yahoo/search/cluster/ClusterSearcher.java b/container-search/src/main/java/com/yahoo/search/cluster/ClusterSearcher.java index 20f56c86f7b..2d05168731a 100644 --- a/container-search/src/main/java/com/yahoo/search/cluster/ClusterSearcher.java +++ b/container-search/src/main/java/com/yahoo/search/cluster/ClusterSearcher.java @@ -58,7 +58,7 @@ public abstract class ClusterSearcher<T> extends PingableSearcher implements Nod this(id, connections, hasher, internal, true); } - public ClusterSearcher(ComponentId id, List<T> connections, Hasher<T> hasher, boolean internal, boolean startPingThread) { + protected ClusterSearcher(ComponentId id, List<T> connections, Hasher<T> hasher, boolean internal, boolean startPingThread) { super(id); this.hasher = hasher; this.monitor = new ClusterMonitor<>(this, startPingThread); @@ -70,7 +70,7 @@ public abstract class ClusterSearcher<T> extends PingableSearcher implements Nod /** Pinging a node, called from ClusterMonitor */ @Override - public final void ping(T p, Executor executor) { + public final void ping(ClusterMonitor<T> clusterMonitor, T p, Executor executor) { log(LogLevel.FINE, "Sending ping to: ", p); Pinger pinger = new Pinger(p); FutureTask<Pong> future = new FutureTask<>(pinger); @@ -80,7 +80,7 @@ public abstract class ClusterSearcher<T> extends PingableSearcher implements Nod Throwable logThrowable = null; try { - pong = future.get(monitor.getConfiguration().getFailLimit(), TimeUnit.MILLISECONDS); + pong = future.get(clusterMonitor.getConfiguration().getFailLimit(), TimeUnit.MILLISECONDS); } catch (InterruptedException e) { pong = new Pong(ErrorMessage.createUnspecifiedError("Ping was interrupted: " + p)); logThrowable = e; @@ -96,10 +96,10 @@ public abstract class ClusterSearcher<T> extends PingableSearcher implements Nod future.cancel(true); if (pong.badResponse()) { - monitor.failed(p, pong.error().get()); + clusterMonitor.failed(p, pong.error().get()); log(LogLevel.FINE, "Failed ping - ", pong); } else { - monitor.responded(p); + clusterMonitor.responded(p); log(LogLevel.FINE, "Answered ping - ", p); } diff --git a/container-search/src/main/java/com/yahoo/search/cluster/NodeManager.java b/container-search/src/main/java/com/yahoo/search/cluster/NodeManager.java index 9b20139e3c5..ffd26e3088d 100644 --- a/container-search/src/main/java/com/yahoo/search/cluster/NodeManager.java +++ b/container-search/src/main/java/com/yahoo/search/cluster/NodeManager.java @@ -19,9 +19,21 @@ public interface NodeManager<T> { /** * Called when a node should be pinged. - * This *must* lead to either a call to NodeMonitor.failed or NodeMonitor.responded + * This *must* lead to either a call to NodeMonitor.failed or NodeMonitor.responded + * @deprecated Use ping(ClusterMonitor<T> clusterMonitor, T node, Executor executor) instead. */ - void ping(T node, Executor executor); + @Deprecated + default void ping(T node, Executor executor) { + throw new IllegalStateException("If you have not overrriden ping(ClusterMonitor<T> clusterMonitor, T node, Executor executor), you should at least have overriden this method."); + } + + /** + * Called when a node should be pinged. + * This *must* lead to either a call to ClusterMonitor.failed or ClusterMonitor.responded + */ + default void ping(ClusterMonitor<T> clusterMonitor, T node, Executor executor) { + ping(node, executor); + } /** Called right after a ping has been issued to each node. This default implementation does nothing. */ default void pingIterationCompleted() {} 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 1f31d807024..91bd5c6da11 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 @@ -11,6 +11,7 @@ import com.yahoo.prelude.fastsearch.VespaBackEndSearcher; import com.yahoo.processing.request.CompoundName; import com.yahoo.search.Query; import com.yahoo.search.Result; +import com.yahoo.search.cluster.ClusterMonitor; import com.yahoo.search.dispatch.SearchPath.InvalidSearchPathException; import com.yahoo.search.dispatch.rpc.RpcInvokerFactory; import com.yahoo.search.dispatch.rpc.RpcPingFactory; @@ -59,6 +60,7 @@ public class Dispatcher extends AbstractComponent { /** A model of the search cluster this dispatches to */ private final SearchCluster searchCluster; + private final ClusterMonitor clusterMonitor; private final LoadBalancer loadBalancer; @@ -97,20 +99,19 @@ public class Dispatcher extends AbstractComponent { ClusterInfoConfig clusterInfoConfig, VipStatus vipStatus, Metric metric) { - this(resourcePool, new SearchCluster(clusterId.stringValue(), dispatchConfig, clusterInfoConfig.nodeCount(), vipStatus, new RpcPingFactory(resourcePool)), + this(resourcePool, new SearchCluster(clusterId.stringValue(), dispatchConfig, clusterInfoConfig.nodeCount(), + vipStatus, new RpcPingFactory(resourcePool)), dispatchConfig, metric); } private Dispatcher(RpcResourcePool resourcePool, SearchCluster searchCluster, DispatchConfig dispatchConfig, Metric metric) { - this(searchCluster, - dispatchConfig, - new RpcInvokerFactory(resourcePool, searchCluster), - metric); + this(new ClusterMonitor<>(searchCluster, true), searchCluster, dispatchConfig, new RpcInvokerFactory(resourcePool, searchCluster), metric); } /* Protected for simple mocking in tests. Beware that searchCluster is shutdown on in deconstruct() */ - protected Dispatcher(SearchCluster searchCluster, + protected Dispatcher(ClusterMonitor clusterMonitor, + SearchCluster searchCluster, DispatchConfig dispatchConfig, InvokerFactory invokerFactory, Metric metric) { @@ -118,14 +119,14 @@ public class Dispatcher extends AbstractComponent { throw new IllegalArgumentException(searchCluster + " is configured with multilevel dispatch, but this is not supported"); this.searchCluster = searchCluster; + this.clusterMonitor = clusterMonitor; this.loadBalancer = new LoadBalancer(searchCluster, dispatchConfig.distributionPolicy() == DispatchConfig.DistributionPolicy.ROUNDROBIN); this.invokerFactory = invokerFactory; this.metric = metric; this.metricContext = metric.createContext(null); this.maxHitsPerNode = dispatchConfig.maxHitsPerNode(); - - searchCluster.startClusterMonitoring(true); + searchCluster.addMonitoring(clusterMonitor); try { while ( ! searchCluster.hasInformationAboutAllNodes()) { Thread.sleep(1); @@ -140,8 +141,8 @@ public class Dispatcher extends AbstractComponent { @Override public void deconstruct() { - /* The seach cluster must be shutdown first as it uses the invokerfactory. */ - searchCluster.shutDown(); + /* The clustermonitor must be shutdown first as it uses the invokerfactory through the searchCluster. */ + clusterMonitor.shutdown(); invokerFactory.release(); } diff --git a/container-search/src/main/java/com/yahoo/search/dispatch/searchcluster/SearchCluster.java b/container-search/src/main/java/com/yahoo/search/dispatch/searchcluster/SearchCluster.java index 54da33ac6dc..d462479226a 100644 --- a/container-search/src/main/java/com/yahoo/search/dispatch/searchcluster/SearchCluster.java +++ b/container-search/src/main/java/com/yahoo/search/dispatch/searchcluster/SearchCluster.java @@ -36,7 +36,6 @@ public class SearchCluster implements NodeManager<Node> { private final ImmutableMap<Integer, Group> groups; private final ImmutableMultimap<String, Node> nodesByHost; private final ImmutableList<Group> orderedGroups; - private final ClusterMonitor<Node> clusterMonitor; private final VipStatus vipStatus; private final PingFactory pingFactory; private long nextLogTime = 0; @@ -51,7 +50,8 @@ public class SearchCluster implements NodeManager<Node> { */ private final Optional<Node> localCorpusDispatchTarget; - public SearchCluster(String clusterId, DispatchConfig dispatchConfig, int containerClusterSize, VipStatus vipStatus, PingFactory pingFactory) { + public SearchCluster(String clusterId, DispatchConfig dispatchConfig, int containerClusterSize, + VipStatus vipStatus, PingFactory pingFactory) { this.clusterId = clusterId; this.dispatchConfig = dispatchConfig; this.vipStatus = vipStatus; @@ -78,30 +78,18 @@ public class SearchCluster implements NodeManager<Node> { this.nodesByHost = nodesByHostBuilder.build(); this.localCorpusDispatchTarget = findLocalCorpusDispatchTarget(HostName.getLocalhost(), - size, - containerClusterSize, - nodesByHost, - groups); - - this.clusterMonitor = new ClusterMonitor<>(this, false); + size, + containerClusterSize, + nodesByHost, + groups); + } + public void addMonitoring(ClusterMonitor clusterMonitor) { for (var group : orderedGroups) { for (var node : group.nodes()) clusterMonitor.add(node, true); } } - public void shutDown() { - clusterMonitor.shutdown(); - } - - public void startClusterMonitoring(boolean startPingThread) { - if (startPingThread) { - clusterMonitor.start(); - } - } - - ClusterMonitor<Node> clusterMonitor() { return clusterMonitor; } - private static Optional<Node> findLocalCorpusDispatchTarget(String selfHostname, int searchClusterSize, int containerClusterSize, @@ -280,7 +268,7 @@ public class SearchCluster implements NodeManager<Node> { /** Used by the cluster monitor to manage node status */ @Override - public void ping(Node node, Executor executor) { + public void ping(ClusterMonitor clusterMonitor, Node node, Executor executor) { if (pingFactory == null) return; // not initialized yet Pinger pinger = pingFactory.createPinger(node, clusterMonitor, new PongCallback(node, clusterMonitor)); 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 4b65e26c6c2..d5e43fba92d 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 @@ -2,6 +2,7 @@ package com.yahoo.prelude.fastsearch.test; import com.yahoo.container.handler.VipStatus; +import com.yahoo.search.cluster.ClusterMonitor; import com.yahoo.search.dispatch.Dispatcher; import com.yahoo.search.dispatch.rpc.RpcInvokerFactory; import com.yahoo.search.dispatch.rpc.RpcPingFactory; @@ -32,7 +33,7 @@ class MockDispatcher extends Dispatcher { } private MockDispatcher(SearchCluster searchCluster, DispatchConfig dispatchConfig, RpcInvokerFactory invokerFactory) { - super(searchCluster, dispatchConfig, invokerFactory, new MockMetric()); + super(new ClusterMonitor<>(searchCluster, true), searchCluster, dispatchConfig, invokerFactory, new MockMetric()); } static DispatchConfig toDispatchConfig(List<Node> nodes) { diff --git a/container-search/src/test/java/com/yahoo/search/dispatch/DispatcherTest.java b/container-search/src/test/java/com/yahoo/search/dispatch/DispatcherTest.java index dad73b9b8a9..5433a28dd6e 100644 --- a/container-search/src/test/java/com/yahoo/search/dispatch/DispatcherTest.java +++ b/container-search/src/test/java/com/yahoo/search/dispatch/DispatcherTest.java @@ -38,9 +38,10 @@ public class DispatcherTest { assertEquals(2, nodes.get(0).key()); return true; }); - Dispatcher disp = new Dispatcher(cl, createDispatchConfig(), invokerFactory, new MockMetric()); + Dispatcher disp = new Dispatcher(new ClusterMonitor(cl, false), cl, createDispatchConfig(), invokerFactory, new MockMetric()); SearchInvoker invoker = disp.getSearchInvoker(q, null); invokerFactory.verifyAllEventsProcessed(); + disp.deconstruct(); } @Test @@ -52,9 +53,10 @@ public class DispatcherTest { } }; MockInvokerFactory invokerFactory = new MockInvokerFactory(cl, (n, a) -> true); - Dispatcher disp = new Dispatcher(cl, createDispatchConfig(), invokerFactory, new MockMetric()); + Dispatcher disp = new Dispatcher(new ClusterMonitor(cl, false), cl, createDispatchConfig(), invokerFactory, new MockMetric()); SearchInvoker invoker = disp.getSearchInvoker(new Query(), null); invokerFactory.verifyAllEventsProcessed(); + disp.deconstruct(); } @Test @@ -68,9 +70,10 @@ public class DispatcherTest { assertTrue(acceptIncompleteCoverage); return true; }); - Dispatcher disp = new Dispatcher(cl, createDispatchConfig(), invokerFactory, new MockMetric()); + Dispatcher disp = new Dispatcher(new ClusterMonitor(cl, false), cl, createDispatchConfig(), invokerFactory, new MockMetric()); SearchInvoker invoker = disp.getSearchInvoker(new Query(), null); invokerFactory.verifyAllEventsProcessed(); + disp.deconstruct(); } @Test @@ -79,8 +82,9 @@ public class DispatcherTest { SearchCluster cl = new MockSearchCluster("1", 2, 1); MockInvokerFactory invokerFactory = new MockInvokerFactory(cl, (n, a) -> false, (n, a) -> false); - Dispatcher disp = new Dispatcher(cl, createDispatchConfig(), invokerFactory, new MockMetric()); + Dispatcher disp = new Dispatcher(new ClusterMonitor(cl, false), cl, createDispatchConfig(), invokerFactory, new MockMetric()); disp.getSearchInvoker(new Query(), null); + disp.deconstruct(); fail("Expected exception"); } catch (IllegalStateException e) { diff --git a/container-search/src/test/java/com/yahoo/search/dispatch/searchcluster/SearchClusterTest.java b/container-search/src/test/java/com/yahoo/search/dispatch/searchcluster/SearchClusterTest.java index 370b4def2c0..766f9ea6c2d 100644 --- a/container-search/src/test/java/com/yahoo/search/dispatch/searchcluster/SearchClusterTest.java +++ b/container-search/src/test/java/com/yahoo/search/dispatch/searchcluster/SearchClusterTest.java @@ -35,6 +35,7 @@ public class SearchClusterTest { final int nodesPerGroup; final VipStatus vipStatus; final SearchCluster searchCluster; + final ClusterMonitor clusterMonitor; final List<AtomicInteger> numDocsPerNode; List<AtomicInteger> pingCounts; @@ -60,10 +61,8 @@ public class SearchClusterTest { } searchCluster = new SearchCluster(clusterId, MockSearchCluster.createDispatchConfig(nodes), nodes.size() / nodesPerGroup, vipStatus, new Factory(nodesPerGroup, numDocsPerNode, pingCounts)); - } - - void startMonitoring() { - searchCluster.startClusterMonitoring(false); + clusterMonitor = new ClusterMonitor(searchCluster, false); + searchCluster.addMonitoring(clusterMonitor); } private int maxPingCount() { @@ -91,7 +90,7 @@ public class SearchClusterTest { int atLeast = maxPingCount() + 1; while (minPingCount < atLeast) { ExecutorService executor = Executors.newCachedThreadPool(); - searchCluster.clusterMonitor().ping(executor); + clusterMonitor.ping(executor); executor.shutdown(); try { boolean completed = executor.awaitTermination(120, TimeUnit.SECONDS); @@ -108,7 +107,7 @@ public class SearchClusterTest { @Override public void close() { - searchCluster.shutDown(); + clusterMonitor.shutdown(); } static class Factory implements PingFactory { @@ -158,7 +157,6 @@ public class SearchClusterTest { assertTrue(test.searchCluster.localCorpusDispatchTarget().isEmpty()); assertFalse(test.vipStatus.isInRotation()); - test.startMonitoring(); test.waitOneFullPingRound(); assertTrue(test.vipStatus.isInRotation()); } @@ -167,7 +165,6 @@ public class SearchClusterTest { @Test public void requireThatZeroDocsAreFine() { try (State test = new State("cluster.1", 2, "a", "b")) { - test.startMonitoring(); test.waitOneFullPingRound(); assertTrue(test.vipStatus.isInRotation()); @@ -189,7 +186,6 @@ public class SearchClusterTest { assertTrue(test.searchCluster.localCorpusDispatchTarget().isPresent()); assertFalse(test.vipStatus.isInRotation()); - test.startMonitoring(); test.waitOneFullPingRound(); assertTrue(test.vipStatus.isInRotation()); } @@ -201,7 +197,6 @@ public class SearchClusterTest { assertTrue(test.searchCluster.localCorpusDispatchTarget().isPresent()); assertFalse(test.vipStatus.isInRotation()); - test.startMonitoring(); test.waitOneFullPingRound(); assertTrue(test.vipStatus.isInRotation()); test.numDocsPerNode.get(0).set(-1); @@ -214,7 +209,6 @@ public class SearchClusterTest { public void requireThatVipStatusDownWhenLocalIsDown() { try (State test = new State("cluster.1",1,HostName.getLocalhost(), "b")) { - test.startMonitoring(); test.waitOneFullPingRound(); assertTrue(test.vipStatus.isInRotation()); assertTrue(test.searchCluster.localCorpusDispatchTarget().isPresent()); @@ -250,7 +244,6 @@ public class SearchClusterTest { List<String> nodeNames = generateNodeNames(numGroups, nodesPerGroup); try (State test = new State("cluster.1", nodesPerGroup, nodeNames)) { - test.startMonitoring(); test.waitOneFullPingRound(); assertTrue(test.vipStatus.isInRotation()); assertTrue(test.searchCluster.localCorpusDispatchTarget().isEmpty()); @@ -289,7 +282,6 @@ public class SearchClusterTest { List<String> nodeNames = generateNodeNames(numGroups, nodesPerGroup); try (State test = new State("cluster.1", nodesPerGroup, nodeNames)) { - test.startMonitoring(); test.waitOneFullPingRound(); assertTrue(test.vipStatus.isInRotation()); assertTrue(test.searchCluster.localCorpusDispatchTarget().isEmpty()); |