import { groupBy , keyBy , memoize } from 'lodash' ; import { KibanaRequest , Logger , SavedObjectsClientContract } from 'kibana/server' ; import { i18n } from '@kbn/i18n' ; import { MlJob } from '@elastic/elasticsearch/lib/api/typesWithBodyKey' ; import { MlClient } from '../ml_client' ; import { JobSelection } from '../../routes/schemas/alerting_schema' ; import { datafeedsProvider , DatafeedsService } from '../../models/job_service/datafeeds' ; import { ALL_JOBS_SELECTION , HEALTH_CHECK_NAMES } from '../../../common/constants/alerts' ; import { DatafeedStats } from '../../../common/types/anomaly_detection_jobs' ; import { GetGuards } from '../../shared_services/shared_services' ; import { AnomalyDetectionJobsHealthAlertContext , DelayedDataResponse , JobsErrorsResponse , JobsHealthExecutorOptions , MmlTestResponse , NotStartedDatafeedResponse , } from './register_jobs_monitoring_rule_type' ; import { getResultJobsHealthRuleConfig , resolveLookbackInterval , } from '../../../common/util/alerts' ; import { AnnotationService } from '../../models/annotation_service/annotation' ; import { annotationServiceProvider } from '../../models/annotation_service' ; import { parseInterval } from '../../../common/util/parse_interval' ; import { isDefined } from '../../../common/types/guards' ; import { jobAuditMessagesProvider , JobAuditMessagesService , } from '../../models/job_audit_messages/job_audit_messages' ; import type { FieldFormatsRegistryProvider } from '../../../common/types/kibana' ; interface TestResult { name : string ; context : AnomalyDetectionJobsHealthAlertContext ; } type TestsResults = TestResult [ ] ; export function jobsHealthServiceProvider ( mlClient , datafeedsService , annotationService , jobAuditMessagesService , getFieldsFormatRegistry , logger ) { /**
     * Provides a callback for date formatting based on the Kibana settings.
     */ const getFormatters = memoize ( async ( ) => { const fieldFormatsRegistry = await getFieldsFormatRegistry ( ) ; const dateFormatter = fieldFormatsRegistry . deserialize ( { id : 'date' } ) ; const bytesFormatter = fieldFormatsRegistry . deserialize ( { id : 'bytes' } ) ; return { dateFormatter : dateFormatter . convert . bind ( dateFormatter ) , bytesFormatter : bytesFormatter . convert . bind ( bytesFormatter ) , } ; } ) ; param includeJobs param excludeJobs const getResultJobs = async ( includeJobs , excludeJobs ? ) : Promise < MlJob [ ] > => { const jobAndGroupIds = [ ... ( includeJobs . jobIds ?? [ ] ) , ... ( includeJobs . groupIds ?? [ ] ) ] ; const includeAllJobs = jobAndGroupIds . some ( ( id ) => id === ALL_JOBS_SELECTION ) ; const jobsResponse = ( await mlClient . getJobs ( { ... ( includeAllJobs ? { } : { job_id : jobAndGroupIds } ) , } ) ) . body . jobs ; let resultJobs = jobsResponse ; if ( excludeJobs && ( ! ! excludeJobs . jobIds . length || ! ! excludeJobs ?. groupIds . length ) ) { const excludedJobAndGroupIds = [ ... ( excludeJobs ?. jobIds ?? [ ] ) , ... ( excludeJobs ?. groupIds ?? [ ] ) , ] ; const excludedJobsResponse = ( await mlClient . getJobs ( { job_id : excludedJobAndGroupIds , } ) ) . body . jobs ; const excludedJobsIds = new Set ( excludedJobsResponse . map ( ( v ) => v . job_id ) ) ; resultJobs = resultJobs . filter ( ( v ) => ! excludedJobsIds . has ( v . job_id ) ) ; } return resultJobs ; } ; param timeInterval param defaultLookbackInterval const getDelayedDataLookbackTimestamp = ( timeInterval , defaultLookbackInterval ) : number => { const currentTimestamp = Date . now ( ) ; const defaultLookbackTimestamp = currentTimestamp - parseInterval ( defaultLookbackInterval ) ! . asMilliseconds ( ) ; const customIntervalOffsetTimestamp = timeInterval ? currentTimestamp - parseInterval ( timeInterval ) ! . asMilliseconds ( ) : null ; return Math . min ( ... [ defaultLookbackTimestamp , customIntervalOffsetTimestamp ] . filter ( isDefined ) ) ; } ; const getJobIds = memoize ( ( jobs ) => jobs . map ( ( j ) => j . job_id ) ) ; const getDatafeeds = memoize ( datafeedsService . getDatafeedByJobId ) ; const getJobStats = memoize ( async ( jobIds ) => ( await mlClient . getJobStats ( { job_id : jobIds . join ( ',' ) } ) ) . body . jobs ) ; /** Gets values for translation string */ const getJobsAlertingMessageValues = < T extends Array < { job_id : string ; } | undefined > > ( results ) => { const jobIds = ( results || [ ] ) . filter ( isDefined ) . map ( ( v ) => v . job_id ) ; return { count : jobIds . length , jobsString : jobIds . join ( ', ' ) , } ; } ; return { param jobIds async getNotStartedDatafeeds ( jobIds ) { const datafeeds = await getDatafeeds ( jobIds ) ; if ( datafeeds ) { const jobsStats = await getJobStats ( jobIds ) ; const { body : { datafeeds : datafeedsStats } , } = await mlClient . getDatafeedStats ( { datafeed_id : datafeeds . map ( ( d ) => d . datafeed_id ) . join ( ',' ) , } ) ; return ( datafeedsStats as DatafeedStats [ ] ) . map ( ( datafeedStats ) => { const jobId = datafeedStats . timing_stats . job_id ; const jobState = jobsStats . find ( ( jobStats ) => jobStats . job_id === jobId ) ?. state ?? 'failed' ; return { datafeed_id : datafeedStats . datafeed_id , datafeed_state : datafeedStats . state , job_id : jobId , job_state : jobState , } ; } ) . filter ( ( datafeedStat ) => { return datafeedStat . job_state === 'opened' && datafeedStat . datafeed_state !== 'started' ; } ) ; } } , param jobIds async getMmlReport ( jobIds ) { const jobsStats = await getJobStats ( jobIds ) ; const { dateFormatter , bytesFormatter } = await getFormatters ( ) ; return jobsStats . filter ( ( j ) => j . state === 'opened' && j . model_size_stats . memory_status !== 'ok' ) . map ( ( { job_id : jobId , model_size_stats : modelSizeStats } ) => { return { job_id : jobId , memory_status : modelSizeStats . memory_status , log_time : dateFormatter ( modelSizeStats . log_time ) , model_bytes : bytesFormatter ( modelSizeStats . model_bytes ) , model_bytes_memory_limit : bytesFormatter ( modelSizeStats . model_bytes_memory_limit ) , peak_model_bytes : bytesFormatter ( modelSizeStats . peak_model_bytes ) , model_bytes_exceeded : bytesFormatter ( modelSizeStats . model_bytes_exceeded ) , } ; } ) ; } , param jobs param timeInterval param docsCount async getDelayedDataReport ( jobs , timeInterval , docsCount ) { const jobIds = getJobIds ( jobs ) ; const datafeeds = await getDatafeeds ( jobIds ) ; const datafeedsMap = keyBy ( datafeeds , 'job_id' ) ; const resultJobs = jobs . filter ( ( j ) => datafeedsMap [ j . job_id ] !== undefined ) ; const resultJobIds = getJobIds ( resultJobs ) ; const jobsMap = keyBy ( resultJobs , 'job_id' ) ; const defaultLookbackInterval = resolveLookbackInterval ( resultJobs , datafeeds ! ) ; const earliestMs = getDelayedDataLookbackTimestamp ( timeInterval , defaultLookbackInterval ) ; const { dateFormatter } = await getFormatters ( ) ; return ( await annotationService . getDelayedDataAnnotations ( { jobIds : resultJobIds , earliestMs , } ) ) . map ( ( v ) => { const match = v . annotation . match ( /Datafeed has missed (\d+)\s/ ) ; const missedDocsCount = match ? parseInt ( match [ 1 ] , 10 ) : 0 ; return { annotation : v . annotation , end_timestamp : v . end_timestamp ! , missed_docs_count : missedDocsCount , job_id : v . job_id , } ; } ) . filter ( ( v ) => { const job = jobsMap [ v . job_id ] ; const datafeed = datafeedsMap [ v . job_id ] ; const isDocCountExceededThreshold = docsCount ? v . missed_docs_count >= docsCount : true ; const jobLookbackInterval = resolveLookbackInterval ( [ job ] , [ datafeed ] ) ; const isEndTimestampWithinRange = v . end_timestamp > getDelayedDataLookbackTimestamp ( timeInterval , jobLookbackInterval ) ; return isDocCountExceededThreshold && isEndTimestampWithinRange ; } ) . map ( ( v ) => { return { ... v , end_timestamp : dateFormatter ( v . end_timestamp ) , } ; } ) ; } , param jobIds param previousStartedAt async getErrorsReport ( jobIds , previousStartedAt ) { const { dateFormatter } = await getFormatters ( ) ; return ( await jobAuditMessagesService . getJobsErrorMessages ( jobIds , previousStartedAt . getTime ( ) ) ) . map ( ( v ) => { return { ... v , errors : v . errors . map ( ( e ) => { return { ... e , timestamp : dateFormatter ( e . timestamp ) , } ; } ) , } ; } ) ; } , /**
         * Retrieves report grouped by test.
         */ async getTestsResults ( executorOptions ) { const { rule , previousStartedAt , params : { testsConfig , includeJobs , excludeJobs } , } = executorOptions ; const config = getResultJobsHealthRuleConfig ( testsConfig ) ; const results = [ ] ; const jobs = await getResultJobs ( includeJobs , excludeJobs ) ; const jobIds = getJobIds ( jobs ) ; if ( jobIds . length === 0 ) { logger . warn ( `Rule "${ rule . name }" does not have associated jobs.` ) ; return results ; } logger . debug ( `Performing health checks for job IDs: ${ jobIds . join ( ', ' ) }` ) ; if ( config . datafeed . enabled ) { const response = await this . getNotStartedDatafeeds ( jobIds ) ; if ( response && response . length > 0 ) { const { count , jobsString } = getJobsAlertingMessageValues ( response ) ; results . push ( { name : HEALTH_CHECK_NAMES . datafeed . name , context : { results : response , message : i18n . translate ( 'xpack.ml.alertTypes.jobsHealthAlertingRule.datafeedStateMessage' , { defaultMessage : 'Datafeed is not started for {count, plural, one {job} other {jobs}} {jobsString}' , values : { count , jobsString } , } ) , } , } ) ; } } if ( config . mml . enabled ) { const response = await this . getMmlReport ( jobIds ) ; if ( response && response . length > 0 ) { const { hard_limit : hardLimitJobs , soft_limit : softLimitJobs } = groupBy ( response , 'memory_status' ) ; const { count : hardLimitCount , jobsString : hardLimitJobsString } = getJobsAlertingMessageValues ( hardLimitJobs ) ; const { count : softLimitCount , jobsString : softLimitJobsString } = getJobsAlertingMessageValues ( softLimitJobs ) ; let message = '' ; if ( hardLimitCount > 0 ) { message = i18n . translate ( 'xpack.ml.alertTypes.jobsHealthAlertingRule.mmlMessage' , { defaultMessage : `{count, plural, one {Job} other {Jobs}} {jobsString} reached the hard model memory limit. Assign the job more memory and restore from a snapshot from prior to reaching the hard limit.` , values : { count : hardLimitCount , jobsString : hardLimitJobsString , } , } ) ; } if ( softLimitCount > 0 ) { if ( message . length > 0 ) { message += '\n' ; } message += i18n . translate ( 'xpack.ml.alertTypes.jobsHealthAlertingRule.mmlSoftLimitMessage' , { defaultMessage : '{count, plural, one {Job} other {Jobs}} {jobsString} reached the soft model memory limit. Assign the job more memory or edit the datafeed filter to limit scope of analysis.' , values : { count : softLimitCount , jobsString : softLimitJobsString , } , } ) ; } results . push ( { name : HEALTH_CHECK_NAMES . mml . name , context : { results : response , message , } , } ) ; } } if ( config . delayedData . enabled ) { const response = await this . getDelayedDataReport ( jobs , config . delayedData . timeInterval , config . delayedData . docsCount ) ; const { count , jobsString } = getJobsAlertingMessageValues ( response ) ; if ( response . length > 0 ) { results . push ( { name : HEALTH_CHECK_NAMES . delayedData . name , context : { results : response , message : i18n . translate ( 'xpack.ml.alertTypes.jobsHealthAlertingRule.delayedDataMessage' , { defaultMessage : '{count, plural, one {Job} other {Jobs}} {jobsString} {count, plural, one {is} other {are}} suffering from delayed data.' , values : { count , jobsString } , } ) , } , } ) ; } } if ( config . errorMessages . enabled && previousStartedAt ) { const response = await this . getErrorsReport ( jobIds , previousStartedAt ) ; if ( response . length > 0 ) { const { count , jobsString } = getJobsAlertingMessageValues ( response ) ; results . push ( { name : HEALTH_CHECK_NAMES . errorMessages . name , context : { results : response , message : i18n . translate ( 'xpack.ml.alertTypes.jobsHealthAlertingRule.errorMessagesMessage' , { defaultMessage : '{count, plural, one {Job} other {Jobs}} {jobsString} {count, plural, one {contains} other {contain}} errors in the messages.' , values : { count , jobsString } , } ) , } , } ) ; } } return results ; } , } ; } export type JobsHealthService = ReturnType < typeof jobsHealthServiceProvider > ; export function getJobsHealthServiceProvider ( getGuards ) { return { jobsHealthServiceProvider ( savedObjectsClient , request , logger ) { return { getTestsResults : async ( ... args ) : ReturnType < JobsHealthService [ 'getTestsResults' ] > => { return await getGuards ( request , savedObjectsClient ) . isFullLicense ( ) . hasMlCapabilities ( [ 'canGetJobs' ] ) . ok ( ( { mlClient , scopedClient , getFieldsFormatRegistry } ) => jobsHealthServiceProvider ( mlClient , datafeedsProvider ( scopedClient , mlClient ) , annotationServiceProvider ( scopedClient ) , jobAuditMessagesProvider ( scopedClient , mlClient ) , getFieldsFormatRegistry , logger ) . getTestsResults ( ... args ) ) ; } , } ; } , } ; } export type JobsHealthServiceProvider = ReturnType < typeof getJobsHealthServiceProvider > ;