[ML] Add API integration tests for start and stop datafeeds (#110961)

* [ML] Add API integration tests for start and stop datafeeds

* [ML] Edits to setup and clean-up steps following review

x-pack/test/api_integration/apis/ml/jobs/force_start_datafeeds.ts

@@ -0,0 +1,249 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import expect from '@kbn/expect';
+import { FtrProviderContext } from '../../../ftr_provider_context';
+import { USER } from '../../../../functional/services/ml/security_common';
+import { JOB_STATE, DATAFEED_STATE } from '../../../../../plugins/ml/common/constants/states';
+import { MULTI_METRIC_JOB_CONFIG, SINGLE_METRIC_JOB_CONFIG, DATAFEED_CONFIG } from './common_jobs';
+import { COMMON_REQUEST_HEADERS } from '../../../../functional/services/ml/common_api';
+
+export default ({ getService }: FtrProviderContext) => {
+  const esArchiver = getService('esArchiver');
+  const supertest = getService('supertestWithoutAuth');
+  const ml = getService('ml');
+
+  const testSetupJobConfigs = [SINGLE_METRIC_JOB_CONFIG, MULTI_METRIC_JOB_CONFIG];
+
+  async function runStartDatafeedsRequest(
+    user: USER,
+    requestBody: object,
+    expectedResponsecode: number
+  ): Promise<Record<string, { started: boolean; error?: string }>> {
+    const { body } = await supertest
+      .post('/api/ml/jobs/force_start_datafeeds')
+      .auth(user, ml.securityCommon.getPasswordForUser(user))
+      .set(COMMON_REQUEST_HEADERS)
+      .send(requestBody)
+      .expect(expectedResponsecode);
+
+    return body;
+  }
+
+  const testDataList = [
+    {
+      testTitle: 'as ML Poweruser',
+      jobIds: [SINGLE_METRIC_JOB_CONFIG.job_id, MULTI_METRIC_JOB_CONFIG.job_id],
+      user: USER.ML_POWERUSER,
+      requestBody: {
+        datafeedIds: [
+          `datafeed-${SINGLE_METRIC_JOB_CONFIG.job_id}`,
+          `datafeed-${MULTI_METRIC_JOB_CONFIG.job_id}`,
+        ],
+        start: 1454803200000, // Starts in real-time from Feb 7 2016 00:00
+      },
+      expected: {
+        responseCode: 200,
+        responseBody: {
+          [`datafeed-${SINGLE_METRIC_JOB_CONFIG.job_id}`]: { started: true },
+          [`datafeed-${MULTI_METRIC_JOB_CONFIG.job_id}`]: { started: true },
+        },
+      },
+    },
+  ];
+
+  const invalidTestDataList = [
+    {
+      testTitle: 'as ML Poweruser with datafeed ID that does not exist',
+      jobIds: [SINGLE_METRIC_JOB_CONFIG.job_id],
+      user: USER.ML_POWERUSER,
+      requestBody: {
+        datafeedIds: [`invalid-datafeed-${SINGLE_METRIC_JOB_CONFIG.job_id}`],
+        start: 1454803200000, // Feb 7 2016 00:00
+      },
+      expected: {
+        responseCode: 200,
+        responseBody: {
+          [`invalid-datafeed-${SINGLE_METRIC_JOB_CONFIG.job_id}`]: { started: false },
+        },
+      },
+    },
+  ];
+
+  const testDataListUnauthorized = [
+    {
+      testTitle: 'as ML Unauthorized user',
+      user: USER.ML_UNAUTHORIZED,
+      jobIds: [SINGLE_METRIC_JOB_CONFIG.job_id, MULTI_METRIC_JOB_CONFIG.job_id],
+      requestBody: {
+        datafeedIds: [
+          `datafeed-${SINGLE_METRIC_JOB_CONFIG.job_id}`,
+          `datafeed-${MULTI_METRIC_JOB_CONFIG.job_id}`,
+        ],
+        start: 1454803200000, // Feb 7 2016 00:00
+        end: 1455235200000, // Feb 12 2016 00:00
+      },
+      expected: {
+        responseCode: 403,
+        error: 'Forbidden',
+      },
+    },
+    {
+      testTitle: 'as ML Viewer',
+      user: USER.ML_VIEWER,
+      jobIds: [SINGLE_METRIC_JOB_CONFIG.job_id, MULTI_METRIC_JOB_CONFIG.job_id],
+      requestBody: {
+        datafeedIds: [
+          `datafeed-${SINGLE_METRIC_JOB_CONFIG.job_id}`,
+          `datafeed-${MULTI_METRIC_JOB_CONFIG.job_id}`,
+        ],
+        start: 1454803200000, // Feb 7 2016 00:00
+        end: 1455235200000, // Feb 12 2016 00:00
+      },
+      expected: {
+        responseCode: 403,
+        error: 'Forbidden',
+      },
+    },
+  ];
+
+  describe('force_start_datafeeds', function () {
+    before(async () => {
+      await esArchiver.loadIfNeeded('x-pack/test/functional/es_archives/ml/farequote');
+      await ml.testResources.setKibanaTimeZoneToUTC();
+
+      for (const job of testSetupJobConfigs) {
+        const datafeedId = `datafeed-${job.job_id}`;
+        await ml.api.createAnomalyDetectionJob(job);
+        await ml.api.createDatafeed({
+          ...DATAFEED_CONFIG,
+          datafeed_id: datafeedId,
+          job_id: job.job_id,
+        });
+      }
+    });
+
+    after(async () => {
+      for (const job of testSetupJobConfigs) {
+        await ml.api.deleteAnomalyDetectionJobES(job.job_id);
+      }
+      await ml.api.cleanMlIndices();
+    });
+
+    describe('rejects requests for unauthorized users', function () {
+      for (const testData of testDataListUnauthorized) {
+        describe('fails to force start supplied datafeed IDs', function () {
+          it(`${testData.testTitle}`, async () => {
+            const body = await runStartDatafeedsRequest(
+              testData.user,
+              testData.requestBody,
+              testData.expected.responseCode
+            );
+
+            expect(body).to.have.property('error').eql(testData.expected.error);
+
+            // check jobs are still closed
+            for (const id of testData.jobIds) {
+              await ml.api.waitForJobState(id, JOB_STATE.CLOSED);
+            }
+
+            // check datafeeds are still stopped
+            for (const id of testData.requestBody.datafeedIds) {
+              await ml.api.waitForDatafeedState(id, DATAFEED_STATE.STOPPED);
+            }
+          });
+        });
+      }
+    });
+
+    describe('starts datafeeds with supplied IDs', function () {
+      for (const testData of testDataList) {
+        it(`${testData.testTitle}`, async () => {
+          const body = await runStartDatafeedsRequest(
+            testData.user,
+            testData.requestBody,
+            testData.expected.responseCode
+          );
+
+          const expectedResponse = testData.expected.responseBody;
+          const expectedRspDatafeedIds = Object.keys(expectedResponse).sort((a, b) =>
+            a.localeCompare(b)
+          );
+          const actualRspDatafeedIds = Object.keys(body).sort((a, b) => a.localeCompare(b));
+
+          expect(actualRspDatafeedIds).to.have.length(expectedRspDatafeedIds.length);
+          expect(actualRspDatafeedIds).to.eql(expectedRspDatafeedIds);
+
+          // check jobs are open
+          for (const id of testData.jobIds) {
+            await ml.api.waitForJobState(id, JOB_STATE.OPENED);
+          }
+
+          // check datafeeds have started
+          for (const id of testData.requestBody.datafeedIds) {
+            await ml.api.waitForDatafeedState(id, DATAFEED_STATE.STARTED);
+          }
+        });
+      }
+    });
+
+    describe('succeeds with datafeed already started', function () {
+      for (const testData of testDataList) {
+        it(`${testData.testTitle}`, async () => {
+          const body = await runStartDatafeedsRequest(
+            testData.user,
+            testData.requestBody,
+            testData.expected.responseCode
+          );
+
+          const expectedResponse = testData.expected.responseBody;
+          const expectedRspDatafeedIds = Object.keys(expectedResponse).sort((a, b) =>
+            a.localeCompare(b)
+          );
+          const actualRspDatafeedIds = Object.keys(body).sort((a, b) => a.localeCompare(b));
+
+          expect(actualRspDatafeedIds).to.have.length(expectedRspDatafeedIds.length);
+          expect(actualRspDatafeedIds).to.eql(expectedRspDatafeedIds);
+
+          // check jobs are still open
+          for (const id of testData.jobIds) {
+            await ml.api.waitForJobState(id, JOB_STATE.OPENED);
+          }
+
+          // check datafeeds are still started
+          for (const id of testData.requestBody.datafeedIds) {
+            await ml.api.waitForDatafeedState(id, DATAFEED_STATE.STARTED);
+          }
+        });
+      }
+    });
+
+    describe('returns expected response for invalid request', function () {
+      for (const testData of invalidTestDataList) {
+        it(`${testData.testTitle}`, async () => {
+          const body = await runStartDatafeedsRequest(
+            testData.user,
+            testData.requestBody,
+            testData.expected.responseCode
+          );
+          const expectedResponse = testData.expected.responseBody;
+          const expectedRspDatafeedIds = Object.keys(expectedResponse).sort((a, b) =>
+            a.localeCompare(b)
+          );
+          const actualRspDatafeedIds = Object.keys(body).sort((a, b) => a.localeCompare(b));
+
+          expect(actualRspDatafeedIds).to.have.length(expectedRspDatafeedIds.length);
+          expect(actualRspDatafeedIds).to.eql(expectedRspDatafeedIds);
+
+          expectedRspDatafeedIds.forEach((id) => {
+            expect(body[id].started).to.eql(expectedResponse[id].started);
+          });
+        });
+      }
+    });
+  });
+};

x-pack/test/api_integration/apis/ml/jobs/force_start_datafeeds_spaces.ts

@@ -0,0 +1,126 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import expect from '@kbn/expect';
+
+import { FtrProviderContext } from '../../../ftr_provider_context';
+import { COMMON_REQUEST_HEADERS } from '../../../../functional/services/ml/common_api';
+import { USER } from '../../../../functional/services/ml/security_common';
+import { DATAFEED_STATE } from '../../../../../plugins/ml/common/constants/states';
+
+export default ({ getService }: FtrProviderContext) => {
+  const esArchiver = getService('esArchiver');
+  const ml = getService('ml');
+  const spacesService = getService('spaces');
+  const supertest = getService('supertestWithoutAuth');
+
+  const idSpace1 = 'space1';
+  const idSpace2 = 'space2';
+  const jobIdSpace1 = `fq_single_${idSpace1}`;
+  const jobIdSpace2 = `fq_single_${idSpace2}`;
+  const datafeedIdSpace1 = `datafeed-${jobIdSpace1}`;
+  const datafeedIdSpace2 = `datafeed-${jobIdSpace2}`;
+  const startMs = 1454803200000; // Feb 7 2016 00:00
+  const endMs = 1455235200000; // Feb 12 2016 00:00
+
+  async function runRequest(
+    space: string,
+    expectedStatusCode: number,
+    datafeedIds: string[],
+    start: number,
+    end: number
+  ): Promise<Record<string, { started: boolean; error?: string }>> {
+    const { body } = await supertest
+      .post(`/s/${space}/api/ml/jobs/force_start_datafeeds`)
+      .auth(
+        USER.ML_POWERUSER_ALL_SPACES,
+        ml.securityCommon.getPasswordForUser(USER.ML_POWERUSER_ALL_SPACES)
+      )
+      .set(COMMON_REQUEST_HEADERS)
+      .send({ datafeedIds, start, end })
+      .expect(expectedStatusCode);
+
+    return body;
+  }
+
+  describe('force_start_datafeeds with spaces', function () {
+    before(async () => {
+      await spacesService.create({ id: idSpace1, name: 'space_one', disabledFeatures: [] });
+      await spacesService.create({ id: idSpace2, name: 'space_two', disabledFeatures: [] });
+
+      await esArchiver.loadIfNeeded('x-pack/test/functional/es_archives/ml/farequote');
+      await ml.testResources.setKibanaTimeZoneToUTC();
+    });
+
+    beforeEach(async () => {
+      const jobConfigSpace1 = ml.commonConfig.getADFqSingleMetricJobConfig(jobIdSpace1);
+      const datafeedConfigSpace1 = ml.commonConfig.getADFqDatafeedConfig(jobIdSpace1);
+      await ml.api.createAnomalyDetectionJob(jobConfigSpace1, idSpace1);
+      await ml.api.createDatafeed(
+        {
+          ...datafeedConfigSpace1,
+          datafeed_id: datafeedIdSpace1,
+          job_id: jobIdSpace1,
+        },
+        idSpace1
+      );
+      const jobConfigSpace2 = ml.commonConfig.getADFqSingleMetricJobConfig(jobIdSpace2);
+      const datafeedConfigSpace2 = ml.commonConfig.getADFqDatafeedConfig(jobIdSpace2);
+      await ml.api.createAnomalyDetectionJob(jobConfigSpace2, idSpace2);
+      await ml.api.createDatafeed(
+        {
+          ...datafeedConfigSpace2,
+          datafeed_id: datafeedIdSpace2,
+          job_id: jobIdSpace2,
+        },
+        idSpace2
+      );
+    });
+
+    afterEach(async () => {
+      await ml.api.closeAnomalyDetectionJob(jobIdSpace1);
+      await ml.api.closeAnomalyDetectionJob(jobIdSpace2);
+      await ml.api.deleteAnomalyDetectionJobES(jobIdSpace1);
+      await ml.api.deleteAnomalyDetectionJobES(jobIdSpace2);
+      await ml.api.cleanMlIndices();
+      await ml.testResources.cleanMLSavedObjects();
+    });
+
+    after(async () => {
+      await spacesService.delete(idSpace1);
+      await spacesService.delete(idSpace2);
+    });
+
+    it('should start single datafeed from same space', async () => {
+      const body = await runRequest(idSpace1, 200, [datafeedIdSpace1], startMs, endMs);
+      expect(body).to.eql({ [datafeedIdSpace1]: { started: true } });
+      await ml.api.waitForDatafeedState(datafeedIdSpace1, DATAFEED_STATE.STARTED);
+    });
+
+    it('should not start single datafeed from different space', async () => {
+      const body = await runRequest(idSpace2, 200, [datafeedIdSpace1], startMs, endMs);
+      expect(body).to.eql({ [datafeedIdSpace1]: { error: 'Job has no datafeed', started: false } });
+      await ml.api.waitForDatafeedState(datafeedIdSpace1, DATAFEED_STATE.STOPPED);
+    });
+
+    it('should only start datafeed from same space when called with a list of datafeeds', async () => {
+      const body = await runRequest(
+        idSpace1,
+        200,
+        [datafeedIdSpace1, datafeedIdSpace2],
+        startMs,
+        endMs
+      );
+      expect(body).to.eql({
+        [datafeedIdSpace1]: { started: true },
+        [datafeedIdSpace2]: { error: 'Job has no datafeed', started: false },
+      });
+      await ml.api.waitForDatafeedState(datafeedIdSpace1, DATAFEED_STATE.STARTED);
+      await ml.api.waitForDatafeedState(datafeedIdSpace2, DATAFEED_STATE.STOPPED);
+    });
+  });
+};

x-pack/test/api_integration/apis/ml/jobs/index.ts

@@ -18,5 +18,9 @@ export default function ({ loadTestFile }: FtrProviderContext) {
     loadTestFile(require.resolve('./close_jobs_spaces'));
     loadTestFile(require.resolve('./delete_jobs_spaces'));
     loadTestFile(require.resolve('./datafeed_preview'));
+    loadTestFile(require.resolve('./force_start_datafeeds'));
+    loadTestFile(require.resolve('./force_start_datafeeds_spaces'));
+    loadTestFile(require.resolve('./stop_datafeeds'));
+    loadTestFile(require.resolve('./stop_datafeeds_spaces'));
   });
 }