import { Subject , Observable , Subscription } from 'rxjs' ; import { filter } from 'rxjs/operators' ; import { Logger , ExecutionContextStart } from '../../../../src/core/server' ; import { Result , asErr , asOk } from './lib/result_type' ; import { TaskManagerConfig } from './config' ; import { asTaskManagerStatEvent , isTaskRunEvent , isTaskPollingCycleEvent } from './task_events' ; import { Middleware } from './lib/middleware' ; import { EphemeralTaskInstance } from './task' ; import { TaskTypeDictionary } from './task_type_dictionary' ; import { TaskLifecycleEvent } from './polling_lifecycle' ; import { EphemeralTaskManagerRunner } from './task_running/ephemeral_task_runner' ; import { TaskPool } from './task_pool' ; export interface EphemeralTaskLifecycleOpts { logger : Logger ; definitions : TaskTypeDictionary ; config : TaskManagerConfig ; middleware : Middleware ; elasticsearchAndSOAvailability$ : Observable < boolean > ; pool : TaskPool ; lifecycleEvent : Observable < TaskLifecycleEvent > ; executionContext : ExecutionContextStart ; } export type EphemeralTaskInstanceRequest = Omit < EphemeralTaskInstance , 'startedAt' > ; export class EphemeralTaskLifecycle { private definitions : TaskTypeDictionary ; private pool : TaskPool ; private lifecycleEvent : Observable < TaskLifecycleEvent > ; private events$ = new Subject < TaskLifecycleEvent > ( ) ; private ephemeralTaskQueue : Array < { task : EphemeralTaskInstanceRequest ; enqueuedAt : number ; } > = [ ] ; private logger : Logger ; private config : TaskManagerConfig ; private middleware : Middleware ; private lifecycleSubscription : Subscription = Subscription . EMPTY ; private readonly executionContext : ExecutionContextStart ; constructor ( { logger , middleware , definitions , pool , lifecycleEvent , config , executionContext , } ) { this . logger = logger ; this . middleware = middleware ; this . definitions = definitions ; this . pool = pool ; this . lifecycleEvent = lifecycleEvent ; this . config = config ; this . executionContext = executionContext ; if ( this . enabled ) { this . lifecycleSubscription = this . lifecycleEvent . pipe ( filter ( ( e ) => { const hasPollingCycleCompleted = isTaskPollingCycleEvent ( e ) ; if ( hasPollingCycleCompleted ) { this . emitEvent ( asTaskManagerStatEvent ( 'queuedEphemeralTasks' , asOk ( this . queuedTasks ) ) ) ; } return ( ( hasPollingCycleCompleted || isTaskRunEvent ( e ) ) && this . queuedTasks > 0 && this . getCapacity ( ) > 0 ) ; } ) ) . subscribe ( async ( e ) => { let overallCapacity = this . getCapacity ( ) ; const capacityByType = new Map < string , number > ( ) ; const tasksWithinCapacity = [ ... this . ephemeralTaskQueue ] . filter ( ( { task } ) => { if ( overallCapacity > 0 ) { if ( ! capacityByType . has ( task . taskType ) ) { capacityByType . set ( task . taskType , this . getCapacity ( task . taskType ) ) ; } if ( capacityByType . get ( task . taskType ) ! > 0 ) { overallCapacity -- ; capacityByType . set ( task . taskType , capacityByType . get ( task . taskType ) ! - 1 ) ; return true ; } } } ) . map ( ( ephemeralTask ) => { const index = this . ephemeralTaskQueue . indexOf ( ephemeralTask ) ; if ( index >= 0 ) { this . ephemeralTaskQueue . splice ( index , 1 ) ; } this . emitEvent ( asTaskManagerStatEvent ( 'ephemeralTaskDelay' , asOk ( Date . now ( ) - ephemeralTask . enqueuedAt ) ) ) ; return this . createTaskRunnerForTask ( ephemeralTask . task ) ; } ) ; if ( tasksWithinCapacity . length ) { this . pool . run ( tasksWithinCapacity ) . then ( ( successTaskPoolRunResult ) => { this . logger . debug ( `Successful ephemeral task lifecycle resulted in: ${ successTaskPoolRunResult }` ) ; } ) . catch ( ( error ) => { this . logger . debug ( `Failed ephemeral task lifecycle resulted in: ${ error }` ) ; } ) ; } } ) ; } } public get enabled ( ) : boolean { return this . config . ephemeral_tasks . enabled ; } public get events ( ) : Observable < TaskLifecycleEvent > { return this . events$ ; } private getCapacity = ( taskType ? ) => taskType && this . definitions . get ( taskType ) ?. maxConcurrency ? Math . max ( Math . min ( this . pool . availableWorkers , this . definitions . get ( taskType ) ! . maxConcurrency ! - this . pool . getOccupiedWorkersByType ( taskType ) ) , 0 ) : this . pool . availableWorkers ; private emitEvent = ( event ) => { this . events$ . next ( event ) ; } ; public attemptToRun ( task ) { if ( this . lifecycleSubscription . closed ) { return asErr ( task ) ; } return pushIntoSetWithTimestamp ( this . ephemeralTaskQueue , this . config . ephemeral_tasks . request_capacity , task ) ; } public get queuedTasks ( ) { return this . ephemeralTaskQueue . length ; } private createTaskRunnerForTask = ( instance ) : EphemeralTaskManagerRunner => { return new EphemeralTaskManagerRunner ( { logger : this . logger , instance : { ... instance , startedAt : new Date ( ) , } , definitions : this . definitions , beforeRun : this . middleware . beforeRun , beforeMarkRunning : this . middleware . beforeMarkRunning , onTaskEvent : this . emitEvent , executionContext : this . executionContext , } ) ; } ; } param set param maxCapacity param value function pushIntoSetWithTimestamp ( set , maxCapacity , task ) { if ( set . length >= maxCapacity ) { return asErr ( task ) ; } set . push ( { task , enqueuedAt : Date . now ( ) } ) ; return asOk ( task ) ; }