[Rollup] Re-enable api integration tests (#33381)

x-pack/plugins/rollup/server/client/elasticsearch_rollup.js

@@ -91,6 +91,12 @@ export const elasticsearchJsPlugin = (Client, config, components) => {
   });
 
   rollup.stopJob = ca({
+    params: {
+      waitForCompletion: {
+        type: 'boolean',
+        name: 'wait_for_completion'
+      }
+    },
     urls: [
       {
         fmt: '/_rollup/job/<%=id%>/_stop',

x-pack/plugins/rollup/server/routes/api/jobs.js

@@ -75,6 +75,7 @@ export function registerJobsRoute(server) {
     handler: async (request) => {
       try {
         const { jobIds } = request.payload;
+
         const callWithRequest = callWithRequestFactory(server, request);
         return await Promise.all(jobIds.map(id => callWithRequest('rollup.startJob', { id })))
           .then(() => ({ success: true }));
@@ -107,9 +108,17 @@ export function registerJobsRoute(server) {
     handler: async (request) => {
       try {
         const { jobIds } = request.payload;
+        // For our API integration tests we need to wait for the jobs to be stopped
+        // in order to be able to delete them sequencially.
+        const { waitForCompletion } = request.query;
         const callWithRequest = callWithRequestFactory(server, request);
-        return await Promise.all(jobIds.map(id => callWithRequest('rollup.stopJob', { id })))
+
+        const stopRollupJob = id => callWithRequest('rollup.stopJob', { id, waitForCompletion: waitForCompletion === 'true' });
+
+        return await Promise
+          .all(jobIds.map(stopRollupJob))
           .then(() => ({ success: true }));
+
       } catch(err) {
         if (isEsError(err)) {
           return wrapEsError(err);
@@ -129,6 +138,7 @@ export function registerJobsRoute(server) {
     handler: async (request) => {
       try {
         const { jobIds } = request.payload;
+
         const callWithRequest = callWithRequestFactory(server, request);
         return await Promise.all(jobIds.map(id => callWithRequest('rollup.deleteJob', { id })))
           .then(() => ({ success: true }));

x-pack/test/api_integration/apis/management/rollup/index.js

@@ -5,8 +5,7 @@
  */
 
 export default function ({ loadTestFile }) {
-  // FLAKY: https://github.com/elastic/kibana/issues/33217
-  describe.skip('rollup', () => {
+  describe('rollup', () => {
     loadTestFile(require.resolve('./rollup'));
     loadTestFile(require.resolve('./index_patterns_extensions'));
     loadTestFile(require.resolve('./rollup_search'));

x-pack/test/api_integration/apis/management/rollup/rollup.js

@@ -16,6 +16,7 @@ export default function ({ getService }) {
   const {
     createIndexWithMappings,
     getJobPayload,
+    loadJobs,
     createJob,
     deleteJob,
     startJob,
@@ -77,9 +78,7 @@ export default function ({ getService }) {
     describe('crud', () => {
       describe('list', () => {
         it('should return an empty array when there are no jobs', async () => {
-          const { body } = await supertest
-            .get(`${API_BASE_PATH}/jobs`)
-            .expect(200);
+          const { body } = await loadJobs().expect(200);
 
           expect(body).to.eql({ jobs: [] });
         });
@@ -117,7 +116,7 @@ export default function ({ getService }) {
           const payload = getJobPayload(indexName);
           await createJob(payload);
 
-          const { body: { jobs } } = await supertest.get(`${API_BASE_PATH}/jobs`);
+          const { body: { jobs } } = await loadJobs();
           const job = jobs.find(job => job.config.id === payload.job.id);
 
           expect(job).not.be(undefined);
@@ -193,7 +192,8 @@ export default function ({ getService }) {
           await createJob(payload);
         });
 
-        it('should delete a job that was created', async () => {
+        it('should delete a job that has been stopped', async () => {
+          await stopJob(jobId);
           const { body } = await deleteJob(jobId).expect(200);
           expect(body).to.eql({ success: true });
         });
@@ -215,7 +215,7 @@ export default function ({ getService }) {
           const payload = getJobPayload(indexName);
           await createJob(payload);
 
-          const { body: { jobs } } = await supertest.get(`${API_BASE_PATH}/jobs`);
+          const { body: { jobs } } = await loadJobs();
           job = jobs.find(job => job.config.id === payload.job.id);
         });
 
@@ -228,7 +228,7 @@ export default function ({ getService }) {
 
           // Fetch the job to make sure it has been started
           const jobId = job.config.id;
-          const { body: { jobs } } = await supertest.get(`${API_BASE_PATH}/jobs`);
+          const { body: { jobs } } = await loadJobs();
           job = jobs.find(job => job.config.id === jobId);
           expect(job.status.job_state).to.eql('started');
         });
@@ -255,7 +255,7 @@ export default function ({ getService }) {
           expect(body).to.eql({ success: true });
 
           // Fetch the job to make sure it has been stopped
-          const { body: { jobs } } = await supertest.get(`${API_BASE_PATH}/jobs`);
+          const { body: { jobs } } = await loadJobs();
           const job = jobs.find(job => job.config.id === jobId);
           expect(job.status.job_state).to.eql('stopped');
         });

x-pack/test/api_integration/apis/management/rollup/rollup.test_helpers.js

@@ -4,7 +4,7 @@
  * you may not use this file except in compliance with the Elastic License.
  */
 
-import { initElasticsearchIndicesHelpers, getRandomString, wait } from './lib';
+import { initElasticsearchIndicesHelpers, getRandomString } from './lib';
 import { API_BASE_PATH, ROLLUP_INDEX_NAME, INDEX_TO_ROLLUP_MAPPINGS } from './constants';
 
 const jobsCreated = [];
@@ -81,63 +81,21 @@ export const registerHelpers = ({ supertest, es }) => {
     const jobIds = Array.isArray(ids) ? ids : [ids];
 
     return supertest
-      .post(`${API_BASE_PATH}/stop`)
+      .post(`${API_BASE_PATH}/stop?waitForCompletion=true`)
       .set('kbn-xsrf', 'xxx')
       .send({ jobIds });
   };
 
   const loadJobs = () => supertest.get(`${API_BASE_PATH}/jobs`);
 
-  const waitForJobsToStop = (attempt = 0) => (
+  const stopAllJobs = () => (
     loadJobs()
       .then(async ({ body: { jobs } }) => {
-        const jobBeingStopped = jobs.filter(job => job.status.job_state !== 'stopped' && job.status.job_state !== 'started');
+        const jobIds = jobs.map(job => job.config.id);
 
-        if (!jobBeingStopped.length) {
-          return;
-        }
+        await stopJob(jobIds);
 
-        if (attempt < 3 && jobBeingStopped.length) {
-          await wait(500);
-          return waitForJobsToStop(++attempt);
-        }
-
-        throw new Error('Error while waiting for Rollup Jobs to stop');
-      }));
-
-  const stopAllJobStarted = (jobIds = jobsStarted, attempt = 0) => (
-    stopJob(jobIds)
-      .then(waitForJobsToStop)
-      .then(loadJobs)
-      .then(({ body: { jobs } }) => {
-        // We make sure that there are no more jobs started
-        // as trying to delete a job that is started will throw an exception
-        const jobsStillStarted = jobs.filter(job => job.status.job_state === 'started').map(job => job.config.id);
-
-        if (jobsStillStarted.length && attempt < 3) {
-          return stopAllJobStarted(jobsStillStarted, ++attempt);
-        } else if(jobsStillStarted.length) {
-          throw new Error('Error trying to stop jobs started');
-        }
-      })
-  );
-
-  const deleteJobsCreated = (ids = jobsCreated, attempt = 0) => (
-    deleteJob(ids)
-      .then((response) => {
-        if (response.status !== 200 && response.status !== 404) {
-          throw response;
-        }
-      })
-      .then(loadJobs)
-      .then(({ body: { jobs } }) => {
-        if (jobs.length && attempt < 3) {
-          // There are still some jobs left to delete.
-          // Call recursively until all rollup jobs are removed.
-          return deleteJobsCreated(jobs.map(job => job.config.id), ++attempt);
-        } else if (jobs.length) {
-          throw new Error('Error trying to delete Jobs created');
-        }
+        return jobIds;
       })
   );
 
@@ -155,7 +113,7 @@ export const registerHelpers = ({ supertest, es }) => {
   const cleanUp = () => (
     Promise.all([
       deleteAllIndices(),
-      stopAllJobStarted().then(deleteJobsCreated),
+      stopAllJobs().then(deleteJob),
       deleteIndicesGeneratedByJobs(),
     ]).catch(err => {
       console.log('ERROR cleaning up!');
@@ -166,6 +124,7 @@ export const registerHelpers = ({ supertest, es }) => {
   return {
     createIndexWithMappings,
     getJobPayload,
+    loadJobs,
     createJob,
     deleteJob,
     startJob,