[Transform] add throttling

client/rest-high-level/src/main/java/org/elasticsearch/client/transform/transforms/SettingsConfig.java

@@ -0,0 +1,150 @@
+/*
+ * 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.transform.transforms;
+
+import org.elasticsearch.common.ParseField;
+import org.elasticsearch.common.xcontent.ConstructingObjectParser;
+import org.elasticsearch.common.xcontent.ToXContentObject;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.XContentParser;
+
+import java.io.IOException;
+import java.util.Objects;
+
+import static org.elasticsearch.common.xcontent.ConstructingObjectParser.optionalConstructorArg;
+
+public class SettingsConfig implements ToXContentObject {
+
+    private static final ParseField MAX_PAGE_SEARCH_SIZE = new ParseField("max_page_search_size");
+    private static final ParseField DOCS_PER_SECOND = new ParseField("docs_per_second");
+    private static final int DEFAULT_MAX_PAGE_SEARCH_SIZE = -1;
+    private static final float DEFAULT_DOCS_PER_SECOND = -1F;
+
+    private final Integer maxPageSearchSize;
+    private final Float docsPerSecond;
+
+    private static final ConstructingObjectParser<SettingsConfig, Void> PARSER = new ConstructingObjectParser<>(
+        "settings_config",
+        true,
+        args -> new SettingsConfig((Integer) args[0], (Float) args[1])
+    );
+
+    static {
+        PARSER.declareIntOrNull(optionalConstructorArg(), DEFAULT_MAX_PAGE_SEARCH_SIZE, MAX_PAGE_SEARCH_SIZE);
+        PARSER.declareFloatOrNull(optionalConstructorArg(), DEFAULT_DOCS_PER_SECOND, DOCS_PER_SECOND);
+    }
+
+    public static SettingsConfig fromXContent(final XContentParser parser) {
+        return PARSER.apply(parser, null);
+    }
+
+    SettingsConfig(Integer maxPageSearchSize, Float docsPerSecond) {
+        this.maxPageSearchSize = maxPageSearchSize;
+        this.docsPerSecond = docsPerSecond;
+    }
+
+    @Override
+    public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+        builder.startObject();
+        if (maxPageSearchSize != null) {
+            if (maxPageSearchSize.equals(DEFAULT_MAX_PAGE_SEARCH_SIZE)) {
+                builder.field(MAX_PAGE_SEARCH_SIZE.getPreferredName(), (Integer) null);
+            } else {
+                builder.field(MAX_PAGE_SEARCH_SIZE.getPreferredName(), maxPageSearchSize);
+            }
+        }
+        if (docsPerSecond != null) {
+            if (docsPerSecond.equals(DEFAULT_DOCS_PER_SECOND)) {
+                builder.field(DOCS_PER_SECOND.getPreferredName(), (Float) null);
+            } else {
+                builder.field(DOCS_PER_SECOND.getPreferredName(), docsPerSecond);
+            }
+        }
+        builder.endObject();
+        return builder;
+    }
+
+    public Integer getMaxPageSearchSize() {
+        return maxPageSearchSize;
+    }
+
+    public Float getDocsPerSecond() {
+        return docsPerSecond;
+    }
+
+    @Override
+    public boolean equals(Object other) {
+        if (other == this) {
+            return true;
+        }
+        if (other == null || other.getClass() != getClass()) {
+            return false;
+        }
+
+        SettingsConfig that = (SettingsConfig) other;
+        return Objects.equals(maxPageSearchSize, that.maxPageSearchSize) && Objects.equals(docsPerSecond, that.docsPerSecond);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(maxPageSearchSize, docsPerSecond);
+    }
+
+    public static Builder builder() {
+        return new Builder();
+    }
+
+    public static class Builder {
+        private Integer maxPageSearchSize;
+        private Float docsPerSecond;
+
+        /**
+         * Sets the paging maximum paging maxPageSearchSize that transform can use when
+         * pulling the data from the source index.
+         *
+         * If OOM is triggered, the paging maxPageSearchSize is dynamically reduced so that the transform can continue to gather data.
+         *
+         * @param maxPageSearchSize Integer value between 10 and 10_000
+         * @return the {@link Builder} with the paging maxPageSearchSize set.
+         */
+        public Builder setMaxPageSearchSize(Integer maxPageSearchSize) {
+            this.maxPageSearchSize = maxPageSearchSize == null ? DEFAULT_MAX_PAGE_SEARCH_SIZE : maxPageSearchSize;
+            return this;
+        }
+
+        /**
+         * Sets the docs per second that transform can use when pulling the data from the source index.
+         *
+         * This setting throttles transform by issuing queries less often, however processing still happens in
+         * batches. A value of 0 disables throttling (default).
+         *
+         * @param docsPerSecond Integer value
+         * @return the {@link Builder} with requestsPerSecond set.
+         */
+        public Builder setRequestsPerSecond(Float docsPerSecond) {
+            this.docsPerSecond = docsPerSecond == null ? DEFAULT_DOCS_PER_SECOND : docsPerSecond;
+            return this;
+        }
+
+        public SettingsConfig build() {
+            return new SettingsConfig(maxPageSearchSize, docsPerSecond);
+        }
+    }
+}

