import { Batch , BatchConfig , BatchTask , JobTask , Message , PoolLoadGenerator , PoolLoadSnapshot , Queue , StorageConfig } from 'azure-services' ; import { ServiceConfiguration , System } from 'common' ; import { inject , injectable } from 'inversify' ; import { isNil , mergeWith } from 'lodash' ; import { GlobalLogger } from 'logger' ; import { BatchPoolLoadSnapshotProvider , BatchTaskCreator , OnDemandPageScanRunResultProvider , ScanMessage } from 'service-library' ; import { OnDemandScanRequestMessage , StorageDocument } from 'storage-documents' ; @ injectable ( ) export class Worker extends BatchTaskCreator { public constructor ( @ inject ( Batch ) batch , @ inject ( Queue ) queue , @ inject ( PoolLoadGenerator ) private readonly poolLoadGenerator , @ inject ( BatchPoolLoadSnapshotProvider ) private readonly batchPoolLoadSnapshotProvider , @ inject ( OnDemandPageScanRunResultProvider ) private readonly onDemandPageScanRunResultProvider , @ inject ( BatchConfig ) batchConfig , @ inject ( ServiceConfiguration ) serviceConfig , @ inject ( StorageConfig ) private readonly storageConfig , @ inject ( GlobalLogger ) logger , system = System ) { super ( batch , queue , batchConfig , serviceConfig , logger , system ) ; } public getQueueName ( ) { return this . storageConfig . scanQueue ; } public async getMessagesForTaskCreation ( ) { const poolMetricsInfo = await this . batch . getPoolMetricsInfo ( ) ; const poolLoadSnapshot = await this . poolLoadGenerator . getPoolLoadSnapshot ( poolMetricsInfo ) ; await this . writePoolLoadSnapshot ( poolLoadSnapshot ) ; let messages = [ ] ; const queueName = this . getQueueName ( ) ; if ( poolLoadSnapshot . tasksIncrementCountPerInterval > 0 ) { const queueMessages = await this . queue . getMessagesWithTotalCount ( queueName , poolLoadSnapshot . tasksIncrementCountPerInterval ) ; if ( queueMessages ?. length > 0 ) { messages = this . convertToScanMessages ( queueMessages ) ; } } this . logger . logInfo ( 'Pool load statistics' , mergeWith ( { } , poolLoadSnapshot , ( t , s ) => s !== undefined ? ( s . constructor . name !== 'Date' ? s . toString ( ) : s . toJSON ( ) ) : 'undefined' ) as any ) ; this . logger . trackEvent ( 'BatchPoolStats' , null , { runningTasks : poolMetricsInfo . load . runningTasks , samplingIntervalInSeconds : poolLoadSnapshot . samplingIntervalInSeconds , maxParallelTasks : poolMetricsInfo . maxTasksPerPool , } ) ; return messages ; } public async onTasksAdded ( tasks ) { this . poolLoadGenerator . setLastTasksIncrementCount ( tasks . length ) ; } public async handleFailedTasks ( failedTasks ) { await Promise . all ( failedTasks . map ( async ( failedTask ) => { const taskArguments = JSON . parse ( failedTask . taskArguments ) as OnDemandScanRequestMessage ; if ( ! isNil ( taskArguments ?. id ) ) { let error = `Task was terminated unexpectedly. Exit code: ${ failedTask . exitCode }` ; error = failedTask . failureInfo !== undefined ? `${ error }, Error category: ${ failedTask . failureInfo . category }, Error details: ${ failedTask . failureInfo . message }` : error ; const pageScanResult = await this . onDemandPageScanRunResultProvider . readScanRun ( taskArguments . id ) ; if ( pageScanResult !== undefined ) { pageScanResult . run = { state : 'failed' , timestamp : failedTask . timestamp . toJSON ( ) , error , } ; await this . onDemandPageScanRunResultProvider . updateScanRun ( pageScanResult ) ; } else { this . logger . logError ( `Unable to find corresponding scan document in a result storage.` , { correlatedBatchTaskId : failedTask . id , taskProperties : JSON . stringify ( failedTask ) , } ) ; } this . logger . logError ( error , { correlatedBatchTaskId : failedTask . id , taskProperties : JSON . stringify ( failedTask ) } ) ; } else { this . logger . logError ( `Unable to update failed scan run result. Task has no scan id run arguments defined.` , { correlatedBatchTaskId : failedTask . id , taskProperties : JSON . stringify ( failedTask ) , } ) ; } } ) ) ; } private async writePoolLoadSnapshot ( poolLoadSnapshot ) { await this . batchPoolLoadSnapshotProvider . writeBatchPoolLoadSnapshot ( { ... ( { } as StorageDocument ) , batchAccountName : this . batchConfig . accountName , ... poolLoadSnapshot , } ) ; } private convertToScanMessages ( messages ) { return messages . map ( ( message ) => { return { scanId : this . parseMessageBody ( message ) . id , messageId : message . messageId , message : message , } ; } ) ; } private parseMessageBody ( message ) { return message . messageText === undefined ? undefined : ( JSON . parse ( message . messageText ) as OnDemandScanRequestMessage ) ; } }