import moment from 'moment' ; import type { IScopedClusterClient } from 'kibana/server' ; import type { QueryDslQueryContainer } from '@elastic/elasticsearch/lib/api/typesWithBodyKey' ; import type * as estypes from '@elastic/elasticsearch/lib/api/typesWithBodyKey' ; import { ML_NOTIFICATION_INDEX_PATTERN } from '../../../common/constants/index_patterns' ; import { MESSAGE_LEVEL } from '../../../common/constants/message_levels' ; import type { JobSavedObjectService } from '../../saved_objects' ; import type { MlClient } from '../../lib/ml_client' ; import type { JobMessage } from '../../../common/types/audit_message' ; import { AuditMessage } from '../../../common/types/anomaly_detection_jobs' ; const SIZE = 1000 ; const LEVEL = { system_info : - 1 , info : 0 , warning : 1 , error : 2 } as const ; type LevelName = keyof typeof LEVEL ; type LevelValue = typeof LEVEL [ keyof typeof LEVEL ] ; const anomalyDetectorTypeFilter = { bool : { should : [ { term : { job_type : 'anomaly_detector' , } , } , { bool : { must_not : { exists : { field : 'job_type' , } , } , } , } , ] , minimum_should_match : 1 , } , } ; export function isClearable ( index ? ) { if ( typeof index === 'string' ) { const match = index . match ( /\d{6}$/ ) ; return match !== null && ! ! match . length && Number ( match [ match . length - 1 ] ) >= 2 ; } return false ; } export type JobsErrorsResponse = Array < { job_id : string ; errors : JobMessage [ ] ; } > ; export type JobAuditMessagesService = ReturnType < typeof jobAuditMessagesProvider > ; export function jobAuditMessagesProvider ( { asInternalUser } , mlClient ) { async function getJobAuditMessages ( jobSavedObjectService , { jobId , from , start , end , } ) { let gte = null ; if ( jobId !== undefined && from === undefined ) { const jobs = await mlClient . getJobs ( { job_id : jobId } ) ; if ( jobs . body . count > 0 && jobs . body . jobs !== undefined ) { gte = moment ( jobs . body . jobs [ 0 ] . create_time ) . valueOf ( ) ; } } else if ( from !== undefined ) { gte = `now-${ from }` ; } let timeFilter = { } ; if ( from !== null ) { timeFilter = { range : { timestamp : { gte , lte : 'now' , } , } , } ; } if ( start !== undefined && end !== undefined ) { timeFilter = { range : { timestamp : { gte : start , lte : end , } , } , } ; } const query = { bool : { filter : [ { bool : { must_not : { term : { level : 'activity' , } , } , } , } , anomalyDetectorTypeFilter , timeFilter , ] , } , } ; if ( jobId !== undefined ) { query . bool . filter . push ( { bool : { should : [ { term : { job_id : '' , } , } , { term : { job_id : jobId , } , } , ] , } , } ) ; } const { body } = await asInternalUser . search < JobMessage > ( { index : ML_NOTIFICATION_INDEX_PATTERN , ignore_unavailable : true , size : SIZE , body : { sort : [ { timestamp : { order : 'desc' } } , { job_id : { order : 'asc' } } ] , query , } , } ) ; let messages = [ ] ; const notificationIndices = [ ] ; if ( ( body . hits . total as estypes . SearchTotalHits ) . value > 0 ) { let notificationIndex ; body . hits . hits . forEach ( ( hit ) => { if ( notificationIndex !== hit . _index && isClearable ( hit . _index ) ) { notificationIndices . push ( hit . _index ) ; notificationIndex = hit . _index ; } messages . push ( hit . _source ! ) ; } ) ; } messages = await jobSavedObjectService . filterJobsForSpace < JobMessage > ( 'anomaly-detector' , messages , 'job_id' ) ; return { messages , notificationIndices } ; } param jobIds async function getAuditMessagesSummary ( jobIds ) { const maxBuckets = 10000 ; const levelsPerJobAggSize = Array . isArray ( jobIds ) && jobIds . length > 0 ? jobIds . length : maxBuckets ; const query = { bool : { filter : [ { range : { timestamp : { gte : 'now-1d' , } , } , } , anomalyDetectorTypeFilter , ... ( Array . isArray ( jobIds ) && jobIds . length > 0 ? [ { terms : { job_id : jobIds , } , } , ] : [ ] ) , ] , must_not : { term : { cleared : true , } , } , } , } ; const { body } = await asInternalUser . search ( { index : ML_NOTIFICATION_INDEX_PATTERN , ignore_unavailable : true , size : 0 , body : { query , aggs : { levelsPerJob : { terms : { field : 'job_id' , size : levelsPerJobAggSize , } , aggs : { levels : { terms : { field : 'level' , } , aggs : { latestMessage : { terms : { field : 'message.raw' , size : 1 , order : { latestMessage : 'desc' , } , } , aggs : { latestMessage : { max : { field : 'timestamp' , } , } , } , } , } , } , } , } , } , } , } ) ; interface LevelsPerJob { key : string ; levels : estypes . AggregationsTermsAggregateBase < { key : LevelName ; latestMessage : estypes . AggregationsTermsAggregateBase < { key : string ; latestMessage : estypes . AggregationsRateAggregate ; } > ; } > ; } let messagesPerJob = [ ] ; const jobMessages = [ ] ; const bodyAgg = body . aggregations as { levelsPerJob ? : estypes . AggregationsTermsAggregateBase < LevelsPerJob > ; } ; if ( ( body . hits . total as estypes . SearchTotalHits ) . value > 0 && bodyAgg && bodyAgg . levelsPerJob && bodyAgg . levelsPerJob . buckets && bodyAgg . levelsPerJob . buckets . length ) { messagesPerJob = bodyAgg . levelsPerJob . buckets as LevelsPerJob [ ] ; } messagesPerJob . forEach ( ( job ) => { if ( job . key !== '' && job . levels && job . levels . buckets && job . levels . buckets . length ) { let highestLevel = 0 ; let highestLevelText = '' ; let msgTime = 0 ; ( job . levels . buckets as Array < { key : LevelName ; latestMessage : estypes . AggregationsTermsAggregateBase < { key : string ; latestMessage : estypes . AggregationsRateAggregate ; } > ; } > ) . forEach ( ( level ) => { const label = level . key ; if ( LEVEL [ label ] > highestLevel ) { highestLevel = LEVEL [ label ] ; if ( level . latestMessage && level . latestMessage . buckets && level . latestMessage . buckets . length ) { ( level . latestMessage . buckets as Array < { key : string ; latestMessage : estypes . AggregationsRateAggregate ; } > ) . forEach ( ( msg ) => { highestLevelText = msg . key ; if ( msg . latestMessage && msg . latestMessage . value_as_string ) { const time = moment ( msg . latestMessage . value_as_string ) ; msgTime = time . valueOf ( ) ; } } ) ; } } } ) ; if ( msgTime !== 0 && highestLevel !== 0 ) { jobMessages . push ( { job_id : job . key , highestLevelText , highestLevel : levelToText ( highestLevel ) , msgTime , } ) ; } } } ) ; return jobMessages ; } const clearedTime = new Date ( ) . getTime ( ) ; async function clearJobAuditMessages ( jobId , notificationIndices ) { const newClearedMessage = { job_id : jobId , job_type : 'anomaly_detection' , level : MESSAGE_LEVEL . INFO , message : 'Cleared set to true for messages in the last 24hrs.' , timestamp : clearedTime , } ; const query = { bool : { filter : [ { range : { timestamp : { gte : 'now-24h' , } , } , } , { term : { job_id : jobId , } , } , ] , } , } ; const promises = [ asInternalUser . updateByQuery ( { index : notificationIndices . join ( ',' ) , ignore_unavailable : true , refresh : false , conflicts : 'proceed' , body : { query , script : { source : 'ctx._source.cleared = true' , lang : 'painless' , } , } , } ) , ... notificationIndices . map ( ( index ) => asInternalUser . index ( { index , body : newClearedMessage , refresh : 'wait_for' , } ) ) , ] ; await Promise . all ( promises ) ; return { success : true , last_cleared : clearedTime } ; } function levelToText ( level ) { return ( Object . keys ( LEVEL ) as LevelName [ ] ) [ Object . values ( LEVEL ) . indexOf ( level ) ] ; } param jobIds async function getJobsErrorMessages ( jobIds , earliestMs ? ) { const { body } = await asInternalUser . search ( { index : ML_NOTIFICATION_INDEX_PATTERN , ignore_unavailable : true , size : 0 , body : { query : { bool : { filter : [ ... ( earliestMs ? [ { range : { timestamp : { gte : earliestMs } } } ] : [ ] ) , { terms : { job_id : jobIds } } , { term : { level : { value : MESSAGE_LEVEL . ERROR } } , } , ] , } , } , aggs : { by_job : { terms : { field : 'job_id' , size : jobIds . length , } , aggs : { latest_errors : { top_hits : { size : 10 , sort : [ { timestamp : { order : 'desc' , } , } , ] , } , } , } , } , } , } , } ) ; const errors = body . aggregations ! . by_job as estypes . AggregationsTermsAggregateBase < { key : string ; doc_count : number ; latest_errors : Pick < estypes . SearchResponse < JobMessage > , 'hits' > ; } > ; return ( errors . buckets as Array < { key : string ; doc_count : number ; latest_errors : Pick < estypes . SearchResponse < JobMessage > , 'hits' > ; } > ) . map ( ( bucket ) => { return { job_id : bucket . key , errors : bucket . latest_errors . hits . hits . map ( ( v ) => v . _source ! ) , } ; } ) ; } return { getJobAuditMessages , getAuditMessagesSummary , clearJobAuditMessages , getJobsErrorMessages , } ; }