import { BehaviorSubject } from 'rxjs' ; import { ml } from '../../../../services/ml_api_service' ; import { mlJobService } from '../../../../services/job_service' ; import { JobCreator } from '../job_creator' ; import { DatafeedId , JobId } from '../../../../../../common/types/anomaly_detection_jobs' ; import { DATAFEED_STATE } from '../../../../../../common/constants/states' ; const REFRESH_INTERVAL_MS = 250 ; const NODE_ASSIGNMENT_CHECK_REFRESH_INTERVAL_MS = 2000 ; const TARGET_PROGRESS_DELTA = 2 ; const REFRESH_RATE_ADJUSTMENT_DELAY_MS = 2000 ; type Progress = number ; export type ProgressSubscriber = ( progress : number ) => void ; export type JobAssignmentSubscriber = ( assigned : boolean ) => void ; export class JobRunner { private _jobId : JobId ; private _datafeedId : DatafeedId ; private _start : number = 0 ; private _end : number = 0 ; private _datafeedState : DATAFEED_STATE = DATAFEED_STATE . STOPPED ; private _refreshInterval : number = REFRESH_INTERVAL_MS ; private _progress$ : BehaviorSubject < Progress > ; private _percentageComplete : Progress = 0 ; private _stopRefreshPoll : { stop : boolean ; } ; private _subscribers : ProgressSubscriber [ ] ; private _datafeedStartTime : number = 0 ; private _performRefreshRateAdjustment : boolean = false ; private _jobAssignedToNode : boolean = false ; private _jobAssignedToNode$ : BehaviorSubject < boolean > ; constructor ( jobCreator ) { this . _jobId = jobCreator . jobId ; this . _datafeedId = jobCreator . datafeedId ; this . _start = jobCreator . start ; this . _end = jobCreator . end ; this . _percentageComplete = 0 ; this . _stopRefreshPoll = jobCreator . stopAllRefreshPolls ; this . _progress$ = new BehaviorSubject ( this . _percentageComplete ) ; this . _jobAssignedToNode$ = new BehaviorSubject ( this . _jobAssignedToNode ) ; this . _subscribers = jobCreator . subscribers ; } public get datafeedState ( ) : DATAFEED_STATE { return this . _datafeedState ; } public set refreshInterval ( v ) { this . _refreshInterval = v ; } public resetInterval ( ) { this . _refreshInterval = REFRESH_INTERVAL_MS ; } private async openJob ( ) { try { const { node } = await mlJobService . openJob ( this . _jobId ) ; this . _jobAssignedToNode = node !== undefined && node . length > 0 ; this . _jobAssignedToNode$ . next ( this . _jobAssignedToNode ) ; } catch ( error ) { throw error ; } } private async _startDatafeed ( start , end , pollProgress ) { try { this . _datafeedStartTime = Date . now ( ) ; const subscriptions = pollProgress === true ? this . _subscribers . map ( ( s ) => this . _progress$ . subscribe ( s ) ) : [ ] ; await this . openJob ( ) ; const { started } = await mlJobService . startDatafeed ( this . _datafeedId , this . _jobId , start , end ) ; this . _datafeedState = DATAFEED_STATE . STARTED ; this . _percentageComplete = 0 ; const checkProgress = async ( ) => { const { isRunning , progress : prog , isJobClosed } = await this . getProgress ( ) ; const progress = prog === 100 && ( isRunning === true || isJobClosed === false ) ? prog - 1 : prog ; this . _adjustRefreshInterval ( progress ) ; this . _percentageComplete = progress ; this . _progress$ . next ( this . _percentageComplete ) ; if ( ( isRunning === true || isJobClosed === false ) && this . _stopRefreshPoll . stop === false ) { setTimeout ( async ( ) => { if ( this . _stopRefreshPoll . stop === false ) { await checkProgress ( ) ; } } , this . _refreshInterval ) ; } else { this . _progress$ . next ( 100 ) ; subscriptions . forEach ( ( s ) => s . unsubscribe ( ) ) ; } } ; const checkJobIsAssigned = async ( ) => { this . _jobAssignedToNode = await this . _isJobAssigned ( ) ; this . _jobAssignedToNode$ . next ( this . _jobAssignedToNode ) ; if ( this . _jobAssignedToNode === true ) { await checkProgress ( ) ; } else { setTimeout ( async ( ) => { if ( this . _stopRefreshPoll . stop === false ) { await checkJobIsAssigned ( ) ; } } , NODE_ASSIGNMENT_CHECK_REFRESH_INTERVAL_MS ) ; } } ; if ( pollProgress === true ) { if ( this . _jobAssignedToNode === true ) { await checkProgress ( ) ; } else { await checkJobIsAssigned ( ) ; } } return started ; } catch ( error ) { throw error ; } } private _adjustRefreshInterval ( progress ) { if ( this . _performRefreshRateAdjustment === false ) { const timeDeltaMs = Date . now ( ) - this . _datafeedStartTime ; if ( timeDeltaMs > REFRESH_RATE_ADJUSTMENT_DELAY_MS ) { this . _performRefreshRateAdjustment = true ; } else { return ; } } const progressDelta = progress - this . _percentageComplete ; if ( progressDelta !== 0 ) { this . _refreshInterval = Math . floor ( this . _refreshInterval * ( TARGET_PROGRESS_DELTA / progressDelta ) ) ; if ( this . _refreshInterval < REFRESH_INTERVAL_MS ) { this . _refreshInterval = REFRESH_INTERVAL_MS ; } } } private async _isJobAssigned ( ) { const { jobs } = await ml . getJobStats ( { jobId : this . _jobId } ) ; return jobs . length > 0 && jobs [ 0 ] . node !== undefined ; } public async startDatafeed ( ) { return await this . _startDatafeed ( this . _start , this . _end , true ) ; } public async startDatafeedInRealTime ( continueJob ) { const start = continueJob ? this . _end : this . _start ; return await this . _startDatafeed ( start , undefined , false ) ; } public async getProgress ( ) { return await ml . jobs . getLookBackProgress ( this . _jobId , this . _start , this . _end ) ; } public subscribeToProgress ( func ) { this . _progress$ . subscribe ( func ) ; } public async isRunning ( ) { const { isRunning } = await this . getProgress ( ) ; return isRunning ; } public isJobAssignedToNode ( ) { return this . _jobAssignedToNode ; } public subscribeToJobAssignment ( func ) { return this . _jobAssignedToNode$ . subscribe ( func ) ; } }