import moment from 'moment' ; import { LevelLogger , ReportingStore } from '../' ; import { ReportingCore } from '../../' ; import { TaskManagerStartContract , TaskRunCreatorFunction } from '../../../../task_manager/server' ; import { numberToDuration } from '../../../common/schema_utils' ; import { ReportingConfigType } from '../../config' ; import { statuses } from '../statuses' ; import { SavedReport } from '../store' ; import { ReportingTask , ReportingTaskStatus , REPORTING_MONITOR_TYPE , ReportTaskParams } from './' ; export class MonitorReportsTask implements ReportingTask { public TYPE = REPORTING_MONITOR_TYPE ; private logger : LevelLogger ; private taskManagerStart ? : TaskManagerStartContract ; private store ? : ReportingStore ; private timeout : moment . Duration ; constructor ( private reporting , private config , parentLogger ) { this . logger = parentLogger . clone ( [ REPORTING_MONITOR_TYPE ] ) ; this . timeout = numberToDuration ( config . queue . timeout ) ; } private async getStore ( ) { if ( this . store ) { return this . store ; } const { store } = await this . reporting . getPluginStartDeps ( ) ; this . store = store ; return store ; } public async init ( taskManager ) { this . taskManagerStart = taskManager ; const scheduleInterval = Math . ceil ( numberToDuration ( this . config . queue . pollInterval ) . asSeconds ( ) ) + 's' ; this . logger . debug ( `Task to monitor for pending reports to run every ${ scheduleInterval }.` ) ; await taskManager . ensureScheduled ( { id : this . TYPE , taskType : this . TYPE , schedule : { interval : scheduleInterval } , state : { } , params : { } , } ) ; } private getTaskRunner ( ) { return ( ) => { return { run : async ( ) => { const reportingStore = await this . getStore ( ) ; try { const recoveredJob = await reportingStore . findStaleReportJob ( ) ; if ( ! recoveredJob ) { return ; } const { _id : jobId , _source : { process_expiration : processExpiration , status } , } = recoveredJob ; if ( ! [ statuses . JOB_STATUS_PENDING , statuses . JOB_STATUS_PROCESSING ] . includes ( status ) ) { throw new Error ( `Invalid job status in the monitoring search result: ${ status }` ) ; } if ( status === statuses . JOB_STATUS_PENDING ) { this . logger . info ( `${ jobId } was scheduled in a previous version and left in [${ status }] status. Rescheduling...` ) ; } if ( status === statuses . JOB_STATUS_PROCESSING ) { const expirationTime = moment ( processExpiration ) ; const overdueValue = moment ( ) . valueOf ( ) - expirationTime . valueOf ( ) ; this . logger . info ( `${ jobId } status is [${ status }] and the expiration time was [${ overdueValue }ms] ago. Rescheduling...` ) ; } const report = new SavedReport ( { ... recoveredJob , ... recoveredJob . _source } ) ; await reportingStore . prepareReportForRetry ( report ) ; await this . rescheduleTask ( report . toReportTaskJSON ( ) , this . logger ) ; } catch ( err ) { this . logger . error ( err ) ; } return ; } , cancel : async ( ) => ( { state : { } } ) , } ; } ; } public getTaskDefinition ( ) { return { type : REPORTING_MONITOR_TYPE , title : 'Reporting: monitor jobs' , createTaskRunner : this . getTaskRunner ( ) , maxAttempts : 1 , timeout : Math . ceil ( this . timeout . asSeconds ( ) ) + 's' , } ; } private async rescheduleTask ( task , logger ) { if ( ! this . taskManagerStart ) { throw new Error ( 'Reporting task runner has not been initialized!' ) ; } logger . info ( `Rescheduling task:${ task . id } to retry.` ) ; const newTask = await this . reporting . scheduleTask ( task ) ; this . reporting . getEventLogger ( { _id : task . id , ... task } , newTask ) . logRetry ( ) ; return newTask ; } public getStatus ( ) { if ( this . taskManagerStart ) { return ReportingTaskStatus . INITIALIZED ; } return ReportingTaskStatus . UNINITIALIZED ; } }