[APM] Anomaly detection integration with transaction duration chart (#…

…71230)

* 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.

* PR feedback

* Code improvements from PR feedback

* handle errors thrown when fetching ml job for current environment

Co-authored-by: Elastic Machine <elasticmachine@users.noreply.github.com>

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

@@ -89,7 +89,6 @@ export function TransactionDetails() {
             <EuiSpacer size="s" />
 
             <TransactionCharts
-              hasMLJob={false}
               charts={transactionChartsData}
               urlParams={urlParams}
               location={location}

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

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

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

@@ -42,7 +42,6 @@ import {
 } from '../../../../../common/transaction_types';
 
 interface TransactionChartProps {
-  hasMLJob: boolean;
   charts: ITransactionChartData;
   location: Location;
   urlParams: IUrlParams;
@@ -96,18 +95,17 @@ export class TransactionCharts extends Component<TransactionChartProps> {
   };
 
   public renderMLHeader(hasValidMlLicense: boolean | undefined) {
-    const { hasMLJob } = this.props;
-    if (!hasValidMlLicense || !hasMLJob) {
+    const { mlJobId } = this.props.charts;
+
+    if (!hasValidMlLicense || !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 +138,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,93 @@
+/*
+ * 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 { Logger } from 'kibana/server';
+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,
+  logger,
+}: {
+  serviceName: string;
+  transactionType: string;
+  intervalString: string;
+  mlBucketSize: number;
+  setup: Setup & SetupTimeRange;
+  jobId: string;
+  logger: Logger;
+}) {
+  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) {
+      logger.info(
+        `Status code "${err.statusCode}" while retrieving ML anomalies for APM`
+      );
+      return;
+    }
+    logger.error('An error occurred while retrieving ML anomalies for APM');
+    logger.error(err);
+  }
+}

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

@@ -0,0 +1,61 @@
+/*
+ * 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 { Logger } from 'kibana/server';
+import { Setup, SetupTimeRange } from '../../../helpers/setup_request';
+
+interface IOptions {
+  setup: Setup & SetupTimeRange;
+  jobId: string;
+  logger: Logger;
+}
+
+interface ESResponse {
+  bucket_span: number;
+}
+
+export async function getMlBucketSize({
+  setup,
+  jobId,
+  logger,
+}: IOptions): Promise<number | undefined> {
+  const { ml, start, end } = setup;
+  if (!ml) {
+    return;
+  }
+
+  const params = {
+    body: {
+      _source: 'bucket_span',
+      size: 1,
+      terminateAfter: 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;
+  } catch (err) {
+    const isHttpError = 'statusCode' in err;
+    if (isHttpError) {
+      return;
+    }
+    logger.error(err);
+  }
+}

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

@@ -3,32 +3,37 @@
  * or more contributor license agreements. Licensed under the Elastic License;
  * you may not use this file except in compliance with the Elastic License.
  */
-
+import { Logger } from 'kibana/server';
+import { isNumber } from 'lodash';
+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 { UIFilters } from '../../../../../typings/ui_filters';
 
 export async function getAnomalySeries({
   serviceName,
   transactionType,
   transactionName,
   timeSeriesDates,
   setup,
+  logger,
+  uiFilters,
 }: {
   serviceName: string;
   transactionType: string | undefined;
   transactionName: string | undefined;
   timeSeriesDates: number[];
   setup: Setup & SetupTimeRange & SetupUIFilters;
-}): Promise<void | AnomalyTimeseries> {
+  logger: Logger;
+  uiFilters: UIFilters;
+}) {
   // don't fetch anomalies for transaction details page
   if (transactionName) {
     return;
@@ -39,8 +44,12 @@ export async function getAnomalySeries({
     return;
   }
 
-  // don't fetch anomalies if uiFilters are applied
-  if (setup.uiFiltersES.length > 0) {
+  // don't fetch anomalies if unknown uiFilters are applied
+  const knownFilters = ['environment', 'serviceName'];
+  const uiFilterNames = Object.keys(uiFilters);
+  if (
+    uiFilterNames.some((uiFilterName) => !knownFilters.includes(uiFilterName))
+  ) {
     return;
   }
 
@@ -55,6 +64,45 @@ export async function getAnomalySeries({
     return;
   }
 
-  // TODO [APM ML] return a series of anomaly scores, upper & lower bounds for the given timeSeriesDates
-  return;
+  let mlJobIds: string[] = [];
+  try {
+    mlJobIds = await getMLJobIds(setup.ml, uiFilters.environment);
+  } catch (error) {
+    logger.error(error);
+    return;
+  }
+
+  // don't fetch anomalies if there are isn't exaclty 1 ML job match for the given environment
+  if (mlJobIds.length !== 1) {
+    return;
+  }
+  const jobId = mlJobIds[0];
+
+  const mlBucketSize = await getMlBucketSize({ setup, jobId, logger });
+  if (!isNumber(mlBucketSize)) {
+    return;
+  }
+
+  const { start, end } = setup;
+  const { intervalString, bucketSize } = getBucketSize(start, end, 'auto');
+
+  const esResponse = await anomalySeriesFetcher({
+    serviceName,
+    transactionType,
+    intervalString,
+    mlBucketSize,
+    setup,
+    jobId,
+    logger,
+  });
+
+  if (esResponse && mlBucketSize > 0) {
+    return anomalySeriesTransform(
+      esResponse,
+      mlBucketSize,
+      bucketSize,
+      timeSeriesDates,
+      jobId
+    );
+  }
 }