Register data node stats from info carried back in search responses by dakrone · Pull Request #25430 · elastic/elasticsearch

dakrone · 2017-06-27T15:08:24Z

This is part of #24915, where we now calculate the EWMA of service time for
tasks in the search threadpool, and send that as well as the current queue size
back to the coordinating node. The coordinating node now tracks this information
for each node in the cluster.

This information will be used in the future the determining the best replica a
search request should be routed to. This change has no user-visible difference.

This is part of elastic#24915, where we now calculate the EWMA of service time for tasks in the search threadpool, and send that as well as the current queue size back to the coordinating node. The coordinating node now tracks this information for each node in the cluster. This information will be used in the future the determining the best replica a search request should be routed to. This change has no user-visible difference.

dakrone · 2017-06-27T15:36:43Z

retest this please

s1monw

left a bunch of comments, thanks lee for the effort!

s1monw · 2017-06-30T08:18:05Z

core/src/main/java/org/elasticsearch/action/search/AbstractSearchAsyncAction.java

        successfulOps.incrementAndGet();
+        long responseDuration = System.nanoTime() - startNanos;
+        String nodeId = result.getSearchShardTarget().getNodeId();
+        if (nodeId != null && collectorService != null) {


why can any of this be null? this looks like a bug?!

why do we do this in here and not in the query wrapper. This action might run some DFS stats rather than queries so I think we should capture it solely in the ResponseListenerWrapper?

if we only capture it in the wrapper we don't need the dependency here no?

s1monw · 2017-06-30T08:29:47Z

core/src/main/java/org/elasticsearch/action/search/ResponseListenerWrapper.java

+ * result to get the piggybacked queue size and service time EWMA, adding those
+ * values to the coordinating nodes' {@code ResponseCollectorService}.
+ */
+public class ResponseListenerWrapper extends SearchActionListener<SearchPhaseResult> {


I think this one can just be an action listener, we can use the searchShardTarget from the delegate listener. there is no need to set the targets twice etc.

I also wonder if we can find a better name for this

s1monw · 2017-06-30T08:30:24Z

core/src/main/java/org/elasticsearch/action/search/ResponseListenerWrapper.java

+ */
+public class ResponseListenerWrapper extends SearchActionListener<SearchPhaseResult> {
+
+    private static final Logger logger = ESLoggerFactory.getLogger(ResponseListenerWrapper.class);


let's not add logger here. if it's only for debugging we can add it when it's needed

s1monw · 2017-06-30T08:31:01Z

core/src/main/java/org/elasticsearch/action/search/ResponseListenerWrapper.java

+        if (queryResult != null && collector != null) {
+            long ewma = queryResult.serviceTimeEWMA();
+            int queueSize = queryResult.nodeQueueSize();
+            String nodeId = listener.searchShardTarget.getNodeId();


this should never be null. Fail if it is, maybe in the ctor already

s1monw · 2017-06-30T08:34:10Z

core/src/main/java/org/elasticsearch/action/search/ResponseListenerWrapper.java

+
+    public ResponseListenerWrapper(SearchActionListener<SearchPhaseResult> listener,
+                                   ResponseCollectorService collector) {
+        super(listener.searchShardTarget, listener.requestIndex);


maybe we just expect an Actionlistener and a node ID string or even better a DiscoveryNode here. We can get it directly from the connection via Connection#getNode() which works with CCS as well then we can reduce visibility again on SearchActionListener

s1monw · 2017-06-30T08:34:23Z

core/src/main/java/org/elasticsearch/action/search/SearchActionListener.java

-    private final int requestIndex;
-    private final SearchShardTarget searchShardTarget;
+
+    protected final int requestIndex;


see comment above regarding visibility

s1monw · 2017-06-30T08:39:00Z

core/src/main/java/org/elasticsearch/node/Node.java

                transportService, indicesService, pluginsService, circuitBreakerService, scriptModule.getScriptService(),
                httpServerTransport, ingestService, clusterService, settingsModule.getSettingsFilter());
            modules.add(b -> {
+                    b.bind(ResponseCollectorService.class).toInstance(responseCollectorService);


I don't think we should do this! Let's please don't add any guice related stuff anymore no matter what it costs.

s1monw · 2017-06-30T08:40:29Z

core/src/main/java/org/elasticsearch/node/ResponseCollectorService.java

+
+    private static final double ALPHA = 0.3;
+
+    private final Map<String, ExponentiallyWeightedMovingAverage> nodeIdToQueueSize;


Can we maybe just have a single map and then sync on the node ID with using some KeyedLock? I think that would be simpler?

s1monw · 2017-06-30T08:42:15Z

core/src/main/java/org/elasticsearch/search/DefaultSearchContext.java

        this.queryResult = new QuerySearchResult(id, shardTarget);
        this.fetchResult = new FetchSearchResult(id, shardTarget);
        this.indexShard = indexShard;
+        this.searchExecutor = indexShard.getThreadPool().executor(ThreadPool.Names.SEARCH);


can we do this on the consumer end and specialize the code in SearchService? I think this is where it's used. We shouldn't really add methods to this class it's big enough

I don't think I understand, I need a way from QueryPhase.execute to access the executor to get the stats out of it, I can't do it on the consumer end since that may be on a different node, and the only thing available in the (static) QueryPhase.execute method is the SearchContext.

Is there a better place to put this?

what I mean is instead of having the searchExecutor member can't we do this in QueryPhase.execute instead so we don't add more than necessary to the search context. it's already huge and we can access indexShard?

That makes sense and is much cleaner, thanks!

s1monw · 2017-06-30T08:46:42Z

core/src/test/java/org/elasticsearch/node/ResponseCollectorServiceIT.java

+ * Test that response stats are collected for queries executed on a real node
+ */
+@ESIntegTestCase.ClusterScope(scope = ESIntegTestCase.Scope.TEST, numDataNodes = 0)
+public class ResponseCollectorServiceIT extends ESIntegTestCase {


instead of doing these old style integration tests should we rather add unittest to do this. I added so many for all the search actions. Let's test all the things in isolation like the wrapper, the query phase etc.

…nListener Also removes the logger

…eping

dakrone · 2017-07-11T18:14:08Z

retest this please

…election-info

dakrone · 2017-07-12T17:07:43Z

@s1monw thanks for taking a look! I've pushed commits addressing all of your feedback (except for the one I left a question about), could you take another look?

s1monw

left some minors looks close!

s1monw · 2017-07-12T18:19:15Z

core/src/main/java/org/elasticsearch/action/search/SearchActionListener.java

+
    private final int requestIndex;
-    private final SearchShardTarget searchShardTarget;
+    protected final SearchShardTarget searchShardTarget;


can you add a getter instead of making this protected?

given my comment below I think this is unneeded and can be private without a getter

s1monw · 2017-07-12T18:21:02Z

core/src/main/java/org/elasticsearch/action/search/SearchExecutionStatsCollector.java

+    private final ResponseCollectorService collector;
+    private final long startNanos;
+
+    public SearchExecutionStatsCollector(SearchActionListener<SearchPhaseResult> listener,


please instead of exposing the searchShardTarget from SearchActionListener can you add a 3rd arg that is the nodeID. we can get it from the passed in connection we talk to instead. then SearchActionListener can also be a ActionListener instead

s1monw · 2017-07-12T18:21:56Z

core/src/main/java/org/elasticsearch/action/search/SearchExecutionStatsCollector.java

+ * result to get the piggybacked queue size and service time EWMA, adding those
+ * values to the coordinating nodes' {@code ResponseCollectorService}.
+ */
+public class SearchExecutionStatsCollector implements ActionListener<SearchPhaseResult> {


can this be pkg privaate and final?

s1monw · 2017-07-12T18:23:47Z

core/src/main/java/org/elasticsearch/action/search/SearchTransportService.java


-    public SearchTransportService(Settings settings, TransportService transportService) {
+    public SearchTransportService(Settings settings, TransportService transportService,
+                                  ResponseCollectorService responseCollectorService) {


you can also remove the dependency by passing a BiFunction<Connection, SearchActionListener, ActionListener> instead of ResponseCollectorService and require it to be non-null then it's easier to test? and we never need to do nullchecks

I played with this, however, it requires making SearchExecutionStatsCollector public instead of package private, since we need to construct it within the BiFunction passed in to SearchTransportService in Node.

I think this is what you had in mind, but let me know if I'm mistaken:

diff --git a/core/src/main/java/org/elasticsearch/action/search/SearchExecutionStatsCollector.java b/core/src/main/java/org/elasticsearch/action/search/SearchExecutionStatsCollector.java index 031e860..96d5e18 100644 --- a/core/src/main/java/org/elasticsearch/action/search/SearchExecutionStatsCollector.java +++ b/core/src/main/java/org/elasticsearch/action/search/SearchExecutionStatsCollector.java @@ -23,8 +23,10 @@ import org.elasticsearch.action.ActionListener; import org.elasticsearch.node.ResponseCollectorService; import org.elasticsearch.search.SearchPhaseResult; import org.elasticsearch.search.query.QuerySearchResult; +import org.elasticsearch.transport.Transport; import java.util.Objects; +import java.util.function.BiFunction; /** * A wrapper of search action listeners (search results) that unwraps the query @@ -47,6 +49,10 @@ final class SearchExecutionStatsCollector implements ActionListener<SearchPhaseR this.nodeId = nodeId; } + public static BiFunction<Transport.Connection, SearchActionListener, ActionListener> makeWrapper(ResponseCollectorService service) { + return (connection, originalListener) -> new SearchExecutionStatsCollector(listener, service, connection.getNode().getId()); + } + @Override public void onResponse(SearchPhaseResult response) { QuerySearchResult queryResult = response.queryResult(); diff --git a/core/src/main/java/org/elasticsearch/action/search/SearchTransportService.java b/core/src/main/java/org/elasticsearch/action/search/SearchTransportService.java index 2587623..3ad1517 100644 --- a/core/src/main/java/org/elasticsearch/action/search/SearchTransportService.java +++ b/core/src/main/java/org/elasticsearch/action/search/SearchTransportService.java @@ -56,6 +56,7 @@ import org.elasticsearch.transport.TransportResponse; import org.elasticsearch.transport.TransportService; import java.io.IOException; +import java.util.function.BiFunction; import java.util.function.Supplier; /** @@ -77,12 +78,13 @@ public class SearchTransportService extends AbstractComponent { private final TransportService transportService; private final ResponseCollectorService responseCollectorService; + private final BiFunction<Transport.Connection, SearchActionListener, ActionListener> responseWrapper; public SearchTransportService(Settings settings, TransportService transportService, - ResponseCollectorService responseCollectorService) { + BiFunction<Transport.Connection, SearchActionListener, ActionListener> responseWrapper) { super(settings); this.transportService = transportService; - this.responseCollectorService = responseCollectorService; + this.responseWrapper = responseWrapper; } public void sendFreeContext(Transport.Connection connection, final long contextId, OriginalIndices originalIndices) { @@ -123,12 +125,7 @@ public class SearchTransportService extends AbstractComponent { final boolean fetchDocuments = request.numberOfShards() == 1; Supplier<SearchPhaseResult> supplier = fetchDocuments ? QueryFetchSearchResult::new : QuerySearchResult::new; - final ActionListener handler; - if (responseCollectorService == null) { - handler = listener; - } else { - handler = new SearchExecutionStatsCollector(listener, responseCollectorService, connection.getNode().getId()); - } + final ActionListener handler = responseWrapper.apply(connection, listener); transportService.sendChildRequest(connection, QUERY_ACTION_NAME, request, task, new ActionListenerResponseHandler<>(handler, supplier)); } diff --git a/core/src/main/java/org/elasticsearch/node/Node.java b/core/src/main/java/org/elasticsearch/node/Node.java index bb2264f..b10f645 100644 --- a/core/src/main/java/org/elasticsearch/node/Node.java +++ b/core/src/main/java/org/elasticsearch/node/Node.java @@ -29,6 +29,7 @@ import org.elasticsearch.ElasticsearchTimeoutException; import org.elasticsearch.Version; import org.elasticsearch.action.ActionModule; import org.elasticsearch.action.GenericAction; +import org.elasticsearch.action.search.SearchExecutionStatsCollector; import org.elasticsearch.action.search.SearchPhaseController; import org.elasticsearch.action.search.SearchTransportService; import org.elasticsearch.action.support.TransportAction; @@ -424,8 +425,8 @@ public class Node implements Closeable { final TransportService transportService = newTransportService(settings, transport, threadPool, networkModule.getTransportInterceptor(), localNodeFactory, settingsModule.getClusterSettings()); final ResponseCollectorService responseCollectorService = new ResponseCollectorService(this.settings, clusterService); - final SearchTransportService searchTransportService = new SearchTransportService(settings, - transportService, responseCollectorService); + final SearchTransportService searchTransportService = new SearchTransportService(settings, transportService, + SearchExecutionStatsCollector.makeWrapper(responseCollectorService)); final Consumer<Binder> httpBind; final HttpServerTransport httpServerTransport; if (networkModule.isHttpEnabled()) {

yes this looks good to me

I played with this, however, it requires making SearchExecutionStatsCollector public instead of package private, since we need to construct it within the BiFunction passed in to SearchTransportService in Node.

I think that is ok!

s1monw · 2017-07-12T18:24:20Z

core/src/main/java/org/elasticsearch/node/ResponseCollectorService.java

+ * tasks executed on each node, making the EWMA of the values available to the
+ * coordinating node.
+ */
+public class ResponseCollectorService extends AbstractComponent implements ClusterStateListener {


s1monw · 2017-07-12T18:25:28Z

core/src/main/java/org/elasticsearch/node/ResponseCollectorService.java

+        });
+    }
+
+    public Map<String, ComputedNodeStats> getAllNodeStatistics() {


this seems to be test only can we move it into a util in tests?

I'm planning on using this method for the very next PR that implements the actual ranking algorithm, without this method the NodeStatistics values would be exposed, which is the mutable one. By putting this method here NodeStatistics can be private and we can expose only the immutable ComputedNodeStats

s1monw · 2017-07-12T18:25:44Z

core/src/main/java/org/elasticsearch/node/ResponseCollectorService.java

+     * node's statistics. This includes the EWMA of queue size, response time,
+     * and service time.
+     */
+    public class ComputedNodeStats {


can this inner class be static and final and private

s1monw · 2017-07-12T18:25:54Z

core/src/main/java/org/elasticsearch/node/ResponseCollectorService.java

+     * time, and service time, however, this class is private and intended only
+     * to be used for the internal accounting of {@code ResponseCollectorService}.
+     */
+    private class NodeStatistics {


should also be static?

s1monw · 2017-07-12T18:26:04Z

core/src/main/java/org/elasticsearch/node/ResponseCollectorService.java

+     * to be used for the internal accounting of {@code ResponseCollectorService}.
+     */
+    private class NodeStatistics {
+        public final String nodeId;


public is obsolete here

serviceTime is non final?! is that expected? just curious...

…tener

…election-info

dakrone · 2017-07-13T19:15:13Z

Thanks for the comments @s1monw! I added more changes based on your feedback

…election-info

s1monw

please apply your proposed patch other than that LGTM

…ervice

…election-info

dakrone · 2017-07-17T17:05:01Z

Thanks for all the reviews @s1monw!

dakrone mentioned this pull request Jun 29, 2017

Implement adaptive replica selection for coordinating nodes performing queries #24915

Closed

5 tasks

dakrone requested a review from s1monw June 29, 2017 15:06

s1monw suggested changes Jun 30, 2017

View reviewed changes

dakrone added 12 commits July 6, 2017 14:46

Move response time timing into ResponseListenerWrapper

ab3a391

Move ResponseListenerWrapper to ActionListener instead of SearchActio…

ab07b7e

…nListener Also removes the logger

Move requestIndex back to private

31c14ed

De-guice-ify ResponseCollectorService \o/

027d139

Undo all changes to SearchQueryThenFetchAsyncAction

33b352d

Remove unneeded response collector from TransportSearchAction

0a6b36a

Undo all changes to SearchDfsQueryThenFetchAsyncAction

f723aed

Completely rewrite the inside of ResponseCollectorService's record ke…

b82f2d1

…eping

Documentation and cleanups for ResponseCollectorService

94e6e9e

Add unit test for collection of queue size and service time

b93d825

Fix Guice construction error

67c6c67

Add basic unit tests for ResponseCollectorService

bbe84d9

dakrone added 7 commits July 11, 2017 13:10

Merge remote-tracking branch 'origin/master' into register-adaptive-s…

a6c7915

…election-info

Fix version constant for the master merge

2f8363a

Fix test compilation after master merge

3a374f1

Add a test for node removal on cluster changed event

38e7510

Remove integration test as there are now unit tests

3b90cf6

Rename ResponseListenerWrapper -> SearchExecutionStatsCollector

9457c2f

Fix line-length

6f6d2d5

dakrone requested a review from s1monw July 12, 2017 17:07

s1monw suggested changes Jul 12, 2017

View reviewed changes

dakrone added 2 commits July 12, 2017 23:37

Make classes private and final where appropriate

670bdba

Pass nodeId into SearchExecutionStatsCollector and use only ActionLis…

cea52fd

…tener

dakrone added 3 commits July 13, 2017 00:04

Get nodeId from connection so searchShardTarget can be private

227fd09

Remove threadpool from SearchContext, get it from IndexShard instead

818570b

Merge remote-tracking branch 'origin/master' into register-adaptive-s…

c490e7d

…election-info

dakrone requested a review from s1monw July 13, 2017 19:15

dakrone added 2 commits July 14, 2017 12:17

Merge remote-tracking branch 'origin/master' into register-adaptive-s…

cc646b5

…election-info

Add missing import

1581bbb

s1monw approved these changes Jul 17, 2017

View reviewed changes

dakrone added 2 commits July 17, 2017 09:28

Use BiFunction for responseWrapper rather than passing in collector s…

b7e1624

…ervice

Merge remote-tracking branch 'origin/master' into register-adaptive-s…

95deff7

…election-info

dakrone merged commit 610ba7e into elastic:master Jul 17, 2017

dakrone added :Search/Search Search-related issues that do not fall into other categories v6.0.0 labels Jul 24, 2017

colings86 added v6.0.0-beta1 and removed v6.0.0 labels Jul 31, 2017

dakrone deleted the register-adaptive-selection-info branch December 13, 2017 20:37


		private static final double ALPHA = 0.3;

		private final Map<String, ExponentiallyWeightedMovingAverage> nodeIdToQueueSize;

Conversation

dakrone commented Jun 27, 2017

Uh oh!

dakrone commented Jun 27, 2017

Uh oh!

s1monw left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dakrone commented Jul 11, 2017

Uh oh!

dakrone commented Jul 12, 2017

Uh oh!

s1monw left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dakrone Jul 13, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

dakrone Jul 13, 2017 •

edited

Loading