Skip to content

Commit 6993ac9

Browse files
ajaymovvaajaymovva
authored
Added Point In Time Node Stats API ServiceLayer Changes (#4030)
* Adds Node Stats api changes for the Point in time Signed-off-by: Ajay Kumar Movva <[email protected]>
1 parent c665e7c commit 6993ac9

File tree

7 files changed

+175
-5
lines changed

7 files changed

+175
-5
lines changed

‎server/src/main/java/org/opensearch/index/search/stats/SearchStats.java

Lines changed: 56 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,7 @@
3232

3333
package org.opensearch.index.search.stats;
3434

35+
import org.opensearch.Version;
3536
import org.opensearch.common.Nullable;
3637
import org.opensearch.common.Strings;
3738
import org.opensearch.common.io.stream.StreamInput;
@@ -77,6 +78,10 @@ public static class Stats implements Writeable, ToXContentFragment {
7778
private long suggestTimeInMillis;
7879
private long suggestCurrent;
7980

81+
private long pitCount;
82+
private long pitTimeInMillis;
83+
private long pitCurrent;
84+
8085
private Stats() {
8186
// for internal use, initializes all counts to 0
8287
}
@@ -91,6 +96,9 @@ public Stats(
9196
long scrollCount,
9297
long scrollTimeInMillis,
9398
long scrollCurrent,
99+
long pitCount,
100+
long pitTimeInMillis,
101+
long pitCurrent,
94102
long suggestCount,
95103
long suggestTimeInMillis,
96104
long suggestCurrent
@@ -110,6 +118,10 @@ public Stats(
110118
this.suggestCount = suggestCount;
111119
this.suggestTimeInMillis = suggestTimeInMillis;
112120
this.suggestCurrent = suggestCurrent;
121+
122+
this.pitCount = pitCount;
123+
this.pitTimeInMillis = pitTimeInMillis;
124+
this.pitCurrent = pitCurrent;
113125
}
114126

115127
private Stats(StreamInput in) throws IOException {
@@ -128,6 +140,12 @@ private Stats(StreamInput in) throws IOException {
128140
suggestCount = in.readVLong();
129141
suggestTimeInMillis = in.readVLong();
130142
suggestCurrent = in.readVLong();
143+
144+
if (in.getVersion().onOrAfter(Version.V_3_0_0)) {
145+
pitCount = in.readVLong();
146+
pitTimeInMillis = in.readVLong();
147+
pitCurrent = in.readVLong();
148+
}
131149
}
132150

133151
public void add(Stats stats) {
@@ -146,6 +164,10 @@ public void add(Stats stats) {
146164
suggestCount += stats.suggestCount;
147165
suggestTimeInMillis += stats.suggestTimeInMillis;
148166
suggestCurrent += stats.suggestCurrent;
167+
168+
pitCount += stats.pitCount;
169+
pitTimeInMillis += stats.pitTimeInMillis;
170+
pitCurrent += stats.pitCurrent;
149171
}
150172

151173
public void addForClosingShard(Stats stats) {
@@ -162,6 +184,10 @@ public void addForClosingShard(Stats stats) {
162184

163185
suggestCount += stats.suggestCount;
164186
suggestTimeInMillis += stats.suggestTimeInMillis;
187+
188+
pitCount += stats.pitCount;
189+
pitTimeInMillis += stats.pitTimeInMillis;
190+
pitCurrent += stats.pitCurrent;
165191
}
166192

167193
public long getQueryCount() {
@@ -212,6 +238,22 @@ public long getScrollCurrent() {
212238
return scrollCurrent;
213239
}
214240

241+
public long getPitCount() {
242+
return pitCount;
243+
}
244+
245+
public TimeValue getPitTime() {
246+
return new TimeValue(pitTimeInMillis);
247+
}
248+
249+
public long getPitTimeInMillis() {
250+
return pitTimeInMillis;
251+
}
252+
253+
public long getPitCurrent() {
254+
return pitCurrent;
255+
}
256+
215257
public long getSuggestCount() {
216258
return suggestCount;
217259
}
@@ -249,6 +291,12 @@ public void writeTo(StreamOutput out) throws IOException {
249291
out.writeVLong(suggestCount);
250292
out.writeVLong(suggestTimeInMillis);
251293
out.writeVLong(suggestCurrent);
294+
295+
if (out.getVersion().onOrAfter(Version.V_3_0_0)) {
296+
out.writeVLong(pitCount);
297+
out.writeVLong(pitTimeInMillis);
298+
out.writeVLong(pitCurrent);
299+
}
252300
}
253301

254302
@Override
@@ -265,6 +313,10 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
265313
builder.humanReadableField(Fields.SCROLL_TIME_IN_MILLIS, Fields.SCROLL_TIME, getScrollTime());
266314
builder.field(Fields.SCROLL_CURRENT, scrollCurrent);
267315

316+
builder.field(Fields.PIT_TOTAL, pitCount);
317+
builder.humanReadableField(Fields.PIT_TIME_IN_MILLIS, Fields.PIT_TIME, getPitTime());
318+
builder.field(Fields.PIT_CURRENT, pitCurrent);
319+
268320
builder.field(Fields.SUGGEST_TOTAL, suggestCount);
269321
builder.humanReadableField(Fields.SUGGEST_TIME_IN_MILLIS, Fields.SUGGEST_TIME, getSuggestTime());
270322
builder.field(Fields.SUGGEST_CURRENT, suggestCurrent);
@@ -385,6 +437,10 @@ static final class Fields {
385437
static final String SCROLL_TIME = "scroll_time";
386438
static final String SCROLL_TIME_IN_MILLIS = "scroll_time_in_millis";
387439
static final String SCROLL_CURRENT = "scroll_current";
440+
static final String PIT_TOTAL = "point_in_time_total";
441+
static final String PIT_TIME = "point_in_time_time";
442+
static final String PIT_TIME_IN_MILLIS = "point_in_time_time_in_millis";
443+
static final String PIT_CURRENT = "point_in_time_current";
388444
static final String SUGGEST_TOTAL = "suggest_total";
389445
static final String SUGGEST_TIME = "suggest_time";
390446
static final String SUGGEST_TIME_IN_MILLIS = "suggest_time_in_millis";

‎server/src/main/java/org/opensearch/index/search/stats/ShardSearchStats.java

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -187,6 +187,18 @@ public void onFreeScrollContext(ReaderContext readerContext) {
187187
totalStats.scrollMetric.inc(TimeUnit.NANOSECONDS.toMicros(System.nanoTime() - readerContext.getStartTimeInNano()));
188188
}
189189

190+
@Override
191+
public void onNewPitContext(ReaderContext readerContext) {
192+
totalStats.pitCurrent.inc();
193+
}
194+
195+
@Override
196+
public void onFreePitContext(ReaderContext readerContext) {
197+
totalStats.pitCurrent.dec();
198+
assert totalStats.pitCurrent.count() >= 0;
199+
totalStats.pitMetric.inc(TimeUnit.NANOSECONDS.toMicros(System.nanoTime() - readerContext.getStartTimeInNano()));
200+
}
201+
190202
/**
191203
* Holder of statistics values
192204
*
@@ -203,10 +215,12 @@ static final class StatsHolder {
203215
* for one-thousand times as long (i.e., scrolls that execute for almost twelve days on average).
204216
*/
205217
final MeanMetric scrollMetric = new MeanMetric();
218+
final MeanMetric pitMetric = new MeanMetric();
206219
final MeanMetric suggestMetric = new MeanMetric();
207220
final CounterMetric queryCurrent = new CounterMetric();
208221
final CounterMetric fetchCurrent = new CounterMetric();
209222
final CounterMetric scrollCurrent = new CounterMetric();
223+
final CounterMetric pitCurrent = new CounterMetric();
210224
final CounterMetric suggestCurrent = new CounterMetric();
211225

212226
SearchStats.Stats stats() {
@@ -220,6 +234,9 @@ SearchStats.Stats stats() {
220234
scrollMetric.count(),
221235
TimeUnit.MICROSECONDS.toMillis(scrollMetric.sum()),
222236
scrollCurrent.count(),
237+
pitMetric.count(),
238+
TimeUnit.MICROSECONDS.toMillis(pitMetric.sum()),
239+
pitCurrent.count(),
223240
suggestMetric.count(),
224241
TimeUnit.NANOSECONDS.toMillis(suggestMetric.sum()),
225242
suggestCurrent.count()

‎server/src/test/java/org/opensearch/index/search/stats/SearchStatsTests.java

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -45,9 +45,9 @@ public void testShardLevelSearchGroupStats() throws Exception {
4545
// let's create two dummy search stats with groups
4646
Map<String, Stats> groupStats1 = new HashMap<>();
4747
Map<String, Stats> groupStats2 = new HashMap<>();
48-
groupStats2.put("group1", new Stats(1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1));
49-
SearchStats searchStats1 = new SearchStats(new Stats(1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1), 0, groupStats1);
50-
SearchStats searchStats2 = new SearchStats(new Stats(1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1), 0, groupStats2);
48+
groupStats2.put("group1", new Stats(1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1));
49+
SearchStats searchStats1 = new SearchStats(new Stats(1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1), 0, groupStats1);
50+
SearchStats searchStats2 = new SearchStats(new Stats(1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1), 0, groupStats2);
5151

5252
// adding these two search stats and checking group stats are correct
5353
searchStats1.add(searchStats2);
@@ -75,6 +75,9 @@ private static void assertStats(Stats stats, long equalTo) {
7575
assertEquals(equalTo, stats.getScrollCount());
7676
assertEquals(equalTo, stats.getScrollTimeInMillis());
7777
assertEquals(equalTo, stats.getScrollCurrent());
78+
assertEquals(equalTo, stats.getPitCount());
79+
assertEquals(equalTo, stats.getPitTimeInMillis());
80+
assertEquals(equalTo, stats.getPitCurrent());
7881
assertEquals(equalTo, stats.getSuggestCount());
7982
assertEquals(equalTo, stats.getSuggestTimeInMillis());
8083
assertEquals(equalTo, stats.getSuggestCurrent());

‎server/src/test/java/org/opensearch/search/CreatePitSingleNodeTests.java

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,9 @@
2424
import org.opensearch.search.builder.PointInTimeBuilder;
2525
import org.opensearch.search.sort.SortOrder;
2626
import org.opensearch.test.OpenSearchSingleNodeTestCase;
27+
import org.opensearch.index.IndexService;
28+
import org.opensearch.index.shard.IndexShard;
29+
import org.opensearch.indices.IndicesService;
2730

2831
import java.util.Map;
2932
import java.util.concurrent.CountDownLatch;
@@ -74,7 +77,11 @@ public void testCreatePITSuccess() throws ExecutionException, InterruptedExcepti
7477

7578
SearchService service = getInstanceFromNode(SearchService.class);
7679
assertEquals(2, service.getActiveContexts());
80+
validatePitStats("index", 1, 0, 0);
81+
validatePitStats("index", 1, 0, 1);
7782
service.doClose(); // this kills the keep-alive reaper we have to reset the node after this test
83+
validatePitStats("index", 0, 1, 0);
84+
validatePitStats("index", 0, 1, 1);
7885
}
7986

8087
public void testCreatePITWithMultipleIndicesSuccess() throws ExecutionException, InterruptedException {
@@ -91,7 +98,12 @@ public void testCreatePITWithMultipleIndicesSuccess() throws ExecutionException,
9198
PitTestsUtil.assertUsingGetAllPits(client(), response.getId(), response.getCreationTime());
9299
assertEquals(4, response.getSuccessfulShards());
93100
assertEquals(4, service.getActiveContexts());
101+
102+
validatePitStats("index", 1, 0, 0);
103+
validatePitStats("index1", 1, 0, 0);
94104
service.doClose();
105+
validatePitStats("index", 0, 1, 0);
106+
validatePitStats("index1", 0, 1, 0);
95107
}
96108

97109
public void testCreatePITWithShardReplicasSuccess() throws ExecutionException, InterruptedException {
@@ -112,7 +124,11 @@ public void testCreatePITWithShardReplicasSuccess() throws ExecutionException, I
112124

113125
SearchService service = getInstanceFromNode(SearchService.class);
114126
assertEquals(2, service.getActiveContexts());
127+
validatePitStats("index", 1, 0, 0);
128+
validatePitStats("index", 1, 0, 1);
115129
service.doClose();
130+
validatePitStats("index", 0, 1, 0);
131+
validatePitStats("index", 0, 1, 1);
116132
}
117133

118134
public void testCreatePITWithNonExistentIndex() {
@@ -198,6 +214,9 @@ public void testPitSearchOnCloseIndex() throws ExecutionException, InterruptedEx
198214
PitTestsUtil.assertUsingGetAllPits(client(), pitResponse.getId(), pitResponse.getCreationTime());
199215
SearchService service = getInstanceFromNode(SearchService.class);
200216
assertEquals(2, service.getActiveContexts());
217+
validatePitStats("index", 1, 0, 0);
218+
validatePitStats("index", 1, 0, 1);
219+
201220
client().admin().indices().prepareClose("index").get();
202221
SearchPhaseExecutionException ex = expectThrows(SearchPhaseExecutionException.class, () -> {
203222
SearchResponse searchResponse = client().prepareSearch()
@@ -246,7 +265,10 @@ public void testMaxOpenPitContexts() throws Exception {
246265
+ "This limit can be set by changing the [search.max_open_pit_context] setting."
247266
)
248267
);
268+
final int maxPitContexts = SearchService.MAX_OPEN_PIT_CONTEXT.get(Settings.EMPTY);
269+
validatePitStats("index", maxPitContexts, 0, 0);
249270
service.doClose();
271+
validatePitStats("index", 0, maxPitContexts, 0);
250272
}
251273

252274
public void testOpenPitContextsConcurrently() throws Exception {
@@ -292,7 +314,9 @@ public void testOpenPitContextsConcurrently() throws Exception {
292314
thread.join();
293315
}
294316
assertThat(service.getActiveContexts(), equalTo(maxPitContexts));
317+
validatePitStats("index", maxPitContexts, 0, 0);
295318
service.doClose();
319+
validatePitStats("index", 0, maxPitContexts, 0);
296320
}
297321

298322
/**
@@ -461,9 +485,11 @@ public void testPitAfterUpdateIndex() throws Exception {
461485
.getTotalHits().value,
462486
Matchers.equalTo(0L)
463487
);
488+
validatePitStats("test", 1, 0, 0);
464489
} finally {
465490
service.doClose();
466491
assertEquals(0, service.getActiveContexts());
492+
validatePitStats("test", 0, 1, 0);
467493
PitTestsUtil.assertGetAllPitsEmpty(client());
468494
}
469495
}
@@ -505,8 +531,21 @@ public void testConcurrentSearches() throws Exception {
505531

506532
SearchService service = getInstanceFromNode(SearchService.class);
507533
assertEquals(2, service.getActiveContexts());
534+
validatePitStats("index", 1, 0, 0);
535+
validatePitStats("index", 1, 0, 1);
508536
service.doClose();
509537
assertEquals(0, service.getActiveContexts());
538+
validatePitStats("index", 0, 1, 0);
539+
validatePitStats("index", 0, 1, 1);
510540
PitTestsUtil.assertGetAllPitsEmpty(client());
511541
}
542+
543+
public void validatePitStats(String index, long expectedPitCurrent, long expectedPitCount, int shardId) throws ExecutionException,
544+
InterruptedException {
545+
IndicesService indicesService = getInstanceFromNode(IndicesService.class);
546+
IndexService indexService = indicesService.indexServiceSafe(resolveIndex(index));
547+
IndexShard indexShard = indexService.getShard(shardId);
548+
assertEquals(expectedPitCurrent, indexShard.searchStats().getTotal().getPitCurrent());
549+
assertEquals(expectedPitCount, indexShard.searchStats().getTotal().getPitCount());
550+
}
512551
}

‎server/src/test/java/org/opensearch/search/DeletePitMultiNodeTests.java

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,8 @@
1111
import org.junit.After;
1212
import org.junit.Before;
1313
import org.opensearch.action.ActionFuture;
14+
import org.opensearch.action.admin.indices.stats.IndicesStatsRequest;
15+
import org.opensearch.action.admin.indices.stats.IndicesStatsResponse;
1416
import org.opensearch.action.ActionListener;
1517
import org.opensearch.action.LatchedActionListener;
1618
import org.opensearch.action.search.CreatePitAction;
@@ -76,6 +78,7 @@ public void testDeletePit() throws Exception {
7678
execute = client().execute(CreatePitAction.INSTANCE, request);
7779
pitResponse = execute.get();
7880
pitIds.add(pitResponse.getId());
81+
validatePitStats("index", 10, 0);
7982
DeletePitRequest deletePITRequest = new DeletePitRequest(pitIds);
8083
ActionFuture<DeletePitResponse> deleteExecute = client().execute(DeletePitAction.INSTANCE, deletePITRequest);
8184
DeletePitResponse deletePITResponse = deleteExecute.get();
@@ -84,6 +87,7 @@ public void testDeletePit() throws Exception {
8487
assertTrue(pitIds.contains(deletePitInfo.getPitId()));
8588
assertTrue(deletePitInfo.isSuccessful());
8689
}
90+
validatePitStats("index", 0, 10);
8791
/**
8892
* Checking deleting the same PIT id again results in succeeded
8993
*/
@@ -102,6 +106,7 @@ public void testDeletePitWithValidAndDeletedIds() throws Exception {
102106
CreatePitResponse pitResponse = execute.get();
103107
List<String> pitIds = new ArrayList<>();
104108
pitIds.add(pitResponse.getId());
109+
validatePitStats("index", 5, 0);
105110

106111
/**
107112
* Delete Pit #1
@@ -113,9 +118,11 @@ public void testDeletePitWithValidAndDeletedIds() throws Exception {
113118
assertTrue(pitIds.contains(deletePitInfo.getPitId()));
114119
assertTrue(deletePitInfo.isSuccessful());
115120
}
121+
validatePitStats("index", 0, 5);
116122
execute = client().execute(CreatePitAction.INSTANCE, request);
117123
pitResponse = execute.get();
118124
pitIds.add(pitResponse.getId());
125+
validatePitStats("index", 5, 5);
119126
/**
120127
* Delete PIT with both Ids #1 (which is deleted) and #2 (which is present)
121128
*/
@@ -126,6 +133,7 @@ public void testDeletePitWithValidAndDeletedIds() throws Exception {
126133
assertTrue(pitIds.contains(deletePitInfo.getPitId()));
127134
assertTrue(deletePitInfo.isSuccessful());
128135
}
136+
validatePitStats("index", 0, 10);
129137
}
130138

131139
public void testDeletePitWithValidAndInvalidIds() throws Exception {
@@ -148,6 +156,8 @@ public void testDeleteAllPits() throws Exception {
148156
client().prepareIndex("index1").setId("1").setSource("field", "value").setRefreshPolicy(IMMEDIATE).execute().get();
149157
ensureGreen();
150158
createPitOnIndex("index1");
159+
validatePitStats("index", 5, 0);
160+
validatePitStats("index1", 5, 0);
151161
DeletePitRequest deletePITRequest = new DeletePitRequest("_all");
152162

153163
/**
@@ -160,6 +170,8 @@ public void testDeleteAllPits() throws Exception {
160170
assertThat(deletePitInfo.getPitId(), not(blankOrNullString()));
161171
assertTrue(deletePitInfo.isSuccessful());
162172
}
173+
validatePitStats("index", 0, 5);
174+
validatePitStats("index1", 0, 5);
163175
client().admin().indices().prepareDelete("index1").get();
164176
}
165177

@@ -324,4 +336,16 @@ public void onFailure(Exception e) {}
324336
}
325337
}
326338

339+
public void validatePitStats(String index, long expectedPitCurrent, long expectedPitCount) throws ExecutionException,
340+
InterruptedException {
341+
IndicesStatsRequest indicesStatsRequest = new IndicesStatsRequest();
342+
indicesStatsRequest.indices(index);
343+
indicesStatsRequest.all();
344+
IndicesStatsResponse indicesStatsResponse = client().admin().indices().stats(indicesStatsRequest).get();
345+
long pitCurrent = indicesStatsResponse.getIndex(index).getTotal().search.getTotal().getPitCurrent();
346+
long pitCount = indicesStatsResponse.getIndex(index).getTotal().search.getTotal().getPitCount();
347+
assertEquals(expectedPitCurrent, pitCurrent);
348+
assertEquals(expectedPitCount, pitCount);
349+
}
350+
327351
}

0 commit comments

Comments
 (0)