Adding concurrent search versions of query count and time metrics

Signed-off-by: Jay Deng <[email protected]>

CHANGELOG.md

@@ -89,7 +89,8 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - [Segment Replication] Support realtime reads for GET requests ([#9212](https://github.com/opensearch-project/OpenSearch/pull/9212))
 - [Feature] Expose term frequency in Painless script score context ([#9081](https://github.com/opensearch-project/OpenSearch/pull/9081))
 - Add support for reading partial files to HDFS repository ([#9513](https://github.com/opensearch-project/OpenSearch/issues/9513))
-- Add support for extensions to search responses using SearchExtBuilder ([#9379](https://github.com/opensearch-project/OpenSearch/pull/9379)) 
+- Add support for extensions to search responses using SearchExtBuilder ([#9379](https://github.com/opensearch-project/OpenSearch/pull/9379))
+- Add concurrent segment search related metrics to node and index stats ([#9622](https://github.com/opensearch-project/OpenSearch/issues/9622))
 
 ### Dependencies
 - Bump `org.apache.logging.log4j:log4j-core` from 2.17.1 to 2.20.0 ([#8307](https://github.com/opensearch-project/OpenSearch/pull/8307))
@@ -181,4 +182,4 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 ### Security
 
 [Unreleased 3.0]: https://github.com/opensearch-project/OpenSearch/compare/2.x...HEAD
-[Unreleased 2.x]: https://github.com/opensearch-project/OpenSearch/compare/2.10...2.x
+[Unreleased 2.x]: https://github.com/opensearch-project/OpenSearch/compare/2.10...2.x

rest-api-spec/src/main/resources/rest-api-spec/test/cat.shards/10_basic.yml

@@ -1,14 +1,13 @@
----
 "Help":
   - skip:
-      version: " - 2.3.99"
-      reason:  point in time stats were added in 2.4.0
+      version: " - 2.9.99"
+      reason:  concurrent search stats were added in 2.10.0
       features: node_selector
   - do:
       cat.shards:
         help: true
       node_selector:
-        version: "2.4.0 - "
+        version: "2.10.0 - "
 
   - match:
       $body: |
@@ -67,6 +66,9 @@
                     search.query_current             .+   \n
                     search.query_time                .+   \n
                     search.query_total               .+   \n
+                    search.concurrent_query_total    .+   \n
+                    search.concurrent_query_time     .+   \n
+                    search.concurrent_query_current  .+   \n
                     search.scroll_current            .+   \n
                     search.scroll_time               .+   \n
                     search.scroll_total              .+   \n

server/src/internalClusterTest/java/org/opensearch/indices/stats/IndexStatsIT.java

@@ -55,6 +55,7 @@
 import org.opensearch.common.io.stream.BytesStreamOutput;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.common.unit.TimeValue;
+import org.opensearch.common.util.FeatureFlags;
 import org.opensearch.core.action.support.DefaultShardOperationFailedException;
 import org.opensearch.core.common.bytes.BytesReference;
 import org.opensearch.core.common.io.stream.StreamOutput;
@@ -143,6 +144,12 @@ public Settings indexSettings() {
             .build();
     }
 
+    @Override
+    // Do not parameterize this test, instead let individual test methods modify concurrent search settings as needed.
+    protected Settings featureFlagSettings() {
+        return Settings.builder().put(super.featureFlagSettings()).put(FeatureFlags.CONCURRENT_SEGMENT_SEARCH, "true").build();
+    }
+
     private Settings.Builder settingsBuilder() {
         return Settings.builder().put(indexSettings());
     }
@@ -1090,6 +1097,45 @@ public void testGroupsParam() throws Exception {
 
     }
 
+    public void testConcurrentQueryCount() throws Exception {
+        int NUM_SHARDS = randomIntBetween(1, 5);
+        createIndex(
+            "test1",
+            Settings.builder()
+                .put(indexSettings())
+                .put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, NUM_SHARDS)
+                .put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 0)
+                .build()
+        );
+        createIndex(
+            "test2",
+            Settings.builder()
+                .put(indexSettings())
+                .put("search.concurrent_segment_search.enabled", false)
+                .put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, NUM_SHARDS)
+                .put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 0)
+                .build()
+        );
+
+        ensureGreen();
+
+        client().prepareIndex("test1").setId(Integer.toString(1)).setSource("foo", "bar").execute().actionGet();
+        client().prepareIndex("test2").setId(Integer.toString(1)).setSource("foo", "bar").execute().actionGet();
+        refresh();
+
+        client().prepareSearch("_all").execute().actionGet();
+        client().prepareSearch("test1").execute().actionGet();
+        client().prepareSearch("test2").execute().actionGet();
+
+        IndicesStatsRequestBuilder builder = client().admin().indices().prepareStats();
+        IndicesStatsResponse stats = builder.execute().actionGet();
+
+        assertEquals(4 * NUM_SHARDS, stats.getTotal().search.getTotal().getQueryCount());
+        assertEquals(2 * NUM_SHARDS, stats.getTotal().search.getTotal().getConcurrentQueryCount());
+        assertThat(stats.getTotal().search.getTotal().getQueryTimeInMillis(), greaterThan(0L));
+        assertThat(stats.getTotal().search.getTotal().getConcurrentQueryTimeInMillis(), greaterThan(0L));
+    }
+
     private static void set(Flag flag, IndicesStatsRequestBuilder builder, boolean set) {
         switch (flag) {
             case Docs:

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

@@ -67,6 +67,10 @@ public static class Stats implements Writeable, ToXContentFragment {
         private long queryTimeInMillis;
         private long queryCurrent;
 
+        private long concurrentQueryCount;
+        private long concurrentQueryTimeInMillis;
+        private long concurrentQueryCurrent;
+
         private long fetchCount;
         private long fetchTimeInMillis;
         private long fetchCurrent;
@@ -91,6 +95,9 @@ public Stats(
             long queryCount,
             long queryTimeInMillis,
             long queryCurrent,
+            long concurrentQueryCount,
+            long concurrentQueryTimeInMillis,
+            long concurrentQueryCurrent,
             long fetchCount,
             long fetchTimeInMillis,
             long fetchCurrent,
@@ -108,6 +115,10 @@ public Stats(
             this.queryTimeInMillis = queryTimeInMillis;
             this.queryCurrent = queryCurrent;
 
+            this.concurrentQueryCount = concurrentQueryCount;
+            this.concurrentQueryTimeInMillis = concurrentQueryTimeInMillis;
+            this.concurrentQueryCurrent = concurrentQueryCurrent;
+
             this.fetchCount = fetchCount;
             this.fetchTimeInMillis = fetchTimeInMillis;
             this.fetchCurrent = fetchCurrent;
@@ -130,6 +141,12 @@ private Stats(StreamInput in) throws IOException {
             queryTimeInMillis = in.readVLong();
             queryCurrent = in.readVLong();
 
+            if (in.getVersion().onOrAfter(Version.V_2_10_0)) {
+                concurrentQueryCount = in.readVLong();
+                concurrentQueryTimeInMillis = in.readVLong();
+                concurrentQueryCurrent = in.readVLong();
+            }
+
             fetchCount = in.readVLong();
             fetchTimeInMillis = in.readVLong();
             fetchCurrent = in.readVLong();
@@ -154,6 +171,10 @@ public void add(Stats stats) {
             queryTimeInMillis += stats.queryTimeInMillis;
             queryCurrent += stats.queryCurrent;
 
+            concurrentQueryCount += stats.concurrentQueryCount;
+            concurrentQueryTimeInMillis += stats.concurrentQueryTimeInMillis;
+            concurrentQueryCurrent += stats.concurrentQueryCurrent;
+
             fetchCount += stats.fetchCount;
             fetchTimeInMillis += stats.fetchTimeInMillis;
             fetchCurrent += stats.fetchCurrent;
@@ -175,6 +196,9 @@ public void addForClosingShard(Stats stats) {
             queryCount += stats.queryCount;
             queryTimeInMillis += stats.queryTimeInMillis;
 
+            concurrentQueryCount += stats.concurrentQueryCount;
+            concurrentQueryTimeInMillis += stats.concurrentQueryTimeInMillis;
+
             fetchCount += stats.fetchCount;
             fetchTimeInMillis += stats.fetchTimeInMillis;
 
@@ -207,6 +231,22 @@ public long getQueryCurrent() {
             return queryCurrent;
         }
 
+        public long getConcurrentQueryCount() {
+            return concurrentQueryCount;
+        }
+
+        public TimeValue getConcurrentQueryTime() {
+            return new TimeValue(concurrentQueryTimeInMillis);
+        }
+
+        public long getConcurrentQueryTimeInMillis() {
+            return concurrentQueryTimeInMillis;
+        }
+
+        public long getConcurrentQueryCurrent() {
+            return concurrentQueryCurrent;
+        }
+
         public long getFetchCount() {
             return fetchCount;
         }
@@ -281,6 +321,12 @@ public void writeTo(StreamOutput out) throws IOException {
             out.writeVLong(queryTimeInMillis);
             out.writeVLong(queryCurrent);
 
+            if (out.getVersion().onOrAfter(Version.V_2_10_0)) {
+                out.writeVLong(concurrentQueryCount);
+                out.writeVLong(concurrentQueryTimeInMillis);
+                out.writeVLong(concurrentQueryCurrent);
+            }
+
             out.writeVLong(fetchCount);
             out.writeVLong(fetchTimeInMillis);
             out.writeVLong(fetchCurrent);
@@ -306,6 +352,10 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
             builder.humanReadableField(Fields.QUERY_TIME_IN_MILLIS, Fields.QUERY_TIME, getQueryTime());
             builder.field(Fields.QUERY_CURRENT, queryCurrent);
 
+            builder.field(Fields.CONCURRENT_QUERY_TOTAL, concurrentQueryCount);
+            builder.humanReadableField(Fields.CONCURRENT_QUERY_TIME_IN_MILLIS, Fields.CONCURRENT_QUERY_TIME, getConcurrentQueryTime());
+            builder.field(Fields.CONCURRENT_QUERY_CURRENT, concurrentQueryCurrent);
+
             builder.field(Fields.FETCH_TOTAL, fetchCount);
             builder.humanReadableField(Fields.FETCH_TIME_IN_MILLIS, Fields.FETCH_TIME, getFetchTime());
             builder.field(Fields.FETCH_CURRENT, fetchCurrent);
@@ -430,6 +480,10 @@ static final class Fields {
         static final String QUERY_TIME = "query_time";
         static final String QUERY_TIME_IN_MILLIS = "query_time_in_millis";
         static final String QUERY_CURRENT = "query_current";
+        static final String CONCURRENT_QUERY_TOTAL = "concurrent_query_total";
+        static final String CONCURRENT_QUERY_TIME = "concurrent_query_time";
+        static final String CONCURRENT_QUERY_TIME_IN_MILLIS = "concurrent_query_time_in_millis";
+        static final String CONCURRENT_QUERY_CURRENT = "concurrent_query_current";
         static final String FETCH_TOTAL = "fetch_total";
         static final String FETCH_TIME = "fetch_time";
         static final String FETCH_TIME_IN_MILLIS = "fetch_time_in_millis";

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

@@ -91,6 +91,9 @@ public void onPreQueryPhase(SearchContext searchContext) {
                 statsHolder.suggestCurrent.inc();
             } else {
                 statsHolder.queryCurrent.inc();
+                if (searchContext.shouldUseConcurrentSearch()) {
+                    statsHolder.concurrentQueryCurrent.inc();
+                }
             }
         });
     }
@@ -104,6 +107,10 @@ public void onFailedQueryPhase(SearchContext searchContext) {
             } else {
                 statsHolder.queryCurrent.dec();
                 assert statsHolder.queryCurrent.count() >= 0;
+                if (searchContext.shouldUseConcurrentSearch()) {
+                    statsHolder.concurrentQueryCurrent.dec();
+                    assert statsHolder.concurrentQueryCurrent.count() >= 0;
+                }
             }
         });
     }
@@ -119,6 +126,11 @@ public void onQueryPhase(SearchContext searchContext, long tookInNanos) {
                 statsHolder.queryMetric.inc(tookInNanos);
                 statsHolder.queryCurrent.dec();
                 assert statsHolder.queryCurrent.count() >= 0;
+                if (searchContext.shouldUseConcurrentSearch()) {
+                    statsHolder.concurrentQueryMetric.inc(tookInNanos);
+                    statsHolder.concurrentQueryCurrent.dec();
+                    assert statsHolder.concurrentQueryCurrent.count() >= 0;
+                }
             }
         });
     }
@@ -206,6 +218,7 @@ public void onFreePitContext(ReaderContext readerContext) {
      */
     static final class StatsHolder {
         final MeanMetric queryMetric = new MeanMetric();
+        final MeanMetric concurrentQueryMetric = new MeanMetric();
         final MeanMetric fetchMetric = new MeanMetric();
         /* We store scroll statistics in microseconds because with nanoseconds we run the risk of overflowing the total stats if there are
          * many scrolls. For example, on a system with 2^24 scrolls that have been executed, each executing for 2^10 seconds, then using
@@ -218,6 +231,7 @@ static final class StatsHolder {
         final MeanMetric pitMetric = new MeanMetric();
         final MeanMetric suggestMetric = new MeanMetric();
         final CounterMetric queryCurrent = new CounterMetric();
+        final CounterMetric concurrentQueryCurrent = new CounterMetric();
         final CounterMetric fetchCurrent = new CounterMetric();
         final CounterMetric scrollCurrent = new CounterMetric();
         final CounterMetric pitCurrent = new CounterMetric();
@@ -228,6 +242,9 @@ SearchStats.Stats stats() {
                 queryMetric.count(),
                 TimeUnit.NANOSECONDS.toMillis(queryMetric.sum()),
                 queryCurrent.count(),
+                concurrentQueryMetric.count(),
+                TimeUnit.NANOSECONDS.toMillis(concurrentQueryMetric.sum()),
+                concurrentQueryCurrent.count(),
                 fetchMetric.count(),
                 TimeUnit.NANOSECONDS.toMillis(fetchMetric.sum()),
                 fetchCurrent.count(),

server/src/main/java/org/opensearch/rest/action/cat/RestIndicesAction.java

@@ -581,6 +581,23 @@ protected Table getTableWithHeader(final RestRequest request) {
             "sibling:pri;alias:sqto,searchQueryTotal;default:false;text-align:right;desc:total query phase ops"
         );
         table.addCell("pri.search.query_total", "default:false;text-align:right;desc:total query phase ops");
+        table.addCell(
+            "search.concurrent_query_current",
+            "sibling:pri;alias:scqc,searchConcurrentQueryCurrent;default:false;text-align:right;desc:current concurrent query phase ops"
+        );
+        table.addCell("pri.search.concurrent_query_current", "default:false;text-align:right;desc:current concurrent query phase ops");
+
+        table.addCell(
+            "search.concurrent_query_time",
+            "sibling:pri;alias:scqti,searchConcurrentQueryTime;default:false;text-align:right;desc:time spent in concurrent query phase"
+        );
+        table.addCell("pri.search.concurrent_query_time", "default:false;text-align:right;desc:time spent in concurrent query phase");
+
+        table.addCell(
+            "search.concurrent_query_total",
+            "sibling:pri;alias:scqto,searchConcurrentQueryTotal;default:false;text-align:right;desc:total query phase ops"
+        );
+        table.addCell("pri.search.concurrent_query_total", "default:false;text-align:right;desc:total query phase ops");
 
         table.addCell(
             "search.scroll_current",
@@ -890,6 +907,15 @@ Table buildTable(
             table.addCell(totalStats.getSearch() == null ? null : totalStats.getSearch().getTotal().getQueryCount());
             table.addCell(primaryStats.getSearch() == null ? null : primaryStats.getSearch().getTotal().getQueryCount());
 
+            table.addCell(totalStats.getSearch() == null ? null : totalStats.getSearch().getTotal().getConcurrentQueryCurrent());
+            table.addCell(primaryStats.getSearch() == null ? null : primaryStats.getSearch().getTotal().getConcurrentQueryCurrent());
+
+            table.addCell(totalStats.getSearch() == null ? null : totalStats.getSearch().getTotal().getConcurrentQueryTime());
+            table.addCell(primaryStats.getSearch() == null ? null : primaryStats.getSearch().getTotal().getConcurrentQueryTime());
+
+            table.addCell(totalStats.getSearch() == null ? null : totalStats.getSearch().getTotal().getConcurrentQueryCount());
+            table.addCell(primaryStats.getSearch() == null ? null : primaryStats.getSearch().getTotal().getConcurrentQueryCount());
+
             table.addCell(totalStats.getSearch() == null ? null : totalStats.getSearch().getTotal().getScrollCurrent());
             table.addCell(primaryStats.getSearch() == null ? null : primaryStats.getSearch().getTotal().getScrollCurrent());
 

server/src/main/java/org/opensearch/rest/action/cat/RestNodesAction.java

@@ -303,6 +303,18 @@ protected Table getTableWithHeader(final RestRequest request) {
         table.addCell("search.query_current", "alias:sqc,searchQueryCurrent;default:false;text-align:right;desc:current query phase ops");
         table.addCell("search.query_time", "alias:sqti,searchQueryTime;default:false;text-align:right;desc:time spent in query phase");
         table.addCell("search.query_total", "alias:sqto,searchQueryTotal;default:false;text-align:right;desc:total query phase ops");
+        table.addCell(
+            "search.concurrent_query_current",
+            "alias:scqc,searchConcurrentQueryCurrent;default:false;text-align:right;desc:current concurrent query phase ops"
+        );
+        table.addCell(
+            "search.concurrent_query_time",
+            "alias:scqti,searchConcurrentQueryTime;default:false;text-align:right;desc:time spent in concurrent query phase"
+        );
+        table.addCell(
+            "search.concurrent_query_total",
+            "alias:scqto,searchConcurrentQueryTotal;default:false;text-align:right;desc:total concurrent query phase ops"
+        );
         table.addCell("search.scroll_current", "alias:scc,searchScrollCurrent;default:false;text-align:right;desc:open scroll contexts");
         table.addCell(
             "search.scroll_time",
@@ -529,6 +541,9 @@ Table buildTable(
             table.addCell(searchStats == null ? null : searchStats.getTotal().getQueryCurrent());
             table.addCell(searchStats == null ? null : searchStats.getTotal().getQueryTime());
             table.addCell(searchStats == null ? null : searchStats.getTotal().getQueryCount());
+            table.addCell(searchStats == null ? null : searchStats.getTotal().getConcurrentQueryCurrent());
+            table.addCell(searchStats == null ? null : searchStats.getTotal().getConcurrentQueryTime());
+            table.addCell(searchStats == null ? null : searchStats.getTotal().getConcurrentQueryCount());
             table.addCell(searchStats == null ? null : searchStats.getTotal().getScrollCurrent());
             table.addCell(searchStats == null ? null : searchStats.getTotal().getScrollTime());
             table.addCell(searchStats == null ? null : searchStats.getTotal().getScrollCount());