import { i18n } from '@kbn/i18n' ; import { combineLatest , Observable , of , Subject , Subscription } from 'rxjs' ; import { isEqual , cloneDeep } from 'lodash' ; import { catchError , debounceTime , distinctUntilChanged , pluck , startWith , switchMap , map , pairwise , filter , skipWhile , } from 'rxjs/operators' ; import { useEffect , useMemo } from 'react' ; import { DEFAULT_MODEL_MEMORY_LIMIT } from '../../../../../../../common/constants/new_job' ; import { ml } from '../../../../../services/ml_api_service' ; import { JobValidator , VALIDATION_DELAY_MS } from '../../job_validator/job_validator' ; import { MLHttpFetchError , MLResponseError , extractErrorMessage , } from '../../../../../../../common/util/errors' ; import { useMlKibana } from '../../../../../contexts/kibana' ; import { JobCreator } from '../job_creator' ; export type CalculatePayload = Parameters < typeof ml . calculateModelMemoryLimit$ > [ 0 ] ; type ModelMemoryEstimator = ReturnType < typeof modelMemoryEstimatorProvider > ; export const modelMemoryEstimatorProvider = ( jobCreator , jobValidator ) => { const modelMemoryCheck$ = new Subject < CalculatePayload > ( ) ; const error$ = new Subject < MLHttpFetchError < MLResponseError > > ( ) ; return { get error$ ( ) : Observable < MLHttpFetchError < MLResponseError > > { return error$ . asObservable ( ) ; } , get updates$ ( ) : Observable < string > { return combineLatest ( [ jobCreator . wizardInitialized$ . pipe ( skipWhile ( ( wizardInitialized ) => wizardInitialized === false ) ) , modelMemoryCheck$ , ] ) . pipe ( map ( ( [ , payload ] ) => payload ) , debounceTime ( VALIDATION_DELAY_MS + 100 ) , map ( cloneDeep ) , distinctUntilChanged ( isEqual ) , filter ( ( ) => jobValidator . isModelMemoryEstimationPayloadValid ) , switchMap ( ( payload ) => { return ml . calculateModelMemoryLimit$ ( payload ) . pipe ( pluck ( 'modelMemoryLimit' ) , catchError ( ( error ) => { console . error ( 'Model memory limit could not be calculated' , error . body ) ; error$ . next ( error ) ; return of ( DEFAULT_MODEL_MEMORY_LIMIT ) ; } ) ) ; } ) ) ; } , update ( payload ) { modelMemoryCheck$ . next ( payload ) ; } , } ; } ; export const useModelMemoryEstimator = ( jobCreator , jobValidator , jobCreatorUpdate , jobCreatorUpdated ) => { const { services : { notifications } , } = useMlKibana ( ) ; const modelMemoryEstimator = useMemo < ModelMemoryEstimator > ( ( ) => modelMemoryEstimatorProvider ( jobCreator , jobValidator ) , [ ] ) ; useEffect ( ( ) => { const subscription = new Subscription ( ) ; subscription . add ( modelMemoryEstimator . updates$ . pipe ( startWith ( jobCreator . modelMemoryLimit ) , pairwise ( ) ) . subscribe ( ( [ previousEstimation , currentEstimation ] ) => { if ( jobCreator . modelMemoryLimit === null || jobCreator . modelMemoryLimit === previousEstimation ) { jobCreator . modelMemoryLimit = currentEstimation ; jobCreatorUpdate ( ) ; } } ) ) ; subscription . add ( modelMemoryEstimator . error$ . subscribe ( ( error ) => { notifications . toasts . addWarning ( { title : i18n . translate ( 'xpack.ml.newJob.wizard.estimateModelMemoryError' , { defaultMessage : 'Model memory limit could not be calculated' , } ) , text : extractErrorMessage ( error ) , } ) ; } ) ) ; return ( ) => { subscription . unsubscribe ( ) ; } ; } , [ ] ) ; useEffect ( ( ) => { modelMemoryEstimator . update ( { datafeedConfig : jobCreator . datafeedConfig , analysisConfig : jobCreator . jobConfig . analysis_config , indexPattern : jobCreator . indexPatternTitle , query : jobCreator . datafeedConfig . query , timeFieldName : jobCreator . jobConfig . data_description . time_field ! , earliestMs : jobCreator . start , latestMs : jobCreator . end , } ) ; } , [ jobCreatorUpdated ] ) ; } ;