Add a counter to node stat api to track shard going from idle to non-idle

CHANGELOG.md

@@ -118,6 +118,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - [Tiered caching] Add Stale keys Management and CacheCleaner to IndicesRequestCache ([#12625](https://github.com/opensearch-project/OpenSearch/pull/12625))
 - [Tiered caching] Add serializer integration to allow ehcache disk cache to use non-primitive values ([#12709](https://github.com/opensearch-project/OpenSearch/pull/12709))
 - [Admission Control] Integrated IO Based AdmissionController to AdmissionControl Framework ([#12583](https://github.com/opensearch-project/OpenSearch/pull/12583))
+- Add a counter to node stat api to track shard going from idle to non-idle ([#12768](https://github.com/opensearch-project/OpenSearch/pull/12768))
 
 ### Dependencies
 - Bump `peter-evans/find-comment` from 2 to 3 ([#12288](https://github.com/opensearch-project/OpenSearch/pull/12288))

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

@@ -163,6 +163,8 @@ public static class Stats implements Writeable, ToXContentFragment {
         private long pitTimeInMillis;
         private long pitCurrent;
 
+        private long searchIdleWakenUpCount;
+
         @Nullable
         private RequestStatsLongHolder requestStatsLongHolder;
 
@@ -193,7 +195,8 @@ public Stats(
             long pitCurrent,
             long suggestCount,
             long suggestTimeInMillis,
-            long suggestCurrent
+            long suggestCurrent,
+            long searchIdleWakenUpCount
         ) {
             this.requestStatsLongHolder = new RequestStatsLongHolder();
             this.queryCount = queryCount;
@@ -220,6 +223,8 @@ public Stats(
             this.pitCount = pitCount;
             this.pitTimeInMillis = pitTimeInMillis;
             this.pitCurrent = pitCurrent;
+
+            this.searchIdleWakenUpCount = searchIdleWakenUpCount;
         }
 
         private Stats(StreamInput in) throws IOException {
@@ -255,6 +260,10 @@ private Stats(StreamInput in) throws IOException {
                 concurrentQueryCurrent = in.readVLong();
                 queryConcurrency = in.readVLong();
             }
+
+            if (in.getVersion().onOrAfter(Version.V_3_0_0)) {
+                searchIdleWakenUpCount = in.readVLong();
+            }
         }
 
         public void add(Stats stats) {
@@ -282,6 +291,8 @@ public void add(Stats stats) {
             pitCount += stats.pitCount;
             pitTimeInMillis += stats.pitTimeInMillis;
             pitCurrent += stats.pitCurrent;
+
+            searchIdleWakenUpCount += stats.searchIdleWakenUpCount;
         }
 
         public void addForClosingShard(Stats stats) {
@@ -306,6 +317,8 @@ public void addForClosingShard(Stats stats) {
             pitTimeInMillis += stats.pitTimeInMillis;
             pitCurrent += stats.pitCurrent;
             queryConcurrency += stats.queryConcurrency;
+
+            searchIdleWakenUpCount += stats.searchIdleWakenUpCount;
         }
 
         public long getQueryCount() {
@@ -412,6 +425,10 @@ public long getSuggestCurrent() {
             return suggestCurrent;
         }
 
+        public long getSearchIdleWakenUpCount() {
+            return searchIdleWakenUpCount;
+        }
+
         public static Stats readStats(StreamInput in) throws IOException {
             return new Stats(in);
         }
@@ -457,6 +474,10 @@ public void writeTo(StreamOutput out) throws IOException {
                 out.writeVLong(concurrentQueryCurrent);
                 out.writeVLong(queryConcurrency);
             }
+
+            if (out.getVersion().onOrAfter(Version.V_3_0_0)) {
+                out.writeVLong(searchIdleWakenUpCount);
+            }
         }
 
         @Override
@@ -486,6 +507,8 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
             builder.humanReadableField(Fields.SUGGEST_TIME_IN_MILLIS, Fields.SUGGEST_TIME, getSuggestTime());
             builder.field(Fields.SUGGEST_CURRENT, suggestCurrent);
 
+            builder.field(Fields.SEARCH_IDLE_WAKEN_UP_TOTAL, searchIdleWakenUpCount);
+
             if (requestStatsLongHolder != null) {
                 builder.startObject(Fields.REQUEST);
 
@@ -654,6 +677,7 @@ static final class Fields {
         static final String TIME = "time";
         static final String CURRENT = "current";
         static final String TOTAL = "total";
+        static final String SEARCH_IDLE_WAKEN_UP_TOTAL = "search_idle_waken_up_total";
 
     }
 

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

@@ -213,6 +213,11 @@ public void onFreePitContext(ReaderContext readerContext) {
         totalStats.pitMetric.inc(TimeUnit.NANOSECONDS.toMicros(System.nanoTime() - readerContext.getStartTimeInNano()));
     }
 
+    @Override
+    public void onNewSearchIdleWakenUp() {
+        totalStats.searchIdleMetric.inc();
+    }
+
     /**
      * Holder of statistics values
      *
@@ -239,6 +244,7 @@ static final class StatsHolder {
         final CounterMetric scrollCurrent = new CounterMetric();
         final CounterMetric pitCurrent = new CounterMetric();
         final CounterMetric suggestCurrent = new CounterMetric();
+        final CounterMetric searchIdleMetric = new CounterMetric();
 
         SearchStats.Stats stats() {
             return new SearchStats.Stats(
@@ -260,7 +266,8 @@ SearchStats.Stats stats() {
                 pitCurrent.count(),
                 suggestMetric.count(),
                 TimeUnit.NANOSECONDS.toMillis(suggestMetric.sum()),
-                suggestCurrent.count()
+                suggestCurrent.count(),
+                searchIdleMetric.count()
             );
         }
     }

server/src/main/java/org/opensearch/index/shard/IndexShard.java

@@ -1859,6 +1859,10 @@ public Engine.Searcher acquireSearcher(String source) {
     }
 
     private void markSearcherAccessed() {
+        if (isSearchIdle()) {
+            SearchOperationListener searchOperationListener = getSearchOperationListener();
+            searchOperationListener.onNewSearchIdleWakenUp();
+        }
         lastSearcherAccess.lazySet(threadPool.relativeTimeInMillis());
     }
 

server/src/main/java/org/opensearch/index/shard/SearchOperationListener.java

@@ -145,6 +145,11 @@
      */
     default void onFreePitContext(ReaderContext readerContext) {}
 
+    /**
+     * Executed when a shard goes from idle to non-idle state
+     */
+    default void onNewSearchIdleWakenUp() {}
+
     /**
      * A Composite listener that multiplexes calls to each of the listeners methods.
      */
@@ -310,5 +315,16 @@
                 }
             }
         }
+
+        @Override
+        public void onNewSearchIdleWakenUp() {
+            for (SearchOperationListener listener : listeners) {
+                try {
+                    listener.onNewSearchIdleWakenUp();
+                } catch (Exception e) {
+                    logger.warn(() -> new ParameterizedMessage("onNewSearchIdleWakenUp listener [{}] failed", listener), e);
+                }
+            }
+        }
     }
 }

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

@@ -57,9 +57,9 @@ public void testShardLevelSearchGroupStats() throws Exception {
         // let's create two dummy search stats with groups
         Map<String, Stats> groupStats1 = new HashMap<>();
         Map<String, Stats> groupStats2 = new HashMap<>();
-        groupStats2.put("group1", new Stats(1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1));
-        SearchStats searchStats1 = new SearchStats(new Stats(1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1), 0, groupStats1);
-        SearchStats searchStats2 = new SearchStats(new Stats(1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1), 0, groupStats2);
+        groupStats2.put("group1", new Stats(1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1));
+        SearchStats searchStats1 = new SearchStats(new Stats(1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1), 0, groupStats1);
+        SearchStats searchStats2 = new SearchStats(new Stats(1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1), 0, groupStats2);
 
         // adding these two search stats and checking group stats are correct
         searchStats1.add(searchStats2);
@@ -128,6 +128,7 @@ private static void assertStats(Stats stats, long equalTo) {
         assertEquals(equalTo, stats.getSuggestCount());
         assertEquals(equalTo, stats.getSuggestTimeInMillis());
         assertEquals(equalTo, stats.getSuggestCurrent());
+        assertEquals(equalTo, stats.getSearchIdleWakenUpCount());
         // avg_concurrency is not summed up across stats
         assertEquals(1, stats.getConcurrentAvgSliceCount(), 0);
     }

server/src/test/java/org/opensearch/index/shard/SearchOperationListenerTests.java

@@ -64,6 +64,7 @@ public void testListenersAreExecuted() {
         AtomicInteger newScrollContext = new AtomicInteger();
         AtomicInteger freeScrollContext = new AtomicInteger();
         AtomicInteger validateSearchContext = new AtomicInteger();
+        AtomicInteger searchIdleWakenUp = new AtomicInteger();
         AtomicInteger timeInNanos = new AtomicInteger(randomIntBetween(0, 10));
         SearchOperationListener listener = new SearchOperationListener() {
             @Override
@@ -133,6 +134,11 @@ public void validateReaderContext(ReaderContext readerContext, TransportRequest
                 assertNotNull(readerContext);
                 validateSearchContext.incrementAndGet();
             }
+
+            @Override
+            public void onNewSearchIdleWakenUp() {
+                searchIdleWakenUp.incrementAndGet();
+            }
         };
 
         SearchOperationListener throwingListener = (SearchOperationListener) Proxy.newProxyInstance(
@@ -169,6 +175,7 @@ public void validateReaderContext(ReaderContext readerContext, TransportRequest
         assertEquals(0, newScrollContext.get());
         assertEquals(0, freeContext.get());
         assertEquals(0, freeScrollContext.get());
+        assertEquals(0, searchIdleWakenUp.get());
         assertEquals(0, validateSearchContext.get());
 
         compositeListener.onFetchPhase(ctx, timeInNanos.get());
@@ -182,6 +189,7 @@ public void validateReaderContext(ReaderContext readerContext, TransportRequest
         assertEquals(0, newScrollContext.get());
         assertEquals(0, freeContext.get());
         assertEquals(0, freeScrollContext.get());
+        assertEquals(0, searchIdleWakenUp.get());
         assertEquals(0, validateSearchContext.get());
 
         compositeListener.onPreQueryPhase(ctx);
@@ -195,6 +203,7 @@ public void validateReaderContext(ReaderContext readerContext, TransportRequest
         assertEquals(0, newScrollContext.get());
         assertEquals(0, freeContext.get());
         assertEquals(0, freeScrollContext.get());
+        assertEquals(0, searchIdleWakenUp.get());
         assertEquals(0, validateSearchContext.get());
 
         compositeListener.onPreFetchPhase(ctx);
@@ -208,6 +217,7 @@ public void validateReaderContext(ReaderContext readerContext, TransportRequest
         assertEquals(0, newScrollContext.get());
         assertEquals(0, freeContext.get());
         assertEquals(0, freeScrollContext.get());
+        assertEquals(0, searchIdleWakenUp.get());
         assertEquals(0, validateSearchContext.get());
 
         compositeListener.onFailedFetchPhase(ctx);
@@ -221,6 +231,7 @@ public void validateReaderContext(ReaderContext readerContext, TransportRequest
         assertEquals(0, newScrollContext.get());
         assertEquals(0, freeContext.get());
         assertEquals(0, freeScrollContext.get());
+        assertEquals(0, searchIdleWakenUp.get());
         assertEquals(0, validateSearchContext.get());
 
         compositeListener.onFailedQueryPhase(ctx);
@@ -234,6 +245,7 @@ public void validateReaderContext(ReaderContext readerContext, TransportRequest
         assertEquals(0, newScrollContext.get());
         assertEquals(0, freeContext.get());
         assertEquals(0, freeScrollContext.get());
+        assertEquals(0, searchIdleWakenUp.get());
         assertEquals(0, validateSearchContext.get());
 
         compositeListener.onNewReaderContext(mock(ReaderContext.class));
@@ -247,6 +259,7 @@ public void validateReaderContext(ReaderContext readerContext, TransportRequest
         assertEquals(0, newScrollContext.get());
         assertEquals(0, freeContext.get());
         assertEquals(0, freeScrollContext.get());
+        assertEquals(0, searchIdleWakenUp.get());
         assertEquals(0, validateSearchContext.get());
 
         compositeListener.onNewScrollContext(mock(ReaderContext.class));
@@ -260,6 +273,7 @@ public void validateReaderContext(ReaderContext readerContext, TransportRequest
         assertEquals(2, newScrollContext.get());
         assertEquals(0, freeContext.get());
         assertEquals(0, freeScrollContext.get());
+        assertEquals(0, searchIdleWakenUp.get());
         assertEquals(0, validateSearchContext.get());
 
         compositeListener.onFreeReaderContext(mock(ReaderContext.class));
@@ -273,6 +287,7 @@ public void validateReaderContext(ReaderContext readerContext, TransportRequest
         assertEquals(2, newScrollContext.get());
         assertEquals(2, freeContext.get());
         assertEquals(0, freeScrollContext.get());
+        assertEquals(0, searchIdleWakenUp.get());
         assertEquals(0, validateSearchContext.get());
 
         compositeListener.onFreeScrollContext(mock(ReaderContext.class));
@@ -286,6 +301,21 @@ public void validateReaderContext(ReaderContext readerContext, TransportRequest
         assertEquals(2, newScrollContext.get());
         assertEquals(2, freeContext.get());
         assertEquals(2, freeScrollContext.get());
+        assertEquals(0, searchIdleWakenUp.get());
+        assertEquals(0, validateSearchContext.get());
+
+        compositeListener.onNewSearchIdleWakenUp();
+        assertEquals(2, preFetch.get());
+        assertEquals(2, preQuery.get());
+        assertEquals(2, failedFetch.get());
+        assertEquals(2, failedQuery.get());
+        assertEquals(2, onQuery.get());
+        assertEquals(2, onFetch.get());
+        assertEquals(2, newContext.get());
+        assertEquals(2, newScrollContext.get());
+        assertEquals(2, freeContext.get());
+        assertEquals(2, freeScrollContext.get());
+        assertEquals(2, searchIdleWakenUp.get());
         assertEquals(0, validateSearchContext.get());
 
         if (throwingListeners == 0) {
@@ -311,6 +341,7 @@ public void validateReaderContext(ReaderContext readerContext, TransportRequest
         assertEquals(2, newScrollContext.get());
         assertEquals(2, freeContext.get());
         assertEquals(2, freeScrollContext.get());
+        assertEquals(2, searchIdleWakenUp.get());
         assertEquals(2, validateSearchContext.get());
     }
 }