import { Subject , Observable , Subscription } from 'rxjs' ; import { pipe } from 'fp-ts/lib/pipeable' ; import { Option , some , map as mapOptional } from 'fp-ts/lib/Option' ; import { tap } from 'rxjs/operators' ; import { UsageCounter } from '../../../../src/plugins/usage_collection/server' ; import type { Logger , ExecutionContextStart } from '../../../../src/core/server' ; import { Result , asErr , mapErr , asOk , map , mapOk } from './lib/result_type' ; import { ManagedConfiguration } from './lib/create_managed_configuration' ; import { TaskManagerConfig } from './config' ; import { TaskMarkRunning , TaskRun , TaskClaim , TaskRunRequest , asTaskRunRequestEvent , TaskPollingCycle , asTaskPollingCycleEvent , TaskManagerStat , asTaskManagerStatEvent , EphemeralTaskRejectedDueToCapacity , } from './task_events' ; import { fillPool , FillPoolResult , TimedFillPoolResult } from './lib/fill_pool' ; import { Middleware } from './lib/middleware' ; import { intervalFromNow } from './lib/intervals' ; import { ConcreteTaskInstance } from './task' ; import { createTaskPoller , PollingError , PollingErrorType , createObservableMonitor , } from './polling' ; import { TaskPool } from './task_pool' ; import { TaskManagerRunner , TaskRunner } from './task_running' ; import { TaskStore } from './task_store' ; import { identifyEsError , isEsCannotExecuteScriptError } from './lib/identify_es_error' ; import { BufferedTaskStore } from './buffered_task_store' ; import { TaskTypeDictionary } from './task_type_dictionary' ; import { delayOnClaimConflicts } from './polling' ; import { TaskClaiming , ClaimOwnershipResult } from './queries/task_claiming' ; export type TaskPollingLifecycleOpts = { logger : Logger ; definitions : TaskTypeDictionary ; taskStore : TaskStore ; config : TaskManagerConfig ; middleware : Middleware ; elasticsearchAndSOAvailability$ : Observable < boolean > ; executionContext : ExecutionContextStart ; usageCounter ? : UsageCounter ; } & ManagedConfiguration ; export type TaskLifecycleEvent = TaskMarkRunning | TaskRun | TaskClaim | TaskRunRequest | TaskPollingCycle | TaskManagerStat | EphemeralTaskRejectedDueToCapacity ; /**
 * The public interface into the task manager system.
 */ export class TaskPollingLifecycle { private definitions : TaskTypeDictionary ; private store : TaskStore ; private taskClaiming : TaskClaiming ; private bufferedStore : BufferedTaskStore ; private readonly executionContext : ExecutionContextStart ; private logger : Logger ; public pool : TaskPool ; private events$ = new Subject < TaskLifecycleEvent > ( ) ; private claimRequests$ = new Subject < Option < string > > ( ) ; private pollingSubscription : Subscription = Subscription . EMPTY ; private middleware : Middleware ; private usageCounter ? : UsageCounter ; /**
     * Initializes the task manager, preventing any further addition of middleware,
     * enabling the task manipulation methods, and beginning the background polling
     * mechanism.
     */ constructor ( { logger , middleware , maxWorkersConfiguration$ , pollIntervalConfiguration$ , elasticsearchAndSOAvailability$ , config , taskStore , definitions , executionContext , usageCounter , } ) { this . logger = logger ; this . middleware = middleware ; this . definitions = definitions ; this . store = taskStore ; this . executionContext = executionContext ; this . usageCounter = usageCounter ; const emitEvent = ( event ) => this . events$ . next ( event ) ; this . bufferedStore = new BufferedTaskStore ( this . store , { bufferMaxOperations : config . max_workers , logger , } ) ; this . pool = new TaskPool ( { logger , maxWorkers$ : maxWorkersConfiguration$ , } ) ; this . pool . load . subscribe ( emitEvent ) ; this . taskClaiming = new TaskClaiming ( { taskStore , maxAttempts : config . max_attempts , excludedTaskTypes : config . unsafe . exclude_task_types , definitions , logger : this . logger , 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 , } ) ; this . taskClaiming . events . subscribe ( emitEvent ) ; const { max_poll_inactivity_cycles : maxPollInactivityCycles , poll_interval : pollInterval } = config ; const pollIntervalDelay$ = delayOnClaimConflicts ( maxWorkersConfiguration$ , pollIntervalConfiguration$ , this . events$ , config . version_conflict_threshold , config . monitored_stats_running_average_window ) . pipe ( tap ( ( delay ) => emitEvent ( asTaskManagerStatEvent ( 'pollingDelay' , asOk ( delay ) ) ) ) ) ; const poller$ = createObservableMonitor < Result < TimedFillPoolResult , PollingError < string > > , Error > ( ( ) => createTaskPoller < string , TimedFillPoolResult > ( { logger , pollInterval$ : pollIntervalConfiguration$ , pollIntervalDelay$ , bufferCapacity : config . request_capacity , getCapacity : ( ) => { const capacity = this . pool . availableWorkers ; if ( ! capacity ) { this . emitEvent ( asTaskManagerStatEvent ( 'load' , asOk ( this . pool . workerLoad ) ) ) ; } return capacity ; } , pollRequests$ : this . claimRequests$ , work : this . pollForWork , workTimeout : pollInterval * maxPollInactivityCycles , } ) , { heartbeatInterval : pollInterval , inactivityTimeout : pollInterval * ( maxPollInactivityCycles + 1 ) , onError : ( error ) => { logger . error ( `[Task Poller Monitor]: ${ error . message }` ) ; } , } ) ; elasticsearchAndSOAvailability$ . subscribe ( ( areESAndSOAvailable ) => { if ( areESAndSOAvailable && ! this . isStarted ) { this . pollingSubscription = this . subscribeToPoller ( poller$ ) ; } else if ( ! areESAndSOAvailable && this . isStarted ) { this . pollingSubscription . unsubscribe ( ) ; this . pool . cancelRunningTasks ( ) ; } } ) ; } public get events ( ) : Observable < TaskLifecycleEvent > { return this . events$ ; } private emitEvent = ( event ) => { this . events$ . next ( event ) ; } ; public attemptToRun ( task ) { this . claimRequests$ . next ( some ( task ) ) ; } private createTaskRunnerForTask = ( instance ) => { return new TaskManagerRunner ( { logger : this . logger , instance , store : this . bufferedStore , definitions : this . definitions , beforeRun : this . middleware . beforeRun , beforeMarkRunning : this . middleware . beforeMarkRunning , onTaskEvent : this . emitEvent , defaultMaxAttempts : this . taskClaiming . maxAttempts , executionContext : this . executionContext , usageCounter : this . usageCounter , } ) ; } ; public get isStarted ( ) { return ! this . pollingSubscription . closed ; } private pollForWork = async ( ... tasksToClaim ) : Promise < TimedFillPoolResult > => { return fillPool ( ( ) => { return claimAvailableTasks ( tasksToClaim . splice ( 0 , this . pool . availableWorkers ) , this . taskClaiming , this . logger ) . pipe ( tap ( mapOk ( ( { timing } ) => { if ( timing ) { this . emitEvent ( asTaskManagerStatEvent ( 'claimDuration' , asOk ( timing . stop - timing . start ) ) ) ; } } ) ) ) ; } , this . createTaskRunnerForTask , async ( tasks ) => { const result = await this . pool . run ( tasks ) ; this . emitEvent ( asTaskManagerStatEvent ( 'load' , asOk ( this . pool . workerLoad ) ) ) ; return result ; } ) ; } ; private subscribeToPoller ( poller$ ) { return poller$ . pipe ( tap ( mapErr ( ( error ) => { if ( error . type === PollingErrorType . RequestCapacityReached ) { pipe ( error . data , mapOptional ( ( id ) => this . emitEvent ( asTaskRunRequestEvent ( id , asErr ( error ) ) ) ) ) ; } this . logger . error ( error . message ) ; } ) ) ) . subscribe ( ( result ) => { this . emitEvent ( map ( result , ( { timing , ... event } ) => asTaskPollingCycleEvent < string > ( asOk ( event ) , timing ) , ( event ) => asTaskPollingCycleEvent < string > ( asErr ( event ) ) ) ) ; } ) ; } } export function claimAvailableTasks ( claimTasksById , taskClaiming , logger ) { return new Observable ( ( observer ) => { taskClaiming . claimAvailableTasksIfCapacityIsAvailable ( { claimOwnershipUntil : intervalFromNow ( '30s' ) ! , claimTasksById , } ) . subscribe ( ( claimResult ) => { observer . next ( claimResult ) ; } , ( ex ) => { if ( isEsCannotExecuteScriptError ( ex ) ) { logger . warn ( `Task Manager cannot operate when inline scripts are disabled in Elasticsearch` ) ; observer . next ( asErr ( FillPoolResult . Failed ) ) ; observer . complete ( ) ; } else { const esError = identifyEsError ( ex ) ; observer . error ( esError . length > 0 ? esError : ex ) ; } } , ( ) => { observer . complete ( ) ; } ) ; } ) ; }