Recent_Search_Load calculation - DfsPhase

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

@@ -567,10 +567,12 @@ public void testSimpleStats() throws Exception {
         // make sure that number of requests in progress is 0
         assertThat(stats.getIndex("test1").getTotal().getIndexing().getTotal().getIndexCurrent(), equalTo(0L));
         assertThat(stats.getIndex("test1").getTotal().getIndexing().getTotal().getDeleteCurrent(), equalTo(0L));
+        assertThat(stats.getIndex("test1").getTotal().getSearch().getTotal().getDfsCurrent(), equalTo(0L));
         assertThat(stats.getIndex("test1").getTotal().getSearch().getTotal().getFetchCurrent(), equalTo(0L));
         assertThat(stats.getIndex("test1").getTotal().getSearch().getTotal().getQueryCurrent(), equalTo(0L));
         assertThat(stats.getIndex("test2").getTotal().getIndexing().getTotal().getIndexCurrent(), equalTo(0L));
         assertThat(stats.getIndex("test2").getTotal().getIndexing().getTotal().getDeleteCurrent(), equalTo(0L));
+        assertThat(stats.getIndex("test2").getTotal().getSearch().getTotal().getDfsCurrent(), equalTo(0L));
         assertThat(stats.getIndex("test2").getTotal().getSearch().getTotal().getFetchCurrent(), equalTo(0L));
         assertThat(stats.getIndex("test2").getTotal().getSearch().getTotal().getQueryCurrent(), equalTo(0L));
 

server/src/internalClusterTest/java/org/elasticsearch/monitor/metrics/IndicesMetricsIT.java

@@ -68,6 +68,9 @@ protected Settings nodeSettings(int nodeOrdinal, Settings otherSettings) {
     static final String STANDARD_INDEX_COUNT = "es.indices.standard.total";
     static final String STANDARD_BYTES_SIZE = "es.indices.standard.size";
     static final String STANDARD_DOCS_COUNT = "es.indices.standard.docs.total";
+    static final String STANDARD_DFS_COUNT = "es.indices.standard.dfs.total";
+    static final String STANDARD_DFS_TIME = "es.indices.standard.dfs.time";
+    static final String STANDARD_DFS_FAILURE = "es.indices.standard.dfs.failure.total";
     static final String STANDARD_QUERY_COUNT = "es.indices.standard.query.total";
     static final String STANDARD_QUERY_TIME = "es.indices.standard.query.time";
     static final String STANDARD_QUERY_FAILURE = "es.indices.standard.query.failure.total";
@@ -245,6 +248,10 @@ public void testIndicesMetrics() {
             telemetry,
             1,
             Map.of(
+                STANDARD_DFS_COUNT,
+                equalTo(search1.getDfsCount()),
+                STANDARD_DFS_TIME,
+                equalTo(search1.getDfsTimeInMillis()),
                 STANDARD_QUERY_COUNT,
                 equalTo(numStandardIndices),
                 STANDARD_QUERY_TIME,
@@ -334,6 +341,8 @@ public void testIndicesMetrics() {
             telemetry,
             4,
             Map.of(
+                STANDARD_DFS_FAILURE,
+                equalTo(0L),
                 STANDARD_QUERY_FAILURE,
                 equalTo(0L),
                 STANDARD_FETCH_FAILURE,

server/src/internalClusterTest/java/org/elasticsearch/search/stats/SearchStatsIT.java

@@ -130,6 +130,8 @@ public void testSimpleStats() throws Exception {
 
         IndicesStatsResponse indicesStats = indicesAdmin().prepareStats().get();
         logger.debug("###### indices search stats: {}", indicesStats.getTotal().getSearch());
+        assertThat(indicesStats.getTotal().getSearch().getTotal().getDfsCount(), equalTo(0L));
+        assertThat(indicesStats.getTotal().getSearch().getTotal().getDfsTimeInMillis(), equalTo(0L));
         assertThat(indicesStats.getTotal().getSearch().getTotal().getQueryCount(), greaterThan(0L));
         assertThat(indicesStats.getTotal().getSearch().getTotal().getQueryTimeInMillis(), greaterThan(0L));
         assertThat(indicesStats.getTotal().getSearch().getTotal().getFetchCount(), greaterThan(0L));
@@ -138,6 +140,8 @@ public void testSimpleStats() throws Exception {
 
         indicesStats = indicesAdmin().prepareStats().setGroups("group1").get();
         assertThat(indicesStats.getTotal().getSearch().getGroupStats(), notNullValue());
+        assertThat(indicesStats.getTotal().getSearch().getGroupStats().get("group1").getDfsCount(), equalTo(0L));
+        assertThat(indicesStats.getTotal().getSearch().getGroupStats().get("group1").getDfsTimeInMillis(), equalTo(0L));
         assertThat(indicesStats.getTotal().getSearch().getGroupStats().get("group1").getQueryCount(), greaterThan(0L));
         assertThat(indicesStats.getTotal().getSearch().getGroupStats().get("group1").getQueryTimeInMillis(), greaterThan(0L));
         assertThat(indicesStats.getTotal().getSearch().getGroupStats().get("group1").getFetchCount(), greaterThan(0L));

server/src/main/java/org/elasticsearch/TransportVersions.java

@@ -326,6 +326,7 @@ static TransportVersion def(int id) {
     public static final TransportVersion ML_INFERENCE_COHERE_API_VERSION = def(9_110_0_00);
     public static final TransportVersion ESQL_PROFILE_INCLUDE_PLAN = def(9_111_0_00);
     public static final TransportVersion MAPPINGS_IN_DATA_STREAMS = def(9_112_0_00);
+    public static final TransportVersion DFS_STATS = def(9_113_00_0);
 
     /*
      * STOP! READ THIS FIRST! No, really,

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

@@ -30,6 +30,10 @@ public class SearchStats implements Writeable, ToXContentFragment {
 
     public static class Stats implements Writeable, ToXContentFragment {
 
+        private long dfsCount;
+        private long dfsTimeInMillis;
+        private long dfsCurrent;
+
         private long queryCount;
         private long queryTimeInMillis;
         private long queryCurrent;
@@ -46,6 +50,7 @@ public static class Stats implements Writeable, ToXContentFragment {
         private long suggestTimeInMillis;
         private long suggestCurrent;
 
+        private long dfsFailure;
         private long queryFailure;
         private long fetchFailure;
 
@@ -58,6 +63,10 @@ private Stats() {
         }
 
         public Stats(
+            long dfsCount,
+            long dfsTimeInMillis,
+            long dfsCurrent,
+            long dfsFailure,
             long queryCount,
             long queryTimeInMillis,
             long queryCurrent,
@@ -74,6 +83,11 @@ public Stats(
             long suggestCurrent,
             double recentSearchLoad
         ) {
+            this.dfsCount = dfsCount;
+            this.dfsTimeInMillis = dfsTimeInMillis;
+            this.dfsCurrent = dfsCurrent;
+            this.dfsFailure = dfsFailure;
+
             this.queryCount = queryCount;
             this.queryTimeInMillis = queryTimeInMillis;
             this.queryCurrent = queryCurrent;
@@ -97,6 +111,13 @@ public Stats(
         }
 
         private Stats(StreamInput in) throws IOException {
+            if (in.getTransportVersion().onOrAfter(TransportVersions.DFS_STATS)) {
+                dfsCount = in.readVLong();
+                dfsTimeInMillis = in.readVLong();
+                dfsCurrent = in.readVLong();
+                dfsFailure = in.readVLong();
+            }
+
             queryCount = in.readVLong();
             queryTimeInMillis = in.readVLong();
             queryCurrent = in.readVLong();
@@ -125,6 +146,13 @@ private Stats(StreamInput in) throws IOException {
 
         @Override
         public void writeTo(StreamOutput out) throws IOException {
+            if (out.getTransportVersion().onOrAfter(TransportVersions.DFS_STATS)) {
+                out.writeVLong(dfsCount);
+                out.writeVLong(dfsTimeInMillis);
+                out.writeVLong(dfsCurrent);
+                out.writeVLong(dfsFailure);
+            }
+
             out.writeVLong(queryCount);
             out.writeVLong(queryTimeInMillis);
             out.writeVLong(queryCurrent);
@@ -152,6 +180,11 @@ public void writeTo(StreamOutput out) throws IOException {
         }
 
         public void add(Stats stats) {
+            dfsCount += stats.dfsCount;
+            dfsTimeInMillis += stats.dfsTimeInMillis;
+            dfsCurrent += stats.dfsCurrent;
+            dfsFailure += stats.dfsFailure;
+
             queryCount += stats.queryCount;
             queryTimeInMillis += stats.queryTimeInMillis;
             queryCurrent += stats.queryCurrent;
@@ -174,6 +207,10 @@ public void add(Stats stats) {
         }
 
         public void addForClosingShard(Stats stats) {
+            dfsCount += stats.dfsCount;
+            dfsTimeInMillis += stats.dfsTimeInMillis;
+            dfsFailure += stats.dfsFailure;
+
             queryCount += stats.queryCount;
             queryTimeInMillis += stats.queryTimeInMillis;
             queryFailure += stats.queryFailure;
@@ -193,6 +230,26 @@ public void addForClosingShard(Stats stats) {
             recentSearchLoad += stats.recentSearchLoad;
         }
 
+        public long getDfsCount() {
+            return dfsCount;
+        }
+
+        public TimeValue getDfsTime() {
+            return new TimeValue(dfsTimeInMillis);
+        }
+
+        public long getDfsTimeInMillis() {
+            return dfsTimeInMillis;
+        }
+
+        public long getDfsCurrent() {
+            return dfsCurrent;
+        }
+
+        public long getDfsFailure() {
+            return dfsFailure;
+        }
+
         public long getQueryCount() {
             return queryCount;
         }
@@ -275,6 +332,11 @@ public static Stats readStats(StreamInput in) throws IOException {
 
         @Override
         public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+            builder.field(Fields.DFS_TOTAL, dfsCount);
+            builder.humanReadableField(Fields.DFS_TIME_IN_MILLIS, Fields.DFS_TIME, getDfsTime());
+            builder.field(Fields.DFS_CURRENT, dfsCurrent);
+            builder.field(Fields.DFS_FAILURE, dfsFailure);
+
             builder.field(Fields.QUERY_TOTAL, queryCount);
             builder.humanReadableField(Fields.QUERY_TIME_IN_MILLIS, Fields.QUERY_TIME, getQueryTime());
             builder.field(Fields.QUERY_CURRENT, queryCurrent);
@@ -303,7 +365,11 @@ public boolean equals(Object o) {
             if (this == o) return true;
             if (o == null || getClass() != o.getClass()) return false;
             Stats that = (Stats) o;
-            return queryCount == that.queryCount
+            return dfsCount == that.dfsCount
+                && dfsTimeInMillis == that.dfsTimeInMillis
+                && dfsCurrent == that.dfsCurrent
+                && dfsFailure == that.dfsFailure
+                && queryCount == that.queryCount
                 && queryTimeInMillis == that.queryTimeInMillis
                 && queryCurrent == that.queryCurrent
                 && queryFailure == that.queryFailure
@@ -323,6 +389,10 @@ public boolean equals(Object o) {
         @Override
         public int hashCode() {
             return Objects.hash(
+                dfsCount,
+                dfsTimeInMillis,
+                dfsCurrent,
+                dfsFailure,
                 queryCount,
                 queryTimeInMillis,
                 queryCurrent,
@@ -437,6 +507,11 @@ static final class Fields {
         static final String SEARCH = "search";
         static final String OPEN_CONTEXTS = "open_contexts";
         static final String GROUPS = "groups";
+        static final String DFS_TOTAL = "dfs_total";
+        static final String DFS_TIME = "dfs_time";
+        static final String DFS_TIME_IN_MILLIS = "dfs_time_in_millis";
+        static final String DFS_CURRENT = "dfs_current";
+        static final String DFS_FAILURE = "dfs_failure";
         static final String QUERY_TOTAL = "query_total";
         static final String QUERY_TIME = "query_time";
         static final String QUERY_TIME_IN_MILLIS = "query_time_in_millis";

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

@@ -117,6 +117,28 @@ public void onFetchPhase(SearchContext searchContext, long tookInNanos) {
         });
     }
 
+    @Override
+    public void onPreDfsPhase(SearchContext searchContext) {
+        computeStats(searchContext, statsHolder -> statsHolder.dfsCurrent.inc());
+    }
+
+    @Override
+    public void onDfsPhase(SearchContext searchContext, long tookInNanos) {
+        computeStats(searchContext, statsHolder -> {
+            statsHolder.recentSearchLoad.addIncrement(tookInNanos, System.nanoTime());
+            statsHolder.dfsMetric.inc(tookInNanos);
+            statsHolder.dfsCurrent.dec();
+        });
+    }
+
+    @Override
+    public void onFailedDfsPhase(SearchContext searchContext) {
+        computeStats(searchContext, statsHolder -> {
+            statsHolder.dfsCurrent.dec();
+            statsHolder.dfsFailure.inc();
+        });
+    }
+
     private void computeStats(SearchContext searchContext, Consumer<StatsHolder> consumer) {
         consumer.accept(totalStats);
         var groupStats = searchContext.groupStats();
@@ -164,6 +186,7 @@ public void onFreeScrollContext(ReaderContext readerContext) {
     }
 
     static final class StatsHolder {
+        final MeanMetric dfsMetric = new MeanMetric();
         final MeanMetric queryMetric = 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
@@ -175,11 +198,13 @@ static final class StatsHolder {
          */
         final MeanMetric scrollMetric = new MeanMetric();
         final MeanMetric suggestMetric = new MeanMetric();
+        final CounterMetric dfsCurrent = new CounterMetric();
         final CounterMetric queryCurrent = new CounterMetric();
         final CounterMetric fetchCurrent = new CounterMetric();
         final CounterMetric scrollCurrent = new CounterMetric();
         final CounterMetric suggestCurrent = new CounterMetric();
 
+        final CounterMetric dfsFailure = new CounterMetric();
         final CounterMetric queryFailure = new CounterMetric();
         final CounterMetric fetchFailure = new CounterMetric();
 
@@ -192,6 +217,10 @@ static final class StatsHolder {
 
         SearchStats.Stats stats() {
             return new SearchStats.Stats(
+                dfsMetric.count(),
+                TimeUnit.NANOSECONDS.toMillis(dfsMetric.sum()),
+                dfsCurrent.count(),
+                dfsFailure.count(),
                 queryMetric.count(),
                 TimeUnit.NANOSECONDS.toMillis(queryMetric.sum()),
                 queryCurrent.count(),

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

@@ -65,6 +65,28 @@ default void onFailedFetchPhase(SearchContext searchContext) {}
      */
     default void onFetchPhase(SearchContext searchContext, long tookInNanos) {}
 
+    /**
+     * Executed before the DFS phase is executed
+     * @param searchContext the current search context
+     */
+    default void onPreDfsPhase(SearchContext searchContext) {}
+
+    /**
+     * Executed after the query DFS successfully finished.
+     * Note: this is not invoked if the DFS phase execution failed.
+     * @param searchContext the current search context
+     * @param tookInNanos the number of nanoseconds the query execution took
+     *
+     * @see #onFailedQueryPhase(SearchContext)
+     */
+    default void onDfsPhase(SearchContext searchContext, long tookInNanos) {}
+
+    /**
+     * Executed if a dfs phased failed.
+     * @param searchContext the current search context
+     */
+    default void onFailedDfsPhase(SearchContext searchContext) {}
+
     /**
      * Executed when a new reader context was created
      * @param readerContext the created context
@@ -182,6 +204,39 @@ public void onFetchPhase(SearchContext searchContext, long tookInNanos) {
             }
         }
 
+        @Override
+        public void onPreDfsPhase(SearchContext searchContext) {
+            for (SearchOperationListener listener : listeners) {
+                try {
+                    listener.onPreDfsPhase(searchContext);
+                } catch (Exception e) {
+                    logger.warn(() -> "onPreDfsPhase listener [" + listener + "] failed", e);
+                }
+            }
+        }
+
+        @Override
+        public void onFailedDfsPhase(SearchContext searchContext) {
+            for (SearchOperationListener listener : listeners) {
+                try {
+                    listener.onFailedDfsPhase(searchContext);
+                } catch (Exception e) {
+                    logger.warn(() -> "onFailedDfsPhase listener [" + listener + "] failed", e);
+                }
+            }
+        }
+
+        @Override
+        public void onDfsPhase(SearchContext searchContext, long tookInNanos) {
+            for (SearchOperationListener listener : listeners) {
+                try {
+                    listener.onDfsPhase(searchContext, tookInNanos);
+                } catch (Exception e) {
+                    logger.warn(() -> "onDfsPhase listener [" + listener + "] failed", e);
+                }
+            }
+        }
+
         @Override
         public void onNewReaderContext(ReaderContext readerContext) {
             for (SearchOperationListener listener : listeners) {