import { snakeCase } from 'lodash' ; import { Logger , LegacyAPICaller , OpenSearchClient } from 'opensearch-dashboards/server' ; import { Collector , CollectorOptions } from './collector' ; import { UsageCollector } from './usage_collector' ; interface CollectorSetConfig { logger : Logger ; maximumWaitTimeForAllCollectorsInS ? : number ; collectors ? : Array < Collector < any , any > > ; } export class CollectorSet { private _waitingForAllCollectorsTimestamp ? : number ; private readonly logger : Logger ; private readonly maximumWaitTimeForAllCollectorsInS : number ; private readonly collectors : Map < string , Collector < any , any > > ; constructor ( { logger , maximumWaitTimeForAllCollectorsInS , collectors = [ ] } ) { this . logger = logger ; this . collectors = new Map ( collectors . map ( ( collector ) => [ collector . type , collector ] ) ) ; this . maximumWaitTimeForAllCollectorsInS = maximumWaitTimeForAllCollectorsInS || 60 ; } public makeStatsCollector = < T , U > ( options ) => { return new Collector ( this . logger , options ) ; } ; public makeUsageCollector = < T , U = T > ( options ) => { return new UsageCollector ( this . logger , options ) ; } ; public registerCollector = < T , U > ( collector ) => { if ( ! ( collector instanceof Collector ) ) { throw new Error ( 'CollectorSet can only have Collector instances registered' ) ; } if ( this . collectors . get ( collector . type ) ) { throw new Error ( `Usage collector's type "${ collector . type }" is duplicated.` ) ; } this . collectors . set ( collector . type , collector ) ; if ( collector . init ) { this . logger . debug ( `Initializing ${ collector . type } collector` ) ; collector . init ( ) ; } } ; public getCollectorByType = ( type ) => { return [ ... this . collectors . values ( ) ] . find ( ( c ) => c . type === type ) ; } ; public isUsageCollector = ( x ) : x is UsageCollector => { return x instanceof UsageCollector ; } ; public areAllCollectorsReady = async ( collectorSet = this ) => { if ( ! ( collectorSet instanceof CollectorSet ) ) { throw new Error ( `areAllCollectorsReady method given bad collectorSet parameter: ` + typeof collectorSet ) ; } const collectors = [ ... collectorSet . collectors . values ( ) ] ; const collectorsWithStatus = await Promise . all ( collectors . map ( async ( collector ) => { return { isReady : await collector . isReady ( ) , collector , } ; } ) ) ; const collectorsTypesNotReady = collectorsWithStatus . filter ( ( collectorWithStatus ) => collectorWithStatus . isReady === false ) . map ( ( collectorWithStatus ) => collectorWithStatus . collector . type ) ; const allReady = collectorsTypesNotReady . length === 0 ; if ( ! allReady && this . maximumWaitTimeForAllCollectorsInS >= 0 ) { const nowTimestamp = + new Date ( ) ; this . _waitingForAllCollectorsTimestamp = this . _waitingForAllCollectorsTimestamp || nowTimestamp ; const timeWaitedInMS = nowTimestamp - this . _waitingForAllCollectorsTimestamp ; const timeLeftInMS = this . maximumWaitTimeForAllCollectorsInS * 1000 - timeWaitedInMS ; if ( timeLeftInMS <= 0 ) { this . logger . debug ( `All collectors are not ready (waiting for ${ collectorsTypesNotReady . join ( ',' ) }) ` + `but we have waited the required ` + `${ this . maximumWaitTimeForAllCollectorsInS }s and will return data from all collectors that are ready.` ) ; return true ; } else { this . logger . debug ( `All collectors are not ready. Waiting for ${ timeLeftInMS }ms longer.` ) ; } } else { this . _waitingForAllCollectorsTimestamp = undefined ; } return allReady ; } ; public bulkFetch = async ( callCluster , opensearchClient , collectors = this . collectors ) => { const responses = await Promise . all ( [ ... collectors . values ( ) ] . map ( async ( collector ) => { this . logger . debug ( `Fetching data from ${ collector . type } collector` ) ; try { return { type : collector . type , result : await collector . fetch ( callCluster , opensearchClient ) , } ; } catch ( err ) { this . logger . warn ( err ) ; this . logger . warn ( `Unable to fetch data from ${ collector . type } collector` ) ; } } ) ) ; return responses . filter ( ( response ) : response is { type : string ; result : unknown ; } => typeof response !== 'undefined' ) ; } ; public getFilteredCollectorSet = ( filter ) => { const filtered = [ ... this . collectors . values ( ) ] . filter ( filter ) ; return this . makeCollectorSetFromArray ( filtered ) ; } ; public bulkFetchUsage = async ( callCluster , opensearchClient ) => { const usageCollectors = this . getFilteredCollectorSet ( ( c ) => c instanceof UsageCollector ) ; return await this . bulkFetch ( callCluster , opensearchClient , usageCollectors . collectors ) ; } ; public toObject = < Result , T > ( statsData = [ ] ) => { return statsData . reduce < Result > ( ( accumulatedStats , { type , result } ) => { return { ... accumulatedStats , [ type ] : result , } ; } , { } as Result ) ; } ; public toApiFieldNames = ( apiData ) : any => { const getValueOrRecurse = ( value ) => { if ( value == null || typeof value !== 'object' ) { return value ; } else { return this . toApiFieldNames ( value ) ; } } ; if ( Array . isArray ( apiData ) ) { return apiData . map ( getValueOrRecurse ) ; } return Object . keys ( apiData ) . reduce ( ( accum , field ) => { const value = apiData [ field ] ; let newName = field ; newName = snakeCase ( newName ) ; newName = newName . replace ( /^(1|5|15)_m/ , '$1m' ) ; newName = newName . replace ( '_in_bytes' , '_bytes' ) ; newName = newName . replace ( '_in_millis' , '_ms' ) ; return { ... accum , [ newName ] : getValueOrRecurse ( value ) , } ; } , { } ) ; } ; public map = ( mapFn ) => { return [ ... this . collectors . values ( ) ] . map ( mapFn ) ; } ; public some = ( someFn ) => { return [ ... this . collectors . values ( ) ] . some ( someFn ) ; } ; private makeCollectorSetFromArray = ( collectors ) => { return new CollectorSet ( { logger : this . logger , maximumWaitTimeForAllCollectorsInS : this . maximumWaitTimeForAllCollectorsInS , collectors , } ) ; } ; }