Count scans in search stats and add metrics for scrolls

Each scroll on a scan causes a query to be executed. This commit adds support for these indirect queries to count against the search stats.
Additionally, this commit adds three new search stats: scroll_count, scroll_time_in_millis, and scroll_current. scroll_count tracks the
number of completed scrolls. scroll_time_in_millis tracks the total time that scrolls were held open. scroll_current tracks the number of
scrolls currently open.

Closes #9109

core/src/main/java/org/elasticsearch/index/search/stats/SearchStats.java

@@ -47,21 +47,38 @@ public class SearchStats implements Streamable, ToXContent {
         private long fetchTimeInMillis;
         private long fetchCurrent;
 
+        private long scrollCount;
+        private long scrollTimeInMillis;
+        private long scrollCurrent;
+
         Stats() {
 
         }
 
-        public Stats(long queryCount, long queryTimeInMillis, long queryCurrent, long fetchCount, long fetchTimeInMillis, long fetchCurrent) {
+        public Stats(
+                long queryCount, long queryTimeInMillis, long queryCurrent,
+                long fetchCount, long fetchTimeInMillis, long fetchCurrent,
+                long scrollCount, long scrollTimeInMillis, long scrollCurrent
+        ) {
             this.queryCount = queryCount;
             this.queryTimeInMillis = queryTimeInMillis;
             this.queryCurrent = queryCurrent;
+
             this.fetchCount = fetchCount;
             this.fetchTimeInMillis = fetchTimeInMillis;
             this.fetchCurrent = fetchCurrent;
+
+            this.scrollCount = scrollCount;
+            this.scrollTimeInMillis = scrollTimeInMillis;
+            this.scrollCurrent = scrollCurrent;
         }
 
         public Stats(Stats stats) {
-            this(stats.queryCount, stats.queryTimeInMillis, stats.queryCurrent, stats.fetchCount, stats.fetchTimeInMillis, stats.fetchCurrent);
+            this(
+                    stats.queryCount, stats.queryTimeInMillis, stats.queryCurrent,
+                    stats.fetchCount, stats.fetchTimeInMillis, stats.fetchCurrent,
+                    stats.scrollCount, stats.scrollTimeInMillis, stats.scrollCurrent
+            );
         }
 
         public void add(Stats stats) {
@@ -72,6 +89,10 @@ public class SearchStats implements Streamable, ToXContent {
             fetchCount += stats.fetchCount;
             fetchTimeInMillis += stats.fetchTimeInMillis;
             fetchCurrent += stats.fetchCurrent;
+
+            scrollCount += stats.scrollCount;
+            scrollTimeInMillis += stats.scrollTimeInMillis;
+            scrollCurrent += stats.scrollCurrent;
         }
 
         public long getQueryCount() {
@@ -106,6 +127,21 @@ public class SearchStats implements Streamable, ToXContent {
             return fetchCurrent;
         }
 
+        public long getScrollCount() {
+            return scrollCount;
+        }
+
+        public TimeValue getScrollTime() {
+            return new TimeValue(scrollTimeInMillis);
+        }
+
+        public long getScrollTimeInMillis() {
+            return scrollTimeInMillis;
+        }
+
+        public long getScrollCurrent() {
+            return scrollCurrent;
+        }
 
         public static Stats readStats(StreamInput in) throws IOException {
             Stats stats = new Stats();
@@ -122,6 +158,10 @@ public class SearchStats implements Streamable, ToXContent {
             fetchCount = in.readVLong();
             fetchTimeInMillis = in.readVLong();
             fetchCurrent = in.readVLong();
+
+            scrollCount = in.readVLong();
+            scrollTimeInMillis = in.readVLong();
+            scrollCurrent = in.readVLong();
         }
 
         @Override
@@ -133,6 +173,10 @@ public class SearchStats implements Streamable, ToXContent {
             out.writeVLong(fetchCount);
             out.writeVLong(fetchTimeInMillis);
             out.writeVLong(fetchCurrent);
+
+            out.writeVLong(scrollCount);
+            out.writeVLong(scrollTimeInMillis);
+            out.writeVLong(scrollCurrent);
         }
 
         @Override
@@ -145,6 +189,10 @@ public class SearchStats implements Streamable, ToXContent {
             builder.timeValueField(Fields.FETCH_TIME_IN_MILLIS, Fields.FETCH_TIME, fetchTimeInMillis);
             builder.field(Fields.FETCH_CURRENT, fetchCurrent);
 
+            builder.field(Fields.SCROLL_TOTAL, scrollCount);
+            builder.timeValueField(Fields.SCROLL_TIME_IN_MILLIS, Fields.SCROLL_TIME, scrollTimeInMillis);
+            builder.field(Fields.SCROLL_CURRENT, scrollCurrent);
+
             return builder;
         }
     }
@@ -233,6 +281,10 @@ public class SearchStats implements Streamable, ToXContent {
         static final XContentBuilderString FETCH_TIME = new XContentBuilderString("fetch_time");
         static final XContentBuilderString FETCH_TIME_IN_MILLIS = new XContentBuilderString("fetch_time_in_millis");
         static final XContentBuilderString FETCH_CURRENT = new XContentBuilderString("fetch_current");
+        static final XContentBuilderString SCROLL_TOTAL = new XContentBuilderString("scroll_total");
+        static final XContentBuilderString SCROLL_TIME = new XContentBuilderString("scroll_time");
+        static final XContentBuilderString SCROLL_TIME_IN_MILLIS = new XContentBuilderString("scroll_time_in_millis");
+        static final XContentBuilderString SCROLL_CURRENT = new XContentBuilderString("scroll_current");
     }
 
     public static SearchStats readSearchStats(StreamInput in) throws IOException {

core/src/main/java/org/elasticsearch/index/search/stats/ShardSearchStats.java

@@ -169,6 +169,15 @@ public final class ShardSearchStats {
         openContexts.dec();
     }
 
+    public void onNewScrollContext(SearchContext context) {
+        totalStats.scrollCurrent.inc();
+    }
+
+    public void onFreeScrollContext(SearchContext context) {
+        totalStats.scrollCurrent.dec();
+        totalStats.scrollMetric.inc(TimeUnit.MILLISECONDS.toNanos(System.currentTimeMillis() - context.nowInMillis()));
+    }
+
     public void onRefreshSettings(Settings settings) {
         slowLogSearchService.onRefreshSettings(settings);
     }
@@ -176,21 +185,27 @@ public final class ShardSearchStats {
     final static class StatsHolder {
         public final MeanMetric queryMetric = new MeanMetric();
         public final MeanMetric fetchMetric = new MeanMetric();
+        public final MeanMetric scrollMetric = new MeanMetric();
         public final CounterMetric queryCurrent = new CounterMetric();
         public final CounterMetric fetchCurrent = new CounterMetric();
+        public final CounterMetric scrollCurrent = new CounterMetric();
 
         public SearchStats.Stats stats() {
-            return new SearchStats.Stats(queryMetric.count(), TimeUnit.NANOSECONDS.toMillis(queryMetric.sum()), queryCurrent.count(),
+            return new SearchStats.Stats(
-                    fetchMetric.count(), TimeUnit.NANOSECONDS.toMillis(fetchMetric.sum()), fetchCurrent.count());
+                    queryMetric.count(), TimeUnit.NANOSECONDS.toMillis(queryMetric.sum()), queryCurrent.count(),
+                    fetchMetric.count(), TimeUnit.NANOSECONDS.toMillis(fetchMetric.sum()), fetchCurrent.count(),
+                    scrollMetric.count(), TimeUnit.NANOSECONDS.toMillis(scrollMetric.sum()), scrollCurrent.count()
+            );
         }
 
         public long totalCurrent() {
-            return queryCurrent.count() + fetchCurrent.count();
+            return queryCurrent.count() + fetchCurrent.count() + scrollCurrent.count();
         }
 
         public void clear() {
             queryMetric.clear();
             fetchMetric.clear();
+            scrollMetric.clear();
         }
     }
 }

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

@@ -267,22 +267,39 @@ public class SearchService extends AbstractLifecycleComponent<SearchService> {
 
     public ScrollQueryFetchSearchResult executeScan(InternalScrollSearchRequest request) {
         final SearchContext context = findContext(request.id());
+        ShardSearchStats shardSearchStats = context.indexShard().searchService();
         contextProcessing(context);
         try {
             processScroll(request, context);
-            if (context.searchType() == SearchType.QUERY_THEN_FETCH) {
+            shardSearchStats.onPreQueryPhase(context);
-                // first scanning, reset the from to 0
+            long time = System.nanoTime();
-                context.searchType(SearchType.SCAN);
+            try {
-                context.from(0);
+                if (context.searchType() == SearchType.QUERY_THEN_FETCH) {
+                    // first scanning, reset the from to 0
+                    context.searchType(SearchType.SCAN);
+                    context.from(0);
+                }
+                queryPhase.execute(context);
+            } catch (Throwable e) {
+                shardSearchStats.onFailedQueryPhase(context);
+                throw ExceptionsHelper.convertToRuntime(e);
             }
-            queryPhase.execute(context);
+            long queryFinishTime = System.nanoTime();
-            shortcutDocIdsToLoadForScanning(context);
+            shardSearchStats.onQueryPhase(context, queryFinishTime - time);
-            fetchPhase.execute(context);
+            shardSearchStats.onPreFetchPhase(context);
-            if (context.scroll() == null || context.fetchResult().hits().hits().length < context.size()) {
+            try {
-                freeContext(request.id());
+                shortcutDocIdsToLoadForScanning(context);
-            } else {
+                fetchPhase.execute(context);
-                contextProcessedSuccessfully(context);
+                if (context.scroll() == null || context.fetchResult().hits().hits().length < context.size()) {
+                    freeContext(request.id());
+                } else {
+                    contextProcessedSuccessfully(context);
+                }
+            } catch (Throwable e) {
+                shardSearchStats.onFailedQueryPhase(context);
+                throw ExceptionsHelper.convertToRuntime(e);
             }
+            shardSearchStats.onFetchPhase(context, System.nanoTime() - queryFinishTime);
             return new ScrollQueryFetchSearchResult(new QueryFetchSearchResult(context.queryResult(), context.fetchResult()), context.shardTarget());
         } catch (Throwable e) {
             logger.trace("Scan phase failed", e);
@@ -569,6 +586,9 @@ public class SearchService extends AbstractLifecycleComponent<SearchService> {
         boolean success = false;
         try {
             putContext(context);
+            if (request.scroll() != null) {
+                context.indexShard().searchService().onNewScrollContext(context);
+            }
             context.indexShard().searchService().onNewContext(context);
             success = true;
             return context;
@@ -643,6 +663,9 @@ public class SearchService extends AbstractLifecycleComponent<SearchService> {
         if (context != null) {
             try {
                 context.indexShard().searchService().onFreeContext(context);
+                if (context.scroll() != null) {
+                    context.indexShard().searchService().onFreeScrollContext(context);
+                }
             } finally {
                 context.close();
             }

core/src/test/java/org/elasticsearch/search/stats/SearchStatsTests.java

@@ -33,11 +33,13 @@ import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.index.query.QueryBuilders;
 import org.elasticsearch.index.search.stats.SearchStats.Stats;
 import org.elasticsearch.script.Script;
+import org.elasticsearch.search.SearchHit;
 import org.elasticsearch.test.ElasticsearchIntegrationTest;
 import org.junit.Test;
 
 import java.util.HashSet;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
 
 import static org.elasticsearch.cluster.metadata.IndexMetaData.SETTING_NUMBER_OF_REPLICAS;
 import static org.elasticsearch.cluster.metadata.IndexMetaData.SETTING_NUMBER_OF_SHARDS;
@@ -178,7 +180,32 @@ public class SearchStatsTests extends ElasticsearchIntegrationTest {
         assertSearchResponse(searchResponse);
 
         indicesStats = client().admin().indices().prepareStats().execute().actionGet();
-        assertThat(indicesStats.getTotal().getSearch().getOpenContexts(), equalTo((long)numAssignedShards("test1")));
+        assertThat(indicesStats.getTotal().getSearch().getOpenContexts(), equalTo((long) numAssignedShards("test1")));
+        assertThat(indicesStats.getTotal().getSearch().getTotal().getScrollCurrent(), equalTo((long) numAssignedShards("test1")));
+
+        // force the scan to complete measuring the time taken
+        // the total time the scroll is open should be greater than this
+        // the number of queries should equal the number of pages in the scan times the number of shards
+        long time = System.nanoTime();
+        int count = 0;
+        while (true) {
+            count++;
+            searchResponse = client().prepareSearchScroll(searchResponse.getScrollId())
+                    .setScroll(TimeValue.timeValueMinutes(2))
+                    .execute().actionGet();
+            if (searchResponse.getHits().getHits().length == 0) {
+                break;
+            }
+        }
+        long total = System.nanoTime() - time;
+
+        indicesStats = client().admin().indices().prepareStats().execute().actionGet();
+        assertThat(indicesStats.getTotal().getSearch().getTotal().getQueryCount(), equalTo(count * (long)numAssignedShards("test1")));
+        assertThat(indicesStats.getTotal().getSearch().getTotal().getScrollCount(), equalTo((long)numAssignedShards("test1")));
+        assertThat(
+                indicesStats.getTotal().getSearch().getTotal().getScrollTimeInMillis(),
+                greaterThan(TimeUnit.NANOSECONDS.toMillis(total * numAssignedShards("test1")))
+        );
 
         // scroll, but with no timeout (so no context)
         searchResponse = client().prepareSearchScroll(searchResponse.getScrollId()).execute().actionGet();

core/src/test/java/org/elasticsearch/search/stats/SearchStatsUnitTests.java

@@ -35,9 +35,9 @@ public class SearchStatsUnitTests extends ElasticsearchTestCase {
         // 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));
+        groupStats2.put("group1", new Stats(1, 1, 1, 1, 1, 1, 1, 1, 1));
-        SearchStats searchStats1 = new SearchStats(new Stats(1, 1, 1, 1, 1, 1), 0, groupStats1);
+        SearchStats searchStats1 = new SearchStats(new Stats(1, 1, 1, 1, 1, 1, 1, 1, 1), 0, groupStats1);
-        SearchStats searchStats2 = new SearchStats(new Stats(1, 1, 1, 1, 1, 1), 0, groupStats2);
+        SearchStats searchStats2 = new SearchStats(new Stats(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);
@@ -62,5 +62,8 @@ public class SearchStatsUnitTests extends ElasticsearchTestCase {
         assertEquals(equalTo, stats.getFetchCount());
         assertEquals(equalTo, stats.getFetchTimeInMillis());
         assertEquals(equalTo, stats.getFetchCurrent());
+        assertEquals(equalTo, stats.getScrollCount());
+        assertEquals(equalTo, stats.getScrollTimeInMillis());
+        assertEquals(equalTo, stats.getScrollCurrent());
     }
 }