Skip to content

Commit 4b97713

Browse files
Poojita-RajPoojita-Raj
authored
Bridging the gap in network overhead measurement in the profiler (#1360)
* Bridging the gap in network overhead measurement in the profiler Signed-off-by: Poojita Raj <[email protected]> * add tests + remove total n/w time field Signed-off-by: Poojita Raj <[email protected]> * refactor network time into a class Signed-off-by: Poojita Raj <[email protected]> * gradle test fix Signed-off-by: Poojita Raj <[email protected]>
1 parent 6cc462b commit 4b97713

File tree

13 files changed

+265
-27
lines changed

13 files changed

+265
-27
lines changed

‎server/src/internalClusterTest/java/org/opensearch/search/profile/ProfilerSingleNodeNetworkTest.java

+72
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,72 @@
1+
/* SPDX-License-Identifier: Apache-2.0
2+
*
3+
* The OpenSearch Contributors require contributions made to
4+
* this file be licensed under the Apache-2.0 license or a
5+
* compatible open source license.
6+
*/
7+
8+
package org.opensearch.search.profile;
9+
10+
import org.apache.lucene.util.English;
11+
import org.opensearch.action.index.IndexRequestBuilder;
12+
import org.opensearch.action.search.SearchResponse;
13+
import org.opensearch.action.search.SearchType;
14+
import org.opensearch.index.query.QueryBuilder;
15+
import org.opensearch.test.OpenSearchSingleNodeTestCase;
16+
17+
import java.util.Arrays;
18+
import java.util.List;
19+
import java.util.Map;
20+
21+
import static org.hamcrest.Matchers.is;
22+
import static org.hamcrest.Matchers.not;
23+
import static org.opensearch.search.profile.query.RandomQueryGenerator.randomQueryBuilder;
24+
25+
public class ProfilerSingleNodeNetworkTest extends OpenSearchSingleNodeTestCase {
26+
27+
/**
28+
* This test checks to make sure in a single node cluster, the network time
29+
* is 0 as expected in the profiler for inbound an doutbound network time.
30+
*/
31+
public void testProfilerNetworkTime() throws Exception {
32+
createIndex("test");
33+
ensureGreen();
34+
35+
int numDocs = randomIntBetween(100, 150);
36+
IndexRequestBuilder[] docs = new IndexRequestBuilder[numDocs];
37+
for (int i = 0; i < numDocs; i++) {
38+
docs[i] = client().prepareIndex("test", "type1", String.valueOf(i)).setSource("field1", English.intToEnglish(i), "field2", i);
39+
}
40+
41+
List<String> stringFields = Arrays.asList("field1");
42+
List<String> numericFields = Arrays.asList("field2");
43+
44+
int iters = between(20, 100);
45+
for (int i = 0; i < iters; i++) {
46+
QueryBuilder q = randomQueryBuilder(stringFields, numericFields, numDocs, 3);
47+
logger.info("Query: {}", q);
48+
49+
SearchResponse resp = client().prepareSearch()
50+
.setQuery(q)
51+
.setTrackTotalHits(true)
52+
.setProfile(true)
53+
.setSearchType(SearchType.QUERY_THEN_FETCH)
54+
.get();
55+
56+
assertNotNull("Profile response element should not be null", resp.getProfileResults());
57+
assertThat("Profile response should not be an empty array", resp.getProfileResults().size(), not(0));
58+
for (Map.Entry<String, ProfileShardResult> shard : resp.getProfileResults().entrySet()) {
59+
assertThat(
60+
"Profile response inbound network time should be 0 in single node clusters",
61+
shard.getValue().getNetworkTime().getInboundNetworkTime(),
62+
is(0L)
63+
);
64+
assertThat(
65+
"Profile response outbound network time should be 0 in single node clusters",
66+
shard.getValue().getNetworkTime().getOutboundNetworkTime(),
67+
is(0L)
68+
);
69+
}
70+
}
71+
}
72+
}

‎server/src/internalClusterTest/java/org/opensearch/search/profile/query/QueryProfilerIT.java

+9-14
Original file line numberDiff line numberDiff line change
@@ -34,11 +34,7 @@
3434

3535
import org.apache.lucene.util.English;
3636
import org.opensearch.action.index.IndexRequestBuilder;
37-
import org.opensearch.action.search.MultiSearchResponse;
38-
import org.opensearch.action.search.SearchRequestBuilder;
39-
import org.opensearch.action.search.SearchResponse;
40-
import org.opensearch.action.search.SearchType;
41-
import org.opensearch.action.search.ShardSearchFailure;
37+
import org.opensearch.action.search.*;
4238
import org.opensearch.common.settings.Settings;
4339
import org.opensearch.index.query.QueryBuilder;
4440
import org.opensearch.index.query.QueryBuilders;
@@ -48,18 +44,15 @@
4844
import org.opensearch.search.sort.SortOrder;
4945
import org.opensearch.test.OpenSearchIntegTestCase;
5046

51-
import java.util.Arrays;
52-
import java.util.HashSet;
53-
import java.util.List;
54-
import java.util.Map;
55-
import java.util.Set;
47+
import java.util.*;
5648

57-
import static org.opensearch.search.profile.query.RandomQueryGenerator.randomQueryBuilder;
58-
import static org.hamcrest.Matchers.emptyOrNullString;
59-
import static org.hamcrest.Matchers.equalTo;
60-
import static org.hamcrest.Matchers.greaterThan;
6149
import static org.hamcrest.Matchers.is;
50+
import static org.hamcrest.Matchers.greaterThan;
51+
import static org.hamcrest.Matchers.greaterThanOrEqualTo;
6252
import static org.hamcrest.Matchers.not;
53+
import static org.hamcrest.Matchers.emptyOrNullString;
54+
import static org.hamcrest.Matchers.equalTo;
55+
import static org.opensearch.search.profile.query.RandomQueryGenerator.randomQueryBuilder;
6356

6457
public class QueryProfilerIT extends OpenSearchIntegTestCase {
6558

@@ -98,6 +91,8 @@ public void testProfileQuery() throws Exception {
9891
assertNotNull("Profile response element should not be null", resp.getProfileResults());
9992
assertThat("Profile response should not be an empty array", resp.getProfileResults().size(), not(0));
10093
for (Map.Entry<String, ProfileShardResult> shard : resp.getProfileResults().entrySet()) {
94+
assertThat(shard.getValue().getNetworkTime().getInboundNetworkTime(), greaterThanOrEqualTo(0L));
95+
assertThat(shard.getValue().getNetworkTime().getOutboundNetworkTime(), greaterThanOrEqualTo(0L));
10196
for (QueryProfileShardResult searchProfiles : shard.getValue().getQueryProfileResults()) {
10297
for (ProfileResult result : searchProfiles.getQueryResults()) {
10398
assertNotNull(result.getQueryName());

‎server/src/main/java/org/opensearch/action/search/SearchExecutionStatsCollector.java

+18
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,7 @@
3535
import org.opensearch.action.ActionListener;
3636
import org.opensearch.node.ResponseCollectorService;
3737
import org.opensearch.search.SearchPhaseResult;
38+
import org.opensearch.search.fetch.QueryFetchSearchResult;
3839
import org.opensearch.search.query.QuerySearchResult;
3940
import org.opensearch.transport.Transport;
4041

@@ -66,7 +67,24 @@ public static BiFunction<Transport.Connection, SearchActionListener, ActionListe
6667

6768
@Override
6869
public void onResponse(SearchPhaseResult response) {
70+
if (response instanceof QueryFetchSearchResult) {
71+
response.queryResult().getShardSearchRequest().setOutboundNetworkTime(0);
72+
response.queryResult().getShardSearchRequest().setInboundNetworkTime(0);
73+
}
6974
QuerySearchResult queryResult = response.queryResult();
75+
if (response.getShardSearchRequest() != null) {
76+
if (response.remoteAddress() != null) {
77+
// update outbound network time for request sent over network for shard requests
78+
response.getShardSearchRequest()
79+
.setOutboundNetworkTime(
80+
Math.max(0, System.currentTimeMillis() - response.getShardSearchRequest().getOutboundNetworkTime())
81+
);
82+
} else {
83+
// reset inbound and outbound network time to 0 for local request for shard requests
84+
response.getShardSearchRequest().setOutboundNetworkTime(0);
85+
response.getShardSearchRequest().setInboundNetworkTime(0);
86+
}
87+
}
7088
if (nodeId != null && queryResult != null) {
7189
final long serviceTimeEWMA = queryResult.serviceTimeEWMA();
7290
final int queueSize = queryResult.nodeQueueSize();

‎server/src/main/java/org/opensearch/action/search/SearchQueryThenFetchAsyncAction.java

+4
Original file line numberDiff line numberDiff line change
@@ -121,6 +121,10 @@ protected void executePhaseOnShard(
121121
final SearchActionListener<SearchPhaseResult> listener
122122
) {
123123
ShardSearchRequest request = rewriteShardSearchRequest(super.buildShardSearchRequest(shardIt));
124+
// update inbound network time with current time before sending request over n/w to data node
125+
if (request != null) {
126+
request.setInboundNetworkTime(System.currentTimeMillis());
127+
}
124128
getSearchTransport().sendExecuteQuery(getConnection(shard.getClusterAlias(), shard.getNodeId()), request, getTask(), listener);
125129
}
126130

‎server/src/main/java/org/opensearch/search/internal/ShardSearchRequest.java

+31
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,7 @@
3333
package org.opensearch.search.internal;
3434

3535
import org.opensearch.LegacyESVersion;
36+
import org.opensearch.Version;
3637
import org.opensearch.action.IndicesRequest;
3738
import org.opensearch.action.OriginalIndices;
3839
import org.opensearch.action.search.SearchRequest;
@@ -90,6 +91,8 @@ public class ShardSearchRequest extends TransportRequest implements IndicesReque
9091
private final float indexBoost;
9192
private final Boolean requestCache;
9293
private final long nowInMillis;
94+
private long inboundNetworkTime;
95+
private long outboundNetworkTime;
9396
private final boolean allowPartialSearchResults;
9497
private final String[] indexRoutings;
9598
private final String preference;
@@ -221,6 +224,8 @@ private ShardSearchRequest(
221224
this.preference = preference;
222225
this.scroll = scroll;
223226
this.nowInMillis = nowInMillis;
227+
this.inboundNetworkTime = 0;
228+
this.outboundNetworkTime = 0;
224229
this.clusterAlias = clusterAlias;
225230
this.originalIndices = originalIndices;
226231
this.readerId = readerId;
@@ -240,6 +245,10 @@ public ShardSearchRequest(StreamInput in) throws IOException {
240245
indexBoost = in.readFloat();
241246
nowInMillis = in.readVLong();
242247
requestCache = in.readOptionalBoolean();
248+
if (in.getVersion().onOrAfter(Version.V_2_0_0)) {
249+
inboundNetworkTime = in.readVLong();
250+
outboundNetworkTime = in.readVLong();
251+
}
243252
clusterAlias = in.readOptionalString();
244253
if (in.getVersion().onOrAfter(LegacyESVersion.V_7_0_0)) {
245254
allowPartialSearchResults = in.readBoolean();
@@ -283,6 +292,8 @@ public ShardSearchRequest(ShardSearchRequest clone) {
283292
this.aliasFilter = clone.aliasFilter;
284293
this.indexBoost = clone.indexBoost;
285294
this.nowInMillis = clone.nowInMillis;
295+
this.inboundNetworkTime = clone.inboundNetworkTime;
296+
this.outboundNetworkTime = clone.outboundNetworkTime;
286297
this.requestCache = clone.requestCache;
287298
this.clusterAlias = clone.clusterAlias;
288299
this.allowPartialSearchResults = clone.allowPartialSearchResults;
@@ -317,6 +328,10 @@ protected final void innerWriteTo(StreamOutput out, boolean asKey) throws IOExce
317328
out.writeVLong(nowInMillis);
318329
}
319330
out.writeOptionalBoolean(requestCache);
331+
if (asKey == false && out.getVersion().onOrAfter(Version.V_2_0_0)) {
332+
out.writeVLong(inboundNetworkTime);
333+
out.writeVLong(outboundNetworkTime);
334+
}
320335
out.writeOptionalString(clusterAlias);
321336
if (out.getVersion().onOrAfter(LegacyESVersion.V_7_0_0)) {
322337
out.writeBoolean(allowPartialSearchResults);
@@ -395,6 +410,22 @@ public long nowInMillis() {
395410
return nowInMillis;
396411
}
397412

413+
public long getInboundNetworkTime() {
414+
return inboundNetworkTime;
415+
}
416+
417+
public void setInboundNetworkTime(long newTime) {
418+
this.inboundNetworkTime = newTime;
419+
}
420+
421+
public long getOutboundNetworkTime() {
422+
return outboundNetworkTime;
423+
}
424+
425+
public void setOutboundNetworkTime(long newTime) {
426+
this.outboundNetworkTime = newTime;
427+
}
428+
398429
public Boolean requestCache() {
399430
return requestCache;
400431
}

‎server/src/main/java/org/opensearch/search/profile/NetworkTime.java

+56
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,56 @@
1+
/*
2+
* SPDX-License-Identifier: Apache-2.0
3+
*
4+
* The OpenSearch Contributors require contributions made to
5+
* this file be licensed under the Apache-2.0 license or a
6+
* compatible open source license.
7+
*/
8+
9+
package org.opensearch.search.profile;
10+
11+
import org.opensearch.Version;
12+
import org.opensearch.common.io.stream.StreamInput;
13+
import org.opensearch.common.io.stream.StreamOutput;
14+
import org.opensearch.common.io.stream.Writeable;
15+
16+
import java.io.IOException;
17+
18+
public class NetworkTime implements Writeable {
19+
private long inboundNetworkTime;
20+
private long outboundNetworkTime;
21+
22+
public NetworkTime(long inboundTime, long outboundTime) {
23+
this.inboundNetworkTime = inboundTime;
24+
this.outboundNetworkTime = outboundTime;
25+
}
26+
27+
public NetworkTime(StreamInput in) throws IOException {
28+
if (in.getVersion().onOrAfter(Version.V_2_0_0)) {
29+
this.inboundNetworkTime = in.readVLong();
30+
this.outboundNetworkTime = in.readVLong();
31+
}
32+
}
33+
34+
public void writeTo(StreamOutput out) throws IOException {
35+
if (out.getVersion().onOrAfter(Version.V_2_0_0)) {
36+
out.writeVLong(inboundNetworkTime);
37+
out.writeVLong(outboundNetworkTime);
38+
}
39+
}
40+
41+
public long getInboundNetworkTime() {
42+
return this.inboundNetworkTime;
43+
}
44+
45+
public long getOutboundNetworkTime() {
46+
return this.outboundNetworkTime;
47+
}
48+
49+
public void setInboundNetworkTime(long newTime) {
50+
this.inboundNetworkTime = newTime;
51+
}
52+
53+
public void setOutboundNetworkTime(long newTime) {
54+
this.outboundNetworkTime = newTime;
55+
}
56+
}

‎server/src/main/java/org/opensearch/search/profile/ProfileShardResult.java

+20-1
Original file line numberDiff line numberDiff line change
@@ -49,9 +49,16 @@ public class ProfileShardResult implements Writeable {
4949

5050
private final AggregationProfileShardResult aggProfileShardResult;
5151

52-
public ProfileShardResult(List<QueryProfileShardResult> queryProfileResults, AggregationProfileShardResult aggProfileShardResult) {
52+
private NetworkTime networkTime;
53+
54+
public ProfileShardResult(
55+
List<QueryProfileShardResult> queryProfileResults,
56+
AggregationProfileShardResult aggProfileShardResult,
57+
NetworkTime networkTime
58+
) {
5359
this.aggProfileShardResult = aggProfileShardResult;
5460
this.queryProfileResults = Collections.unmodifiableList(queryProfileResults);
61+
this.networkTime = networkTime;
5562
}
5663

5764
public ProfileShardResult(StreamInput in) throws IOException {
@@ -63,6 +70,7 @@ public ProfileShardResult(StreamInput in) throws IOException {
6370
}
6471
this.queryProfileResults = Collections.unmodifiableList(queryProfileResults);
6572
this.aggProfileShardResult = new AggregationProfileShardResult(in);
73+
this.networkTime = new NetworkTime(in);
6674
}
6775

6876
@Override
@@ -72,6 +80,7 @@ public void writeTo(StreamOutput out) throws IOException {
7280
queryShardResult.writeTo(out);
7381
}
7482
aggProfileShardResult.writeTo(out);
83+
networkTime.writeTo(out);
7584
}
7685

7786
public List<QueryProfileShardResult> getQueryProfileResults() {
@@ -81,4 +90,14 @@ public List<QueryProfileShardResult> getQueryProfileResults() {
8190
public AggregationProfileShardResult getAggregationProfileResults() {
8291
return aggProfileShardResult;
8392
}
93+
94+
public NetworkTime getNetworkTime() {
95+
return networkTime;
96+
}
97+
98+
public void setNetworkTime(NetworkTime newTime) {
99+
networkTime.setInboundNetworkTime(newTime.getInboundNetworkTime());
100+
networkTime.setOutboundNetworkTime(newTime.getOutboundNetworkTime());
101+
}
102+
84103
}

0 commit comments

Comments
 (0)