import { IScopedClusterClient } from 'kibana/server' ; import { JobSavedObjectService } from '../../saved_objects' ; import { JobType } from '../../../common/types/saved_objects' ; import { Job , Datafeed } from '../../../common/types/anomaly_detection_jobs' ; import { searchProvider } from './search' ; import { DataFrameAnalyticsConfig } from '../../../common/types/data_frame_analytics' ; import { MLJobNotFound } from './errors' ; import { MlClient , MlClientParams , MlGetADParams , MlGetDFAParams , MlGetDatafeedParams , } from './types' ; export function getMlClient ( client , jobSavedObjectService ) { const mlClient = client . asInternalUser . ml ; async function jobIdsCheck ( jobType , p , allowWildcards = false ) { const jobIds = jobType === 'anomaly-detector' ? getADJobIdsFromRequest ( p ) : getDFAJobIdsFromRequest ( p ) ; if ( jobIds . length ) { await checkIds ( jobType , jobIds , allowWildcards ) ; } } async function checkIds ( jobType , jobIds , allowWildcards = false ) { const filteredJobIds = await jobSavedObjectService . filterJobIdsForSpace ( jobType , jobIds ) ; let missingIds = jobIds . filter ( ( j ) => filteredJobIds . indexOf ( j ) === - 1 ) ; if ( allowWildcards === true && missingIds . join ( ) . match ( '\\*' ) !== null ) { missingIds = missingIds . filter ( ( id ) => id . match ( '\\*' ) === null ) ; } if ( missingIds . length ) { throw new MLJobNotFound ( `No known job with id '${ missingIds . join ( ',' ) }'` ) ; } } async function groupIdsCheck ( p , allJobs , filteredJobIds ) { const ids = getADJobIdsFromRequest ( p ) ; if ( ids . length ) { const responseGroupIds = [ ... new Set ( allJobs . map ( ( j ) => j . groups ?? [ ] ) . flat ( ) ) ] ; const requestedGroupIds = [ ] ; const requestedJobIds = [ ] ; ids . forEach ( ( id ) => { if ( responseGroupIds . includes ( id ) ) { requestedGroupIds . push ( id ) ; } else { requestedJobIds . push ( id ) ; } } ) ; const groupIdsFromFilteredJobs = [ ... new Set ( allJobs . filter ( ( j ) => filteredJobIds . includes ( j . job_id ) ) . map ( ( j ) => j . groups ?? [ ] ) . flat ( ) ) , ] ; const groupsIdsThatDidNotMatch = requestedGroupIds . filter ( ( id ) => groupIdsFromFilteredJobs . includes ( id ) === false ) ; if ( groupsIdsThatDidNotMatch . length ) { throw new MLJobNotFound ( `No known job with id '${ groupsIdsThatDidNotMatch . join ( ',' ) }'` ) ; } if ( requestedJobIds . length ) { await checkIds ( 'anomaly-detector' , requestedJobIds , true ) ; } } } async function groupIdsCheckFromJobStats ( filteredJobIds , ... p ) { const ids = getADJobIdsFromRequest ( p ) ; if ( ids . length ) { const { body } = await mlClient . getJobs ( ... p ) ; await groupIdsCheck ( p , body . jobs , filteredJobIds ) ; } } async function datafeedIdsCheck ( p , allowWildcards = false ) { const datafeedIds = getDatafeedIdsFromRequest ( p ) ; if ( datafeedIds . length ) { const filteredDatafeedIds = await jobSavedObjectService . filterDatafeedIdsForSpace ( datafeedIds ) ; let missingIds = datafeedIds . filter ( ( j ) => filteredDatafeedIds . indexOf ( j ) === - 1 ) ; if ( allowWildcards === true && missingIds . join ( ) . match ( '\\*' ) !== null ) { missingIds = missingIds . filter ( ( id ) => id . match ( '\\*' ) === null ) ; } if ( missingIds . length ) { throw new MLJobNotFound ( `No known datafeed with id '${ missingIds . join ( ',' ) }'` ) ; } } } return { async closeJob ( ... p ) { await jobIdsCheck ( 'anomaly-detector' , p ) ; return mlClient . closeJob ( ... p ) ; } , async deleteCalendar ( ... p ) { return mlClient . deleteCalendar ( ... p ) ; } , async deleteCalendarEvent ( ... p ) { return mlClient . deleteCalendarEvent ( ... p ) ; } , async deleteCalendarJob ( ... p ) { return mlClient . deleteCalendarJob ( ... p ) ; } , async deleteDataFrameAnalytics ( ... p ) { await jobIdsCheck ( 'data-frame-analytics' , p ) ; const resp = await mlClient . deleteDataFrameAnalytics ( ... p ) ; return resp ; } , async deleteDatafeed ( ... p ) { await datafeedIdsCheck ( p ) ; const resp = await mlClient . deleteDatafeed ( ... p ) ; const [ datafeedId ] = getDatafeedIdsFromRequest ( p ) ; if ( datafeedId !== undefined ) { await jobSavedObjectService . deleteDatafeed ( datafeedId ) ; } return resp ; } , async deleteExpiredData ( ... p ) { await jobIdsCheck ( 'anomaly-detector' , p ) ; return mlClient . deleteExpiredData ( ... p ) ; } , async deleteFilter ( ... p ) { return mlClient . deleteFilter ( ... p ) ; } , async deleteForecast ( ... p ) { await jobIdsCheck ( 'anomaly-detector' , p ) ; return mlClient . deleteForecast ( ... p ) ; } , async deleteJob ( ... p ) { await jobIdsCheck ( 'anomaly-detector' , p ) ; const resp = await mlClient . deleteJob ( ... p ) ; return resp ; } , async deleteModelSnapshot ( ... p ) { await jobIdsCheck ( 'anomaly-detector' , p ) ; return mlClient . deleteModelSnapshot ( ... p ) ; } , async deleteTrainedModel ( ... p ) { return mlClient . deleteTrainedModel ( ... p ) ; } , async estimateModelMemory ( ... p ) { return mlClient . estimateModelMemory ( ... p ) ; } , async evaluateDataFrame ( ... p ) { return mlClient . evaluateDataFrame ( ... p ) ; } , async explainDataFrameAnalytics ( ... p ) { await jobIdsCheck ( 'data-frame-analytics' , p ) ; return mlClient . explainDataFrameAnalytics ( ... p ) ; } , async flushJob ( ... p ) { await jobIdsCheck ( 'anomaly-detector' , p ) ; return mlClient . flushJob ( ... p ) ; } , async forecast ( ... p ) { await jobIdsCheck ( 'anomaly-detector' , p ) ; return mlClient . forecast ( ... p ) ; } , async getBuckets ( ... p ) { await jobIdsCheck ( 'anomaly-detector' , p ) ; return mlClient . getBuckets ( ... p ) ; } , async getCalendarEvents ( ... p ) { await jobIdsCheck ( 'anomaly-detector' , p ) ; return mlClient . getCalendarEvents ( ... p ) ; } , async getCalendars ( ... p ) { const { body } = await mlClient . getCalendars ( ... p ) ; const { body : { jobs : allJobs } , } = await mlClient . getJobs < { jobs : Job [ ] ; } > ( ) ; const allJobIds = allJobs . map ( ( j ) => j . job_id ) ; const calJobIds = [ ... new Set ( body . calendars . map ( ( c ) => c . job_ids ) . flat ( ) ) ] ; const groups = calJobIds . filter ( ( j ) => allJobIds . includes ( j ) === false ) ; const filteredJobIds = await jobSavedObjectService . filterJobIdsForSpace ( 'anomaly-detector' , calJobIds ) ; const calendars = body . calendars . map ( ( c ) => ( { ... c , job_ids : c . job_ids . filter ( ( id ) => filteredJobIds . includes ( id ) || groups . includes ( id ) ) , total_job_count : calJobIds . length , } ) ) ; return { body : { ... body , calendars } } ; } , async getCategories ( ... p ) { await jobIdsCheck ( 'anomaly-detector' , p ) ; return mlClient . getCategories ( ... p ) ; } , async getDataFrameAnalytics ( ... p ) { await jobIdsCheck ( 'data-frame-analytics' , p , true ) ; try { const { body } = await mlClient . getDataFrameAnalytics < { data_frame_analytics : DataFrameAnalyticsConfig [ ] ; } > ( ... p ) ; const jobs = await jobSavedObjectService . filterJobsForSpace < DataFrameAnalyticsConfig > ( 'data-frame-analytics' , body . data_frame_analytics , 'id' ) ; return { body : { ... body , count : jobs . length , data_frame_analytics : jobs } } ; } catch ( error ) { if ( error . statusCode === 404 ) { throw new MLJobNotFound ( error . body . error . reason ) ; } throw error . body ?? error ; } } , async getDataFrameAnalyticsStats ( ... p ) { await jobIdsCheck ( 'data-frame-analytics' , p , true ) ; try { const { body } = ( await mlClient . getDataFrameAnalyticsStats ( ... p ) ) as unknown as { body : { data_frame_analytics : DataFrameAnalyticsConfig [ ] ; } ; } ; const jobs = await jobSavedObjectService . filterJobsForSpace < DataFrameAnalyticsConfig > ( 'data-frame-analytics' , body . data_frame_analytics , 'id' ) ; return { body : { ... body , count : jobs . length , data_frame_analytics : jobs } } ; } catch ( error ) { if ( error . statusCode === 404 ) { throw new MLJobNotFound ( error . body . error . reason ) ; } throw error . body ?? error ; } } , async getDatafeedStats ( ... p ) { await datafeedIdsCheck ( p , true ) ; try { const { body } = await mlClient . getDatafeedStats ( ... p ) ; const datafeeds = await jobSavedObjectService . filterDatafeedsForSpace ( 'anomaly-detector' , body . datafeeds , 'datafeed_id' ) ; return { body : { ... body , count : datafeeds . length , datafeeds } } ; } catch ( error ) { if ( error . statusCode === 404 ) { throw new MLJobNotFound ( error . body . error . reason ) ; } throw error . body ?? error ; } } , async getDatafeeds ( ... p ) { await datafeedIdsCheck ( p , true ) ; try { const { body } = await mlClient . getDatafeeds ( ... p ) ; const datafeeds = await jobSavedObjectService . filterDatafeedsForSpace < Datafeed > ( 'anomaly-detector' , body . datafeeds , 'datafeed_id' ) ; return { body : { ... body , count : datafeeds . length , datafeeds } } ; } catch ( error ) { if ( error . statusCode === 404 ) { throw new MLJobNotFound ( error . body . error . reason ) ; } throw error . body ?? error ; } } , async getFilters ( ... p ) { return mlClient . getFilters ( ... p ) ; } , async getInfluencers ( ... p ) { await jobIdsCheck ( 'anomaly-detector' , p ) ; return mlClient . getInfluencers ( ... p ) ; } , async getJobStats ( ... p ) { try { const { body } = await mlClient . getJobStats ( ... p ) ; const jobs = await jobSavedObjectService . filterJobsForSpace ( 'anomaly-detector' , body . jobs , 'job_id' ) ; await groupIdsCheckFromJobStats ( jobs . map ( ( j ) => j . job_id ) , ... p ) ; return { body : { ... body , count : jobs . length , jobs } } ; } catch ( error ) { if ( error instanceof MLJobNotFound ) { throw error ; } if ( error . statusCode === 404 ) { throw new MLJobNotFound ( error . body . error . reason ) ; } throw error . body ?? error ; } } , async getJobs ( ... p ) { try { const { body } = await mlClient . getJobs < { jobs : Job [ ] ; } > ( ... p ) ; const jobs = await jobSavedObjectService . filterJobsForSpace < Job > ( 'anomaly-detector' , body . jobs , 'job_id' ) ; await groupIdsCheck ( p , body . jobs , jobs . map ( ( j ) => j . job_id ) ) ; return { body : { ... body , count : jobs . length , jobs } } ; } catch ( error ) { if ( error instanceof MLJobNotFound ) { throw error ; } if ( error . statusCode === 404 ) { throw new MLJobNotFound ( error . body . error . reason ) ; } throw error . body ?? error ; } } , async getModelSnapshots ( ... p ) { await jobIdsCheck ( 'anomaly-detector' , p ) ; return mlClient . getModelSnapshots ( ... p ) ; } , async getOverallBuckets ( ... p ) { await jobIdsCheck ( 'anomaly-detector' , p ) ; return mlClient . getOverallBuckets ( ... p ) ; } , async getRecords ( ... p ) { await jobIdsCheck ( 'anomaly-detector' , p ) ; return mlClient . getRecords ( ... p ) ; } , async getTrainedModels ( ... p ) { return mlClient . getTrainedModels ( ... p ) ; } , async getTrainedModelsStats ( ... p ) { return mlClient . getTrainedModelsStats ( ... p ) ; } , async startTrainedModelDeployment ( ... p ) { return mlClient . startTrainedModelDeployment ( ... p ) ; } , async stopTrainedModelDeployment ( ... p ) { return mlClient . stopTrainedModelDeployment ( ... p ) ; } , async info ( ... p ) { return mlClient . info ( ... p ) ; } , async openJob ( ... p ) { await jobIdsCheck ( 'anomaly-detector' , p ) ; return mlClient . openJob ( ... p ) ; } , async postCalendarEvents ( ... p ) { return mlClient . postCalendarEvents ( ... p ) ; } , async postData ( ... p ) { await jobIdsCheck ( 'anomaly-detector' , p ) ; return mlClient . postData ( ... p ) ; } , async previewDatafeed ( ... p ) { await datafeedIdsCheck ( p ) ; return mlClient . previewDatafeed ( ... p ) ; } , async putCalendar ( ... p ) { return mlClient . putCalendar ( ... p ) ; } , async putCalendarJob ( ... p ) { return mlClient . putCalendarJob ( ... p ) ; } , async putDataFrameAnalytics ( ... p ) { const resp = await mlClient . putDataFrameAnalytics ( ... p ) ; const [ analyticsId ] = getDFAJobIdsFromRequest ( p ) ; if ( analyticsId !== undefined ) { await jobSavedObjectService . createDataFrameAnalyticsJob ( analyticsId ) ; } return resp ; } , async putDatafeed ( ... p ) { const resp = await mlClient . putDatafeed ( ... p ) ; const [ datafeedId ] = getDatafeedIdsFromRequest ( p ) ; const jobId = getJobIdFromBody ( p ) ; if ( datafeedId !== undefined && jobId !== undefined ) { await jobSavedObjectService . addDatafeed ( datafeedId , jobId ) ; } return resp ; } , async putFilter ( ... p ) { return mlClient . putFilter ( ... p ) ; } , async putJob ( ... p ) { const resp = await mlClient . putJob ( ... p ) ; const [ jobId ] = getADJobIdsFromRequest ( p ) ; if ( jobId !== undefined ) { await jobSavedObjectService . createAnomalyDetectionJob ( jobId ) ; } return resp ; } , async putTrainedModel ( ... p ) { return mlClient . putTrainedModel ( ... p ) ; } , async revertModelSnapshot ( ... p ) { await jobIdsCheck ( 'anomaly-detector' , p ) ; return mlClient . revertModelSnapshot ( ... p ) ; } , async setUpgradeMode ( ... p ) { return mlClient . setUpgradeMode ( ... p ) ; } , async startDataFrameAnalytics ( ... p ) { await jobIdsCheck ( 'data-frame-analytics' , p ) ; return mlClient . startDataFrameAnalytics ( ... p ) ; } , async startDatafeed ( ... p ) { await datafeedIdsCheck ( p ) ; return mlClient . startDatafeed ( ... p ) ; } , async stopDataFrameAnalytics ( ... p ) { await jobIdsCheck ( 'data-frame-analytics' , p ) ; return mlClient . stopDataFrameAnalytics ( ... p ) ; } , async stopDatafeed ( ... p ) { await datafeedIdsCheck ( p ) ; return mlClient . stopDatafeed ( ... p ) ; } , async updateDataFrameAnalytics ( ... p ) { await jobIdsCheck ( 'data-frame-analytics' , p ) ; return mlClient . updateDataFrameAnalytics ( ... p ) ; } , async updateDatafeed ( ... p ) { await datafeedIdsCheck ( p ) ; if ( p . length === 0 || p [ 0 ] === undefined ) { throw new Error ( 'Incorrect arguments supplied' ) ; } const { datafeed_id : id , body } = p [ 0 ] ; return client . asInternalUser . transport . request ( { method : 'POST' , path : `/_ml/datafeeds/${ id }/_update` , body , } ) ; } , async updateFilter ( ... p ) { return mlClient . updateFilter ( ... p ) ; } , async updateJob ( ... p ) { await jobIdsCheck ( 'anomaly-detector' , p ) ; return mlClient . updateJob ( ... p ) ; } , async resetJob ( ... p ) { await jobIdsCheck ( 'anomaly-detector' , p ) ; return mlClient . resetJob ( ... p ) ; } , async updateModelSnapshot ( ... p ) { await jobIdsCheck ( 'anomaly-detector' , p ) ; return mlClient . updateModelSnapshot ( ... p ) ; } , async validate ( ... p ) { return mlClient . validate ( ... p ) ; } , async validateDetector ( ... p ) { return mlClient . validateDetector ( ... p ) ; } , ... searchProvider ( client , jobSavedObjectService ) , } as MlClient ; } function getDFAJobIdsFromRequest ( [ params ] ) { const ids = params ?. id ?. split ( ',' ) ; return ids || [ ] ; } function getADJobIdsFromRequest ( [ params ] ) { const ids = typeof params ?. job_id === 'string' ? params ?. job_id . split ( ',' ) : params ?. job_id ; return ids || [ ] ; } function getDatafeedIdsFromRequest ( [ params ] ) { const ids = typeof params ?. datafeed_id === 'string' ? params ?. datafeed_id . split ( ',' ) : params ?. datafeed_id ; return ids || [ ] ; } function getJobIdFromBody ( p ) { const [ params ] = p ; return params ?. body ?. job_id ; }