[ML] Functional tests - stablize AD job creation checks (#66733)

With this PR, anomaly detection jobs are checked to have a processed_record_count > 0 before waiting for a stopped datafeed.
elastic · May 18, 2020 · 7364f7f · 7364f7f
1 parent 1c72279
commit 7364f7f
Show file tree

Hide file tree

Showing 3 changed files with 41 additions and 5 deletions.
diff --git a/x-pack/test/api_integration/apis/ml/modules/setup_module.ts b/x-pack/test/api_integration/apis/ml/modules/setup_module.ts
@@ -236,6 +236,9 @@ export default ({ getService }: FtrProviderContext) => {
             const datafeedId = `datafeed-${job.jobId}`;
             await ml.api.waitForAnomalyDetectionJobToExist(job.jobId);
             await ml.api.waitForDatafeedToExist(datafeedId);
+            if (testData.requestBody.startDatafeed === true) {
+              await ml.api.waitForADJobRecordCountToBePositive(job.jobId);
+            }
             await ml.api.waitForJobState(job.jobId, job.jobState);
             await ml.api.waitForDatafeedState(datafeedId, job.datafeedState);
           }

diff --git a/x-pack/test/functional/services/machine_learning/api.ts b/x-pack/test/functional/services/machine_learning/api.ts
@@ -138,18 +138,24 @@ export function MachineLearningAPIProvider({ getService }: FtrProviderContext) {
     },
 
     async getJobState(jobId: string): Promise<JOB_STATE> {
-      log.debug(`Fetching job state for job ${jobId}`);
-      const jobStats = await esSupertest
-        .get(`/_ml/anomaly_detectors/${jobId}/_stats`)
-        .expect(200)
-        .then((res: any) => res.body);
+      const jobStats = await this.getADJobStats(jobId);
 
       expect(jobStats.jobs).to.have.length(1);
       const state: JOB_STATE = jobStats.jobs[0].state;
 
       return state;
     },
 
+    async getADJobStats(jobId: string): Promise<any> {
+      log.debug(`Fetching anomaly detection job stats for job ${jobId}...`);
+      const jobStats = await esSupertest
+        .get(`/_ml/anomaly_detectors/${jobId}/_stats`)
+        .expect(200)
+        .then((res: any) => res.body);
+
+      return jobStats;
+    },
+
     async waitForJobState(jobId: string, expectedJobState: JOB_STATE) {
       await retry.waitForWithTimeout(
         `job state to be ${expectedJobState}`,
@@ -390,5 +396,31 @@ export function MachineLearningAPIProvider({ getService }: FtrProviderContext) {
 
       await this.waitForDataFrameAnalyticsJobToExist(analyticsId);
     },
+
+    async getADJobRecordCount(jobId: string): Promise<number> {
+      const jobStats = await this.getADJobStats(jobId);
+
+      expect(jobStats.jobs).to.have.length(1);
+      const processedRecordCount: number = jobStats.jobs[0].data_counts.processed_record_count;
+
+      return processedRecordCount;
+    },
+
+    async waitForADJobRecordCountToBePositive(jobId: string) {
+      await retry.waitForWithTimeout(
+        `'${jobId}' to have processed_record_count > 0`,
+        10 * 1000,
+        async () => {
+          const processedRecordCount = await this.getADJobRecordCount(jobId);
+          if (processedRecordCount > 0) {
+            return true;
+          } else {
+            throw new Error(
+              `expected anomaly detection job '${jobId}' to have processed_record_count > 0 (got ${processedRecordCount})`
+            );
+          }
+        }
+      );
+    },
   };
 }
diff --git a/x-pack/test/functional/services/machine_learning/job_management.ts b/x-pack/test/functional/services/machine_learning/job_management.ts
@@ -43,6 +43,7 @@ export function MachineLearningJobManagementProvider(
     },
 
     async waitForJobCompletion(jobId: string) {
+      await mlApi.waitForADJobRecordCountToBePositive(jobId);
       await mlApi.waitForDatafeedState(`datafeed-${jobId}`, DATAFEED_STATE.STOPPED);
       await mlApi.waitForJobState(jobId, JOB_STATE.CLOSED);
     },