Reintroduces the previous anomaly detection ML integration back into the

transaction duration chart in the service details screen. Support the
latest APM anoamly detection by environment jobs.

x-pack/plugins/apm/public/components/app/TransactionOverview/index.tsx

@@ -130,8 +130,7 @@ export function TransactionOverview() {
             <EuiSpacer size="s" />
 
             <TransactionCharts
-              // TODO [APM ML] set hasMLJob prop when ML integration is reintroduced:
-              hasMLJob={false}
+              hasMLJob={transactionCharts.mlJobId !== undefined}
               charts={transactionCharts}
               location={location}
               urlParams={urlParams}

x-pack/plugins/apm/public/components/shared/charts/TransactionCharts/index.tsx

@@ -96,18 +96,19 @@ export class TransactionCharts extends Component<TransactionChartProps> {
   };
 
   public renderMLHeader(hasValidMlLicense: boolean | undefined) {
-    const { hasMLJob } = this.props;
-    if (!hasValidMlLicense || !hasMLJob) {
+    const {
+      hasMLJob,
+      charts: { mlJobId },
+    } = this.props;
+    if (!hasValidMlLicense || !hasMLJob || !mlJobId) {
       return null;
     }
 
-    const { serviceName, kuery } = this.props.urlParams;
+    const { serviceName, kuery, transactionType } = this.props.urlParams;
     if (!serviceName) {
       return null;
     }
 
-    const linkedJobId = ''; // TODO [APM ML] link to ML job id for the selected environment
-
     const hasKuery = !isEmpty(kuery);
     const icon = hasKuery ? (
       <EuiIconTip
@@ -140,7 +141,13 @@ export class TransactionCharts extends Component<TransactionChartProps> {
               }
             )}{' '}
           </span>
-          <MLJobLink jobId={linkedJobId}>View Job</MLJobLink>
+          <MLJobLink
+            jobId={mlJobId}
+            serviceName={serviceName}
+            transactionType={transactionType}
+          >
+            View Job
+          </MLJobLink>
         </ShiftedEuiText>
       </EuiFlexItem>
     );

x-pack/plugins/apm/public/selectors/chartSelectors.ts

@@ -33,6 +33,7 @@ export interface ITpmBucket {
 export interface ITransactionChartData {
   tpmSeries: ITpmBucket[];
   responseTimeSeries: TimeSeries[];
+  mlJobId: string | undefined;
 }
 
 const INITIAL_DATA = {
@@ -62,6 +63,7 @@ export function getTransactionCharts(
   return {
     tpmSeries,
     responseTimeSeries,
+    mlJobId: anomalyTimeseries?.jobId,
   };
 }
 

x-pack/plugins/apm/server/lib/transactions/charts/get_anomaly_data/fetcher.ts

@@ -0,0 +1,105 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+import { PromiseReturnType } from '../../../../../../observability/typings/common';
+import { Setup, SetupTimeRange } from '../../../helpers/setup_request';
+
+export type ESResponse = Exclude<
+  PromiseReturnType<typeof anomalySeriesFetcher>,
+  undefined
+>;
+
+export async function anomalySeriesFetcher({
+  serviceName,
+  transactionType,
+  intervalString,
+  mlBucketSize,
+  setup,
+  jobId,
+}: {
+  serviceName: string;
+  transactionType: string;
+  intervalString: string;
+  mlBucketSize: number;
+  setup: Setup & SetupTimeRange;
+  jobId: string;
+}) {
+  const { ml, start, end } = setup;
+  if (!ml) {
+    return;
+  }
+
+  // move the start back with one bucket size, to ensure to get anomaly data in the beginning
+  // this is required because ML has a minimum bucket size (default is 900s) so if our buckets are smaller, we might have several null buckets in the beginning
+  const newStart = start - mlBucketSize * 1000;
+
+  const params = {
+    body: {
+      size: 0,
+      query: {
+        bool: {
+          filter: [
+            { term: { job_id: jobId } },
+            { exists: { field: 'bucket_span' } },
+            {
+              term: {
+                result_type: 'model_plot',
+              },
+            },
+            {
+              term: {
+                partition_field_value: serviceName,
+              },
+            },
+            {
+              term: {
+                by_field_value: transactionType,
+              },
+            },
+            {
+              range: {
+                timestamp: {
+                  gte: newStart,
+                  lte: end,
+                  format: 'epoch_millis',
+                },
+              },
+            },
+          ],
+        },
+      },
+      aggs: {
+        ml_avg_response_times: {
+          date_histogram: {
+            field: 'timestamp',
+            fixed_interval: intervalString,
+            min_doc_count: 0,
+            extended_bounds: {
+              min: newStart,
+              max: end,
+            },
+          },
+          aggs: {
+            anomaly_score: { max: { field: 'anomaly_score' } },
+            lower: { min: { field: 'model_lower' } },
+            upper: { max: { field: 'model_upper' } },
+          },
+        },
+      },
+    },
+  };
+
+  try {
+    const response = await ml.mlSystem.mlAnomalySearch(params);
+    return response;
+  } catch (err) {
+    const isHttpError = 'statusCode' in err;
+    if (isHttpError) {
+      return;
+    }
+    throw err;
+  }
+}

x-pack/plugins/apm/server/lib/transactions/charts/get_anomaly_data/get_ml_bucket_size.ts

@@ -0,0 +1,65 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+import { Setup, SetupTimeRange } from '../../../helpers/setup_request';
+
+interface IOptions {
+  serviceName: string;
+  transactionType: string;
+  setup: Setup & SetupTimeRange;
+  jobId: string;
+}
+
+interface ESResponse {
+  bucket_span: number;
+}
+
+export async function getMlBucketSize({
+  serviceName,
+  transactionType,
+  setup,
+  jobId,
+}: IOptions): Promise<number> {
+  const { ml, start, end } = setup;
+  if (!ml) {
+    return 0;
+  }
+
+  const params = {
+    body: {
+      _source: 'bucket_span',
+      size: 1,
+      query: {
+        bool: {
+          filter: [
+            { term: { job_id: jobId } },
+            { exists: { field: 'bucket_span' } },
+            {
+              range: {
+                timestamp: {
+                  gte: start,
+                  lte: end,
+                  format: 'epoch_millis',
+                },
+              },
+            },
+          ],
+        },
+      },
+    },
+  };
+
+  try {
+    const resp = await ml.mlSystem.mlAnomalySearch<ESResponse>(params);
+    return resp.hits.hits[0]?._source.bucket_span || 0;
+  } catch (err) {
+    const isHttpError = 'statusCode' in err;
+    if (isHttpError) {
+      return 0;
+    }
+    throw err;
+  }
+}

x-pack/plugins/apm/server/lib/transactions/charts/get_anomaly_data/index.ts

@@ -3,32 +3,36 @@
  * or more contributor license agreements. Licensed under the Elastic License;
  * you may not use this file except in compliance with the Elastic License.
  */
-
+import { getBucketSize } from '../../../helpers/get_bucket_size';
 import {
   Setup,
   SetupTimeRange,
   SetupUIFilters,
 } from '../../../helpers/setup_request';
-import { Coordinate, RectCoordinate } from '../../../../../typings/timeseries';
-
-interface AnomalyTimeseries {
-  anomalyBoundaries: Coordinate[];
-  anomalyScore: RectCoordinate[];
-}
+import { anomalySeriesFetcher } from './fetcher';
+import { getMlBucketSize } from './get_ml_bucket_size';
+import { anomalySeriesTransform } from './transform';
+import { getMLJobIds } from '../../../service_map/get_service_anomalies';
+import {
+  SERVICE_ENVIRONMENT,
+  SERVICE_NAME,
+} from '../../../../../common/elasticsearch_fieldnames';
 
 export async function getAnomalySeries({
   serviceName,
   transactionType,
   transactionName,
+  environment,
   timeSeriesDates,
   setup,
 }: {
   serviceName: string;
   transactionType: string | undefined;
   transactionName: string | undefined;
+  environment: string | undefined;
   timeSeriesDates: number[];
   setup: Setup & SetupTimeRange & SetupUIFilters;
-}): Promise<void | AnomalyTimeseries> {
+}) {
   // don't fetch anomalies for transaction details page
   if (transactionName) {
     return;
@@ -39,9 +43,16 @@ export async function getAnomalySeries({
     return;
   }
 
-  // don't fetch anomalies if uiFilters are applied
   if (setup.uiFiltersES.length > 0) {
-    return;
+    // filter out known uiFilters like service.environment & service.name
+    const unknownFilters = setup.uiFiltersES.filter(
+      (uiFilter) =>
+        !uiFilter.term?.[SERVICE_ENVIRONMENT] && !uiFilter.terms?.[SERVICE_NAME]
+    );
+    // don't fetch anomalies if unknown uiFilters are applied
+    if (unknownFilters.length > 0) {
+      return;
+    }
   }
 
   // don't fetch anomalies if the ML plugin is not setup
@@ -55,6 +66,40 @@ export async function getAnomalySeries({
     return;
   }
 
-  // TODO [APM ML] return a series of anomaly scores, upper & lower bounds for the given timeSeriesDates
-  return;
+  const mlJobIds = await getMLJobIds(setup.ml, environment);
+
+  // don't fetch anomalies if there are more than 1 ML jobs for the given environment
+  if (mlJobIds.length > 1) {
+    return;
+  }
+  const jobId = mlJobIds[0];
+
+  const mlBucketSize = await getMlBucketSize({
+    serviceName,
+    transactionType,
+    setup,
+    jobId,
+  });
+
+  const { start, end } = setup;
+  const { intervalString, bucketSize } = getBucketSize(start, end, 'auto');
+
+  const esResponse = await anomalySeriesFetcher({
+    serviceName,
+    transactionType,
+    intervalString,
+    mlBucketSize,
+    setup,
+    jobId,
+  });
+
+  return esResponse
+    ? anomalySeriesTransform(
+        esResponse,
+        mlBucketSize,
+        bucketSize,
+        timeSeriesDates,
+        jobId
+      )
+    : undefined;
 }