import { Observable } from 'rxjs' ; import { map } from 'rxjs/operators' ; import { each , get } from 'lodash' ; import { Dictionary } from '../../../../common/types/common' ; import { ML_MEDIAN_PERCENTS } from '../../../../common/util/job_utils' ; import { Datafeed , JobId } from '../../../../common/types/anomaly_detection_jobs' ; import { MlApiServices } from '../ml_api_service' ; import { CriteriaField } from './index' ; import { findAggField } from '../../../../common/util/validation_utils' ; import { getDatafeedAggregations } from '../../../../common/util/datafeed_utils' ; import { aggregationTypeTransform , EntityField } from '../../../../common/util/anomaly_utils' ; import { ES_AGGREGATION } from '../../../../common/constants/aggregation_types' ; import { isPopulatedObject } from '../../../../common/util/object_utils' ; import { InfluencersFilterQuery } from '../../../../common/types/es_client' ; import { RecordForInfluencer } from './results_service' ; import { isRuntimeMappings } from '../../../../common' ; import { ErrorType } from '../../../../common/util/errors' ; export interface ResultResponse { success : boolean ; error ? : ErrorType ; } export interface MetricData extends ResultResponse { results : Record < string , any > ; } export interface FieldDefinition { /**
     * Field name.
     */ name : string | number ; /**
     * Field distinct values.
     */ values : Array < { value : any ; maxRecordScore ? : number ; } > ; } type FieldTypes = 'partition_field' | 'over_field' | 'by_field' ; export type PartitionFieldsDefinition = { [ field in FieldTypes ] : FieldDefinition ; } ; export interface ModelPlotOutput extends ResultResponse { results : Record < string , any > ; } export interface RecordsForCriteria extends ResultResponse { records : any [ ] ; } export interface ScheduledEventsByBucket extends ResultResponse { events : Record < string , any > ; } export function resultsServiceRxProvider ( mlApiServices ) { return { getMetricData ( index , entityFields , query , metricFunction , metricFieldName , summaryCountFieldName , timeFieldName , earliestMs , latestMs , intervalMs , datafeedConfig ? ) { const scriptFields = datafeedConfig ?. script_fields ; const aggFields = getDatafeedAggregations ( datafeedConfig ) ; const shouldCriteria = [ ] ; const mustCriteria = [ { range : { [ timeFieldName ] : { gte : earliestMs , lte : latestMs , format : 'epoch_millis' , } , } , } , ... ( query ? [ query ] : [ ] ) , ] ; entityFields . forEach ( ( entity ) => { if ( entity . fieldValue . length !== 0 ) { mustCriteria . push ( { term : { [ entity . fieldName ] : entity . fieldValue , } , } ) ; } else { shouldCriteria . push ( { bool : { must : [ { term : { [ entity . fieldName ] : '' , } , } , ] , } , } ) ; shouldCriteria . push ( { bool : { must_not : [ { exists : { field : entity . fieldName } , } , ] , } , } ) ; } } ) ; const body = { query : { bool : { must : mustCriteria , } , } , size : 0 , _source : false , aggs : { byTime : { date_histogram : { field : timeFieldName , fixed_interval : `${ intervalMs }ms` , min_doc_count : 0 , } , } , } , ... ( isRuntimeMappings ( datafeedConfig ?. runtime_mappings ) ? { runtime_mappings : datafeedConfig ?. runtime_mappings } : { } ) , } ; if ( shouldCriteria . length > 0 ) { body . query . bool . should = shouldCriteria ; body . query . bool . minimum_should_match = shouldCriteria . length / 2 ; } body . aggs . byTime . aggs = { } ; if ( metricFieldName !== undefined && metricFieldName !== '' && metricFunction ) { const metricAgg = { [ metricFunction ] : { } , } ; if ( scriptFields !== undefined && scriptFields [ metricFieldName ] !== undefined ) { metricAgg [ metricFunction ] . script = scriptFields [ metricFieldName ] . script ; } else { metricAgg [ metricFunction ] . field = metricFieldName ; } if ( metricFunction === 'percentiles' ) { metricAgg [ metricFunction ] . percents = [ ML_MEDIAN_PERCENTS ] ; } if ( isPopulatedObject ( aggFields ) ) { const accessor = Object . keys ( aggFields ) [ 0 ] ; const tempAggs = { ... ( aggFields [ accessor ] . aggs ?? aggFields [ accessor ] . aggregations ) } ; const foundValue = findAggField ( tempAggs , metricFieldName ) ; if ( foundValue !== undefined ) { tempAggs . metric = foundValue ; delete tempAggs [ metricFieldName ] ; } body . aggs . byTime . aggs = tempAggs ; } else { body . aggs . byTime . aggs . metric = metricAgg ; } } else { if ( summaryCountFieldName !== undefined && metricFunction === ES_AGGREGATION . CARDINALITY ) { if ( typeof aggFields === 'object' && Object . keys ( aggFields ) . length > 0 ) { const accessor = Object . keys ( aggFields ) [ 0 ] ; const tempAggs = { ... ( aggFields [ accessor ] . aggs ?? aggFields [ accessor ] . aggregations ) } ; const foundCardinalityField = findAggField ( tempAggs , summaryCountFieldName ) ; if ( foundCardinalityField !== undefined ) { tempAggs . metric = foundCardinalityField ; } body . aggs . byTime . aggs = tempAggs ; } } } return mlApiServices . esSearch$ ( { index , body } ) . pipe ( map ( ( resp ) => { const obj = { success : true , results : { } } ; const dataByTime = resp ?. aggregations ?. byTime ?. buckets ?? [ ] ; dataByTime . forEach ( ( dataForTime ) => { if ( metricFunction === 'count' ) { obj . results [ dataForTime . key ] = dataForTime . doc_count ; } else { const value = dataForTime ?. metric ?. value ; const values = dataForTime ?. metric ?. values ; if ( dataForTime . doc_count === 0 ) { obj . results [ dataForTime . key ] = null ; } else if ( value !== undefined ) { obj . results [ dataForTime . key ] = value ; } else if ( values !== undefined ) { const medianValues = values [ ML_MEDIAN_PERCENTS ] ; obj . results [ dataForTime . key ] = ! isNaN ( medianValues ) ? medianValues : null ; } else { obj . results [ dataForTime . key ] = null ; } } } ) ; return obj ; } ) ) ; } , getModelPlotOutput ( jobId , detectorIndex , criteriaFields , earliestMs , latestMs , intervalMs , aggType ? ) { const obj = { success : true , results : { } , } ; const modelAggs = aggType === undefined ? { max : 'max' , min : 'min' } : { max : aggType . max , min : aggType . min , } ; const mustCriteria = [ { term : { job_id : jobId } , } , { range : { timestamp : { gte : earliestMs , lte : latestMs , format : 'epoch_millis' , } , } , } , ] ; each ( criteriaFields , ( criteria ) => { mustCriteria . push ( { term : { [ criteria . fieldName ] : criteria . fieldValue , } , } ) ; } ) ; const shouldCriteria = [ { term : { detector_index : detectorIndex } , } , { bool : { must_not : [ { exists : { field : 'detector_index' } , } , ] , } , } , ] ; return mlApiServices . results . anomalySearch$ ( { size : 0 , body : { query : { bool : { filter : [ { query_string : { query : 'result_type:model_plot' , analyze_wildcard : true , } , } , { bool : { must : mustCriteria , should : shouldCriteria , minimum_should_match : 1 , } , } , ] , } , } , aggs : { times : { date_histogram : { field : 'timestamp' , fixed_interval : `${ intervalMs }ms` , min_doc_count : 0 , } , aggs : { actual : { avg : { field : 'actual' , } , } , modelUpper : { [ modelAggs . max ] : { field : 'model_upper' , } , } , modelLower : { [ modelAggs . min ] : { field : 'model_lower' , } , } , } , } , } , } , } , [ jobId ] ) . pipe ( map ( ( resp ) => { const aggregationsByTime = get ( resp , [ 'aggregations' , 'times' , 'buckets' ] , [ ] ) ; each ( aggregationsByTime , ( dataForTime ) => { const time = dataForTime . key ; const modelUpper = get ( dataForTime , [ 'modelUpper' , 'value' ] ) ; const modelLower = get ( dataForTime , [ 'modelLower' , 'value' ] ) ; const actual = get ( dataForTime , [ 'actual' , 'value' ] ) ; obj . results [ time ] = { actual , modelUpper : modelUpper === undefined || isFinite ( modelUpper ) === false ? null : modelUpper , modelLower : modelLower === undefined || isFinite ( modelLower ) === false ? null : modelLower , } ; } ) ; return obj ; } ) ) ; } , getRecordsForCriteria ( jobIds , criteriaFields , threshold , earliestMs , latestMs , maxResults , functionDescription ? ) { const obj = { success : true , records : [ ] } ; 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 = '' ; each ( jobIds , ( jobId , i ) => { if ( i > 0 ) { jobIdFilterStr += ' OR ' ; } jobIdFilterStr += 'job_id:' ; jobIdFilterStr += jobId ; } ) ; boolCriteria . push ( { query_string : { analyze_wildcard : false , query : jobIdFilterStr , } , } ) ; } each ( criteriaFields , ( criteria ) => { boolCriteria . push ( { term : { [ criteria . fieldName ] : criteria . fieldValue , } , } ) ; } ) ; if ( functionDescription !== undefined ) { const mlFunctionToPlotIfMetric = functionDescription !== undefined ? aggregationTypeTransform . toML ( functionDescription ) : functionDescription ; boolCriteria . push ( { term : { function_description : mlFunctionToPlotIfMetric , } , } ) ; } return mlApiServices . results . anomalySearch$ ( { size : maxResults !== undefined ? maxResults : 100 , body : { query : { bool : { filter : [ { query_string : { query : 'result_type:record' , analyze_wildcard : false , } , } , { bool : { must : boolCriteria , } , } , ] , } , } , sort : [ { record_score : { order : 'desc' } } ] , } , } , jobIds ) . pipe ( map ( ( resp ) => { if ( resp . hits . total . value > 0 ) { each ( resp . hits . hits , ( hit ) => { obj . records . push ( hit . _source ) ; } ) ; } return obj ; } ) ) ; } , getScheduledEventsByBucket ( jobIds , earliestMs , latestMs , intervalMs , maxJobs , maxEvents ) { const obj = { success : true , events : { } , } ; const boolCriteria = [ { range : { timestamp : { gte : earliestMs , lte : latestMs , format : 'epoch_millis' , } , } , } , { exists : { field : 'scheduled_events' } , } , ] ; if ( jobIds && jobIds . length > 0 && ! ( jobIds . length === 1 && jobIds [ 0 ] === '*' ) ) { let jobIdFilterStr = '' ; each ( jobIds , ( jobId , i ) => { jobIdFilterStr += `${ i > 0 ? ' OR ' : '' }job_id:${ jobId }` ; } ) ; boolCriteria . push ( { query_string : { analyze_wildcard : false , query : jobIdFilterStr , } , } ) ; } return mlApiServices . results . anomalySearch$ ( { size : 0 , body : { query : { bool : { filter : [ { query_string : { query : 'result_type:bucket' , analyze_wildcard : false , } , } , { bool : { must : boolCriteria , } , } , ] , } , } , aggs : { jobs : { terms : { field : 'job_id' , min_doc_count : 1 , size : maxJobs , } , aggs : { times : { date_histogram : { field : 'timestamp' , fixed_interval : `${ intervalMs }ms` , min_doc_count : 1 , } , aggs : { events : { terms : { field : 'scheduled_events' , size : maxEvents , } , } , } , } , } , } , } , } , } , jobIds ) . pipe ( map ( ( resp ) => { const dataByJobId = get ( resp , [ 'aggregations' , 'jobs' , 'buckets' ] , [ ] ) ; each ( dataByJobId , ( dataForJob ) => { const jobId = dataForJob . key ; const resultsForTime = { } ; const dataByTime = get ( dataForJob , [ 'times' , 'buckets' ] , [ ] ) ; each ( dataByTime , ( dataForTime ) => { const time = dataForTime . key ; const events = get ( dataForTime , [ 'events' , 'buckets' ] ) ; resultsForTime [ time ] = events . map ( ( e ) => e . key ) ; } ) ; obj . events [ jobId ] = resultsForTime ; } ) ; return obj ; } ) ) ; } , fetchPartitionFieldsValues ( jobId , searchTerm , criteriaFields , earliestMs , latestMs ) { return mlApiServices . results . fetchPartitionFieldsValues ( jobId , searchTerm , criteriaFields , earliestMs , latestMs ) ; } , getRecordsForInfluencer$ ( jobIds , influencers , threshold , earliestMs , latestMs , maxResults , influencersFilterQuery ) { const obj = { success : true , records : [ ] as RecordForInfluencer [ ] } ; 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 = '' ; each ( jobIds , ( jobId , i ) => { if ( i > 0 ) { jobIdFilterStr += ' OR ' ; } jobIdFilterStr += 'job_id:' ; jobIdFilterStr += jobId ; } ) ; boolCriteria . push ( { query_string : { analyze_wildcard : false , query : jobIdFilterStr , } , } ) ; } 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 , } , } ) ; } return mlApiServices . results . anomalySearch$ ( { size : maxResults !== undefined ? maxResults : 100 , body : { query : { bool : { filter : [ { query_string : { query : 'result_type:record' , analyze_wildcard : false , } , } , { bool : { must : boolCriteria , } , } , ] , } , } , sort : [ { record_score : { order : 'desc' } } ] , } , } , jobIds ) . pipe ( map ( ( resp ) => { if ( resp . hits . total . value > 0 ) { each ( resp . hits . hits , ( hit ) => { obj . records . push ( hit . _source ) ; } ) ; } return obj ; } ) ) ; } , } ; }