Enable Shard-Level Search-load rate metric

server/src/internalClusterTest/java/org/elasticsearch/index/shard/IndexShardIT.java

@@ -53,6 +53,7 @@
 import org.elasticsearch.index.flush.FlushStats;
 import org.elasticsearch.index.mapper.MapperMetrics;
 import org.elasticsearch.index.mapper.SourceToParse;
+import org.elasticsearch.index.search.stats.SearchStatsSettings;
 import org.elasticsearch.index.seqno.RetentionLeaseSyncer;
 import org.elasticsearch.index.seqno.SequenceNumbers;
 import org.elasticsearch.index.translog.TestTranslog;
@@ -638,7 +639,8 @@ public static final IndexShard newIndexShard(
             System::nanoTime,
             null,
             MapperMetrics.NOOP,
-            new IndexingStatsSettings(ClusterSettings.createBuiltInClusterSettings())
+            new IndexingStatsSettings(ClusterSettings.createBuiltInClusterSettings()),
+            new SearchStatsSettings(ClusterSettings.createBuiltInClusterSettings())
         );
     }
 

server/src/main/java/module-info.java

@@ -137,6 +137,7 @@
     exports org.elasticsearch.action.ingest;
     exports org.elasticsearch.action.resync;
     exports org.elasticsearch.action.search;
+    exports org.elasticsearch.action.search.load;
     exports org.elasticsearch.action.support;
     exports org.elasticsearch.action.support.broadcast;
     exports org.elasticsearch.action.support.broadcast.node;

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

