Add delayed datacheck to the datafeed job runner

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

@@ -62,6 +62,8 @@ public class DatafeedConfig implements ToXContentObject {
     public static final ParseField AGGREGATIONS = new ParseField("aggregations");
     public static final ParseField SCRIPT_FIELDS = new ParseField("script_fields");
     public static final ParseField CHUNKING_CONFIG = new ParseField("chunking_config");
+    public static final ParseField DELAYED_DATA_CHECK_WINDOW = new ParseField("delayed_data_check_window");
+    public static final ParseField SHOULD_RUN_DELAYED_DATA_CHECK = new ParseField("should_run_delayed_data_check");
 
     public static final ConstructingObjectParser<Builder, Void> PARSER = new ConstructingObjectParser<>(
         "datafeed_config", true, a -> new Builder((String)a[0], (String)a[1]));
@@ -88,6 +90,9 @@ public class DatafeedConfig implements ToXContentObject {
         }, SCRIPT_FIELDS);
         PARSER.declareInt(Builder::setScrollSize, SCROLL_SIZE);
         PARSER.declareObject(Builder::setChunkingConfig, ChunkingConfig.PARSER, CHUNKING_CONFIG);
+        PARSER.declareString((builder, val) -> builder.setDelayedDataCheckWindow(
+            TimeValue.parseTimeValue(val, DELAYED_DATA_CHECK_WINDOW.getPreferredName())), DELAYED_DATA_CHECK_WINDOW);
+        PARSER.declareBoolean(Builder::setShouldRunDelayedDataCheck, SHOULD_RUN_DELAYED_DATA_CHECK);
     }
 
     private static BytesReference parseBytes(XContentParser parser) throws IOException {
@@ -108,9 +113,16 @@ private static BytesReference parseBytes(XContentParser parser) throws IOExcepti
     private final Integer scrollSize;
     private final ChunkingConfig chunkingConfig;
 
+    /**
+     * The window of time to check for missing data
+     */
+    private final TimeValue delayedDataCheckWindow;
+    private final boolean shouldRunDelayedDataCheck;
+
     private DatafeedConfig(String id, String jobId, TimeValue queryDelay, TimeValue frequency, List<String> indices, List<String> types,
                            BytesReference query, BytesReference aggregations, List<SearchSourceBuilder.ScriptField> scriptFields,
-                           Integer scrollSize, ChunkingConfig chunkingConfig) {
+                           Integer scrollSize, ChunkingConfig chunkingConfig, TimeValue delayedDataCheckWindow,
+                           boolean shouldRunDelayedDataCheck) {
         this.id = id;
         this.jobId = jobId;
         this.queryDelay = queryDelay;
@@ -122,6 +134,8 @@ private DatafeedConfig(String id, String jobId, TimeValue queryDelay, TimeValue
         this.scriptFields = scriptFields == null ? null : Collections.unmodifiableList(scriptFields);
         this.scrollSize = scrollSize;
         this.chunkingConfig = chunkingConfig;
+        this.delayedDataCheckWindow = delayedDataCheckWindow;
+        this.shouldRunDelayedDataCheck = shouldRunDelayedDataCheck;
     }
 
     public String getId() {
@@ -168,6 +182,21 @@ public ChunkingConfig getChunkingConfig() {
         return chunkingConfig;
     }
 
+    /**
+     * The window of time in which to check for latent data
+     * @return The delayed data check window
+     */
+    public TimeValue getDelayedDataCheckWindow() {
+        return delayedDataCheckWindow;
+    }
+
+    /**
+     * Should we check for delayed data
+     */
+    public boolean getShouldRunDelayedDataCheck() {
+        return shouldRunDelayedDataCheck;
+    }
+
     @Override
     public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
         builder.startObject();
@@ -204,6 +233,10 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
         if (chunkingConfig != null) {
             builder.field(CHUNKING_CONFIG.getPreferredName(), chunkingConfig);
         }
+        if (delayedDataCheckWindow != null) {
+            builder.field(DELAYED_DATA_CHECK_WINDOW.getPreferredName(), delayedDataCheckWindow.getStringRep());
+        }
+        builder.field(SHOULD_RUN_DELAYED_DATA_CHECK.getPreferredName(), shouldRunDelayedDataCheck);
 
         builder.endObject();
         return builder;
@@ -244,7 +277,9 @@ public boolean equals(Object other) {
             && Objects.equals(this.scrollSize, that.scrollSize)
             && Objects.equals(asMap(this.aggregations), asMap(that.aggregations))
             && Objects.equals(this.scriptFields, that.scriptFields)
-            && Objects.equals(this.chunkingConfig, that.chunkingConfig);
+            && Objects.equals(this.chunkingConfig, that.chunkingConfig)
+            && Objects.equals(this.delayedDataCheckWindow, that.delayedDataCheckWindow)
+            && Objects.equals(this.shouldRunDelayedDataCheck, that.shouldRunDelayedDataCheck);
     }
 
     /**
@@ -255,7 +290,7 @@ public boolean equals(Object other) {
     @Override
     public int hashCode() {
         return Objects.hash(id, jobId, frequency, queryDelay, indices, types, asMap(query), scrollSize, asMap(aggregations), scriptFields,
-            chunkingConfig);
+            chunkingConfig, delayedDataCheckWindow, shouldRunDelayedDataCheck);
     }
 
     public static Builder builder(String id, String jobId) {
@@ -275,6 +310,8 @@ public static class Builder {
         private List<SearchSourceBuilder.ScriptField> scriptFields;
         private Integer scrollSize;
         private ChunkingConfig chunkingConfig;
+        private TimeValue delayedDataCheckWindow;
+        private boolean shouldRunDelayedDataCheck;
 
         public Builder(String id, String jobId) {
             this.id = Objects.requireNonNull(id, ID.getPreferredName());
@@ -293,6 +330,8 @@ public Builder(DatafeedConfig config) {
             this.scriptFields = config.scriptFields;
             this.scrollSize = config.scrollSize;
             this.chunkingConfig = config.chunkingConfig;
+            this.delayedDataCheckWindow = config.getDelayedDataCheckWindow();
+            this.shouldRunDelayedDataCheck = config.getShouldRunDelayedDataCheck();
         }
 
         public Builder setIndices(List<String> indices) {
@@ -366,9 +405,38 @@ public Builder setChunkingConfig(ChunkingConfig chunkingConfig) {
             return this;
         }
 
+        /**
+         * This determines how far in the past we look for data being indexed too late for the datafeed to pick it up.
+         *
+         * We query the index to the latest finalized bucket from this TimeValue in the past looking to see if any data has been indexed
+         * since the data was read with the Datafeed.
+         *
+         * The window must be larger than the {@link org.elasticsearch.client.ml.job.config.AnalysisConfig#bucketSpan}, less than
+         * 24 hours, and span less than 10,000x buckets.
+         *
+         * @param delayedDataCheckWindow The time length in the past from the latest finalized bucket to look for latent data
+         *                               Defaults to 2 hours.
+         */
+        public Builder setDelayedDataCheckWindow(TimeValue delayedDataCheckWindow) {
+            this.delayedDataCheckWindow = delayedDataCheckWindow;
+            return this;
+        }
+
+        /**
+         * When running the datafeed in real-time, should there be additional checks for data being indexed after the datafeed
+         * reads from the index.
+         *
+         * @param shouldRunDelayedDataCheck when {@code false} no checks are made for latent data in the real-time datafeed
+         *                                  Defaults to {@code true}
+         */
+        public Builder setShouldRunDelayedDataCheck(boolean shouldRunDelayedDataCheck) {
+            this.shouldRunDelayedDataCheck = shouldRunDelayedDataCheck;
+            return this;
+        }
+
         public DatafeedConfig build() {
             return new DatafeedConfig(id, jobId, queryDelay, frequency, indices, types, query, aggregations, scriptFields, scrollSize,
-                chunkingConfig);
+                chunkingConfig, delayedDataCheckWindow, shouldRunDelayedDataCheck);
         }
 
         private static BytesReference xContentToBytes(ToXContentObject object) throws IOException {

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

@@ -77,6 +77,10 @@ public class DatafeedUpdate implements ToXContentObject {
         }, DatafeedConfig.SCRIPT_FIELDS);
         PARSER.declareInt(Builder::setScrollSize, DatafeedConfig.SCROLL_SIZE);
         PARSER.declareObject(Builder::setChunkingConfig, ChunkingConfig.PARSER, DatafeedConfig.CHUNKING_CONFIG);
+        PARSER.declareString((builder, val) -> builder.setDelayedDataCheckWindow(
+            TimeValue.parseTimeValue(val, DatafeedConfig.DELAYED_DATA_CHECK_WINDOW.getPreferredName())),
+            DatafeedConfig.DELAYED_DATA_CHECK_WINDOW);
+        PARSER.declareBoolean(Builder::setShouldRunDelayedDataCheck, DatafeedConfig.SHOULD_RUN_DELAYED_DATA_CHECK);
     }
 
     private static BytesReference parseBytes(XContentParser parser) throws IOException {
@@ -96,10 +100,13 @@ private static BytesReference parseBytes(XContentParser parser) throws IOExcepti
     private final List<SearchSourceBuilder.ScriptField> scriptFields;
     private final Integer scrollSize;
     private final ChunkingConfig chunkingConfig;
+    private final TimeValue delayedDataCheckWindow;
+    private final Boolean shouldRunDelayedDataCheck;
 
     private DatafeedUpdate(String id, String jobId, TimeValue queryDelay, TimeValue frequency, List<String> indices, List<String> types,
                            BytesReference query, BytesReference aggregations, List<SearchSourceBuilder.ScriptField> scriptFields,
-                           Integer scrollSize, ChunkingConfig chunkingConfig) {
+                           Integer scrollSize, ChunkingConfig chunkingConfig, TimeValue delayedDataCheckWindow,
+                           Boolean shouldRunDelayedDataCheck) {
         this.id = id;
         this.jobId = jobId;
         this.queryDelay = queryDelay;
@@ -111,6 +118,8 @@ private DatafeedUpdate(String id, String jobId, TimeValue queryDelay, TimeValue
         this.scriptFields = scriptFields;
         this.scrollSize = scrollSize;
         this.chunkingConfig = chunkingConfig;
+        this.delayedDataCheckWindow = delayedDataCheckWindow;
+        this.shouldRunDelayedDataCheck = shouldRunDelayedDataCheck;
     }
 
     /**
@@ -146,6 +155,10 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
             }
             builder.endObject();
         }
+        if (delayedDataCheckWindow != null) {
+            builder.field(DatafeedConfig.DELAYED_DATA_CHECK_WINDOW.getPreferredName(), delayedDataCheckWindow.getStringRep());
+        }
+        addOptionalField(builder, DatafeedConfig.SHOULD_RUN_DELAYED_DATA_CHECK, shouldRunDelayedDataCheck);
         addOptionalField(builder, DatafeedConfig.SCROLL_SIZE, scrollSize);
         addOptionalField(builder, DatafeedConfig.CHUNKING_CONFIG, chunkingConfig);
         builder.endObject();
@@ -198,6 +211,14 @@ public ChunkingConfig getChunkingConfig() {
         return chunkingConfig;
     }
 
+    public TimeValue getDelayedDataCheckWindow() {
+        return delayedDataCheckWindow;
+    }
+
+    public Boolean getShouldRunDelayedDataCheck() {
+        return shouldRunDelayedDataCheck;
+    }
+
     private static Map<String, Object> asMap(BytesReference bytesReference) {
         return bytesReference == null ? null : XContentHelper.convertToMap(bytesReference, true, XContentType.JSON).v2();
     }
@@ -232,6 +253,8 @@ public boolean equals(Object other) {
             && Objects.equals(asMap(this.query), asMap(that.query))
             && Objects.equals(this.scrollSize, that.scrollSize)
             && Objects.equals(asMap(this.aggregations), asMap(that.aggregations))
+            && Objects.equals(this.delayedDataCheckWindow, that.delayedDataCheckWindow)
+            && Objects.equals(this.shouldRunDelayedDataCheck, that.shouldRunDelayedDataCheck)
             && Objects.equals(this.scriptFields, that.scriptFields)
             && Objects.equals(this.chunkingConfig, that.chunkingConfig);
     }
@@ -244,7 +267,7 @@ public boolean equals(Object other) {
     @Override
     public int hashCode() {
         return Objects.hash(id, jobId, frequency, queryDelay, indices, types, asMap(query), scrollSize, asMap(aggregations), scriptFields,
-            chunkingConfig);
+            chunkingConfig, delayedDataCheckWindow, shouldRunDelayedDataCheck);
     }
 
     public static Builder builder(String id) {
@@ -264,6 +287,8 @@ public static class Builder {
         private List<SearchSourceBuilder.ScriptField> scriptFields;
         private Integer scrollSize;
         private ChunkingConfig chunkingConfig;
+        private TimeValue delayedDataCheckWindow;
+        private Boolean shouldRunDelayedDataCheck;
 
         public Builder(String id) {
             this.id = Objects.requireNonNull(id, DatafeedConfig.ID.getPreferredName());
@@ -281,6 +306,8 @@ public Builder(DatafeedUpdate config) {
             this.scriptFields = config.scriptFields;
             this.scrollSize = config.scrollSize;
             this.chunkingConfig = config.chunkingConfig;
+            this.delayedDataCheckWindow = config.delayedDataCheckWindow;
+            this.shouldRunDelayedDataCheck = config.shouldRunDelayedDataCheck;
         }
 
         public Builder setJobId(String jobId) {
@@ -359,9 +386,19 @@ public Builder setChunkingConfig(ChunkingConfig chunkingConfig) {
             return this;
         }
 
+        public Builder setDelayedDataCheckWindow(TimeValue delayedDataCheckWindow) {
+            this.delayedDataCheckWindow = delayedDataCheckWindow;
+            return this;
+        }
+
+        public Builder setShouldRunDelayedDataCheck(boolean shouldRunDelayedDataCheck) {
+            this.shouldRunDelayedDataCheck = shouldRunDelayedDataCheck;
+            return this;
+        }
+
         public DatafeedUpdate build() {
             return new DatafeedUpdate(id, jobId, queryDelay, frequency, indices, types, query, aggregations, scriptFields, scrollSize,
-                chunkingConfig);
+                chunkingConfig, delayedDataCheckWindow, shouldRunDelayedDataCheck);
         }
 
         private static BytesReference xContentToBytes(ToXContentObject object) throws IOException {

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

@@ -576,6 +576,17 @@ public void testPutDatafeed() throws Exception {
             datafeedBuilder.setQueryDelay(TimeValue.timeValueMinutes(1)); // <1>
             // end::put-datafeed-config-set-query-delay
 
+            // tag::put-datafeed-config-set-should-run-delayed-data-check
+            datafeedBuilder.setShouldRunDelayedDataCheck(true); // <1>
+            // end::put-datafeed-config-set-should-run-delayed-data-check
+
+            // no need to accidentally trip internal validations due to job bucket size
+            datafeedBuilder.setShouldRunDelayedDataCheck(false);
+
+            // tag::put-datafeed-config-set-delayed-data-check-window
+            datafeedBuilder.setDelayedDataCheckWindow(TimeValue.timeValueMinutes(90)); // <1>
+            // end::put-datafeed-config-set-delayed-data-check-window
+
             List<SearchSourceBuilder.ScriptField> scriptFields = Collections.emptyList();
             // tag::put-datafeed-config-set-script-fields
             datafeedBuilder.setScriptFields(scriptFields); // <1>

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

@@ -103,6 +103,11 @@ public static DatafeedConfig.Builder createRandomBuilder() {
         if (randomBoolean()) {
             builder.setChunkingConfig(ChunkingConfigTests.createRandomizedChunk());
         }
+        boolean shouldRunDelayedDataCheck = randomBoolean();
+        builder.setShouldRunDelayedDataCheck(shouldRunDelayedDataCheck);
+        if (shouldRunDelayedDataCheck || randomBoolean()) {
+            builder.setDelayedDataCheckWindow(new TimeValue(randomLongBetween(bucketSpanMillis,bucketSpanMillis*10)));
+        }
         return builder;
     }
 

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

@@ -83,6 +83,11 @@ public static DatafeedUpdate createRandom() {
         if (randomBoolean()) {
             builder.setChunkingConfig(ChunkingConfigTests.createRandomizedChunk());
         }
+        boolean shouldRunDelayedDataCheck = randomBoolean();
+        builder.setShouldRunDelayedDataCheck(shouldRunDelayedDataCheck);
+        if (shouldRunDelayedDataCheck || randomBoolean()) {
+            builder.setDelayedDataCheckWindow(new TimeValue(randomLongBetween(300_001, 600_000)));
+        }
         return builder.build();
     }
 

docs/java-rest/high-level/ml/put-datafeed.asciidoc

@@ -63,6 +63,21 @@ include-tagged::{doc-tests-file}[{api}-config-set-query-delay]
 --------------------------------------------------
 <1> The time interval behind real time that data is queried.
 
+["source","java",subs="attributes,callouts,macros"]
+--------------------------------------------------
+include-tagged::{doc-tests-file}[{api}-config-set-should-run-delayed-data-check]
+--------------------------------------------------
+<1> Should we look for data that is ingested too late when running the datafeed
+in real time. Defaults to `true`
+
+["source","java",subs="attributes,callouts,macros"]
+--------------------------------------------------
+include-tagged::{doc-tests-file}[{api}-config-set-delayed-data-check-window]
+--------------------------------------------------
+<1> How far in the past from the latest finalized bucket to check for delayed data being indexed.
+The window must be larger than the Job's bucket size, but smaller than 24 hours, and span less than
+10,000 buckets. Defaults to 2 hours.
+
 ["source","java",subs="attributes,callouts,macros"]
 --------------------------------------------------
 include-tagged::{doc-tests-file}[{api}-config-set-script-fields]

docs/reference/ml/apis/datafeedresource.asciidoc

@@ -64,6 +64,20 @@ A {dfeed} resource has the following properties:
   example: `[]`. This property is provided for backwards compatibility with
   releases earlier than 6.0.0. For more information, see <<removal-of-types>>.  
 
+`delayed_data_check_window`::
+  (time units) The time window to query that ends at the latest finalized bucket
+  when determining if any data was indexed too late for the {dfeed} to see it. For
+  example, supply `3h` to check for latent data in the {dfeed} `indices` over the
+  last `3h` since the latest finalized bucket. Only valid in real-time analysis
+  and when `should_run_delayed_data_check` is `true`. Defaults to 2 hours.
+  This value must be larger than the bucket_span, less than 24 hours, and span
+  fewer than 10,000 buckets.
+
+`should_run_delayed_data_check`::
+  (boolean) When running a real-time {dfeed}, should the `indices` be checked
+  for latent data. `delayed_data_check_window` must be set when this is `true`.
+  Defaults to `true`
+
 [[ml-datafeed-chunking-config]]
 ==== Chunking Configuration Objects
 

docs/reference/ml/apis/put-datafeed.asciidoc

@@ -78,6 +78,20 @@ You must create a job before you create a {dfeed}.  You can associate only one
   For example: `[]`. This property is provided for backwards compatibility with
   releases earlier than 6.0.0. For more information, see <<removal-of-types>>.
 
+`delayed_data_check_window`::
+  (time units) The time window to query that ends at the latest finalized bucket
+  when determining if any data was indexed too late for the {dfeed} to see it. For
+  example, supply `3h` to check for latent data in the {dfeed} `indices` over the
+  last `3h` since the latest finalized bucket. Only valid in real-time analysis
+  and when `should_run_delayed_data_check` is `true`. Defaults to 2 hours.
+  This value must be larger than the bucket_span, less than 24 hours, and span
+  fewer than 10,000 buckets.
+
+`should_run_delayed_data_check`::
+  (boolean) When running a real-time {dfeed}, should the `indices` be checked
+  for latent data. `delayed_data_check_window` must be set when this is `true`.
+  Defaults true `true`
+
 For more information about these properties,
 see <<ml-datafeed-resource>>.