import { Observable , Subject } from 'rxjs' ; import moment , { Duration } from 'moment' ; import { padStart } from 'lodash' ; import { Logger } from '../../../../src/core/server' ; import { TaskRunner } from './task_running' ; import { isTaskSavedObjectNotFoundError } from './lib/is_task_not_found_error' ; import { TaskManagerStat } from './task_events' ; interface Opts { maxWorkers$ : Observable < number > ; logger : Logger ; } export enum TaskPoolRunResult { NoTaskWereRan = 'NoTaskWereRan' , RunningAllClaimedTasks = 'RunningAllClaimedTasks' , RunningAtCapacity = 'RunningAtCapacity' , RanOutOfCapacity = 'RanOutOfCapacity' } const VERSION_CONFLICT_MESSAGE = 'Task has been claimed by another Kibana service' ; /**
 * Runs tasks in batches, taking costs into account.
 */ export class TaskPool { private maxWorkers : number = 0 ; private tasksInPool = new Map < string , TaskRunner > ( ) ; private logger : Logger ; private load$ = new Subject < TaskManagerStat > ( ) ; param { Opts } opts prop prop constructor ( opts ) { this . logger = opts . logger ; opts . maxWorkers$ . subscribe ( ( maxWorkers ) => { this . logger . debug ( `Task pool now using ${ maxWorkers } as the max worker value` ) ; this . maxWorkers = maxWorkers ; } ) ; } public get load ( ) : Observable < TaskManagerStat > { return this . load$ ; } /**
     * Gets how many workers are currently in use.
     */ public get occupiedWorkers ( ) { return this . tasksInPool . size ; } /**
     * Gets % of workers in use
     */ public get workerLoad ( ) { return this . maxWorkers ? Math . round ( ( this . occupiedWorkers * 100 ) / this . maxWorkers ) : 100 ; } /**
     * Gets how many workers are currently available.
     */ public get availableWorkers ( ) { this . cancelExpiredTasks ( ) ; return this . maxWorkers - this . occupiedWorkers ; } /**
     * Gets how many workers are currently in use by type.
     */ public getOccupiedWorkersByType ( type ) { return [ ... this . tasksInPool . values ( ) ] . reduce ( ( count , runningTask ) => ( runningTask . definition . type === type ? ++ count : count ) , 0 ) ; } param { TaskRunner [ ] } tasks returns { Promise < boolean > } public run = async ( tasks ) : Promise < TaskPoolRunResult > => { const [ tasksToRun , leftOverTasks ] = partitionListByCount ( tasks , this . availableWorkers ) ; if ( tasksToRun . length ) { await Promise . all ( tasksToRun . filter ( ( taskRunner ) => ! Array . from ( this . tasksInPool . keys ( ) ) . some ( ( executionId ) => taskRunner . isSameTask ( executionId ) ) ) . map ( async ( taskRunner ) => { this . tasksInPool . set ( taskRunner . taskExecutionId , taskRunner ) ; return taskRunner . markTaskAsRunning ( ) . then ( ( hasTaskBeenMarkAsRunning ) => hasTaskBeenMarkAsRunning ? this . handleMarkAsRunning ( taskRunner ) : this . handleFailureOfMarkAsRunning ( taskRunner , { name : 'TaskPoolVersionConflictError' , message : VERSION_CONFLICT_MESSAGE , } ) ) . catch ( ( err ) => this . handleFailureOfMarkAsRunning ( taskRunner , err ) ) ; } ) ) ; } if ( leftOverTasks . length ) { if ( this . availableWorkers ) { return this . run ( leftOverTasks ) ; } return TaskPoolRunResult . RanOutOfCapacity ; } else if ( ! this . availableWorkers ) { return TaskPoolRunResult . RunningAtCapacity ; } return TaskPoolRunResult . RunningAllClaimedTasks ; } ; public cancelRunningTasks ( ) { this . logger . debug ( 'Cancelling running tasks.' ) ; for ( const task of this . tasksInPool . values ( ) ) { this . cancelTask ( task ) ; } } private handleMarkAsRunning ( taskRunner ) { taskRunner . run ( ) . catch ( ( err ) => { const errorLogLine = `Task ${ taskRunner . toString ( ) } failed in attempt to run: ${ err . message }` ; if ( isTaskSavedObjectNotFoundError ( err , taskRunner . id ) ) { this . logger . debug ( errorLogLine ) ; } else { this . logger . warn ( errorLogLine ) ; } } ) . then ( ( ) => { this . tasksInPool . delete ( taskRunner . taskExecutionId ) ; } ) ; } private handleFailureOfMarkAsRunning ( task , err ) { this . tasksInPool . delete ( task . taskExecutionId ) ; this . logger . error ( `Failed to mark Task ${ task . toString ( ) } as running: ${ err . message }` ) ; } private cancelExpiredTasks ( ) { for ( const taskRunner of this . tasksInPool . values ( ) ) { if ( taskRunner . isExpired ) { this . logger . warn ( `Cancelling task ${ taskRunner . toString ( ) } as it expired at ${ taskRunner . expiration . toISOString ( ) }${ taskRunner . startedAt ? ` after running for ${ durationAsString ( moment . duration ( moment ( new Date ( ) ) . utc ( ) . diff ( taskRunner . startedAt ) ) ) }` : `` }${ taskRunner . definition . timeout ? ` (with timeout set at ${ taskRunner . definition . timeout })` : `` }.` ) ; this . cancelTask ( taskRunner ) ; } } } private async cancelTask ( task ) { try { this . logger . debug ( `Cancelling task ${ task . toString ( ) }.` ) ; this . tasksInPool . delete ( task . taskExecutionId ) ; await task . cancel ( ) ; } catch ( err ) { this . logger . error ( `Failed to cancel task ${ task . toString ( ) }: ${ err }` ) ; } } } function partitionListByCount < T > ( list , count ) { const listInCount = list . splice ( 0 , count ) ; return [ listInCount , list ] ; } function durationAsString ( duration ) { const [ m , s ] = [ duration . minutes ( ) , duration . seconds ( ) ] . map ( ( value ) => padStart ( `${ value }` , 2 , '0' ) ) ; return `${ m }m ${ s }s` ; }