import { ReplaySubject } from 'rxjs' ; import { first } from 'rxjs/operators' ; import { CoreService } from '../../types' ; import { CoreContext } from '../core_context' ; import { Logger } from '../logging' ; import { InternalHttpServiceSetup } from '../http' ; import { InternalMetricsServiceSetup , InternalMetricsServiceStart , OpsMetrics } from './types' ; import { OpsMetricsCollector } from './ops_metrics_collector' ; import { opsConfig , OpsConfigType } from './ops_config' ; import { getEcsOpsMetricsLog } from './logging' ; export interface MetricsServiceSetupDeps { http : InternalHttpServiceSetup ; } internal export class MetricsService implements CoreService < InternalMetricsServiceSetup , InternalMetricsServiceStart > { private readonly logger : Logger ; private readonly opsMetricsLogger : Logger ; private metricsCollector ? : OpsMetricsCollector ; private collectInterval ? : NodeJS . Timeout ; private metrics$ = new ReplaySubject < OpsMetrics > ( 1 ) ; private service ? : InternalMetricsServiceSetup ; constructor ( private readonly coreContext ) { this . logger = coreContext . logger . get ( 'metrics' ) ; this . opsMetricsLogger = coreContext . logger . get ( 'metrics' , 'ops' ) ; } public async setup ( { http } ) { const config = await this . coreContext . configService . atPath < OpsConfigType > ( opsConfig . path ) . pipe ( first ( ) ) . toPromise ( ) ; this . metricsCollector = new OpsMetricsCollector ( http . server , { logger : this . logger , ... config . cGroupOverrides , } ) ; await this . refreshMetrics ( ) ; this . collectInterval = setInterval ( ( ) => { this . refreshMetrics ( ) ; } , config . interval . asMilliseconds ( ) ) ; const metricsObservable = this . metrics$ . asObservable ( ) ; this . service = { collectionInterval : config . interval . asMilliseconds ( ) , getOpsMetrics$ : ( ) => metricsObservable , } ; return this . service ; } public async start ( ) { if ( ! this . service ) { throw new Error ( '#setup() needs to be run first' ) ; } return this . service ; } private async refreshMetrics ( ) { const metrics = await this . metricsCollector ! . collect ( ) ; const { message , meta } = getEcsOpsMetricsLog ( metrics ) ; this . opsMetricsLogger . debug ( message ! , meta ) ; this . metricsCollector ! . reset ( ) ; this . metrics$ . next ( metrics ) ; } public async stop ( ) { if ( this . collectInterval ) { clearInterval ( this . collectInterval ) ; } this . metrics$ . complete ( ) ; } }