client/rest-high-level/src/main/java/org/elasticsearch/client/transform/transforms/TransformConfig.java

@@ -48,6 +48,7 @@ public class TransformConfig implements ToXContentObject {
     public static final ParseField FREQUENCY = new ParseField("frequency");
     public static final ParseField DESCRIPTION = new ParseField("description");
     public static final ParseField SYNC = new ParseField("sync");
+    public static final ParseField SETTINGS = new ParseField("settings");
     public static final ParseField VERSION = new ParseField("version");
     public static final ParseField CREATE_TIME = new ParseField("create_time");
     // types of transforms
@@ -58,45 +59,61 @@ public class TransformConfig implements ToXContentObject {
     private final DestConfig dest;
     private final TimeValue frequency;
     private final SyncConfig syncConfig;
+    private final SettingsConfig settings;
     private final PivotConfig pivotConfig;
     private final String description;
     private final Version transformVersion;
     private final Instant createTime;
 
-    public static final ConstructingObjectParser<TransformConfig, Void> PARSER =
-            new ConstructingObjectParser<>("transform", true,
-                (args) -> {
-                    String id = (String) args[0];
-                    SourceConfig source = (SourceConfig) args[1];
-                    DestConfig dest = (DestConfig) args[2];
-                    TimeValue frequency = (TimeValue) args[3];
-                    SyncConfig syncConfig = (SyncConfig) args[4];
-                    PivotConfig pivotConfig = (PivotConfig) args[5];
-                    String description = (String)args[6];
-                    Instant createTime = (Instant)args[7];
-                    String transformVersion = (String)args[8];
-                    return new TransformConfig(id,
-                        source,
-                        dest,
-                        frequency,
-                        syncConfig,
-                        pivotConfig,
-                        description,
-                        createTime,
-                        transformVersion);
-                });
+    public static final ConstructingObjectParser<TransformConfig, Void> PARSER = new ConstructingObjectParser<>(
+        "transform",
+        true,
+        (args) -> {
+            String id = (String) args[0];
+            SourceConfig source = (SourceConfig) args[1];
+            DestConfig dest = (DestConfig) args[2];
+            TimeValue frequency = (TimeValue) args[3];
+            SyncConfig syncConfig = (SyncConfig) args[4];
+            PivotConfig pivotConfig = (PivotConfig) args[5];
+            String description = (String) args[6];
+            SettingsConfig settings = (SettingsConfig) args[7];
+            Instant createTime = (Instant) args[8];
+            String transformVersion = (String) args[9];
+            return new TransformConfig(
+                id,
+                source,
+                dest,
+                frequency,
+                syncConfig,
+                pivotConfig,
+                description,
+                settings,
+                createTime,
+                transformVersion
+            );
+        }
+    );
 
     static {
         PARSER.declareString(constructorArg(), ID);
         PARSER.declareObject(constructorArg(), (p, c) -> SourceConfig.PARSER.apply(p, null), SOURCE);
         PARSER.declareObject(constructorArg(), (p, c) -> DestConfig.PARSER.apply(p, null), DEST);
-        PARSER.declareField(optionalConstructorArg(), p -> TimeValue.parseTimeValue(p.text(), FREQUENCY.getPreferredName()),
-            FREQUENCY, ObjectParser.ValueType.STRING);
+        PARSER.declareField(
+            optionalConstructorArg(),
+            p -> TimeValue.parseTimeValue(p.text(), FREQUENCY.getPreferredName()),
+            FREQUENCY,
+            ObjectParser.ValueType.STRING
+        );
         PARSER.declareObject(optionalConstructorArg(), (p, c) -> parseSyncConfig(p), SYNC);
         PARSER.declareObject(optionalConstructorArg(), (p, c) -> PivotConfig.fromXContent(p), PIVOT_TRANSFORM);
         PARSER.declareString(optionalConstructorArg(), DESCRIPTION);
-        PARSER.declareField(optionalConstructorArg(),
-            p -> TimeUtil.parseTimeFieldToInstant(p, CREATE_TIME.getPreferredName()), CREATE_TIME, ObjectParser.ValueType.VALUE);
+        PARSER.declareObject(optionalConstructorArg(), (p, c) -> SettingsConfig.fromXContent(p), SETTINGS);
+        PARSER.declareField(
+            optionalConstructorArg(),
+            p -> TimeUtil.parseTimeFieldToInstant(p, CREATE_TIME.getPreferredName()),
+            CREATE_TIME,
+            ObjectParser.ValueType.VALUE
+        );
         PARSER.declareString(optionalConstructorArg(), VERSION);
     }
 
@@ -108,7 +125,6 @@ private static SyncConfig parseSyncConfig(XContentParser parser) throws IOExcept
         return syncConfig;
     }
 
-
     public static TransformConfig fromXContent(final XContentParser parser) {
         return PARSER.apply(parser, null);
     }
@@ -125,25 +141,29 @@ public static TransformConfig fromXContent(final XContentParser parser) {
      * @return A TransformConfig to preview, NOTE it will have a {@code null} id, destination and index.
      */
     public static TransformConfig forPreview(final SourceConfig source, final PivotConfig pivotConfig) {
-        return new TransformConfig(null, source, null, null, null, pivotConfig, null, null, null);
+        return new TransformConfig(null, source, null, null, null, pivotConfig, null, null, null, null);
     }
 
-    TransformConfig(final String id,
-                    final SourceConfig source,
-                    final DestConfig dest,
-                    final TimeValue frequency,
-                    final SyncConfig syncConfig,
-                    final PivotConfig pivotConfig,
-                    final String description,
-                    final Instant createTime,
-                    final String version) {
+    TransformConfig(
+        final String id,
+        final SourceConfig source,
+        final DestConfig dest,
+        final TimeValue frequency,
+        final SyncConfig syncConfig,
+        final PivotConfig pivotConfig,
+        final String description,
+        final SettingsConfig settings,
+        final Instant createTime,
+        final String version
+    ) {
         this.id = id;
         this.source = source;
         this.dest = dest;
         this.frequency = frequency;
         this.syncConfig = syncConfig;
         this.pivotConfig = pivotConfig;
         this.description = description;
+        this.settings = settings;
         this.createTime = createTime == null ? null : Instant.ofEpochMilli(createTime.toEpochMilli());
         this.transformVersion = version == null ? null : Version.fromString(version);
     }
@@ -185,6 +205,11 @@ public String getDescription() {
         return description;
     }
 
+    @Nullable
+    public SettingsConfig getSettings() {
+        return settings;
+    }
+
     @Override
     public XContentBuilder toXContent(final XContentBuilder builder, final Params params) throws IOException {
         builder.startObject();
@@ -211,6 +236,9 @@ public XContentBuilder toXContent(final XContentBuilder builder, final Params pa
         if (description != null) {
             builder.field(DESCRIPTION.getPreferredName(), description);
         }
+        if (settings != null) {
+            builder.field(SETTINGS.getPreferredName(), settings);
+        }
         if (createTime != null) {
             builder.timeField(CREATE_TIME.getPreferredName(), CREATE_TIME.getPreferredName() + "_string", createTime.toEpochMilli());
         }
@@ -240,13 +268,14 @@ public boolean equals(Object other) {
             && Objects.equals(this.description, that.description)
             && Objects.equals(this.syncConfig, that.syncConfig)
             && Objects.equals(this.transformVersion, that.transformVersion)
+            && Objects.equals(this.settings, that.settings)
             && Objects.equals(this.createTime, that.createTime)
             && Objects.equals(this.pivotConfig, that.pivotConfig);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(id, source, dest, frequency, syncConfig, pivotConfig, description);
+        return Objects.hash(id, source, dest, frequency, syncConfig, settings, createTime, transformVersion, pivotConfig, description);
     }
 
     @Override
@@ -266,6 +295,7 @@ public static class Builder {
         private TimeValue frequency;
         private SyncConfig syncConfig;
         private PivotConfig pivotConfig;
+        private SettingsConfig settings;
         private String description;
 
         public Builder setId(String id) {
@@ -303,8 +333,13 @@ public Builder setDescription(String description) {
             return this;
         }
 
+        public Builder setSettings(SettingsConfig settings) {
+            this.settings = settings;
+            return this;
+        }
+
         public TransformConfig build() {
-            return new TransformConfig(id, source, dest, frequency, syncConfig, pivotConfig, description, null, null);
+            return new TransformConfig(id, source, dest, frequency, syncConfig, pivotConfig, description, settings, null, null);
         }
     }
 }