@@ -273,6 +273,7 @@ static TransportVersion def(int id) {
     public static final TransportVersion INFERENCE_CUSTOM_SERVICE_ADDED = def(9_084_0_00);
     public static final TransportVersion ESQL_LIMIT_ROW_SIZE = def(9_085_0_00);
     public static final TransportVersion ESQL_REGEX_MATCH_WITH_CASE_INSENSITIVITY = def(9_086_0_00);
+    public static final TransportVersion EWMR_STATS = def(9_087_0_00);
 
     /*
      * STOP! READ THIS FIRST! No, really,

server/src/main/java/org/elasticsearch/action/ActionModule.java

@@ -183,6 +183,8 @@
 import org.elasticsearch.action.search.TransportSearchAction;
 import org.elasticsearch.action.search.TransportSearchScrollAction;
 import org.elasticsearch.action.search.TransportSearchShardsAction;
+import org.elasticsearch.action.search.load.ShardSearchLoadStatsAction;
+import org.elasticsearch.action.search.load.TransportShardSearchLoadStatsAction;
 import org.elasticsearch.action.support.ActionFilter;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.AutoCreateIndex;
@@ -745,6 +747,7 @@ public <Request extends ActionRequest, Response extends ActionResponse> void reg
         actions.register(FieldUsageStatsAction.INSTANCE, TransportFieldUsageAction.class);
         actions.register(MasterHistoryAction.INSTANCE, MasterHistoryAction.TransportAction.class);
         actions.register(CoordinationDiagnosticsAction.INSTANCE, CoordinationDiagnosticsAction.TransportAction.class);
+        actions.register(ShardSearchLoadStatsAction.INSTANCE, TransportShardSearchLoadStatsAction.class);
 
         // Indexed scripts
         actions.register(TransportPutStoredScriptAction.TYPE, TransportPutStoredScriptAction.class);

server/src/main/java/org/elasticsearch/action/search/load/ShardSearchLoadStats.java

@@ -0,0 +1,122 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.action.search.load;
+
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.transport.Transports;
+
+import java.io.IOException;
+import java.util.Objects;
+
+/**
+ * ShardStats class represents the statistics of a shard in an index.
+ * It contains information such as the index name, shard ID, allocation ID, and EWMA rate.
+ */
+public class ShardSearchLoadStats implements Writeable {
+
+    private final String indexName;
+
+    private final Integer shardId;
+
+    private final String allocationId;
+
+    private final Double emwRate;
+
+    /**
+     * Constructor to create a ShardStats object from a StreamInput.
+     *
+     * @param in the StreamInput to read from
+     * @throws IOException if an I/O error occurs
+     */
+    public ShardSearchLoadStats(StreamInput in) throws IOException {
+        assert Transports.assertNotTransportThread("O(#shards) work must always fork to an appropriate executor");
+        this.indexName = in.readString();
+        this.shardId = in.readVInt();
+        this.allocationId = in.readString();
+        this.emwRate = in.readDouble();
+    }
+
+    /**
+     * Constructor to create a ShardStats object with the given parameters.
+     *
+     * @param indexName   the name of the index
+     * @param shardId     the ID of the shard
+     * @param allocationId the allocation ID of the shard
+     * @param ewma        the EWMA rate of the shard
+     */
+    public ShardSearchLoadStats(String indexName, Integer shardId, String allocationId, Double ewma) {
+        this.indexName = indexName;
+        this.shardId = shardId;
+        this.allocationId = allocationId;
+        this.emwRate = ewma;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        ShardSearchLoadStats that = (ShardSearchLoadStats) o;
+        return Objects.equals(indexName, that.indexName)
+            && Objects.equals(shardId, that.shardId)
+            && Objects.equals(allocationId, that.allocationId)
+            && Objects.equals(emwRate, that.emwRate);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(indexName, shardId, allocationId, emwRate);
+    }
+
+    /**
+     * Returns the index name of the shard.
+     *
+     * @return the index name
+     */
+    public String getIndexName() {
+        return this.indexName;
+    }
+
+    /**
+     * Returns the shard ID of the shard.
+     *
+     * @return the shard ID
+     */
+    public Integer getShardId() {
+        return this.shardId;
+    }
+
+    /**
+     * Returns the allocation ID of the shard.
+     *
+     * @return the allocation ID
+     */
+    public String getAllocationId() {
+        return this.allocationId;
+    }
+
+    /**
+     * Returns the EWMA rate of the shard.
+     *
+     * @return the EWMA rate
+     */
+    public Double getEwmRate() {
+        return this.emwRate;
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        out.writeString(indexName);
+        out.writeVInt(shardId);
+        out.writeString(allocationId);
+        out.writeDouble(emwRate);
+    }
+}

server/src/main/java/org/elasticsearch/action/search/load/ShardSearchLoadStatsAction.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.action.search.load;
+
+import org.elasticsearch.action.ActionType;
+import org.elasticsearch.action.RemoteClusterActionType;
+
+/**
+ * Action definition for retrieving shard-level search load statistics.
+ * <p>
+ * This action serves as a marker for executing {@link TransportShardSearchLoadStatsAction}
+ * </p>
+ */
+public class ShardSearchLoadStatsAction extends ActionType<ShardSearchLoadStatsResponse> {
+
+    /**
+     * Singleton instance of the action type.
+     */
+    public static final ShardSearchLoadStatsAction INSTANCE = new ShardSearchLoadStatsAction();
+    public static final String NAME = "internal:search/stats";
+    public static final RemoteClusterActionType<ShardSearchLoadStatsResponse> REMOTE_TYPE = new RemoteClusterActionType<>(
+        NAME,
+        ShardSearchLoadStatsResponse::new
+    );
+
+    private ShardSearchLoadStatsAction() {
+        super(NAME);
+    }
+}

server/src/main/java/org/elasticsearch/action/search/load/ShardSearchLoadStatsResponse.java

@@ -0,0 +1,81 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch.action.search.load;
+
+import org.elasticsearch.action.support.DefaultShardOperationFailedException;
+import org.elasticsearch.action.support.broadcast.ChunkedBroadcastResponse;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.xcontent.ToXContent;
+
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * Response to a shard stats request.
+ */
+public class ShardSearchLoadStatsResponse extends ChunkedBroadcastResponse {
+
+    private final ShardSearchLoadStats[] shards;
+
+    /**
+     * Constructor to create a ShardStatsResponse object from a StreamInput.
+     *
+     * @param in the StreamInput to read from
+     * @throws IOException if an I/O error occurs
+     */
+    ShardSearchLoadStatsResponse(StreamInput in) throws IOException {
+        super(in);
+        shards = in.readArray(ShardSearchLoadStats::new, ShardSearchLoadStats[]::new);
+    }
+
+    /**
+     * Constructor to create a ShardStatsResponse object with the given parameters.
+     *
+     * @param shards          the array of shard stats
+     * @param totalShards     the total number of shards
+     * @param successfulShards the number of successful shards
+     * @param failedShards    the number of failed shards
+     * @param shardFailures   the list of shard failures
+     */
+    ShardSearchLoadStatsResponse(
+        ShardSearchLoadStats[] shards,
+        int totalShards,
+        int successfulShards,
+        int failedShards,
+        List<DefaultShardOperationFailedException> shardFailures
+    ) {
+        super(totalShards, successfulShards, failedShards, shardFailures);
+        this.shards = shards;
+        Objects.requireNonNull(shards);
+    }
+
+    /**
+     * Returns the array of shard stats.
+     *
+     * @return the array of shard stats
+     */
+    public ShardSearchLoadStats[] getShards() {
+        return shards;
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        super.writeTo(out);
+        out.writeArray(shards);
+    }
+
+    @Override
+    protected Iterator<ToXContent> customXContentChunks(ToXContent.Params params) {
+        return null;
+    }
+}