import { interval , merge , of , Observable } from 'rxjs' ; import { filter , mergeScan , map , scan , distinctUntilChanged , startWith } from 'rxjs/operators' ; import { SavedObjectsErrorHelpers } from '../../../../../src/core/server' ; import { Logger } from '../../../../../src/core/server' ; import { isEsCannotExecuteScriptError } from './identify_es_error' ; const FLUSH_MARKER = Symbol ( 'flush' ) ; export const ADJUST_THROUGHPUT_INTERVAL = 10 * 1000 ; const MAX_WORKERS_DECREASE_PERCENTAGE = 0.8 ; const MAX_WORKERS_INCREASE_PERCENTAGE = 1.05 ; const POLL_INTERVAL_DECREASE_PERCENTAGE = 0.95 ; const POLL_INTERVAL_INCREASE_PERCENTAGE = 1.2 ; interface ManagedConfigurationOpts { logger : Logger ; startingMaxWorkers : number ; startingPollInterval : number ; errors$ : Observable < Error > ; } export interface ManagedConfiguration { maxWorkersConfiguration$ : Observable < number > ; pollIntervalConfiguration$ : Observable < number > ; } export function createManagedConfiguration ( { logger , startingMaxWorkers , startingPollInterval , errors$ , } ) { const errorCheck$ = countErrors ( errors$ , ADJUST_THROUGHPUT_INTERVAL ) ; return { maxWorkersConfiguration$ : errorCheck$ . pipe ( createMaxWorkersScan ( logger , startingMaxWorkers ) , startWith ( startingMaxWorkers ) , distinctUntilChanged ( ) ) , pollIntervalConfiguration$ : errorCheck$ . pipe ( createPollIntervalScan ( logger , startingPollInterval ) , startWith ( startingPollInterval ) , distinctUntilChanged ( ) ) , } ; } function createMaxWorkersScan ( logger , startingMaxWorkers ) { return scan ( ( previousMaxWorkers , errorCount ) => { let newMaxWorkers ; if ( errorCount > 0 ) { newMaxWorkers = Math . max ( Math . floor ( previousMaxWorkers * MAX_WORKERS_DECREASE_PERCENTAGE ) , 1 ) ; } else { newMaxWorkers = Math . min ( startingMaxWorkers , Math . ceil ( previousMaxWorkers * MAX_WORKERS_INCREASE_PERCENTAGE ) ) ; } if ( newMaxWorkers !== previousMaxWorkers ) { logger . debug ( `Max workers configuration changing from ${ previousMaxWorkers } to ${ newMaxWorkers } after seeing ${ errorCount } "too many request" and/or "execute [inline] script" error(s)` ) ; if ( previousMaxWorkers === startingMaxWorkers ) { logger . warn ( `Max workers configuration is temporarily reduced after Elasticsearch returned ${ errorCount } "too many request" and/or "execute [inline] script" error(s).` ) ; } } return newMaxWorkers ; } , startingMaxWorkers ) ; } function createPollIntervalScan ( logger , startingPollInterval ) { return scan ( ( previousPollInterval , errorCount ) => { let newPollInterval ; if ( errorCount > 0 ) { newPollInterval = Math . ceil ( previousPollInterval * POLL_INTERVAL_INCREASE_PERCENTAGE ) ; } else { newPollInterval = Math . max ( startingPollInterval , Math . floor ( previousPollInterval * POLL_INTERVAL_DECREASE_PERCENTAGE ) ) ; } if ( newPollInterval !== previousPollInterval ) { logger . debug ( `Poll interval configuration changing from ${ previousPollInterval } to ${ newPollInterval } after seeing ${ errorCount } "too many request" and/or "execute [inline] script" error(s)` ) ; if ( previousPollInterval === startingPollInterval ) { logger . warn ( `Poll interval configuration is temporarily increased after Elasticsearch returned ${ errorCount } "too many request" and/or "execute [inline] script" error(s).` ) ; } } return newPollInterval ; } , startingPollInterval ) ; } function countErrors ( errors$ , countInterval ) { return merge ( interval ( countInterval ) . pipe ( map ( ( ) => FLUSH_MARKER ) ) , errors$ . pipe ( filter ( ( e ) => SavedObjectsErrorHelpers . isTooManyRequestsError ( e ) || isEsCannotExecuteScriptError ( e ) ) ) ) . pipe ( mergeScan ( ( { count } , next ) => { return next === FLUSH_MARKER ? of ( emitErrorCount ( count ) , resetErrorCount ( ) ) : of ( incementErrorCount ( count ) ) ; } , emitErrorCount ( 0 ) ) , filter ( isEmitEvent ) , map ( ( { count } ) => count ) ) ; } function emitErrorCount ( count ) { return { tag : 'emit' , count , } ; } function isEmitEvent ( event ) { return event . tag === 'emit' ; } function incementErrorCount ( count ) { return { tag : 'inc' , count : count + 1 , } ; } function resetErrorCount ( ) { return { tag : 'initial' , count : 0 , } ; }