Use bucket aggregation for anomaly distribution. Issue: #91 (#126)

public/pages/Dashboard/Components/AnomaliesDistribution.tsx

@@ -16,8 +16,7 @@ import { DetectorListItem } from '../../../models/interfaces';
 import { useState, useEffect } from 'react';
 import {
   fillOutColors,
-  visualizeAnomalyResultForSunburstChart,
-  getLatestAnomalyResultsForDetectorsByTimeRange,
+  getAnomalyDistributionForDetectorsByTimeRange,
 } from '../utils/utils';
 import ContentPanel from '../../../components/ContentPanel/ContentPanel';
 import {
@@ -36,7 +35,6 @@ import React from 'react';
 import { TIME_RANGE_OPTIONS } from '../../Dashboard/utils/constants';
 import { get, isEmpty } from 'lodash';
 import { searchES } from '../../../redux/reducers/elasticsearch';
-import { MAX_DETECTORS, MAX_ANOMALIES } from '../../../utils/constants';
 import { AD_DOC_FIELDS } from '../../../../server/utils/constants';
 export interface AnomaliesDistributionChartProps {
   selectedDetectors: DetectorListItem[];
@@ -47,7 +45,9 @@ export const AnomaliesDistributionChart = (
 ) => {
   const dispatch = useDispatch();
 
-  const [anomalyResults, setAnomalyResults] = useState([] as object[]);
+  const [anomalyDistribution, setAnomalyDistribution] = useState(
+    [] as object[]
+  );
 
   // TODO: try to find a better way of using redux,
   // which can leverage redux, and also get rid of issue with multiple redux on same page,
@@ -64,24 +64,19 @@ export const AnomaliesDistributionChart = (
 
   const getAnomalyResult = async (currentDetectors: DetectorListItem[]) => {
     setAnomalyResultsLoading(true);
-    const latestAnomalyResult = await getLatestAnomalyResultsForDetectorsByTimeRange(
+
+    const distributionResult = await getAnomalyDistributionForDetectorsByTimeRange(
       searchES,
       props.selectedDetectors,
       timeRange,
       dispatch,
       0,
-      MAX_ANOMALIES,
-      MAX_DETECTORS,
       false
     );
-
-    const nonZeroAnomalyResult = latestAnomalyResult.filter(
-      anomalyData => get(anomalyData, AD_DOC_FIELDS.ANOMALY_GRADE, 0) > 0
-    );
-    setAnomalyResults(nonZeroAnomalyResult);
+    setAnomalyDistribution(distributionResult);
 
     const resultDetectors = getFinalDetectors(
-      nonZeroAnomalyResult,
+      distributionResult,
       props.selectedDetectors
     );
     setIndicesNumber(getFinalIndices(resultDetectors).size);
@@ -99,11 +94,11 @@ export const AnomaliesDistributionChart = (
   };
 
   const getFinalDetectors = (
-    finalLiveAnomalyResult: object[],
+    finalAnomalyResult: object[],
     detectorList: DetectorListItem[]
   ): DetectorListItem[] => {
     const detectorSet = new Set<string>();
-    finalLiveAnomalyResult.forEach(anomalyResult => {
+    finalAnomalyResult.forEach(anomalyResult => {
       detectorSet.add(get(anomalyResult, AD_DOC_FIELDS.DETECTOR_ID, ''));
     });
 
@@ -174,14 +169,11 @@ export const AnomaliesDistributionChart = (
       ) : (
         <EuiFlexGroup justifyContent="center">
           <EuiFlexItem grow={false}>
-            {isEmpty(anomalyResults) ? null : (
+            {isEmpty(anomalyDistribution) ? null : (
               <Chart className="anomalies-distribution-sunburst">
                 <Partition
                   id="Anomalies by index and detector"
-                  data={visualizeAnomalyResultForSunburstChart(
-                    anomalyResults,
-                    finalDetectors
-                  )}
+                  data={anomalyDistribution}
                   valueAccessor={(d: Datum) => d.count as number}
                   valueFormatter={(d: number) => d.toString()}
                   layers={[

public/pages/Dashboard/utils/utils.tsx

@@ -35,6 +35,7 @@ import { APIAction } from 'public/redux/middleware/types';
 import { Dispatch } from 'redux';
 import { EuiBasicTableColumn } from '@elastic/eui';
 import { SHOW_DECIMAL_NUMBER_THRESHOLD } from './constants';
+import { MAX_DETECTORS } from '../../../pages/utils/constants';
 
 /**
  * Get the recent anomaly result query for the last timeRange period(Date-Math)
@@ -363,57 +364,6 @@ export const anomalousDetectorsStaticColumn = [
   },
 ] as EuiBasicTableColumn<any>[];
 
-export const visualizeAnomalyResultForSunburstChart = (
-  anomalyResults: any[],
-  detectors: DetectorListItem[]
-): object[] => {
-  const detectorAnomalyResultMap = buildDetectorAnomalyResultMap(
-    anomalyResults,
-    detectors
-  );
-  const visualizedResult = [] as object[];
-  for (let detectorInfo of detectorAnomalyResultMap.values()) {
-    visualizedResult.push(detectorInfo);
-  }
-  return visualizedResult;
-};
-
-const buildDetectorAnomalyResultMap = (
-  anomalyResults: any[],
-  detectors: DetectorListItem[]
-): Map<string, object> => {
-  const detectorAndIdMap = buildDetectorAndIdMap(detectors);
-  const detectorAnomalyResultMap = new Map();
-  anomalyResults.forEach(anomalyResult => {
-    const detectorId = get(anomalyResult, AD_DOC_FIELDS.DETECTOR_ID, '');
-    const detector = detectorAndIdMap.get(detectorId);
-    if (detectorAnomalyResultMap.has(detectorId)) {
-      const detectorInfo = detectorAnomalyResultMap.get(detectorId);
-      let currentCount = get(detectorInfo, 'count', 0);
-      currentCount++;
-      detectorAnomalyResultMap.set(
-        detectorId,
-        Object.assign({}, detectorInfo, { count: currentCount })
-      );
-    } else {
-      detectorAnomalyResultMap.set(detectorId, {
-        [AD_DOC_FIELDS.DETECTOR_NAME]: get(
-          anomalyResult,
-          AD_DOC_FIELDS.DETECTOR_NAME,
-          ''
-        ),
-        [AD_DOC_FIELDS.INDICES]: get(
-          detector,
-          AD_DOC_FIELDS.INDICES,
-          ''
-        ).toString(),
-        count: 1,
-      });
-    }
-  });
-  return detectorAnomalyResultMap;
-};
-
 export const visualizeAnomalyResultForXYChart = (
   anomalyResult: any
 ): object => {
@@ -635,3 +585,66 @@ const selectLatestAnomalousDetectorIds = (
 
   return new Set(Array.from(anomalousDetectorIds).slice(0, neededDetectorNum));
 };
+
+export const getAnomalyDistributionForDetectorsByTimeRange = async (
+  func: (request: any) => APIAction,
+  selectedDetectors: DetectorListItem[],
+  timeRange: string,
+  dispatch: Dispatch<any>,
+  threshold: number,
+  checkLastIndexOnly: boolean
+) => {
+  const aggregationName = 'anomaly_dist';
+  const detectorAndIdMap = buildDetectorAndIdMap(selectedDetectors);
+
+  const getResultQuery = buildGetAnomalyResultQueryByRange(
+    timeRange,
+    0,
+    0,
+    threshold,
+    checkLastIndexOnly
+  );
+  const anomaly_dist_aggs = {
+    aggs: {
+      [aggregationName]: {
+        terms: {
+          field: AD_DOC_FIELDS.DETECTOR_ID,
+          size: MAX_DETECTORS,
+        },
+      },
+    },
+  };
+  const finalQuery = Object.assign({}, getResultQuery, anomaly_dist_aggs);
+
+  const result = await dispatch(func(finalQuery));
+
+  const detectorsAggResults = get(
+    result,
+    `data.response.aggregations.${aggregationName}.buckets`,
+    []
+  );
+
+  const finalDetectorDistributionResult = [] as object[];
+  for (let detectorResult of detectorsAggResults) {
+    const detectorId = get(detectorResult, 'key', '');
+    if (detectorAndIdMap.has(detectorId)) {
+      const detector = detectorAndIdMap.get(detectorId);
+      finalDetectorDistributionResult.push({
+        [AD_DOC_FIELDS.DETECTOR_ID]: detectorId,
+        [AD_DOC_FIELDS.DETECTOR_NAME]: get(
+          detector,
+          AD_DOC_FIELDS.DETECTOR_NAME,
+          ''
+        ),
+        [AD_DOC_FIELDS.INDICES]: get(
+          detector,
+          AD_DOC_FIELDS.INDICES,
+          ''
+        ).toString(),
+        count: get(detectorResult, 'doc_count', 0),
+      });
+    }
+  }
+
+  return finalDetectorDistributionResult;
+};

server/routes/elasticsearch.ts

@@ -53,13 +53,15 @@ const executeSearch = async (
       size = 0,
       sort = undefined,
       collapse = undefined,
+      aggs = undefined,
       rawQuery = undefined,
     } = req.payload as {
       index: string;
       query?: object;
       size?: number;
       sort?: object;
       collapse?: object;
+      aggs?: object;
       rawQuery: object;
     };
     const requestBody = rawQuery
@@ -68,6 +70,7 @@ const executeSearch = async (
           query: query,
           ...(sort !== undefined && { sort: sort }),
           ...(collapse !== undefined && { collapse: collapse }),
+          ...(aggs !== undefined && { aggs: aggs }),
         };
 
     const params: SearchParams = { index, size, body: requestBody };