import { sortBy , slice , get , cloneDeep } from 'lodash' ; import moment from 'moment' ; import Boom from '@hapi/boom' ; import { IScopedClusterClient } from 'kibana/server' ; import { buildAnomalyTableItems } from './build_anomaly_table_items' ; import { ANOMALIES_TABLE_DEFAULT_QUERY_SIZE } from '../../../common/constants/search' ; import { getPartitionFieldsValuesFactory } from './get_partition_fields_values' ; import { AnomaliesTableRecord , AnomalyCategorizerStatsDoc , AnomalyRecordDoc , } from '../../../common/types/anomalies' ; import { JOB_ID , PARTITION_FIELD_VALUE } from '../../../common/constants/anomalies' ; import { GetStoppedPartitionResult , GetDatafeedResultsChartDataResult , defaultSearchQuery , DatafeedResultsChartDataParams , } from '../../../common/types/results' ; import { MlJobsResponse } from '../../../common/types/job_service' ; import type { MlClient } from '../../lib/ml_client' ; import { datafeedsProvider } from '../job_service/datafeeds' ; import { annotationServiceProvider } from '../annotation_service' ; import { showActualForFunction , showTypicalForFunction } from '../../../common/util/anomaly_utils' ; const DEFAULT_MAX_EXAMPLES = 500 ; export interface CriteriaField { fieldType ? : string ; fieldName : string ; fieldValue : any ; } interface Influencer { fieldName : string ; fieldValue : any ; } param source export function getTypicalAndActualValues ( source ) { const result = { } ; const functionDescription = source . function_description || '' ; const causes = source . causes || [ ] ; if ( showActualForFunction ( functionDescription ) ) { if ( source . actual !== undefined ) { result . actual = source . actual ; } else { if ( causes . length === 1 ) { result . actual = causes [ 0 ] . actual ; } } } if ( showTypicalForFunction ( functionDescription ) ) { if ( source . typical !== undefined ) { result . typical = source . typical ; } else { if ( causes . length === 1 ) { result . typical = causes [ 0 ] . typical ; } } } return result ; } export function resultsServiceProvider ( mlClient , client ? ) { async function getAnomaliesTableData ( jobIds , criteriaFields , influencers , aggregationInterval , threshold , earliestMs , latestMs , dateFormatTz , maxRecords = ANOMALIES_TABLE_DEFAULT_QUERY_SIZE , maxExamples = DEFAULT_MAX_EXAMPLES , influencersFilterQuery ? , functionDescription ? ) { const boolCriteria = [ { range : { timestamp : { gte : earliestMs , lte : latestMs , format : 'epoch_millis' , } , } , } , { range : { record_score : { gte : threshold , } , } , } , ] ; if ( jobIds && jobIds . length > 0 && ! ( jobIds . length === 1 && jobIds [ 0 ] === '*' ) ) { let jobIdFilterStr = '' ; jobIds . forEach ( ( jobId , i ) => { if ( i > 0 ) { jobIdFilterStr += ' OR ' ; } jobIdFilterStr += 'job_id:' ; jobIdFilterStr += jobId ; } ) ; boolCriteria . push ( { query_string : { analyze_wildcard : false , query : jobIdFilterStr , } , } ) ; } criteriaFields . forEach ( ( criteria ) => { boolCriteria . push ( { term : { [ criteria . fieldName ] : criteria . fieldValue , } , } ) ; } ) ; if ( functionDescription !== undefined ) { boolCriteria . push ( { term : { function_description : functionDescription , } , } ) ; } if ( influencersFilterQuery !== undefined ) { boolCriteria . push ( influencersFilterQuery ) ; } if ( influencers . length > 0 ) { boolCriteria . push ( { bool : { should : influencers . map ( ( influencer ) => { return { nested : { path : 'influencers' , query : { bool : { must : [ { match : { 'influencers.influencer_field_name' : influencer . fieldName , } , } , { match : { 'influencers.influencer_field_values' : influencer . fieldValue , } , } , ] , } , } , } , } ; } ) , minimum_should_match : 1 , } , } ) ; } const { body } = await mlClient . anomalySearch ( { size : maxRecords , body : { query : { bool : { filter : [ { query_string : { query : 'result_type:record' , analyze_wildcard : false , } , } , { bool : { must : boolCriteria , } , } , ] , } , } , sort : [ { record_score : { order : 'desc' } } ] , } , } , [ ] ) ; const tableData = { anomalies : [ ] , interval : 'second' , } ; if ( body . hits . total . value > 0 ) { let records = [ ] ; body . hits . hits . forEach ( ( hit ) => { records . push ( hit . _source ) ; } ) ; records = sortBy ( records , 'timestamp' ) ; tableData . interval = aggregationInterval ; if ( aggregationInterval === 'auto' ) { const earliest = moment ( records [ 0 ] . timestamp ) ; const latest = moment ( records [ records . length - 1 ] . timestamp ) ; const daysDiff = latest . diff ( earliest , 'days' ) ; tableData . interval = daysDiff < 2 ? 'hour' : 'day' ; } tableData . anomalies = buildAnomalyTableItems ( records , tableData . interval , dateFormatTz ) ; const categoryAnomalies = tableData . anomalies . filter ( ( item ) => item . entityName === 'mlcategory' ) ; if ( categoryAnomalies . length > 0 ) { tableData . examplesByJobId = { } ; const categoryIdsByJobId = { } ; categoryAnomalies . forEach ( ( anomaly ) => { if ( categoryIdsByJobId [ anomaly . jobId ] === undefined ) { categoryIdsByJobId [ anomaly . jobId ] = [ ] ; } if ( categoryIdsByJobId [ anomaly . jobId ] . indexOf ( anomaly . entityValue ) === - 1 ) { categoryIdsByJobId [ anomaly . jobId ] . push ( anomaly . entityValue ) ; } } ) ; const categoryJobIds = Object . keys ( categoryIdsByJobId ) ; await Promise . all ( categoryJobIds . map ( async ( jobId ) => { const examplesByCategoryId = await getCategoryExamples ( jobId , categoryIdsByJobId [ jobId ] , maxExamples ) ; if ( tableData . examplesByJobId !== undefined ) { tableData . examplesByJobId [ jobId ] = examplesByCategoryId ; } } ) ) ; } } return tableData ; } async function getMaxAnomalyScore ( jobIds = [ ] , earliestMs , latestMs ) { const boolCriteria = [ { range : { timestamp : { gte : earliestMs , lte : latestMs , format : 'epoch_millis' , } , } , } , ] ; if ( jobIds . length > 0 ) { let jobIdFilterStr = '' ; jobIds . forEach ( ( jobId , i ) => { if ( i > 0 ) { jobIdFilterStr += ' OR ' ; } jobIdFilterStr += 'job_id:' ; jobIdFilterStr += jobId ; } ) ; boolCriteria . push ( { query_string : { analyze_wildcard : false , query : jobIdFilterStr , } , } ) ; } const query = { size : 0 , body : { query : { bool : { filter : [ { query_string : { query : 'result_type:bucket' , analyze_wildcard : false , } , } , { bool : { must : boolCriteria , } , } , ] , } , } , aggs : { max_score : { max : { field : 'anomaly_score' , } , } , } , } , } ; const { body } = await mlClient . anomalySearch ( query , [ ] ) ; const maxScore = get ( body , [ 'aggregations' , 'max_score' , 'value' ] , null ) ; return { maxScore } ; } async function getLatestBucketTimestampByJob ( jobIds = [ ] ) { const filter = [ { term : { result_type : 'bucket' , } , } , ] ; if ( jobIds . length > 0 && ! ( jobIds . length === 1 && jobIds [ 0 ] === '*' ) ) { let jobIdFilterStr = '' ; jobIds . forEach ( ( jobId , i ) => { if ( i > 0 ) { jobIdFilterStr += ' OR ' ; } jobIdFilterStr += 'job_id:' ; jobIdFilterStr += jobId ; } ) ; filter . push ( { query_string : { analyze_wildcard : false , query : jobIdFilterStr , } , } ) ; } const maxJobs = 10000 ; const { body } = await mlClient . anomalySearch ( { size : 0 , body : { query : { bool : { filter , } , } , aggs : { byJobId : { terms : { field : 'job_id' , size : maxJobs , } , aggs : { maxTimestamp : { max : { field : 'timestamp' , } , } , } , } , } , } , } , [ ] ) ; const bucketsByJobId = get ( body , [ 'aggregations' , 'byJobId' , 'buckets' ] , [ ] ) ; const timestampByJobId = { } ; bucketsByJobId . forEach ( ( bucket ) => { timestampByJobId [ bucket . key ] = bucket . maxTimestamp . value ; } ) ; return timestampByJobId ; } async function getCategoryExamples ( jobId , categoryIds , maxExamples ) { const { body } = await mlClient . anomalySearch ( { size : ANOMALIES_TABLE_DEFAULT_QUERY_SIZE , body : { query : { bool : { filter : [ { term : { job_id : jobId } } , { terms : { category_id : categoryIds } } ] , } , } , } , } , [ ] ) ; const examplesByCategoryId = { } ; if ( body . hits . total . value > 0 ) { body . hits . hits . forEach ( ( hit ) => { if ( maxExamples ) { examplesByCategoryId [ hit . _source . category_id ] = slice ( hit . _source . examples , 0 , Math . min ( hit . _source . examples . length , maxExamples ) ) ; } else { examplesByCategoryId [ hit . _source . category_id ] = hit . _source . examples ; } } ) ; } return examplesByCategoryId ; } async function getCategoryDefinition ( jobId , categoryId ) { const { body } = await mlClient . anomalySearch < any > ( { size : 1 , body : { query : { bool : { filter : [ { term : { job_id : jobId } } , { term : { category_id : categoryId } } ] , } , } , } , } , [ ] ) ; const definition = { categoryId , terms : null , regex : null , examples : [ ] } ; if ( body . hits . total . value > 0 ) { const source = body . hits . hits [ 0 ] . _source ; definition . categoryId = source . category_id ; definition . regex = source . regex ; definition . terms = source . terms ; definition . examples = source . examples ; } return definition ; } async function getCategorizerStats ( jobId , partitionByValue ? ) { const mustMatchClauses = [ { match : { result_type : 'categorizer_stats' , } , } , ] ; if ( typeof partitionByValue === 'string' ) { mustMatchClauses . push ( { match : { partition_by_value : partitionByValue , } , } ) ; } const { body } = await mlClient . anomalySearch < AnomalyCategorizerStatsDoc > ( { body : { query : { bool : { must : mustMatchClauses , filter : [ { term : { job_id : jobId , } , } , ] , } , } , } , } , [ ] ) ; return body ? body . hits . hits . map ( ( r ) => r . _source ) : [ ] ; } async function getCategoryStoppedPartitions ( jobIds , fieldToBucket = PARTITION_FIELD_VALUE ) { let finalResults = { jobs : { } , } ; const { body } = await mlClient . getJobs < MlJobsResponse > ( { job_id : jobIds . join ( ) , } ) ; if ( ! body || body . jobs . length < 1 ) { throw Boom . notFound ( `Unable to find anomaly detector jobs ${ jobIds . join ( ', ' ) }` ) ; } const jobIdsWithStopOnWarnSet = body . jobs . filter ( ( jobConfig ) => jobConfig . analysis_config ?. per_partition_categorization ?. stop_on_warn === true ) . map ( ( j ) => j . job_id ) ; let aggs ; if ( fieldToBucket === JOB_ID ) { aggs = { unique_terms : { terms : { field : JOB_ID , } , } , } ; } else { aggs = { jobs : { terms : { field : JOB_ID , } , aggs : { unique_stopped_partitions : { terms : { field : PARTITION_FIELD_VALUE , } , } , } , } , } ; } if ( jobIdsWithStopOnWarnSet . length > 0 ) { const mustMatchClauses = [ { match : { result_type : 'categorizer_stats' , } , } , { match : { categorization_status : 'warn' , } , } , ] ; const { body : results } = await mlClient . anomalySearch < any > ( { size : 0 , body : { query : { bool : { must : mustMatchClauses , filter : [ { terms : { job_id : jobIdsWithStopOnWarnSet , } , } , ] , } , } , aggs , } , } , [ ] ) ; if ( fieldToBucket === JOB_ID ) { finalResults = { jobs : results . aggregations ?. unique_terms ?. buckets . map ( ( b ) => b . key ) , } ; } else if ( fieldToBucket === PARTITION_FIELD_VALUE ) { const jobs = jobIdsWithStopOnWarnSet . reduce ( ( obj , jobId ) => { obj [ jobId ] = [ ] ; return obj ; } , { } ) ; results . aggregations . jobs . buckets . forEach ( ( bucket ) => { jobs [ bucket . key ] = bucket . unique_stopped_partitions . buckets . map ( ( b ) => b . key ) ; } ) ; finalResults . jobs = jobs ; } } return finalResults ; } async function getDatafeedResultsChartData ( { jobId , start , end , } ) { const finalResults = { bucketResults : [ ] , datafeedResults : [ ] , annotationResultsRect : [ ] , annotationResultsLine : [ ] , modelSnapshotResultsLine : [ ] , } ; const { getDatafeedByJobId } = datafeedsProvider ( client ! , mlClient ) ; const [ datafeedConfig , { body : jobsResponse } ] = await Promise . all ( [ getDatafeedByJobId ( jobId ) , mlClient . getJobs ( { job_id : jobId } ) , ] ) ; if ( jobsResponse && ( jobsResponse . count === 0 || jobsResponse . jobs === undefined ) ) { throw Boom . notFound ( `Job with the id "${ jobId }" not found` ) ; } const jobConfig = jobsResponse . jobs [ 0 ] ; const timefield = jobConfig . data_description . time_field ! ; const bucketSpan = jobConfig . analysis_config . bucket_span ; if ( datafeedConfig === undefined ) { return finalResults ; } const rangeFilter = { range : { [ timefield ] : { gte : start , lte : end } , } , } ; let datafeedQueryClone = datafeedConfig . query !== undefined ? cloneDeep ( datafeedConfig . query ) : defaultSearchQuery ; if ( datafeedQueryClone . bool !== undefined ) { if ( datafeedQueryClone . bool . filter === undefined ) { datafeedQueryClone . bool . filter = [ ] ; } if ( Array . isArray ( datafeedQueryClone . bool . filter ) ) { datafeedQueryClone . bool . filter . push ( rangeFilter ) ; } else { const filterQuery = cloneDeep ( datafeedQueryClone . bool . filter ) ; datafeedQueryClone . bool . filter = [ filterQuery , rangeFilter ] ; } } else { datafeedQueryClone = { bool : { must : [ datafeedQueryClone ] , filter : [ rangeFilter ] } } ; } const esSearchRequest = { index : datafeedConfig . indices . join ( ',' ) , body : { query : datafeedQueryClone , ... ( datafeedConfig . runtime_mappings ? { runtime_mappings : datafeedConfig . runtime_mappings } : { } ) , aggs : { doc_count_by_bucket_span : { date_histogram : { field : timefield , fixed_interval : bucketSpan , } , } , } , size : 0 , } , ... ( datafeedConfig ?. indices_options ?? { } ) , } ; if ( client ) { const { body : { aggregations } , } = await client . asCurrentUser . search ( esSearchRequest ) ; finalResults . datafeedResults = aggregations ?. doc_count_by_bucket_span ?. buckets . map ( ( result ) => [ result . key , result . doc_count , ] ) || [ ] ; } const { getAnnotations } = annotationServiceProvider ( client ! ) ; const [ bucketResp , annotationResp , { body : modelSnapshotsResp } ] = await Promise . all ( [ mlClient . getBuckets ( { job_id : jobId , body : { desc : true , start : String ( start ) , end : String ( end ) , page : { from : 0 , size : 1000 } } , } ) , getAnnotations ( { jobIds : [ jobId ] , earliestMs : start , latestMs : end , maxAnnotations : 1000 , } ) , mlClient . getModelSnapshots ( { job_id : jobId , start : String ( start ) , end : String ( end ) , } ) , ] ) ; const bucketResults = bucketResp ?. body ?. buckets ?? [ ] ; bucketResults . forEach ( ( dataForTime ) => { const timestamp = Number ( dataForTime ?. timestamp ) ; const eventCount = dataForTime ?. event_count ; finalResults . bucketResults . push ( [ timestamp , eventCount ] ) ; } ) ; const annotationResults = annotationResp . annotations [ jobId ] || [ ] ; annotationResults . forEach ( ( annotation ) => { const timestamp = Number ( annotation ?. timestamp ) ; const endTimestamp = Number ( annotation ?. end_timestamp ) ; if ( timestamp === endTimestamp ) { finalResults . annotationResultsLine . push ( { dataValue : timestamp , details : annotation . annotation , } ) ; } else { finalResults . annotationResultsRect . push ( { coordinates : { x0 : timestamp , x1 : endTimestamp , } , details : annotation . annotation , header : timestamp , } ) ; } } ) ; const modelSnapshots = modelSnapshotsResp ?. model_snapshots ?? [ ] ; modelSnapshots . forEach ( ( modelSnapshot ) => { const timestamp = Number ( modelSnapshot ?. timestamp ) ; finalResults . modelSnapshotResultsLine . push ( { dataValue : timestamp , details : modelSnapshot . description , } ) ; } ) ; return finalResults ; } return { getAnomaliesTableData , getCategoryDefinition , getCategoryExamples , getLatestBucketTimestampByJob , getMaxAnomalyScore , getPartitionFieldsValues : getPartitionFieldsValuesFactory ( mlClient ) , getCategorizerStats , getCategoryStoppedPartitions , getDatafeedResultsChartData , } ; }