import { isString , TSError , toString , pDelay , Assignment , toHumanTime } from '@terascope/job-components' ; import autoBind from 'auto-bind' ; import Client from './client' ; import { ClientConfig , SearchQuery , SearchOptions , PausedResponse , StoppedResponse , StopQuery , Execution , ResumeResponse , ExecutionStatus , StateErrors , RequestOptions , RecoverQuery , ControllerState , ClusterState , WorkerJobProcesses , ClusterProcess , ChangeWorkerResponse , ChangeWorkerQueryParams , JobIDResponse } from './interfaces' ; export default class Ex extends Client { private readonly _exId : string ; constructor ( config , exId ) { validateExId ( exId ) ; super ( config ) ; autoBind ( this ) ; this . _exId = exId ; } id ( ) { return this . _exId ; } async stop ( query ? , searchOptions = { } ) { const options = this . makeOptions ( query , searchOptions ) ; return this . post ( `/ex/${ this . _exId }/_stop` , null , options ) ; } async pause ( query ? , searchOptions = { } ) { const options = this . makeOptions ( query , searchOptions ) ; return this . post ( `/ex/${ this . _exId }/_pause` , null , options ) ; } async resume ( query ? , searchOptions = { } ) { const options = this . makeOptions ( query , searchOptions ) ; return this . post ( `/ex/${ this . _exId }/_resume` , null , options ) ; } async recover ( query = { } , searchOptions = { } ) { const options = this . makeOptions ( query , searchOptions ) ; const result = await this . post ( `/ex/${ this . _exId }/_recover` , null , options ) ; if ( ! result . ex_id ) { const { ex_id : exId } = await this . get ( `/jobs/${ result . job_id }/ex` ) ; return new Ex ( this . _config , exId ) ; } return new Ex ( this . _config , result . ex_id ) ; } async status ( requestOptions ? ) { const { _status : status } = await this . config ( requestOptions ) ; return status ; } async slicer ( requestOptions = { } ) { return this . get ( `/ex/${ this . _exId }/slicer` , requestOptions ) ; } async controller ( requestOptions = { } ) { return this . get ( `/ex/${ this . _exId }/controller` , requestOptions ) ; } async config ( requestOptions = { } ) { return this . get ( `/ex/${ this . _exId }` , requestOptions ) ; } async workers ( requestOptions = { } ) { const state = await this . get ( '/cluster/state' , requestOptions ) ; return filterProcesses < WorkerJobProcesses > ( state , this . _exId , 'worker' ) ; } async errors ( options ? ) { return this . get ( `/ex/${ this . _exId }/errors` , { searchParams : options , } as SearchOptions ) ; } async changeWorkers ( action , workerNum , requestOptions = { } ) { if ( action == null || workerNum == null ) { throw new TSError ( 'Change workers requires action and count' , { statusCode : 400 } ) ; } if ( ! [ 'add' , 'remove' , 'total' ] . includes ( action ) ) { throw new TSError ( 'Change workers requires action to be one of add, remove, or total' , { statusCode : 400 } ) ; } const query = { [ action ] : workerNum } ; requestOptions . responseType = 'text' ; const options = this . makeOptions ( query , requestOptions ) ; const response = await this . post ( `/ex/${ this . _exId }/_workers` , null , options ) ; if ( typeof response === 'string' ) { try { return this . parse ( response ) ; } catch ( err ) { } } return response ; } async waitForStatus ( target , intervalMs = 1000 , timeoutMs = 0 , requestOptions = { } ) { const terminal = { [ ExecutionStatus . terminated ] : true , [ ExecutionStatus . failed ] : true , [ ExecutionStatus . rejected ] : true , [ ExecutionStatus . completed ] : true , [ ExecutionStatus . stopped ] : true , } ; const startTime = Date . now ( ) ; const options = Object . assign ( { } , { responseType : 'json' , timeout : intervalMs < 1000 ? 1000 : intervalMs , } , requestOptions ) ; const checkStatus = async ( ) : Promise < ExecutionStatus > => { let result ; try { result = await this . status ( options ) ; } catch ( err ) { if ( /(timeout|timedout)/i . test ( toString ( err ) ) ) { await pDelay ( intervalMs ) ; return checkStatus ( ) ; } throw err ; } if ( result === target ) { return result ; } if ( terminal [ result ] ) { throw new TSError ( `Execution cannot reach the target status, "${ target }", because it is in the terminal state, "${ result }"` , { context : { lastStatus : result } } ) ; } const elapsed = Date . now ( ) - startTime ; if ( timeoutMs > 0 && elapsed >= timeoutMs ) { throw new TSError ( `Execution status failed to change from status "${ result }" to "${ target }" within ${ toHumanTime ( timeoutMs ) }` , { context : { lastStatus : result } } ) ; } await pDelay ( intervalMs ) ; return checkStatus ( ) ; } ; return checkStatus ( ) ; } } function validateExId ( exId ? ) { if ( ! exId ) { throw new TSError ( 'Ex requires exId' , { statusCode : 400 } ) ; } if ( ! isString ( exId ) ) { throw new TSError ( 'Ex requires exId to be a string' , { statusCode : 400 } ) ; } } function filterProcesses < T > ( state , exId , type ) { const results = [ ] ; for ( const node of Object . values ( state ) ) { node . active . forEach ( ( child ) => { const { assignment , ex_id : procExId } = child ; if ( ( assignment && assignment === type ) && ( procExId && procExId === exId ) ) { const jobProcess = Object . assign ( { } , child , { node_id : node . node_id } ) ; results . push ( jobProcess as any ) ; } } ) ; } return results ; }