HLRC: ML Adding get datafeed stats API (#34271)

* HLRC: ML Adding get datafeed stats API

* addressing PR comments

* fixing field exclusion filter

* removing unnecessary whitespace

client/rest-high-level/src/main/java/org/elasticsearch/client/MLRequestConverters.java

@@ -38,6 +38,7 @@ import org.elasticsearch.client.ml.GetBucketsRequest;
 import org.elasticsearch.client.ml.GetCalendarsRequest;
 import org.elasticsearch.client.ml.GetCategoriesRequest;
 import org.elasticsearch.client.ml.GetDatafeedRequest;
+import org.elasticsearch.client.ml.GetDatafeedStatsRequest;
 import org.elasticsearch.client.ml.GetInfluencersRequest;
 import org.elasticsearch.client.ml.GetJobRequest;
 import org.elasticsearch.client.ml.GetJobStatsRequest;
@@ -260,6 +261,23 @@ final class MLRequestConverters {
         return request;
     }
 
+    static Request getDatafeedStats(GetDatafeedStatsRequest getDatafeedStatsRequest) {
+        String endpoint = new EndpointBuilder()
+            .addPathPartAsIs("_xpack")
+            .addPathPartAsIs("ml")
+            .addPathPartAsIs("datafeeds")
+            .addPathPart(Strings.collectionToCommaDelimitedString(getDatafeedStatsRequest.getDatafeedIds()))
+            .addPathPartAsIs("_stats")
+            .build();
+        Request request = new Request(HttpGet.METHOD_NAME, endpoint);
+
+        RequestConverters.Params params = new RequestConverters.Params(request);
+        if (getDatafeedStatsRequest.isAllowNoDatafeeds() != null) {
+            params.putParam("allow_no_datafeeds", Boolean.toString(getDatafeedStatsRequest.isAllowNoDatafeeds()));
+        }
+        return request;
+    }
+
     static Request previewDatafeed(PreviewDatafeedRequest previewDatafeedRequest) {
         String endpoint = new EndpointBuilder()
             .addPathPartAsIs("_xpack")

client/rest-high-level/src/main/java/org/elasticsearch/client/MachineLearningClient.java

@@ -38,6 +38,8 @@ import org.elasticsearch.client.ml.GetCategoriesRequest;
 import org.elasticsearch.client.ml.GetCategoriesResponse;
 import org.elasticsearch.client.ml.GetDatafeedRequest;
 import org.elasticsearch.client.ml.GetDatafeedResponse;
+import org.elasticsearch.client.ml.GetDatafeedStatsRequest;
+import org.elasticsearch.client.ml.GetDatafeedStatsResponse;
 import org.elasticsearch.client.ml.GetInfluencersRequest;
 import org.elasticsearch.client.ml.GetInfluencersResponse;
 import org.elasticsearch.client.ml.GetJobRequest;
@@ -183,7 +185,7 @@ public final class MachineLearningClient {
     }
 
     /**
-     * Gets one or more Machine Learning job configuration info, asynchronously.
+     * Gets usage statistics for one or more Machine Learning jobs, asynchronously.
      * <p>
      * For additional info
      * see <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/ml-get-job-stats.html">Get job stats docs</a>
@@ -651,6 +653,26 @@ public final class MachineLearningClient {
             Collections.emptySet());
     }
 
+    /**
+     * Gets statistics for one or more Machine Learning datafeeds
+     * <p>
+     * For additional info
+     * see <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/ml-get-datafeed-stats.html">Get datafeed stats docs</a>
+     *
+     * @param request {@link GetDatafeedStatsRequest} Request containing a list of datafeedId(s) and additional options
+     * @param options Additional request options (e.g. headers), use {@link RequestOptions#DEFAULT} if nothing needs to be customized
+     * @return {@link GetDatafeedStatsResponse} response object containing
+     * the {@link org.elasticsearch.client.ml.datafeed.DatafeedStats} objects and the number of datafeeds found
+     * @throws IOException when there is a serialization issue sending the request or receiving the response
+     */
+    public GetDatafeedStatsResponse getDatafeedStats(GetDatafeedStatsRequest request, RequestOptions options) throws IOException {
+        return restHighLevelClient.performRequestAndParseEntity(request,
+            MLRequestConverters::getDatafeedStats,
+            options,
+            GetDatafeedStatsResponse::fromXContent,
+            Collections.emptySet());
+    }
+
     /**
      * Previews the given Machine Learning Datafeed
      * <p>
@@ -672,6 +694,27 @@ public final class MachineLearningClient {
             Collections.emptySet());
     }
 
+    /**
+     * Gets statistics for one or more Machine Learning datafeeds, asynchronously.
+     * <p>
+     * For additional info
+     * see <a href="https://www.elastic.co/guide/en/elasticsearch/reference/current/ml-get-datafeed-stats.html">Get datafeed stats docs</a>
+     *
+     * @param request  {@link GetDatafeedStatsRequest} Request containing a list of datafeedId(s) and additional options
+     * @param options  Additional request options (e.g. headers), use {@link RequestOptions#DEFAULT} if nothing needs to be customized
+     * @param listener Listener to be notified with {@link GetDatafeedStatsResponse} upon request completion
+     */
+    public void getDatafeedStatsAsync(GetDatafeedStatsRequest request,
+                                      RequestOptions options,
+                                      ActionListener<GetDatafeedStatsResponse> listener) {
+        restHighLevelClient.performRequestAsyncAndParseEntity(request,
+            MLRequestConverters::getDatafeedStats,
+            options,
+            GetDatafeedStatsResponse::fromXContent,
+            listener,
+            Collections.emptySet());
+    }
+
     /**
      * Previews the given Machine Learning Datafeed asynchronously and notifies the listener on completion
      * <p>

client/rest-high-level/src/main/java/org/elasticsearch/client/ml/GetDatafeedStatsRequest.java

@@ -0,0 +1,147 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.elasticsearch.client.ml;
+
+import org.elasticsearch.action.ActionRequest;
+import org.elasticsearch.action.ActionRequestValidationException;
+import org.elasticsearch.client.ml.datafeed.DatafeedConfig;
+import org.elasticsearch.common.ParseField;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.xcontent.ConstructingObjectParser;
+import org.elasticsearch.common.xcontent.ObjectParser;
+import org.elasticsearch.common.xcontent.ToXContent;
+import org.elasticsearch.common.xcontent.ToXContentObject;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * Request object to get {@link org.elasticsearch.client.ml.datafeed.DatafeedStats} by their respective datafeedIds
+ *
+ * {@code _all} explicitly gets all the datafeeds' statistics in the cluster
+ * An empty request (no {@code datafeedId}s) implicitly gets all the datafeeds' statistics in the cluster
+ */
+public class GetDatafeedStatsRequest extends ActionRequest implements ToXContentObject {
+
+    public static final ParseField ALLOW_NO_DATAFEEDS = new ParseField("allow_no_datafeeds");
+
+    @SuppressWarnings("unchecked")
+    public static final ConstructingObjectParser<GetDatafeedStatsRequest, Void> PARSER = new ConstructingObjectParser<>(
+        "get_datafeed_stats_request", a -> new GetDatafeedStatsRequest((List<String>) a[0]));
+
+    static {
+        PARSER.declareField(ConstructingObjectParser.constructorArg(),
+            p -> Arrays.asList(Strings.commaDelimitedListToStringArray(p.text())),
+            DatafeedConfig.ID, ObjectParser.ValueType.STRING_ARRAY);
+        PARSER.declareBoolean(GetDatafeedStatsRequest::setAllowNoDatafeeds, ALLOW_NO_DATAFEEDS);
+    }
+
+    private static final String ALL_DATAFEEDS = "_all";
+
+    private final List<String> datafeedIds;
+    private Boolean allowNoDatafeeds;
+
+    /**
+     * Explicitly gets all datafeeds statistics
+     *
+     * @return a {@link GetDatafeedStatsRequest} for all existing datafeeds
+     */
+    public static GetDatafeedStatsRequest getAllDatafeedStatsRequest(){
+        return new GetDatafeedStatsRequest(ALL_DATAFEEDS);
+    }
+
+    GetDatafeedStatsRequest(List<String> datafeedIds) {
+        if (datafeedIds.stream().anyMatch(Objects::isNull)) {
+            throw new NullPointerException("datafeedIds must not contain null values");
+        }
+        this.datafeedIds = new ArrayList<>(datafeedIds);
+    }
+
+    /**
+     * Get the specified Datafeed's statistics via their unique datafeedIds
+     *
+     * @param datafeedIds must be non-null and each datafeedId must be non-null
+     */
+    public GetDatafeedStatsRequest(String... datafeedIds) {
+        this(Arrays.asList(datafeedIds));
+    }
+
+    /**
+     * All the datafeedIds for which to get statistics
+     */
+    public List<String> getDatafeedIds() {
+        return datafeedIds;
+    }
+
+    public Boolean isAllowNoDatafeeds() {
+        return this.allowNoDatafeeds;
+    }
+
+    /**
+     * Whether to ignore if a wildcard expression matches no datafeeds.
+     *
+     * This includes {@code _all} string or when no datafeeds have been specified
+     *
+     * @param allowNoDatafeeds When {@code true} ignore if wildcard or {@code _all} matches no datafeeds. Defaults to {@code true}
+     */
+    public void setAllowNoDatafeeds(boolean allowNoDatafeeds) {
+        this.allowNoDatafeeds = allowNoDatafeeds;
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(datafeedIds, allowNoDatafeeds);
+    }
+
+    @Override
+    public boolean equals(Object other) {
+        if (this == other) {
+            return true;
+        }
+
+        if (other == null || getClass() != other.getClass()) {
+            return false;
+        }
+
+        GetDatafeedStatsRequest that = (GetDatafeedStatsRequest) other;
+        return Objects.equals(datafeedIds, that.datafeedIds) &&
+            Objects.equals(allowNoDatafeeds, that.allowNoDatafeeds);
+    }
+
+    @Override
+    public ActionRequestValidationException validate() {
+        return null;
+    }
+
+    @Override
+    public XContentBuilder toXContent(XContentBuilder builder, ToXContent.Params params) throws IOException {
+        builder.startObject();
+        builder.field(DatafeedConfig.ID.getPreferredName(), Strings.collectionToCommaDelimitedString(datafeedIds));
+        if (allowNoDatafeeds != null) {
+            builder.field(ALLOW_NO_DATAFEEDS.getPreferredName(), allowNoDatafeeds);
+        }
+        builder.endObject();
+        return builder;
+    }
+
+}

client/rest-high-level/src/main/java/org/elasticsearch/client/ml/GetDatafeedStatsResponse.java

@@ -0,0 +1,89 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.elasticsearch.client.ml;
+
+import org.elasticsearch.client.ml.datafeed.DatafeedStats;
+import org.elasticsearch.common.ParseField;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.xcontent.ConstructingObjectParser;
+import org.elasticsearch.common.xcontent.XContentParser;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Objects;
+
+import static org.elasticsearch.common.xcontent.ConstructingObjectParser.constructorArg;
+
+/**
+ * Contains a {@link List} of the found {@link DatafeedStats} objects and the total count found
+ */
+public class GetDatafeedStatsResponse extends AbstractResultResponse<DatafeedStats> {
+
+    public static final ParseField RESULTS_FIELD = new ParseField("datafeeds");
+
+    @SuppressWarnings("unchecked")
+    public static final ConstructingObjectParser<GetDatafeedStatsResponse, Void> PARSER =
+        new ConstructingObjectParser<>("get_datafeed_stats_response",
+            true,
+            a -> new GetDatafeedStatsResponse((List<DatafeedStats>) a[0], (long) a[1]));
+
+    static {
+        PARSER.declareObjectArray(constructorArg(), DatafeedStats.PARSER, RESULTS_FIELD);
+        PARSER.declareLong(constructorArg(), COUNT);
+    }
+
+    GetDatafeedStatsResponse(List<DatafeedStats> results, long count) {
+        super(RESULTS_FIELD, results, count);
+    }
+
+    /**
+     * The collection of {@link DatafeedStats} objects found in the query
+     */
+    public List<DatafeedStats> datafeedStats() {
+        return results;
+    }
+
+    public static GetDatafeedStatsResponse fromXContent(XContentParser parser) throws IOException {
+        return PARSER.parse(parser, null);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(results, count);
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (this == obj) {
+            return true;
+        }
+
+        if (obj == null || getClass() != obj.getClass()) {
+            return false;
+        }
+
+        GetDatafeedStatsResponse other = (GetDatafeedStatsResponse) obj;
+        return Objects.equals(results, other.results) && count == other.count;
+    }
+
+    @Override
+    public final String toString() {
+        return Strings.toString(this);
+    }
+}

client/rest-high-level/src/main/java/org/elasticsearch/client/ml/datafeed/DatafeedState.java

@@ -0,0 +1,42 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.elasticsearch.client.ml.datafeed;
+
+import org.elasticsearch.common.ParseField;
+
+import java.util.Locale;
+
+/**
+ * Datafeed State POJO
+ */
+public enum DatafeedState {
+
+    STARTED, STOPPED, STARTING, STOPPING;
+
+    public static final ParseField STATE = new ParseField("state");
+
+    public static DatafeedState fromString(String name) {
+        return valueOf(name.trim().toUpperCase(Locale.ROOT));
+    }
+
+    @Override
+    public String toString() {
+        return name().toLowerCase(Locale.ROOT);
+    }
+}

client/rest-high-level/src/main/java/org/elasticsearch/client/ml/datafeed/DatafeedStats.java

@@ -0,0 +1,136 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.elasticsearch.client.ml.datafeed;
+
+import org.elasticsearch.client.ml.NodeAttributes;
+import org.elasticsearch.common.Nullable;
+import org.elasticsearch.common.ParseField;
+import org.elasticsearch.common.xcontent.ConstructingObjectParser;
+import org.elasticsearch.common.xcontent.ToXContent;
+import org.elasticsearch.common.xcontent.ToXContentObject;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * Datafeed Statistics POJO
+ */
+public class DatafeedStats implements ToXContentObject {
+
+    private final String datafeedId;
+    private final DatafeedState datafeedState;
+    @Nullable
+    private final NodeAttributes node;
+    @Nullable
+    private final String assignmentExplanation;
+
+    public static final ParseField ASSIGNMENT_EXPLANATION = new ParseField("assignment_explanation");
+    public static final ParseField NODE = new ParseField("node");
+
+    public static final ConstructingObjectParser<DatafeedStats, Void> PARSER = new ConstructingObjectParser<>("datafeed_stats",
+    true,
+    a -> {
+        String datafeedId = (String)a[0];
+        DatafeedState datafeedState = DatafeedState.fromString((String)a[1]);
+        NodeAttributes nodeAttributes = (NodeAttributes)a[2];
+        String assignmentExplanation = (String)a[3];
+        return new DatafeedStats(datafeedId, datafeedState, nodeAttributes, assignmentExplanation);
+    } );
+
+    static {
+        PARSER.declareString(ConstructingObjectParser.constructorArg(), DatafeedConfig.ID);
+        PARSER.declareString(ConstructingObjectParser.constructorArg(), DatafeedState.STATE);
+        PARSER.declareObject(ConstructingObjectParser.optionalConstructorArg(), NodeAttributes.PARSER, NODE);
+        PARSER.declareString(ConstructingObjectParser.optionalConstructorArg(), ASSIGNMENT_EXPLANATION);
+    }
+
+    public DatafeedStats(String datafeedId, DatafeedState datafeedState, @Nullable NodeAttributes node,
+                         @Nullable String assignmentExplanation) {
+        this.datafeedId = Objects.requireNonNull(datafeedId);
+        this.datafeedState = Objects.requireNonNull(datafeedState);
+        this.node = node;
+        this.assignmentExplanation = assignmentExplanation;
+    }
+
+    public String getDatafeedId() {
+        return datafeedId;
+    }
+
+    public DatafeedState getDatafeedState() {
+        return datafeedState;
+    }
+
+    public NodeAttributes getNode() {
+        return node;
+    }
+
+    public String getAssignmentExplanation() {
+        return assignmentExplanation;
+    }
+
+    @Override
+    public XContentBuilder toXContent(XContentBuilder builder, ToXContent.Params params) throws IOException {
+        builder.startObject();
+        builder.field(DatafeedConfig.ID.getPreferredName(), datafeedId);
+        builder.field(DatafeedState.STATE.getPreferredName(), datafeedState.toString());
+        if (node != null) {
+            builder.startObject("node");
+            builder.field("id", node.getId());
+            builder.field("name", node.getName());
+            builder.field("ephemeral_id", node.getEphemeralId());
+            builder.field("transport_address", node.getTransportAddress());
+
+            builder.startObject("attributes");
+            for (Map.Entry<String, String> entry : node.getAttributes().entrySet()) {
+                if (entry.getKey().startsWith("ml.")) {
+                    builder.field(entry.getKey(), entry.getValue());
+                }
+            }
+            builder.endObject();
+            builder.endObject();
+        }
+        if (assignmentExplanation != null) {
+            builder.field(ASSIGNMENT_EXPLANATION.getPreferredName(), assignmentExplanation);
+        }
+        builder.endObject();
+        return builder;
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(datafeedId, datafeedState.toString(), node, assignmentExplanation);
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (obj == null) {
+            return false;
+        }
+        if (getClass() != obj.getClass()) {
+            return false;
+        }
+        DatafeedStats other = (DatafeedStats) obj;
+        return Objects.equals(datafeedId, other.datafeedId) &&
+            Objects.equals(this.datafeedState, other.datafeedState) &&
+            Objects.equals(this.node, other.node) &&
+            Objects.equals(this.assignmentExplanation, other.assignmentExplanation);
+    }
+}

client/rest-high-level/src/test/java/org/elasticsearch/client/MLRequestConvertersTests.java

@@ -34,6 +34,7 @@ import org.elasticsearch.client.ml.GetBucketsRequest;
 import org.elasticsearch.client.ml.GetCalendarsRequest;
 import org.elasticsearch.client.ml.GetCategoriesRequest;
 import org.elasticsearch.client.ml.GetDatafeedRequest;
+import org.elasticsearch.client.ml.GetDatafeedStatsRequest;
 import org.elasticsearch.client.ml.GetInfluencersRequest;
 import org.elasticsearch.client.ml.GetJobRequest;
 import org.elasticsearch.client.ml.GetJobStatsRequest;
@@ -294,6 +295,23 @@ public class MLRequestConvertersTests extends ESTestCase {
         }
     }
 
+    public void testGetDatafeedStats() {
+        GetDatafeedStatsRequest getDatafeedStatsRequestRequest = new GetDatafeedStatsRequest();
+
+        Request request = MLRequestConverters.getDatafeedStats(getDatafeedStatsRequestRequest);
+
+        assertEquals(HttpGet.METHOD_NAME, request.getMethod());
+        assertEquals("/_xpack/ml/datafeeds/_stats", request.getEndpoint());
+        assertFalse(request.getParameters().containsKey("allow_no_datafeeds"));
+
+        getDatafeedStatsRequestRequest = new GetDatafeedStatsRequest("datafeed1", "datafeeds*");
+        getDatafeedStatsRequestRequest.setAllowNoDatafeeds(true);
+        request = MLRequestConverters.getDatafeedStats(getDatafeedStatsRequestRequest);
+
+        assertEquals("/_xpack/ml/datafeeds/datafeed1,datafeeds*/_stats", request.getEndpoint());
+        assertEquals(Boolean.toString(true), request.getParameters().get("allow_no_datafeeds"));
+    }
+
     public void testPreviewDatafeed() {
         PreviewDatafeedRequest datafeedRequest = new PreviewDatafeedRequest("datafeed_1");
         Request request = MLRequestConverters.previewDatafeed(datafeedRequest);

client/rest-high-level/src/test/java/org/elasticsearch/client/MachineLearningIT.java

@@ -41,6 +41,8 @@ import org.elasticsearch.client.ml.GetCalendarsRequest;
 import org.elasticsearch.client.ml.GetCalendarsResponse;
 import org.elasticsearch.client.ml.GetDatafeedRequest;
 import org.elasticsearch.client.ml.GetDatafeedResponse;
+import org.elasticsearch.client.ml.GetDatafeedStatsRequest;
+import org.elasticsearch.client.ml.GetDatafeedStatsResponse;
 import org.elasticsearch.client.ml.GetJobRequest;
 import org.elasticsearch.client.ml.GetJobResponse;
 import org.elasticsearch.client.ml.GetJobStatsRequest;
@@ -65,6 +67,8 @@ import org.elasticsearch.client.ml.UpdateJobRequest;
 import org.elasticsearch.client.ml.calendars.Calendar;
 import org.elasticsearch.client.ml.calendars.CalendarTests;
 import org.elasticsearch.client.ml.datafeed.DatafeedConfig;
+import org.elasticsearch.client.ml.datafeed.DatafeedState;
+import org.elasticsearch.client.ml.datafeed.DatafeedStats;
 import org.elasticsearch.client.ml.job.config.AnalysisConfig;
 import org.elasticsearch.client.ml.job.config.DataDescription;
 import org.elasticsearch.client.ml.job.config.Detector;
@@ -569,6 +573,76 @@ public class MachineLearningIT extends ESRestHighLevelClientTestCase {
         }
     }
 
+    public void testGetDatafeedStats() throws Exception {
+        String jobId1 = "ml-get-datafeed-stats-test-id-1";
+        String jobId2 = "ml-get-datafeed-stats-test-id-2";
+        String indexName = "datafeed_stats_data_1";
+
+        // Set up the index
+        CreateIndexRequest createIndexRequest = new CreateIndexRequest(indexName);
+        createIndexRequest.mapping("doc", "timestamp", "type=date", "total", "type=long");
+        highLevelClient().indices().create(createIndexRequest, RequestOptions.DEFAULT);
+
+        // create the job and the datafeed
+        Job job1 = buildJob(jobId1);
+        putJob(job1);
+        openJob(job1);
+
+        Job job2 = buildJob(jobId2);
+        putJob(job2);
+
+        String datafeedId1 = createAndPutDatafeed(jobId1, indexName);
+        String datafeedId2 = createAndPutDatafeed(jobId2, indexName);
+
+        MachineLearningClient machineLearningClient = highLevelClient().machineLearning();
+
+        machineLearningClient.startDatafeed(new StartDatafeedRequest(datafeedId1), RequestOptions.DEFAULT);
+
+        GetDatafeedStatsRequest request = new GetDatafeedStatsRequest(datafeedId1);
+
+        // Test getting specific
+        GetDatafeedStatsResponse response =
+            execute(request, machineLearningClient::getDatafeedStats, machineLearningClient::getDatafeedStatsAsync);
+
+        assertEquals(1, response.count());
+        assertThat(response.datafeedStats(), hasSize(1));
+        assertThat(response.datafeedStats().get(0).getDatafeedId(), equalTo(datafeedId1));
+        assertThat(response.datafeedStats().get(0).getDatafeedState().toString(), equalTo(DatafeedState.STARTED.toString()));
+
+        // Test getting all explicitly
+        request = GetDatafeedStatsRequest.getAllDatafeedStatsRequest();
+        response = execute(request, machineLearningClient::getDatafeedStats, machineLearningClient::getDatafeedStatsAsync);
+
+        assertTrue(response.count() >= 2L);
+        assertTrue(response.datafeedStats().size() >= 2L);
+        assertThat(response.datafeedStats().stream().map(DatafeedStats::getDatafeedId).collect(Collectors.toList()),
+            hasItems(datafeedId1, datafeedId2));
+
+        // Test getting all implicitly
+        response =
+            execute(new GetDatafeedStatsRequest(), machineLearningClient::getDatafeedStats, machineLearningClient::getDatafeedStatsAsync);
+
+        assertTrue(response.count() >= 2L);
+        assertTrue(response.datafeedStats().size() >= 2L);
+        assertThat(response.datafeedStats().stream().map(DatafeedStats::getDatafeedId).collect(Collectors.toList()),
+            hasItems(datafeedId1, datafeedId2));
+
+        // Test getting all with wildcard
+        request = new GetDatafeedStatsRequest("ml-get-datafeed-stats-test-id-*");
+        response = execute(request, machineLearningClient::getDatafeedStats, machineLearningClient::getDatafeedStatsAsync);
+        assertEquals(2L, response.count());
+        assertThat(response.datafeedStats(), hasSize(2));
+        assertThat(response.datafeedStats().stream().map(DatafeedStats::getDatafeedId).collect(Collectors.toList()),
+            hasItems(datafeedId1, datafeedId2));
+
+        // Test when allow_no_jobs is false
+        final GetDatafeedStatsRequest erroredRequest = new GetDatafeedStatsRequest("datafeeds-that-do-not-exist*");
+        erroredRequest.setAllowNoDatafeeds(false);
+        ElasticsearchStatusException exception = expectThrows(ElasticsearchStatusException.class,
+            () -> execute(erroredRequest, machineLearningClient::getDatafeedStats, machineLearningClient::getDatafeedStatsAsync));
+        assertThat(exception.status().getStatus(), equalTo(404));
+    }
+
     public void testPreviewDatafeed() throws Exception {
         String jobId = "test-preview-datafeed";
         String indexName = "preview_data_1";

client/rest-high-level/src/test/java/org/elasticsearch/client/documentation/MlClientDocumentationIT.java

@@ -51,6 +51,8 @@ import org.elasticsearch.client.ml.GetCategoriesRequest;
 import org.elasticsearch.client.ml.GetCategoriesResponse;
 import org.elasticsearch.client.ml.GetDatafeedRequest;
 import org.elasticsearch.client.ml.GetDatafeedResponse;
+import org.elasticsearch.client.ml.GetDatafeedStatsRequest;
+import org.elasticsearch.client.ml.GetDatafeedStatsResponse;
 import org.elasticsearch.client.ml.GetInfluencersRequest;
 import org.elasticsearch.client.ml.GetInfluencersResponse;
 import org.elasticsearch.client.ml.GetJobRequest;
@@ -81,6 +83,7 @@ import org.elasticsearch.client.ml.UpdateJobRequest;
 import org.elasticsearch.client.ml.calendars.Calendar;
 import org.elasticsearch.client.ml.datafeed.ChunkingConfig;
 import org.elasticsearch.client.ml.datafeed.DatafeedConfig;
+import org.elasticsearch.client.ml.datafeed.DatafeedStats;
 import org.elasticsearch.client.ml.job.config.AnalysisConfig;
 import org.elasticsearch.client.ml.job.config.AnalysisLimits;
 import org.elasticsearch.client.ml.job.config.DataDescription;
@@ -885,6 +888,82 @@ public class MlClientDocumentationIT extends ESRestHighLevelClientTestCase {
         }
     }
 
+    public void testGetDatafeedStats() throws Exception {
+        RestHighLevelClient client = highLevelClient();
+
+        Job job = MachineLearningIT.buildJob("get-machine-learning-datafeed-stats1");
+        client.machineLearning().putJob(new PutJobRequest(job), RequestOptions.DEFAULT);
+
+        Job secondJob = MachineLearningIT.buildJob("get-machine-learning-datafeed-stats2");
+        client.machineLearning().putJob(new PutJobRequest(secondJob), RequestOptions.DEFAULT);
+        String datafeedId1 = job.getId() + "-feed";
+        String indexName = "datafeed_stats_data_2";
+        CreateIndexRequest createIndexRequest = new CreateIndexRequest(indexName);
+        createIndexRequest.mapping("doc", "timestamp", "type=date", "total", "type=long");
+        highLevelClient().indices().create(createIndexRequest, RequestOptions.DEFAULT);
+        DatafeedConfig datafeed = DatafeedConfig.builder(datafeedId1, job.getId())
+            .setTypes(Arrays.asList("doc"))
+            .setIndices(indexName)
+            .build();
+        client.machineLearning().putDatafeed(new PutDatafeedRequest(datafeed), RequestOptions.DEFAULT);
+
+        String datafeedId2 = secondJob.getId() + "-feed";
+        DatafeedConfig secondDatafeed = DatafeedConfig.builder(datafeedId2, secondJob.getId())
+            .setTypes(Arrays.asList("doc"))
+            .setIndices(indexName)
+            .build();
+        client.machineLearning().putDatafeed(new PutDatafeedRequest(secondDatafeed), RequestOptions.DEFAULT);
+
+        {
+            //tag::x-pack-ml-get-datafeed-stats-request
+            GetDatafeedStatsRequest request =
+                new GetDatafeedStatsRequest("get-machine-learning-datafeed-stats1-feed", "get-machine-learning-datafeed*"); // <1>
+            request.setAllowNoDatafeeds(true); // <2>
+            //end::x-pack-ml-get-datafeed-stats-request
+
+            //tag::x-pack-ml-get-datafeed-stats-execute
+            GetDatafeedStatsResponse response = client.machineLearning().getDatafeedStats(request, RequestOptions.DEFAULT);
+            //end::x-pack-ml-get-datafeed-stats-execute
+
+            //tag::x-pack-ml-get-datafeed-stats-response
+            long numberOfDatafeedStats = response.count(); // <1>
+            List<DatafeedStats> datafeedStats = response.datafeedStats(); // <2>
+            //end::x-pack-ml-get-datafeed-stats-response
+
+            assertEquals(2, response.count());
+            assertThat(response.datafeedStats(), hasSize(2));
+            assertThat(response.datafeedStats().stream().map(DatafeedStats::getDatafeedId).collect(Collectors.toList()),
+                containsInAnyOrder(datafeed.getId(), secondDatafeed.getId()));
+        }
+        {
+            GetDatafeedStatsRequest request = new GetDatafeedStatsRequest("*");
+
+            // tag::x-pack-ml-get-datafeed-stats-listener
+            ActionListener<GetDatafeedStatsResponse> listener = new ActionListener<GetDatafeedStatsResponse>() {
+                @Override
+                public void onResponse(GetDatafeedStatsResponse response) {
+                    // <1>
+                }
+
+                @Override
+                public void onFailure(Exception e) {
+                    // <2>
+                }
+            };
+            // end::x-pack-ml-get-datafeed-stats-listener
+
+            // Replace the empty listener by a blocking listener in test
+            final CountDownLatch latch = new CountDownLatch(1);
+            listener = new LatchedActionListener<>(listener, latch);
+
+            // tag::x-pack-ml-get-datafeed-stats-execute-async
+            client.machineLearning().getDatafeedStatsAsync(request, RequestOptions.DEFAULT, listener); // <1>
+            // end::x-pack-ml-get-datafeed-stats-execute-async
+
+            assertTrue(latch.await(30L, TimeUnit.SECONDS));
+        }
+    }
+
     public void testGetBuckets() throws IOException, InterruptedException {
         RestHighLevelClient client = highLevelClient();
 

client/rest-high-level/src/test/java/org/elasticsearch/client/ml/GetDatafeedStatsRequestTests.java

@@ -0,0 +1,69 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.elasticsearch.client.ml;
+
+import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.test.AbstractXContentTestCase;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+public class GetDatafeedStatsRequestTests extends AbstractXContentTestCase<GetDatafeedStatsRequest> {
+
+    public void testAllDatafeedsRequest() {
+        GetDatafeedStatsRequest request = GetDatafeedStatsRequest.getAllDatafeedStatsRequest();
+
+        assertEquals(request.getDatafeedIds().size(), 1);
+        assertEquals(request.getDatafeedIds().get(0), "_all");
+    }
+
+    public void testNewWithDatafeedId() {
+        Exception exception = expectThrows(NullPointerException.class, () -> new GetDatafeedStatsRequest("datafeed", null));
+        assertEquals(exception.getMessage(), "datafeedIds must not contain null values");
+    }
+
+    @Override
+    protected GetDatafeedStatsRequest createTestInstance() {
+        int datafeedCount = randomIntBetween(0, 10);
+        List<String> datafeedIds = new ArrayList<>(datafeedCount);
+
+        for (int i = 0; i < datafeedCount; i++) {
+            datafeedIds.add(randomAlphaOfLength(10));
+        }
+
+        GetDatafeedStatsRequest request = new GetDatafeedStatsRequest(datafeedIds);
+
+        if (randomBoolean()) {
+            request.setAllowNoDatafeeds(randomBoolean());
+        }
+
+        return request;
+    }
+
+    @Override
+    protected GetDatafeedStatsRequest doParseInstance(XContentParser parser) throws IOException {
+        return GetDatafeedStatsRequest.PARSER.parse(parser, null);
+    }
+
+    @Override
+    protected boolean supportsUnknownFields() {
+        return false;
+    }
+}

client/rest-high-level/src/test/java/org/elasticsearch/client/ml/GetDatafeedStatsResponseTests.java

@@ -0,0 +1,59 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.elasticsearch.client.ml;
+
+import org.elasticsearch.client.ml.datafeed.DatafeedStats;
+import org.elasticsearch.client.ml.datafeed.DatafeedStatsTests;
+import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.test.AbstractXContentTestCase;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.function.Predicate;
+
+public class GetDatafeedStatsResponseTests extends AbstractXContentTestCase<GetDatafeedStatsResponse> {
+
+    @Override
+    protected GetDatafeedStatsResponse createTestInstance() {
+
+        int count = randomIntBetween(1, 5);
+        List<DatafeedStats> results = new ArrayList<>(count);
+        for(int i = 0; i < count; i++) {
+            results.add(DatafeedStatsTests.createRandomInstance());
+        }
+
+        return new GetDatafeedStatsResponse(results, count);
+    }
+
+    @Override
+    protected GetDatafeedStatsResponse doParseInstance(XContentParser parser) throws IOException {
+        return GetDatafeedStatsResponse.fromXContent(parser);
+    }
+
+    @Override
+    protected boolean supportsUnknownFields() {
+        return true;
+    }
+
+    @Override
+    protected Predicate<String> getRandomFieldsExcludeFilter() {
+        return field -> !field.isEmpty();
+    }
+}

client/rest-high-level/src/test/java/org/elasticsearch/client/ml/datafeed/DatafeedStatsTests.java

@@ -0,0 +1,75 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.elasticsearch.client.ml.datafeed;
+
+import org.elasticsearch.client.ml.NodeAttributes;
+import org.elasticsearch.client.ml.NodeAttributesTests;
+import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.test.AbstractXContentTestCase;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.function.Predicate;
+
+public class DatafeedStatsTests extends AbstractXContentTestCase<DatafeedStats> {
+
+    public static DatafeedStats createRandomInstance() {
+        String datafeedId = DatafeedConfigTests.randomValidDatafeedId();
+        DatafeedState datafeedState =
+            randomFrom(DatafeedState.STARTED, DatafeedState.STARTING, DatafeedState.STOPPED, DatafeedState.STOPPING);
+        NodeAttributes nodeAttributes = null;
+        if (randomBoolean()) {
+            NodeAttributes randomAttributes = NodeAttributesTests.createRandom();
+            int numberOfAttributes = randomIntBetween(1, 10);
+            Map<String, String> attributes = new HashMap<>(numberOfAttributes);
+            for(int i = 0; i < numberOfAttributes; i++) {
+                String val = randomAlphaOfLength(10);
+                attributes.put("ml.key-"+i, val);
+            }
+            nodeAttributes = new NodeAttributes(randomAttributes.getId(),
+                randomAttributes.getName(),
+                randomAttributes.getEphemeralId(),
+                randomAttributes.getTransportAddress(),
+                attributes);
+        }
+        String assignmentReason = randomBoolean() ? randomAlphaOfLength(10) : null;
+        return new DatafeedStats(datafeedId, datafeedState, nodeAttributes, assignmentReason);
+    }
+
+    @Override
+    protected DatafeedStats createTestInstance() {
+        return createRandomInstance();
+    }
+
+    @Override
+    protected DatafeedStats doParseInstance(XContentParser parser) throws IOException {
+        return DatafeedStats.PARSER.apply(parser, null);
+    }
+
+    @Override
+    protected boolean supportsUnknownFields() {
+        return true;
+    }
+
+    @Override
+    protected Predicate<String> getRandomFieldsExcludeFilter() {
+        return field -> field.equals("node.attributes");
+    }
+}

docs/java-rest/high-level/ml/get-datafeed-stats.asciidoc

@@ -0,0 +1,67 @@
+[[java-rest-high-x-pack-ml-get-datafeed-stats]]
+=== Get Datafeed Stats API
+
+The Get Datafeed Stats API provides the ability to get any number of
+ {ml} datafeed's statistics in the cluster.
+It accepts a `GetDatafeedStatsRequest` object and responds
+with a `GetDatafeedStatsResponse` object.
+
+[[java-rest-high-x-pack-ml-get-datafeed-stats-request]]
+==== Get Datafeed Stats Request
+
+A `GetDatafeedStatsRequest` object can have any number of `datafeedId`
+entries. However, they all must be non-null. An empty list is the same as
+requesting statistics for all datafeeds.
+
+["source","java",subs="attributes,callouts,macros"]
+--------------------------------------------------
+include-tagged::{doc-tests}/MlClientDocumentationIT.java[x-pack-ml-get-datafeed-stats-request]
+--------------------------------------------------
+<1> Constructing a new request referencing existing `datafeedIds`, can contain wildcards
+<2> Whether to ignore if a wildcard expression matches no datafeeds.
+ (This includes `_all` string or when no datafeeds have been specified)
+
+[[java-rest-high-x-pack-ml-get-datafeed-stats-execution]]
+==== Execution
+
+The request can be executed through the `MachineLearningClient` contained
+in the `RestHighLevelClient` object, accessed via the `machineLearningClient()` method.
+
+["source","java",subs="attributes,callouts,macros"]
+--------------------------------------------------
+include-tagged::{doc-tests}/MlClientDocumentationIT.java[x-pack-ml-get-datafeed-stats-execute]
+--------------------------------------------------
+
+[[java-rest-high-x-pack-ml-get-datafeed-stats-execution-async]]
+==== Asynchronous Execution
+
+The request can also be executed asynchronously:
+
+["source","java",subs="attributes,callouts,macros"]
+--------------------------------------------------
+include-tagged::{doc-tests}/MlClientDocumentationIT.java[x-pack-ml-get-datafeed-stats-execute-async]
+--------------------------------------------------
+<1> The `GetDatafeedStatsRequest` to execute and the `ActionListener` to use when
+the execution completes
+
+The method does not block and returns immediately. The passed `ActionListener` is used
+to notify the caller of completion. A typical `ActionListener` for `GetDatafeedStatsResponse` may
+look like
+
+["source","java",subs="attributes,callouts,macros"]
+--------------------------------------------------
+include-tagged::{doc-tests}/MlClientDocumentationIT.java[x-pack-ml-get-datafeed-stats-listener]
+--------------------------------------------------
+<1> `onResponse` is called back when the action is completed successfully
+<2> `onFailure` is called back when some unexpected error occurs
+
+[[java-rest-high-x-pack-ml-get-datafeed-stats-response]]
+==== Get Datafeed Stats Response
+The returned `GetDatafeedStatsResponse` contains the requested datafeed statistics:
+
+["source","java",subs="attributes,callouts,macros"]
+--------------------------------------------------
+include-tagged::{doc-tests}/MlClientDocumentationIT.java[x-pack-ml-get-datafeed-stats-response]
+--------------------------------------------------
+<1> `count()` indicates the number of datafeeds statistics found
+<2> `datafeedStats()` is the collection of {ml} `DatafeedStats` objects found

docs/java-rest/high-level/supported-apis.asciidoc

@@ -238,6 +238,7 @@ The Java High Level REST Client supports the following Machine Learning APIs:
 * <<{upid}-preview-datafeed>>
 * <<{upid}-start-datafeed>>
 * <<{upid}-stop-datafeed>>
+* <<{upid}-get-datafeed-stats>>
 * <<{upid}-forecast-job>>
 * <<{upid}-delete-forecast>>
 * <<{upid}-get-buckets>>
@@ -263,6 +264,7 @@ include::ml/delete-datafeed.asciidoc[]
 include::ml/preview-datafeed.asciidoc[]
 include::ml/start-datafeed.asciidoc[]
 include::ml/stop-datafeed.asciidoc[]
+include::ml/get-datafeed-stats.asciidoc[]
 include::ml/get-job-stats.asciidoc[]
 include::ml/forecast-job.asciidoc[]
 include::ml/delete-forecast.asciidoc[]