import { map , filter , startWith , buffer , share } from 'rxjs/operators' ; import { JsonObject } from '@kbn/utility-types' ; import { combineLatest , Observable , zip } from 'rxjs' ; import { isOk , Ok } from '../lib/result_type' ; import { AggregatedStat , AggregatedStatProvider } from './runtime_statistics_aggregator' ; import { EphemeralTaskLifecycle } from '../ephemeral_task_lifecycle' ; import { TaskLifecycleEvent } from '../polling_lifecycle' ; import { isTaskRunEvent , isTaskManagerStatEvent } from '../task_events' ; import { AveragedStat , calculateRunningAverage , createRunningAveragedStat , } from './task_run_calcultors' ; import { HealthStatus } from './monitoring_stats_stream' ; export interface EphemeralTaskStat extends JsonObject { queuedTasks : number [ ] ; executionsPerCycle : number [ ] ; load : number [ ] ; delay : number [ ] ; } export interface SummarizedEphemeralTaskStat extends JsonObject { queuedTasks : AveragedStat ; executionsPerCycle : AveragedStat ; load : AveragedStat ; } export function createEphemeralTaskAggregator ( ephemeralTaskLifecycle , runningAverageWindowSize , maxWorkers ) { const ephemeralTaskRunEvents$ = ephemeralTaskLifecycle . events . pipe ( filter ( ( taskEvent ) => isTaskRunEvent ( taskEvent ) ) ) ; const ephemeralQueueSizeEvents$ = ephemeralTaskLifecycle . events . pipe ( filter ( ( taskEvent ) => isTaskManagerStatEvent ( taskEvent ) && taskEvent . id === 'queuedEphemeralTasks' && isOk < number , never > ( taskEvent . event ) ) , map < TaskLifecycleEvent , number > ( ( taskEvent ) => { return ( taskEvent . event as unknown as Ok < number > ) . value ; } ) , share ( ) ) ; const ephemeralQueueExecutionsPerCycleQueue = createRunningAveragedStat < number > ( runningAverageWindowSize ) ; const ephemeralQueuedTasksQueue = createRunningAveragedStat < number > ( runningAverageWindowSize ) ; const ephemeralTaskLoadQueue = createRunningAveragedStat < number > ( runningAverageWindowSize ) ; const ephemeralPollingCycleBasedStats$ = zip ( ephemeralTaskRunEvents$ . pipe ( buffer ( ephemeralQueueSizeEvents$ ) , map ( ( taskEvents ) => taskEvents . length ) ) , ephemeralQueueSizeEvents$ ) . pipe ( map ( ( [ tasksRanSincePreviousQueueSize , ephemeralQueueSize ] ) => ( { queuedTasks : ephemeralQueuedTasksQueue ( ephemeralQueueSize ) , executionsPerCycle : ephemeralQueueExecutionsPerCycleQueue ( tasksRanSincePreviousQueueSize ) , load : ephemeralTaskLoadQueue ( calculateWorkerLoad ( maxWorkers , tasksRanSincePreviousQueueSize ) ) , } ) ) , startWith ( { queuedTasks : [ ] , executionsPerCycle : [ ] , load : [ ] , } ) ) ; const ephemeralTaskDelayQueue = createRunningAveragedStat < number > ( runningAverageWindowSize ) ; const ephemeralTaskDelayEvents$ = ephemeralTaskLifecycle . events . pipe ( filter ( ( taskEvent ) => isTaskManagerStatEvent ( taskEvent ) && taskEvent . id === 'ephemeralTaskDelay' && isOk < number , never > ( taskEvent . event ) ) , map < TaskLifecycleEvent , number [ ] > ( ( taskEvent ) => { return ephemeralTaskDelayQueue ( ( taskEvent . event as unknown as Ok < number > ) . value ) ; } ) , startWith ( [ ] ) ) ; return combineLatest ( [ ephemeralPollingCycleBasedStats$ , ephemeralTaskDelayEvents$ ] ) . pipe ( map ( ( [ stats , delay ] ) => { return { key : 'ephemeral' , value : { ... stats , delay } , } as AggregatedStat < EphemeralTaskStat > ; } ) ) ; } function calculateWorkerLoad ( maxWorkers , tasksExecuted ) { return Math . round ( ( tasksExecuted * 100 ) / maxWorkers ) ; } export function summarizeEphemeralStat ( { queuedTasks , executionsPerCycle , load , delay , } ) { return { value : { queuedTasks : calculateRunningAverage ( queuedTasks . length ? queuedTasks : [ 0 ] ) , load : calculateRunningAverage ( load . length ? load : [ 0 ] ) , executionsPerCycle : calculateRunningAverage ( executionsPerCycle . length ? executionsPerCycle : [ 0 ] ) , delay : calculateRunningAverage ( delay . length ? delay : [ 0 ] ) , } , status : HealthStatus . OK , } ; }