[APM] Fixes duplicate ML job creation for existing environments (elas…

…tic#85023) (elastic#93098)

* [APM] Fixes duplicate ML job creation for existing environments (elastic#85023)

* Removes commented out test code.

* Adds API integration tests

* clean up code for readability

x-pack/plugins/apm/server/lib/anomaly_detection/create_anomaly_detection_jobs.ts

@@ -19,6 +19,7 @@ import {
 } from '../../../common/elasticsearch_fieldnames';
 import { APM_ML_JOB_GROUP, ML_MODULE_ID_APM_TRANSACTION } from './constants';
 import { withApmSpan } from '../../utils/with_apm_span';
+import { getAnomalyDetectionJobs } from './get_anomaly_detection_jobs';
 
 export async function createAnomalyDetectionJobs(
   setup: Setup,
@@ -38,14 +39,19 @@ export async function createAnomalyDetectionJobs(
     throw Boom.forbidden(ML_ERRORS.ML_NOT_AVAILABLE_IN_SPACE);
   }
 
+  const uniqueMlJobEnvs = await getUniqueMlJobEnvs(setup, environments, logger);
+  if (uniqueMlJobEnvs.length === 0) {
+    return [];
+  }
+
   return withApmSpan('create_anomaly_detection_jobs', async () => {
     logger.info(
-      `Creating ML anomaly detection jobs for environments: [${environments}].`
+      `Creating ML anomaly detection jobs for environments: [${uniqueMlJobEnvs}].`
     );
 
     const indexPatternName = indices['apm_oss.transactionIndices'];
     const responses = await Promise.all(
-      environments.map((environment) =>
+      uniqueMlJobEnvs.map((environment) =>
         createAnomalyDetectionJob({ ml, environment, indexPatternName })
       )
     );
@@ -105,3 +111,24 @@ async function createAnomalyDetectionJob({
     });
   });
 }
+
+async function getUniqueMlJobEnvs(
+  setup: Setup,
+  environments: string[],
+  logger: Logger
+) {
+  // skip creation of duplicate ML jobs
+  const jobs = await getAnomalyDetectionJobs(setup, logger);
+  const existingMlJobEnvs = jobs.map(({ environment }) => environment);
+  const requestedExistingMlJobEnvs = environments.filter((env) =>
+    existingMlJobEnvs.includes(env)
+  );
+
+  if (requestedExistingMlJobEnvs.length) {
+    logger.warn(
+      `Skipping creation of existing ML jobs for environments: [${requestedExistingMlJobEnvs}]}`
+    );
+  }
+
+  return environments.filter((env) => !existingMlJobEnvs.includes(env));
+}

x-pack/plugins/apm/server/routes/settings/anomaly_detection.ts

@@ -66,6 +66,7 @@ export const createAnomalyDetectionJobsRoute = createRoute({
     }
 
     await createAnomalyDetectionJobs(setup, environments, context.logger);
+
     notifyFeatureUsage({
       licensingPlugin: context.licensing,
       featureName: 'ml',

x-pack/test/apm_api_integration/tests/settings/anomaly_detection/write_user.ts

@@ -6,6 +6,7 @@
  */
 
 import expect from '@kbn/expect';
+import { countBy } from 'lodash';
 import { registry } from '../../../common/registry';
 import { FtrProviderContext } from '../../../common/ftr_provider_context';
 
@@ -49,7 +50,26 @@ export default function apiTest({ getService }: FtrProviderContext) {
 
           const { body } = await getJobs();
           expect(body.hasLegacyJobs).to.be(false);
-          expect(body.jobs.map((job: any) => job.environment)).to.eql(['production', 'staging']);
+          expect(countBy(body.jobs, 'environment')).to.eql({
+            production: 1,
+            staging: 1,
+          });
+        });
+
+        describe('with existing ML jobs', () => {
+          before(async () => {
+            await createJobs(['production', 'staging']);
+          });
+          it('skips duplicate job creation', async () => {
+            await createJobs(['production', 'test']);
+
+            const { body } = await getJobs();
+            expect(countBy(body.jobs, 'environment')).to.eql({
+              production: 1,
+              staging: 1,
+              test: 1,
+            });
+          });
         });
       });
     });