import apm from 'elastic-apm-node' ; import uuid from 'uuid' ; import { withSpan } from '@kbn/apm-utils' ; import { identity , defaults , flow } from 'lodash' ; import { Logger , SavedObjectsErrorHelpers , ExecutionContextStart , } from '../../../../../src/core/server' ; import { UsageCounter } from '../../../../../src/plugins/usage_collection/server' ; import { Middleware } from '../lib/middleware' ; import { asOk , asErr , mapErr , eitherAsync , unwrap , isOk , mapOk , Result , promiseResult , } from '../lib/result_type' ; import { TaskRun , TaskMarkRunning , asTaskRunEvent , asTaskMarkRunningEvent , startTaskTimer , TaskTiming , TaskPersistence , } from '../task_events' ; import { intervalFromDate , maxIntervalFromDate } from '../lib/intervals' ; import { CancelFunction , CancellableTask , ConcreteTaskInstance , isFailedRunResult , SuccessfulRunResult , FailedRunResult , FailedTaskResult , TaskDefinition , TaskStatus , } from '../task' ; import { TaskTypeDictionary } from '../task_type_dictionary' ; import { isUnrecoverableError } from './errors' ; const defaultBackoffPerFailure = 5 * 60 * 1000 ; export const EMPTY_RUN_RESULT = { state : { } } ; export const TASK_MANAGER_RUN_TRANSACTION_TYPE = 'task-run' ; export const TASK_MANAGER_TRANSACTION_TYPE = 'task-manager' ; export const TASK_MANAGER_TRANSACTION_TYPE_MARK_AS_RUNNING = 'mark-task-as-running' ; export interface TaskRunner { isExpired : boolean ; expiration : Date ; startedAt : Date | null ; definition : TaskDefinition ; cancel : CancelFunction ; markTaskAsRunning : ( ) => Promise < boolean > ; run : ( ) => Promise < Result < SuccessfulRunResult , FailedRunResult > > ; id : string ; taskExecutionId : string ; stage : string ; isEphemeral ? : boolean ; toString : ( ) => string ; isSameTask : ( executionId : string ) => boolean ; } export enum TaskRunningStage { PENDING = 'PENDING' , READY_TO_RUN = 'READY_TO_RUN' , RAN = 'RAN' } export interface TaskRunning < Stage extends TaskRunningStage , Instance > { timestamp : Date ; stage : Stage ; task : Instance ; } export interface Updatable { update ( doc ) : Promise < ConcreteTaskInstance > ; remove ( id ) : Promise < void > ; } type Opts = { logger : Logger ; definitions : TaskTypeDictionary ; instance : ConcreteTaskInstance ; store : Updatable ; onTaskEvent ? : ( event : TaskRun | TaskMarkRunning ) => void ; defaultMaxAttempts : number ; executionContext : ExecutionContextStart ; usageCounter ? : UsageCounter ; } & Pick < Middleware , 'beforeRun' | 'beforeMarkRunning' > ; export enum TaskRunResult { Success = 'Success' , SuccessRescheduled = 'Success' , RetryScheduled = 'RetryScheduled' , Failed = 'Failed' } export type ConcreteTaskInstanceWithStartedAt = ConcreteTaskInstance & { startedAt : Date ; } ; export type PendingTask = TaskRunning < TaskRunningStage . PENDING , ConcreteTaskInstance > ; export type ReadyToRunTask = TaskRunning < TaskRunningStage . READY_TO_RUN , ConcreteTaskInstanceWithStartedAt > ; export type RanTask = TaskRunning < TaskRunningStage . RAN , ConcreteTaskInstance > ; export type TaskRunningInstance = PendingTask | ReadyToRunTask | RanTask ; export class implements TaskRunner export class TaskManagerRunner implements TaskRunner { private task ? : CancellableTask ; private instance : TaskRunningInstance ; private definitions : TaskTypeDictionary ; private logger : Logger ; private bufferedTaskStore : Updatable ; private beforeRun : Middleware [ 'beforeRun' ] ; private beforeMarkRunning : Middleware [ 'beforeMarkRunning' ] ; private onTaskEvent : ( event : TaskRun | TaskMarkRunning ) => void ; private defaultMaxAttempts : number ; private uuid : string ; private readonly executionContext : ExecutionContextStart ; private usageCounter ? : UsageCounter ; param { Opts } opts prop prop prop prop prop memberof constructor ( { instance , definitions , logger , store , beforeRun , beforeMarkRunning , defaultMaxAttempts , onTaskEvent = identity , executionContext , usageCounter , } ) { this . instance = asPending ( sanitizeInstance ( instance ) ) ; this . definitions = definitions ; this . logger = logger ; this . bufferedTaskStore = store ; this . beforeRun = beforeRun ; this . beforeMarkRunning = beforeMarkRunning ; this . onTaskEvent = onTaskEvent ; this . defaultMaxAttempts = defaultMaxAttempts ; this . executionContext = executionContext ; this . usageCounter = usageCounter ; this . uuid = uuid . v4 ( ) ; } /**
     * Gets the id of this task instance.
     */ public get id ( ) { return this . instance . task . id ; } /**
     * Gets the execution id of this task instance.
     */ public get taskExecutionId ( ) { return `${ this . id }::${ this . uuid }` ; } param id public isSameTask ( executionId ) { return executionId . startsWith ( this . id ) ; } /**
     * Gets the task type of this task instance.
     */ public get taskType ( ) { return this . instance . task . taskType ; } /**
     * Get the stage this TaskRunner is at
     */ public get stage ( ) { return this . instance . stage ; } /**
     * Gets the task defintion from the dictionary.
     */ public get definition ( ) { return this . definitions . get ( this . taskType ) ; } /**
     * Gets the time at which this task will expire.
     */ public get expiration ( ) { return intervalFromDate ( isReadyToRun ( this . instance ) ? this . instance . task . startedAt : this . instance . timestamp , this . definition . timeout ) ! ; } /**
     * Gets the duration of the current task run
     */ public get startedAt ( ) { return this . instance . task . startedAt ; } /**
     * Gets whether or not this task has run longer than its expiration setting allows.
     */ public get isExpired ( ) { return this . expiration < new Date ( ) ; } /**
     * Returns a log-friendly representation of this task.
     */ public toString ( ) { return `${ this . taskType } "${ this . id }"` ; } returns { Promise < Result < SuccessfulRunResult , FailedRunResult > > } public async run ( ) { if ( ! isReadyToRun ( this . instance ) ) { throw new Error ( `Running task ${ this } failed as it ${ isPending ( this . instance ) ? `isn't ready to be ran` : `has already been ran` }` ) ; } this . logger . debug ( `Running task ${ this }` ) ; const apmTrans = apm . startTransaction ( this . taskType , TASK_MANAGER_RUN_TRANSACTION_TYPE , { childOf : this . instance . task . traceparent , } ) ; const modifiedContext = await this . beforeRun ( { taskInstance : this . instance . task , } ) ; const stopTaskTimer = startTaskTimer ( ) ; try { this . task = this . definition . createTaskRunner ( modifiedContext ) ; const ctx = { type : 'task manager' , name : `run ${ this . instance . task . taskType }` , id : this . instance . task . id , description : 'run task' , } ; const result = await this . executionContext . withContext ( ctx , ( ) => withSpan ( { name : 'run' , type : 'task manager' } , ( ) => this . task ! . run ( ) ) ) ; const validatedResult = this . validateResult ( result ) ; const processedResult = await withSpan ( { name : 'process result' , type : 'task manager' } , ( ) => this . processResult ( validatedResult , stopTaskTimer ( ) ) ) ; if ( apmTrans ) apmTrans . end ( 'success' ) ; return processedResult ; } catch ( err ) { this . logger . error ( `Task ${ this } failed: ${ err }` ) ; const processedResult = await withSpan ( { name : 'process result' , type : 'task manager' } , ( ) => this . processResult ( asErr ( { error : err , state : modifiedContext . taskInstance . state } ) , stopTaskTimer ( ) ) ) ; if ( apmTrans ) apmTrans . end ( 'failure' ) ; return processedResult ; } } returns { Promise < boolean > } public async markTaskAsRunning ( ) { if ( ! isPending ( this . instance ) ) { throw new Error ( `Marking task ${ this } as running has failed as it ${ isReadyToRun ( this . instance ) ? `is already running` : `has already been ran` }` ) ; } const apmTrans = apm . startTransaction ( TASK_MANAGER_TRANSACTION_TYPE_MARK_AS_RUNNING , TASK_MANAGER_TRANSACTION_TYPE ) ; apmTrans ?. addLabels ( { entityId : this . taskType } ) ; const now = new Date ( ) ; try { const { taskInstance } = await this . beforeMarkRunning ( { taskInstance : this . instance . task , } ) ; const attempts = taskInstance . attempts + 1 ; const ownershipClaimedUntil = taskInstance . retryAt ; const { id } = taskInstance ; const timeUntilClaimExpires = howManyMsUntilOwnershipClaimExpires ( ownershipClaimedUntil ) ; if ( timeUntilClaimExpires < 0 ) { this . logger . debug ( `[Task Runner] Task ${ id } started after ownership expired (${ Math . abs ( timeUntilClaimExpires ) }ms after expiry)` ) ; } this . instance = asReadyToRun ( ( await this . bufferedTaskStore . update ( { ... taskInstance , status : TaskStatus . Running , startedAt : now , attempts , retryAt : ( this . instance . task . schedule ? maxIntervalFromDate ( now , this . instance . task . schedule . interval , this . definition . timeout ) : this . getRetryDelay ( { attempts , error : new Error ( 'Task timeout' ) , addDuration : this . definition . timeout , } ) ) ?? null , } ) ) as ConcreteTaskInstanceWithStartedAt ) ; const timeUntilClaimExpiresAfterUpdate = howManyMsUntilOwnershipClaimExpires ( ownershipClaimedUntil ) ; if ( timeUntilClaimExpiresAfterUpdate < 0 ) { this . logger . debug ( `[Task Runner] Task ${ id } ran after ownership expired (${ Math . abs ( timeUntilClaimExpiresAfterUpdate ) }ms after expiry)` ) ; } if ( apmTrans ) apmTrans . end ( 'success' ) ; this . onTaskEvent ( asTaskMarkRunningEvent ( this . id , asOk ( this . instance . task ) ) ) ; return true ; } catch ( error ) { if ( apmTrans ) apmTrans . end ( 'failure' ) ; this . onTaskEvent ( asTaskMarkRunningEvent ( this . id , asErr ( error ) ) ) ; if ( ! SavedObjectsErrorHelpers . isConflictError ( error ) ) { if ( ! SavedObjectsErrorHelpers . isNotFoundError ( error ) ) { mapErr ( ( errReleaseClaim ) => { this . logger . error ( `[Task Runner] Task ${ this . id } failed to release claim after failure: ${ errReleaseClaim }` ) ; } , await this . releaseClaimAndIncrementAttempts ( ) ) ; } throw error ; } } return false ; } returns { Promise < void > } public async cancel ( ) { const { task } = this ; if ( task ?. cancel ) { this . task = undefined ; return task . cancel ( ) ; } this . logger . debug ( `The task ${ this } is not cancellable.` ) ; } private validateResult ( result ? ) { return isFailedRunResult ( result ) ? asErr ( { ... result , error : result . error } ) : asOk ( result || EMPTY_RUN_RESULT ) ; } private async releaseClaimAndIncrementAttempts ( ) { return promiseResult ( this . bufferedTaskStore . update ( { ... this . instance . task , status : TaskStatus . Idle , attempts : this . instance . task . attempts + 1 , startedAt : null , retryAt : null , ownerId : null , } ) ) ; } private shouldTryToScheduleRetry ( ) { if ( this . instance . task . schedule ) { return true ; } if ( this . isExpired ) { this . logger . warn ( `Skipping reschedule for task ${ this } due to the task expiring` ) ; return false ; } const maxAttempts = this . definition . maxAttempts || this . defaultMaxAttempts ; return this . instance . task . attempts < maxAttempts ; } private rescheduleFailedRun = ( failureResult ) : Result < SuccessfulRunResult , FailedTaskResult > => { const { state , error } = failureResult ; if ( this . shouldTryToScheduleRetry ( ) && ! isUnrecoverableError ( error ) ) { const { schedule , attempts } = this . instance . task ; const reschedule = failureResult . runAt ? { runAt : failureResult . runAt } : failureResult . schedule ? { schedule : failureResult . schedule } : schedule ? { schedule } : { runAt : this . getRetryDelay ( { attempts , error , } ) , } ; if ( reschedule . runAt || reschedule . schedule ) { return asOk ( { state , attempts , ... reschedule , } ) ; } } return asErr ( { status : TaskStatus . Failed } ) ; } ; private async processResultForRecurringTask ( result ) { const hasTaskRunFailed = isOk ( result ) ; const fieldUpdates = flow ( mapErr ( this . rescheduleFailedRun ) , mapOk ( ( { runAt , schedule : reschedule , state , attempts = 0 } ) => { const { startedAt , schedule } = this . instance . task ; return asOk ( { runAt : runAt || intervalFromDate ( startedAt ! , reschedule ?. interval ?? schedule ?. interval ) ! , state , schedule : reschedule ?? schedule , attempts , status : TaskStatus . Idle , } ) ; } ) , unwrap ) ( result ) ; if ( ! this . isExpired ) { this . instance = asRan ( await this . bufferedTaskStore . update ( defaults ( { ... fieldUpdates , startedAt : null , retryAt : null , ownerId : null , } , this . instance . task ) ) ) ; } else { this . usageCounter ?. incrementCounter ( { counterName : `taskManagerUpdateSkippedDueToTaskExpiration` , counterType : 'taskManagerTaskRunner' , incrementBy : 1 , } ) ; } return fieldUpdates . status === TaskStatus . Failed ? TaskRunResult . Failed : hasTaskRunFailed ? TaskRunResult . SuccessRescheduled : TaskRunResult . RetryScheduled ; } private async processResultWhenDone ( ) { try { await this . bufferedTaskStore . remove ( this . id ) ; this . instance = asRan ( this . instance . task ) ; } catch ( err ) { if ( err . statusCode === 404 ) { this . logger . warn ( `Task cleanup of ${ this } failed in processing. Was remove called twice?` ) ; } else { throw err ; } } return TaskRunResult . Success ; } private async processResult ( result , taskTiming ) { const { task } = this . instance ; await eitherAsync ( result , async ( { runAt , schedule } ) => { this . onTaskEvent ( asTaskRunEvent ( this . id , asOk ( { task , persistence : schedule || task . schedule ? TaskPersistence . Recurring : TaskPersistence . NonRecurring , result : await ( runAt || schedule || task . schedule ? this . processResultForRecurringTask ( result ) : this . processResultWhenDone ( ) ) , } ) , taskTiming ) ) ; } , async ( { error } ) => { this . onTaskEvent ( asTaskRunEvent ( this . id , asErr ( { task , persistence : task . schedule ? TaskPersistence . Recurring : TaskPersistence . NonRecurring , result : await this . processResultForRecurringTask ( result ) , error , } ) , taskTiming ) ) ; } ) ; return result ; } private getRetryDelay ( { error , attempts , addDuration , } ) { const retry = this . definition . getRetry ?. ( attempts , error ) ?? true ; let result ; if ( retry instanceof Date ) { result = retry ; } else if ( retry === true ) { result = new Date ( Date . now ( ) + attempts * defaultBackoffPerFailure ) ; } if ( addDuration && result ) { result = intervalFromDate ( result , addDuration ) ! ; } return result ; } } function sanitizeInstance ( instance ) { return { ... instance , params : instance . params || { } , state : instance . state || { } , } ; } function howManyMsUntilOwnershipClaimExpires ( ownershipClaimedUntil ) { return ownershipClaimedUntil ? ownershipClaimedUntil . getTime ( ) - Date . now ( ) : 0 ; } type InstanceOf < S extends TaskRunningStage , T > = T extends TaskRunning < S , infer I > ? I : never ; export function isPending ( taskRunning ) { return taskRunning . stage === TaskRunningStage . PENDING ; } export function asPending ( task ) { return { timestamp : new Date ( ) , stage : TaskRunningStage . PENDING , task , } ; } export function isReadyToRun ( taskRunning ) { return taskRunning . stage === TaskRunningStage . READY_TO_RUN ; } export function asReadyToRun ( task ) { return { timestamp : new Date ( ) , stage : TaskRunningStage . READY_TO_RUN , task , } ; } export function asRan ( task ) { return { timestamp : new Date ( ) , stage : TaskRunningStage . RAN , task , } ; }