[7.17][ML] defend against negative datafeed start times (#100332)

* [ML] Defend against negative datafeed start times (#100284)

A negative start time in the datafeed can cause significant disruption
to an entire cluster. This PR checks that the start time is greater
than or equal to 0 and throws an exception otherwise.

* Adjust backported test for 7.17

Author: edsavage

docs/changelog/100284.yaml

@@ -0,0 +1,5 @@
+pr: 100284
+summary: Defend against negative datafeed start times
+area: Machine Learning
+type: bug
+issues: []

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

@@ -193,6 +193,9 @@ public static DatafeedParams parseRequest(String datafeedId, XContentParser pars
 
         public DatafeedParams(String datafeedId, long startTime) {
             this.datafeedId = ExceptionsHelper.requireNonNull(datafeedId, DatafeedConfig.ID.getPreferredName());
+            if (startTime < 0) {
+                throw new IllegalArgumentException("[" + START_TIME.getPreferredName() + "] must not be negative [" + startTime + "].");
+            }
             this.startTime = startTime;
         }
 

x-pack/plugin/ml/qa/native-multi-node-tests/src/javaRestTest/java/org/elasticsearch/xpack/ml/integration/DatafeedJobsIT.java

@@ -32,6 +32,7 @@
 import org.elasticsearch.xpack.core.ml.action.GetJobsStatsAction;
 import org.elasticsearch.xpack.core.ml.action.KillProcessAction;
 import org.elasticsearch.xpack.core.ml.action.PutJobAction;
+import org.elasticsearch.xpack.core.ml.action.StartDatafeedAction;
 import org.elasticsearch.xpack.core.ml.action.StopDatafeedAction;
 import org.elasticsearch.xpack.core.ml.datafeed.ChunkingConfig;
 import org.elasticsearch.xpack.core.ml.datafeed.DatafeedConfig;
@@ -832,4 +833,36 @@ private void startRealtime(String jobId, Integer maxEmptySearches) throws Except
             assertThat(dataCounts.getOutOfOrderTimeStampCount(), equalTo(0L));
         }, 30, TimeUnit.SECONDS);
     }
+
+    public void testStartDatafeed_GivenNegativeStartTime_Returns408() throws Exception {
+        client().admin().indices().prepareCreate("data-1").addMapping("type", "time", "type=date").get();
+        long numDocs = 100;
+        long now = System.currentTimeMillis();
+        long oneWeekAgo = now - 604800000;
+        long negativeStartTime = -1000;
+        indexDocs(logger, "data-1", numDocs, oneWeekAgo, now);
+
+        String jobId = "job-for-start-datafeed-timeout";
+        String datafeedId = jobId + "-datafeed";
+
+        Job.Builder job = createScheduledJob(jobId);
+        putJob(job);
+        openJob(job.getId());
+        assertBusy(() -> assertEquals(getJobStats(job.getId()).get(0).getState(), JobState.OPENED));
+
+        DatafeedConfig.Builder datafeedConfigBuilder = createDatafeedBuilder(
+            job.getId() + "-datafeed",
+            job.getId(),
+            Collections.singletonList("data-1")
+        );
+        DatafeedConfig datafeedConfig = datafeedConfigBuilder.build();
+        putDatafeed(datafeedConfig);
+
+        IllegalArgumentException e = expectThrows(
+            IllegalArgumentException.class,
+            () -> new StartDatafeedAction.Request(datafeedId, negativeStartTime)
+        );
+
+        assertThat(e.getMessage(), equalTo("[start] must not be negative [-1000]."));
+    }
 }