[ML] relax throttling on expired data cleanup (#56711) (#56895)

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

@@ -167,13 +167,13 @@ static Request closeJob(CloseJobRequest closeJobRequest) throws IOException {
         return request;
     }
 
-    static Request deleteExpiredData(DeleteExpiredDataRequest deleteExpiredDataRequest) {
+    static Request deleteExpiredData(DeleteExpiredDataRequest deleteExpiredDataRequest) throws IOException {
         String endpoint = new EndpointBuilder()
             .addPathPartAsIs("_ml")
             .addPathPartAsIs("_delete_expired_data")
             .build();
         Request request = new Request(HttpDelete.METHOD_NAME, endpoint);
-
+        request.setEntity(createEntity(deleteExpiredDataRequest, REQUEST_BODY_CONTENT_TYPE));
         return request;
     }
 

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

@@ -20,20 +20,82 @@
 
 import org.elasticsearch.action.ActionRequest;
 import org.elasticsearch.action.ActionRequestValidationException;
+import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.common.xcontent.ToXContentObject;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+
+import java.io.IOException;
+import java.util.Objects;
 
 /**
  * Request to delete expired model snapshots and forecasts
  */
-public class DeleteExpiredDataRequest extends ActionRequest {
+public class DeleteExpiredDataRequest extends ActionRequest implements ToXContentObject {
 
+    static final String REQUESTS_PER_SECOND = "requests_per_second";
+    static final String TIMEOUT = "timeout";
+    private final Float requestsPerSecond;
+    private final TimeValue timeout;
    /**
      * Create a new request to delete expired data
      */
     public DeleteExpiredDataRequest() {
+        this(null, null);
+    }
+
+    public DeleteExpiredDataRequest(Float requestsPerSecond, TimeValue timeout) {
+        this.requestsPerSecond = requestsPerSecond;
+        this.timeout = timeout;
+    }
+
+    /**
+     * The requests allowed per second in the underlying Delete by Query requests executed.
+     *
+     * `-1.0f` indicates that the standard nightly cleanup behavior should be ran.
+     *         Throttling scales according to the number of data nodes.
+     * `null` is default and means no throttling will occur.
+     */
+    public Float getRequestsPerSecond() {
+        return requestsPerSecond;
+    }
+
+    /**
+     * Indicates how long the deletion request will run until it timesout.
+     *
+     * Default value is 8 hours.
+     */
+    public TimeValue getTimeout() {
+        return timeout;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        DeleteExpiredDataRequest that = (DeleteExpiredDataRequest) o;
+        return Objects.equals(requestsPerSecond, that.requestsPerSecond) &&
+            Objects.equals(timeout, that.timeout);
     }
 
     @Override
     public ActionRequestValidationException validate() {
         return null;
     }
+
+    public int hashCode() {
+        return Objects.hash(requestsPerSecond, timeout);
+    }
+
+    @Override
+    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+        builder.startObject();
+        if (requestsPerSecond != null) {
+            builder.field(REQUESTS_PER_SECOND, requestsPerSecond);
+        }
+        if (timeout != null) {
+            builder.field(TIMEOUT, timeout.getStringRep());
+        }
+        builder.endObject();
+        return builder;
+    }
 }

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

@@ -214,12 +214,16 @@ public void testCloseJob() throws Exception {
             requestEntityToString(request));
     }
 
-    public void testDeleteExpiredData() {
-        DeleteExpiredDataRequest deleteExpiredDataRequest = new DeleteExpiredDataRequest();
+    public void testDeleteExpiredData() throws Exception {
+        float requestsPerSec = randomBoolean() ? -1.0f : (float)randomDoubleBetween(0.0, 100000.0, false);
+        DeleteExpiredDataRequest deleteExpiredDataRequest = new DeleteExpiredDataRequest(
+            requestsPerSec,
+            TimeValue.timeValueHours(1));
 
         Request request = MLRequestConverters.deleteExpiredData(deleteExpiredDataRequest);
         assertEquals(HttpDelete.METHOD_NAME, request.getMethod());
         assertEquals("/_ml/_delete_expired_data", request.getEndpoint());
+        assertEquals("{\"requests_per_second\":" + requestsPerSec + ",\"timeout\":\"1h\"}", requestEntityToString(request));
     }
 
     public void testDeleteJob() {

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

@@ -2035,7 +2035,11 @@ public void testDeleteExpiredData() throws IOException, InterruptedException {
         MachineLearningIT.buildJob(jobId);
        {
             // tag::delete-expired-data-request
-            DeleteExpiredDataRequest request = new DeleteExpiredDataRequest(); // <1>
+            DeleteExpiredDataRequest request = new DeleteExpiredDataRequest( // <1>
+               1000.0f, // <2>
+               TimeValue.timeValueHours(12) // <3>
+            );
+
             // end::delete-expired-data-request
 
             // tag::delete-expired-data-execute

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

@@ -0,0 +1,62 @@
+/*
+ * 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.ParseField;
+import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.common.xcontent.ConstructingObjectParser;
+import org.elasticsearch.common.xcontent.ObjectParser;
+import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.test.AbstractXContentTestCase;
+
+import java.io.IOException;
+
+
+public class DeleteExpiredDataRequestTests extends AbstractXContentTestCase<DeleteExpiredDataRequest> {
+
+    private static ConstructingObjectParser<DeleteExpiredDataRequest, Void> PARSER = new ConstructingObjectParser<>(
+        "delete_expired_data_request",
+        true,
+        (a) -> new DeleteExpiredDataRequest((Float) a[0], (TimeValue) a[1])
+    );
+    static {
+        PARSER.declareFloat(ConstructingObjectParser.optionalConstructorArg(),
+            new ParseField(DeleteExpiredDataRequest.REQUESTS_PER_SECOND));
+        PARSER.declareField(ConstructingObjectParser.optionalConstructorArg(),
+            (p, c) -> TimeValue.parseTimeValue(p.text(), DeleteExpiredDataRequest.TIMEOUT),
+            new ParseField(DeleteExpiredDataRequest.TIMEOUT),
+            ObjectParser.ValueType.STRING);
+    }
+
+    @Override
+    protected DeleteExpiredDataRequest createTestInstance() {
+        return new DeleteExpiredDataRequest(randomBoolean() ? null : randomFloat(),
+            randomBoolean() ? null : TimeValue.parseTimeValue(randomTimeValue(), "test"));
+    }
+
+    @Override
+    protected DeleteExpiredDataRequest doParseInstance(XContentParser parser) throws IOException {
+        return PARSER.apply(parser, null);
+    }
+
+    @Override
+    protected boolean supportsUnknownFields() {
+        return true;
+    }
+}

docs/java-rest/high-level/ml/delete-expired-data.asciidoc

@@ -21,6 +21,10 @@ A `DeleteExpiredDataRequest` object does not require any arguments.
 include-tagged::{doc-tests-file}[{api}-request]
 ---------------------------------------------------
 <1> Constructing a new request.
+<2> Providing requests per second throttling for the
+    deletion processes. Default is no throttling.
+<3> Setting how long the deletion processes will be allowed
+    to run before they are canceled. Default value is `8h` (8 hours).
 
 [id="{upid}-{api}-response"]
 ==== Delete Expired Data Response

docs/reference/ml/anomaly-detection/apis/delete-expired-data.asciidoc

@@ -27,6 +27,17 @@ Deletes all job results, model snapshots and forecast data that have exceeded
 their `retention days` period. Machine learning state documents that are not
 associated with any job are also deleted.
 
+[[ml-delete-expired-data-request-body]]
+==== {api-request-body-title}
+
+`requests_per_second`::
+(Optional, float) The desired requests per second for the deletion processes.
+The default behavior is no throttling.
+
+`timeout`::
+(Optional, string) How long can the underlying delete processes run until they are canceled.
+The default value is `8h` (8 hours).
+
 [[ml-delete-expired-data-example]]
 ==== {api-examples-title}
 

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/action/DeleteExpiredDataAction.java

@@ -5,6 +5,7 @@
  */
 package org.elasticsearch.xpack.core.ml.action;
 
+import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionRequest;
 import org.elasticsearch.action.ActionRequestBuilder;
 import org.elasticsearch.action.ActionRequestValidationException;
@@ -14,6 +15,8 @@
 import org.elasticsearch.common.ParseField;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.common.xcontent.ObjectParser;
 import org.elasticsearch.common.xcontent.ToXContentObject;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 
@@ -31,20 +34,94 @@ private DeleteExpiredDataAction() {
 
     public static class Request extends ActionRequest {
 
+        public static final ParseField REQUESTS_PER_SECOND = new ParseField("requests_per_second");
+        public static final ParseField TIMEOUT = new ParseField("timeout");
+
+        public static final ObjectParser<Request, Void> PARSER = new ObjectParser<>(
+            "delete_expired_data_request",
+            false,
+            Request::new);
+
+        static {
+            PARSER.declareFloat(Request::setRequestsPerSecond, REQUESTS_PER_SECOND);
+            PARSER.declareString((obj, value) -> obj.setTimeout(TimeValue.parseTimeValue(value, TIMEOUT.getPreferredName())),
+                TIMEOUT);
+        }
+
+        private Float requestsPerSecond;
+        private TimeValue timeout;
+
         public Request() {}
 
+        public Request(Float requestsPerSecond, TimeValue timeValue) {
+            this.requestsPerSecond = requestsPerSecond;
+            this.timeout = timeValue;
+        }
+
         public Request(StreamInput in) throws IOException {
             super(in);
+            if (in.getVersion().onOrAfter(Version.V_7_8_0)) {
+                this.requestsPerSecond = in.readOptionalFloat();
+                this.timeout = in.readOptionalTimeValue();
+            } else {
+                this.requestsPerSecond = null;
+                this.timeout = null;
+            }
+        }
+
+        public Float getRequestsPerSecond() {
+            return requestsPerSecond;
+        }
+
+        public TimeValue getTimeout() {
+            return timeout;
+        }
+
+        public Request setRequestsPerSecond(Float requestsPerSecond) {
+            this.requestsPerSecond = requestsPerSecond;
+            return this;
+        }
+
+        public Request setTimeout(TimeValue timeout) {
+            this.timeout = timeout;
+            return this;
         }
 
         @Override
         public ActionRequestValidationException validate() {
+            if (this.requestsPerSecond != null && this.requestsPerSecond != -1.0f && this.requestsPerSecond <= 0) {
+                ActionRequestValidationException requestValidationException = new ActionRequestValidationException();
+                requestValidationException.addValidationError("[requests_per_second] must either be -1 or greater than 0");
+                return requestValidationException;
+            }
             return null;
         }
+
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) return true;
+            if (o == null || getClass() != o.getClass()) return false;
+            Request request = (Request) o;
+            return Objects.equals(requestsPerSecond, request.requestsPerSecond)
+                && Objects.equals(timeout, request.timeout);
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(requestsPerSecond, timeout);
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            super.writeTo(out);
+            if (out.getVersion().onOrAfter(Version.V_7_8_0)) {
+                out.writeOptionalFloat(requestsPerSecond);
+                out.writeOptionalTimeValue(timeout);
+            }
+        }
     }
 
     static class RequestBuilder extends ActionRequestBuilder<Request, Response> {
-
         RequestBuilder(ElasticsearchClient client, DeleteExpiredDataAction action) {
             super(client, action, new Request());
         }

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/ml/action/DeleteExpiredDataActionRequestTests.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;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.xpack.core.ml.action;
+
+import org.elasticsearch.Version;
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.xpack.core.ml.AbstractBWCWireSerializationTestCase;
+import org.elasticsearch.xpack.core.ml.action.DeleteExpiredDataAction.Request;
+
+public class DeleteExpiredDataActionRequestTests extends AbstractBWCWireSerializationTestCase<Request> {
+
+    @Override
+    protected Request createTestInstance() {
+        return new Request(
+            randomBoolean() ? null : randomFloat(),
+            randomBoolean() ? null : TimeValue.parseTimeValue(randomTimeValue(), "test")
+        );
+    }
+
+    @Override
+    protected Writeable.Reader<Request> instanceReader() {
+        return Request::new;
+    }
+
+    @Override
+    protected Request mutateInstanceForVersion(Request instance, Version version) {
+        if (version.before(Version.V_7_8_0)) {
+            return new Request();
+        }
+        return instance;
+    }
+}