import Boom from '@hapi/boom' ; import { each , get } from 'lodash' ; import { IScopedClusterClient } from 'kibana/server' ; import type * as estypes from '@elastic/elasticsearch/lib/api/typesWithBodyKey' ; import { ANNOTATION_EVENT_USER , ANNOTATION_TYPE } from '../../../common/constants/annotations' ; import { PARTITION_FIELDS } from '../../../common/constants/anomalies' ; import { ML_ANNOTATIONS_INDEX_ALIAS_READ , ML_ANNOTATIONS_INDEX_ALIAS_WRITE , } from '../../../common/constants/index_patterns' ; import { Annotation , Annotations , isAnnotation , isAnnotations , getAnnotationFieldName , getAnnotationFieldValue , } from '../../../common/types/annotations' ; import { JobId } from '../../../common/types/anomaly_detection_jobs' ; interface EsResult { _source : Annotation ; _id : string ; } export interface IndexAnnotationArgs { jobIds : string [ ] ; earliestMs : number | null ; latestMs : number | null ; maxAnnotations : number ; detectorIndex ? : number ; entities ? : any [ ] ; event ? : Annotation [ 'event' ] ; } export interface GetParams { index : string ; size : number ; body : object ; track_total_hits : boolean ; } export interface GetResponse { success : true ; annotations : Record < JobId , Annotations > ; totalCount : number ; } export interface IndexParams { index : string ; body : Annotation ; refresh : boolean | 'wait_for' | undefined ; require_alias ? : boolean ; id ? : string ; } export interface DeleteParams { index : string ; refresh : boolean | 'wait_for' | undefined ; id : string ; } export interface AggByJob { key : string ; doc_count : number ; latest_delayed : Pick < estypes . SearchResponse < Annotation > , 'hits' > ; } export function annotationProvider ( { asInternalUser } ) { async function indexAnnotation ( annotation , username ) { if ( isAnnotation ( annotation ) === false ) { return Promise . reject ( new Error ( 'invalid annotation format' ) ) ; } if ( annotation . create_time === undefined ) { annotation . create_time = new Date ( ) . getTime ( ) ; annotation . create_username = username ; } annotation . modified_time = new Date ( ) . getTime ( ) ; annotation . modified_username = username ; const params = { index : ML_ANNOTATIONS_INDEX_ALIAS_WRITE , body : annotation , refresh : 'wait_for' , require_alias : true , } ; if ( typeof annotation . _id !== 'undefined' ) { params . id = annotation . _id ; delete params . body . _id ; delete params . body . key ; } const { body } = await asInternalUser . index ( params ) ; return body ; } async function getAnnotations ( { jobIds , earliestMs , latestMs , maxAnnotations , detectorIndex , entities , event , } ) { const obj = { success : true , annotations : { } , totalCount : 0 , } ; const boolCriteria = [ ] ; if ( earliestMs !== null && latestMs !== null ) { boolCriteria . push ( { bool : { must_not : [ { bool : { filter : [ { range : { timestamp : { lte : earliestMs , format : 'epoch_millis' , } , } , } , { range : { end_timestamp : { lte : earliestMs , format : 'epoch_millis' , } , } , } , ] , } , } , { bool : { filter : [ { range : { timestamp : { gte : latestMs , format : 'epoch_millis' , } , } , } , { range : { end_timestamp : { gte : latestMs , format : 'epoch_millis' , } , } , } , ] , } , } , ] , } , } ) ; } boolCriteria . push ( { exists : { field : 'annotation' } , } ) ; if ( event ) { boolCriteria . push ( { term : { event } , } ) ; } 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 , } , } ) ; } let shouldClauses ; if ( detectorIndex !== undefined && Array . isArray ( entities ) ) { const beExactMatch = [ ] ; beExactMatch . push ( { term : { detector_index : detectorIndex , } , } ) ; entities . forEach ( ( { fieldName , fieldType , fieldValue } ) => { beExactMatch . push ( { term : { [ getAnnotationFieldName ( fieldType ) ] : fieldName , } , } ) ; beExactMatch . push ( { term : { [ getAnnotationFieldValue ( fieldType ) ] : fieldValue , } , } ) ; } ) ; const haveAnyPartitionFields = [ ] ; PARTITION_FIELDS . forEach ( ( field ) => { haveAnyPartitionFields . push ( { exists : { field : getAnnotationFieldName ( field ) , } , } ) ; haveAnyPartitionFields . push ( { exists : { field : getAnnotationFieldValue ( field ) , } , } ) ; } ) ; shouldClauses = [ { bool : { must_not : haveAnyPartitionFields } } , { bool : { must : beExactMatch } } , ] ; } const params = { index : ML_ANNOTATIONS_INDEX_ALIAS_READ , size : maxAnnotations , track_total_hits : true , body : { query : { bool : { filter : [ { query_string : { query : `type:${ ANNOTATION_TYPE . ANNOTATION }` , analyze_wildcard : false , } , } , { bool : { must : boolCriteria , } , } , ] , ... ( shouldClauses ? { should : shouldClauses , minimum_should_match : 1 } : { } ) , } , } , } , } ; try { const { body } = await asInternalUser . search ( params ) ; if ( body . error !== undefined && body . message !== undefined ) { throw new Error ( `Annotations couldn't be retrieved from Elasticsearch.` ) ; } obj . totalCount = body . hits . total . value ; const docs = get ( body , [ 'hits' , 'hits' ] , [ ] ) . map ( ( d ) => { return { ... d . _source , event : d . _source ?. event ?? ANNOTATION_EVENT_USER , _id : d . _id , } as Annotation ; } ) ; if ( isAnnotations ( docs ) === false ) { throw new Error ( `Annotations didn't pass integrity check.` ) ; } docs . forEach ( ( doc ) => { const jobId = doc . job_id ; if ( typeof obj . annotations [ jobId ] === 'undefined' ) { obj . annotations [ jobId ] = [ ] ; } obj . annotations [ jobId ] . push ( doc ) ; } ) ; return obj ; } catch ( error ) { throw Boom . badRequest ( error ) ; } } param jobIds param earliestMs async function getDelayedDataAnnotations ( { jobIds , earliestMs , } ) { const params = { index : ML_ANNOTATIONS_INDEX_ALIAS_READ , size : 0 , body : { query : { bool : { filter : [ ... ( earliestMs ? [ { range : { end_timestamp : { gte : earliestMs } } } ] : [ ] ) , { term : { event : { value : 'delayed_data' } } , } , { terms : { job_id : jobIds } } , ] , } , } , aggs : { by_job : { terms : { field : 'job_id' , size : jobIds . length } , aggs : { latest_delayed : { top_hits : { size : 1 , sort : [ { end_timestamp : { order : 'desc' , } , } , ] , } , } , } , } , } , } , } ; const { body } = await asInternalUser . search < Annotation > ( params ) ; const annotations = ( ( body . aggregations ! . by_job as estypes . AggregationsTermsAggregateBase < AggByJob > ) . buckets as AggByJob [ ] ) . map ( ( bucket ) => { return bucket . latest_delayed . hits . hits [ 0 ] . _source ! ; } ) ; return annotations ; } async function deleteAnnotation ( id ) { const searchParams = { index : ML_ANNOTATIONS_INDEX_ALIAS_READ , size : 1 , body : { query : { ids : { values : [ id ] , } , } , } , } ; const { body } = await asInternalUser . search ( searchParams ) ; const totalCount = typeof body . hits . total === 'number' ? body . hits . total : body . hits . total ! . value ; if ( totalCount === 0 ) { throw Boom . notFound ( `Cannot find annotation with ID ${ id }` ) ; } const index = body . hits . hits [ 0 ] . _index ; const deleteParams = { index , id , refresh : 'wait_for' , } ; const { body : deleteResponse } = await asInternalUser . delete ( deleteParams ) ; return deleteResponse ; } return { getAnnotations , indexAnnotation , deleteAnnotation , getDelayedDataAnnotations , } ; } export type AnnotationService = ReturnType < typeof annotationProvider > ;