import fs from 'fs' ; import Boom from '@hapi/boom' ; import numeral from '@elastic/numeral' ; import type { KibanaRequest , IScopedClusterClient , SavedObjectsClientContract , } from 'kibana/server' ; import moment from 'moment' ; import { merge } from 'lodash' ; import type { DataViewsService } from '../../../../../../src/plugins/data_views/common' ; import type { AnalysisLimits } from '../../../common/types/anomaly_detection_jobs' ; import { getAuthorizationHeader } from '../../lib/request_authorization' ; import type { MlInfoResponse } from '../../../common/types/ml_server_info' ; import type { MlClient } from '../../lib/ml_client' ; import { ML_MODULE_SAVED_OBJECT_TYPE } from '../../../common/types/saved_objects' ; import type { KibanaObjects , KibanaObjectConfig , ModuleDatafeed , ModuleJob , Module , FileBasedModule , Logo , JobOverride , DatafeedOverride , GeneralJobsOverride , DatafeedResponse , JobResponse , KibanaObjectResponse , DataRecognizerConfigResponse , GeneralDatafeedsOverride , JobSpecificOverride , } from '../../../common/types/modules' ; import { isGeneralJobOverride } from '../../../common/types/modules' ; import { getLatestDataOrBucketTimestamp , prefixDatafeedId , splitIndexPatternNames , } from '../../../common/util/job_utils' ; import { mlLog } from '../../lib/log' ; import { calculateModelMemoryLimitProvider } from '../calculate_model_memory_limit' ; import { fieldsServiceProvider } from '../fields_service' ; import { jobServiceProvider } from '../job_service' ; import { resultsServiceProvider } from '../results_service' ; import type { JobExistResult , JobStat } from '../../../common/types/data_recognizer' ; import type { MlJobsStatsResponse } from '../../../common/types/job_service' ; import type { Datafeed } from '../../../common/types/anomaly_detection_jobs' ; import type { JobSavedObjectService } from '../../saved_objects' ; import { isDefined } from '../../../common/types/guards' ; import { isPopulatedObject } from '../../../common/util/object_utils' ; const ML_DIR = 'ml' ; const KIBANA_DIR = 'kibana' ; const INDEX_PATTERN_ID = 'INDEX_PATTERN_ID' ; const INDEX_PATTERN_NAME = 'INDEX_PATTERN_NAME' ; export const SAVED_OBJECT_TYPES = { DASHBOARD : 'dashboard' , SEARCH : 'search' , VISUALIZATION : 'visualization' , } ; function isModule ( arg ) { return isPopulatedObject ( arg ) && Array . isArray ( arg . jobs ) && arg . jobs [ 0 ] ?. config !== undefined ; } function isFileBasedModule ( arg ) { return isPopulatedObject ( arg ) && Array . isArray ( arg . jobs ) && arg . jobs [ 0 ] ?. file !== undefined ; } interface Config { dirName ? : string ; module : FileBasedModule | Module ; isSavedObject : boolean ; } export interface RecognizeResult { id : string ; title : string ; query : any ; description : string ; logo : Logo ; } interface ObjectExistResult { id : string ; type : string ; exists ? : boolean ; } interface ObjectExistResponse { id : string ; type : string ; exists : boolean ; savedObject ? : { id : string ; type : string ; attributes : KibanaObjectConfig ; } ; } interface SaveResults { jobs : JobResponse [ ] ; datafeeds : DatafeedResponse [ ] ; savedObjects : KibanaObjectResponse [ ] ; } export class DataRecognizer { private _client : IScopedClusterClient ; private _mlClient : MlClient ; private _savedObjectsClient : SavedObjectsClientContract ; private _jobSavedObjectService : JobSavedObjectService ; private _dataViewsService : DataViewsService ; private _request : KibanaRequest ; private _authorizationHeader : object ; private _modulesDir = `${ __dirname }/modules` ; private _indexPatternName : string = '' ; private _indexPatternId : string | undefined = undefined ; private _jobsService : ReturnType < typeof jobServiceProvider > ; private _resultsService : ReturnType < typeof resultsServiceProvider > ; private _calculateModelMemoryLimit : ReturnType < typeof calculateModelMemoryLimitProvider > ; /**
     * List of the module jobs that require model memory estimation
     */ private _jobsForModelMemoryEstimation : Array < { job : ModuleJob ; query : any ; } > = [ ] ; constructor ( mlClusterClient , mlClient , savedObjectsClient , dataViewsService , jobSavedObjectService , request ) { this . _client = mlClusterClient ; this . _mlClient = mlClient ; this . _savedObjectsClient = savedObjectsClient ; this . _dataViewsService = dataViewsService ; this . _jobSavedObjectService = jobSavedObjectService ; this . _request = request ; this . _authorizationHeader = getAuthorizationHeader ( request ) ; this . _jobsService = jobServiceProvider ( mlClusterClient , mlClient ) ; this . _resultsService = resultsServiceProvider ( mlClient ) ; this . _calculateModelMemoryLimit = calculateModelMemoryLimitProvider ( mlClusterClient , mlClient ) ; } private async _listDirs ( dirName ) { const dirs = [ ] ; return new Promise ( ( resolve , reject ) => { fs . readdir ( dirName , ( err , fileNames ) => { if ( err ) { reject ( err ) ; } fileNames . forEach ( ( fileName ) => { const path = `${ dirName }/${ fileName }` ; if ( fs . lstatSync ( path ) . isDirectory ( ) ) { dirs . push ( fileName ) ; } } ) ; resolve ( dirs ) ; } ) ; } ) ; } private async _readFile ( fileName ) { return new Promise ( ( resolve , reject ) => { fs . readFile ( fileName , 'utf-8' , ( err , content ) => { if ( err ) { reject ( err ) ; } else { resolve ( content ) ; } } ) ; } ) ; } private async _loadConfigs ( ) { const configs = [ ] ; const dirs = await this . _listDirs ( this . _modulesDir ) ; await Promise . all ( dirs . map ( async ( dir ) => { let file ; try { file = await this . _readFile ( `${ this . _modulesDir }/${ dir }/manifest.json` ) ; } catch ( error ) { mlLog . warn ( `Data recognizer skipping folder ${ dir } as manifest.json cannot be read` ) ; } if ( file !== undefined ) { try { configs . push ( { dirName : dir , module : JSON . parse ( file ) , isSavedObject : false , } ) ; } catch ( error ) { mlLog . warn ( `Data recognizer error parsing ${ dir }/manifest.json. ${ error }` ) ; } } } ) ) ; const savedObjectConfigs = ( await this . _loadSavedObjectModules ( ) ) . map ( ( module ) => ( { module , isSavedObject : true , } ) ) ; return [ ... configs , ... savedObjectConfigs ] ; } private async _loadSavedObjectModules ( ) { const jobs = await this . _savedObjectsClient . find < Module > ( { type : ML_MODULE_SAVED_OBJECT_TYPE , perPage : 10000 , } ) ; return jobs . saved_objects . map ( ( o ) => o . attributes ) ; } private async _findConfig ( id ) { const configs = await this . _loadConfigs ( ) ; return configs . find ( ( i ) => i . module . id === id ) ; } public async findMatches ( indexPattern ) { const manifestFiles = await this . _loadConfigs ( ) ; const results = [ ] ; await Promise . all ( manifestFiles . map ( async ( i ) => { const moduleConfig = i . module ; let match = false ; try { match = await this . _searchForFields ( moduleConfig , indexPattern ) ; } catch ( error ) { mlLog . warn ( `Data recognizer error running query defined for module ${ moduleConfig . id }. ${ error }` ) ; } if ( match === true ) { let logo = null ; if ( moduleConfig . logo ) { logo = moduleConfig . logo ; } else if ( moduleConfig . logoFile ) { try { const logoFile = await this . _readFile ( `${ this . _modulesDir }/${ i . dirName }/${ moduleConfig . logoFile }` ) ; logo = JSON . parse ( logoFile ) ; } catch ( e ) { logo = null ; } } results . push ( { id : moduleConfig . id , title : moduleConfig . title , query : moduleConfig . query , description : moduleConfig . description , logo , } ) ; } } ) ) ; results . sort ( ( res1 , res2 ) => res1 . id . localeCompare ( res2 . id ) ) ; return results ; } private async _searchForFields ( moduleConfig , indexPattern ) { if ( moduleConfig . query === undefined ) { return false ; } const index = indexPattern ; const size = 0 ; const searchBody = { query : moduleConfig . query , } ; const { body } = await this . _client . asCurrentUser . search ( { index , size , body : searchBody , } ) ; return body . hits . total . value > 0 ; } public async listModules ( ) { const manifestFiles = await this . _loadConfigs ( ) ; manifestFiles . sort ( ( a , b ) => a . module . id . localeCompare ( b . module . id ) ) ; const configs = [ ] ; for ( const config of manifestFiles ) { if ( config . isSavedObject ) { configs . push ( config . module ) ; } else { configs . push ( await this . getModule ( config . module . id ) ) ; } } return configs as Module [ ] ; } public async getModule ( id , prefix = '' ) { let module = null ; let dirName = null ; const config = await this . _findConfig ( id ) ; if ( config !== undefined ) { module = config . module ; dirName = config . dirName ?? null ; } else { throw Boom . notFound ( `Module with the id "${ id }" not found` ) ; } const jobs = [ ] ; const datafeeds = [ ] ; const kibana = { } ; if ( isModule ( module ) ) { const tempJobs = module . jobs . map ( ( j ) => ( { id : `${ prefix }${ j . id }` , config : j . config , } ) ) ; jobs . push ( ... tempJobs ) ; const tempDatafeeds = module . datafeeds . map ( ( d ) => { const jobId = `${ prefix }${ d . job_id }` ; return { id : prefixDatafeedId ( d . id , prefix ) , job_id : jobId , config : { ... d . config , job_id : jobId , } , } ; } ) ; datafeeds . push ( ... tempDatafeeds ) ; } else if ( isFileBasedModule ( module ) ) { const tempJobs = module . jobs . map ( async ( job ) => { try { const jobConfig = await this . _readFile ( `${ this . _modulesDir }/${ dirName }/${ ML_DIR }/${ job . file }` ) ; return { id : `${ prefix }${ job . id }` , config : JSON . parse ( jobConfig ) , } ; } catch ( error ) { mlLog . warn ( `Data recognizer error loading config for job ${ job . id } for module ${ id }. ${ error }` ) ; } } ) ; jobs . push ( ... ( await Promise . all ( tempJobs ) ) . filter ( isDefined ) ) ; const tempDatafeed = module . datafeeds . map ( async ( datafeed ) => { try { const datafeedConfigString = await this . _readFile ( `${ this . _modulesDir }/${ dirName }/${ ML_DIR }/${ datafeed . file }` ) ; const datafeedConfig = JSON . parse ( datafeedConfigString ) as Datafeed ; datafeedConfig . job_id = `${ prefix }${ datafeed . job_id }` ; return { id : prefixDatafeedId ( datafeed . id , prefix ) , job_id : datafeedConfig . job_id , config : datafeedConfig , } ; } catch ( error ) { mlLog . warn ( `Data recognizer error loading config for datafeed ${ datafeed . id } for module ${ id }. ${ error }` ) ; } } ) ; datafeeds . push ( ... ( await Promise . all ( tempDatafeed ) ) . filter ( isDefined ) ) ; } if ( module . kibana !== undefined ) { const kKeys = Object . keys ( module . kibana ) as Array < keyof FileBasedModule [ 'kibana' ] > ; await Promise . all ( kKeys . map ( async ( key ) => { kibana [ key ] = [ ] ; if ( isFileBasedModule ( module ) ) { await Promise . all ( module . kibana [ key ] . map ( async ( obj ) => { try { const kConfigString = await this . _readFile ( `${ this . _modulesDir }/${ dirName }/${ KIBANA_DIR }/${ key }/${ obj . file }` ) ; const kId = obj . file . replace ( '.json' , '' ) ; const kConfig = JSON . parse ( kConfigString ) ; kibana [ key ] ! . push ( { id : kId , title : kConfig . title , config : kConfig , } ) ; } catch ( error ) { mlLog . warn ( `Data recognizer error loading config for ${ key } ${ obj . id } for module ${ id }. ${ error }` ) ; } } ) ) ; } } ) ) ; } return { ... module , jobs , datafeeds , kibana , } ; } public async setup ( moduleId , jobPrefix ? , groups ? , indexPatternName ? , query ? , useDedicatedIndex ? , startDatafeed ? , start ? , end ? , jobOverrides ? , datafeedOverrides ? , estimateModelMemory = true , applyToAllSpaces = false ) { const moduleConfig = await this . getModule ( moduleId , jobPrefix ) ; if ( indexPatternName === undefined && moduleConfig . defaultIndexPattern === undefined ) { throw Boom . badRequest ( `No index pattern configured in "${ moduleId }" configuration file and no index pattern passed to the endpoint` ) ; } this . _indexPatternName = indexPatternName === undefined ? moduleConfig . defaultIndexPattern : indexPatternName ; this . _indexPatternId = await this . _getIndexPatternId ( this . _indexPatternName ) ; if ( this . _indexPatternId === undefined && this . _doJobUrlsContainIndexPatternId ( moduleConfig ) ) { throw Boom . badRequest ( `Module's jobs contain custom URLs which require a Kibana data view (${ this . _indexPatternName }) which cannot be found.` ) ; } if ( this . _indexPatternId === undefined && this . _doSavedObjectsContainIndexPatternId ( moduleConfig ) ) { throw Boom . badRequest ( `Module's saved objects contain custom URLs which require a Kibana data view (${ this . _indexPatternName }) which cannot be found.` ) ; } const results = this . _createResultsTemplate ( moduleConfig ) ; const saveResults = { jobs : [ ] as JobResponse [ ] , datafeeds : [ ] as DatafeedResponse [ ] , savedObjects : [ ] as KibanaObjectResponse [ ] , } ; this . _jobsForModelMemoryEstimation = moduleConfig . jobs . map ( ( job ) => ( { job , query : moduleConfig . datafeeds . find ( ( d ) => d . config . job_id === job . id ) ?. config . query ?? null , } ) ) ; this . applyJobConfigOverrides ( moduleConfig , jobOverrides , jobPrefix ) ; this . applyDatafeedConfigOverrides ( moduleConfig , datafeedOverrides , jobPrefix ) ; this . _updateDatafeedIndices ( moduleConfig ) ; this . _updateJobUrlIndexPatterns ( moduleConfig ) ; await this . _updateModelMemoryLimits ( moduleConfig , estimateModelMemory , start , end ) ; if ( moduleConfig . jobs && moduleConfig . jobs . length ) { if ( Array . isArray ( groups ) ) { moduleConfig . jobs . forEach ( ( job ) => ( job . config . groups = groups ) ) ; } if ( useDedicatedIndex === true ) { moduleConfig . jobs . forEach ( ( job ) => ( job . config . results_index_name = job . id ) ) ; } saveResults . jobs = await this . _saveJobs ( moduleConfig . jobs , applyToAllSpaces ) ; } if ( moduleConfig . datafeeds && moduleConfig . datafeeds . length ) { if ( typeof query === 'object' && query !== null ) { moduleConfig . datafeeds . forEach ( ( df ) => { df . config . query = query ; } ) ; } saveResults . datafeeds = await this . _saveDatafeeds ( moduleConfig . datafeeds ) ; if ( startDatafeed ) { const savedDatafeeds = moduleConfig . datafeeds . filter ( ( df ) => { const datafeedResult = saveResults . datafeeds . find ( ( d ) => d . id === df . id ) ; return datafeedResult !== undefined && datafeedResult . success === true ; } ) ; const startResults = await this . _startDatafeeds ( savedDatafeeds , start , end ) ; saveResults . datafeeds . forEach ( ( df ) => { const startedDatafeed = startResults [ df . id ] ; if ( startedDatafeed !== undefined ) { df . started = startedDatafeed . started ; df . awaitingMlNodeAllocation = startedDatafeed . awaitingMlNodeAllocation ; if ( startedDatafeed . error !== undefined ) { df . error = startedDatafeed . error ; } } } ) ; } } if ( moduleConfig . kibana ) { this . _updateSavedObjectIndexPatterns ( moduleConfig ) ; const savedObjects = await this . _createSavedObjectsToSave ( moduleConfig ) ; this . _updateKibanaResults ( results . kibana , savedObjects ) ; try { saveResults . savedObjects = await this . _saveKibanaObjects ( savedObjects ) ; } catch ( error ) { this . _populateKibanaResultErrors ( results . kibana , error . output ?. payload ) ; } } this . _updateResults ( results , saveResults ) ; return results ; } public async dataRecognizerJobsExist ( moduleId ) { const results = { } as JobExistResult ; const module = await this . getModule ( moduleId ) ; if ( module && module . jobs ) { const jobIds = module . jobs . map ( ( job ) => `*${ job . id }` ) ; const jobInfo = await this . _jobsService . jobsExist ( jobIds ) ; const doJobsExist = Object . values ( jobInfo ) . every ( ( j ) => j . exists === true ) ; results . jobsExist = doJobsExist ; if ( doJobsExist === true ) { const { body } = await this . _mlClient . getJobStats < MlJobsStatsResponse > ( { job_id : jobIds . join ( ) , } ) ; const jobStatsJobs = [ ] ; if ( body . jobs && body . jobs . length > 0 ) { const foundJobIds = body . jobs . map ( ( job ) => job . job_id ) ; const latestBucketTimestampsByJob = await this . _resultsService . getLatestBucketTimestampByJob ( foundJobIds ) ; body . jobs . forEach ( ( job ) => { const jobStat = { id : job . job_id , } as JobStat ; if ( job . data_counts ) { jobStat . earliestTimestampMs = job . data_counts . earliest_record_timestamp ; jobStat . latestTimestampMs = job . data_counts . latest_record_timestamp ; jobStat . latestResultsTimestampMs = getLatestDataOrBucketTimestamp ( jobStat . latestTimestampMs , latestBucketTimestampsByJob [ job . job_id ] as number ) ; } jobStatsJobs . push ( jobStat ) ; } ) ; } results . jobs = jobStatsJobs ; } } return results ; } private async _getIndexPatternId ( name ) { try { const dataViews = await this . _dataViewsService . find ( name ) ; return dataViews . find ( ( d ) => d . title === name ) ?. id ; } catch ( error ) { mlLog . warn ( `Error loading data views, ${ error }` ) ; return ; } } private async _createSavedObjectsToSave ( moduleConfig ) { const savedObjectExistResults = await this . _checkIfSavedObjectsExist ( moduleConfig . kibana ) ; Object . keys ( moduleConfig . kibana ) . forEach ( ( type ) => { moduleConfig . kibana [ type ] ! . forEach ( ( configItem ) => { const existsResult = savedObjectExistResults . find ( ( o ) => o . id === configItem . id ) ; if ( existsResult !== undefined ) { configItem . exists = existsResult . exists ; if ( existsResult . exists === false ) { existsResult . savedObject = { type , id : configItem . id , attributes : configItem . config , } ; } } } ) ; } ) ; return savedObjectExistResults ; } private _updateKibanaResults ( kibanaSaveResults , objectExistResults ) { ( Object . keys ( kibanaSaveResults ) as Array < keyof DataRecognizerConfigResponse [ 'kibana' ] > ) . forEach ( ( type ) => { kibanaSaveResults [ type ] . forEach ( ( resultItem ) => { const i = objectExistResults . find ( ( o ) => o . id === resultItem . id && o . type === type ) ; resultItem . exists = i !== undefined && i . exists ; } ) ; } ) ; } private _populateKibanaResultErrors ( kibanaSaveResults , error ) { const errorObj = error === undefined ? { message : 'Unknown error when creating saved object' } : error ; ( Object . keys ( kibanaSaveResults ) as Array < keyof DataRecognizerConfigResponse [ 'kibana' ] > ) . forEach ( ( type ) => { kibanaSaveResults [ type ] . forEach ( ( resultItem ) => { if ( resultItem . exists === false ) { resultItem . error = errorObj ; } } ) ; } ) ; } private async _checkIfSavedObjectsExist ( kibanaObjects ) { const types = Object . keys ( kibanaObjects ) ; const results = await Promise . all ( types . map ( async ( type ) => { const existingObjects = await this . _loadExistingSavedObjects ( type ) ; return kibanaObjects [ type ] ! . map ( ( obj ) => { const existingObject = existingObjects . saved_objects . find ( ( o ) => o . attributes && o . attributes . title === obj . title ) ; return { id : obj . id , type , exists : existingObject !== undefined , } ; } ) ; } ) ) ; return ( [ ] as ObjectExistResponse [ ] ) . concat ( ... results ) ; } private _loadExistingSavedObjects ( type ) { return this . _savedObjectsClient . find < any > ( { type , perPage : 1000 } ) ; } private async _saveKibanaObjects ( objectExistResults ) { let results = { saved_objects : [ ] as any [ ] } ; const filteredSavedObjects = objectExistResults . filter ( ( o ) => o . exists === false ) . map ( ( o ) => o . savedObject ! ) ; if ( filteredSavedObjects . length ) { results = await this . _savedObjectsClient . bulkCreate ( filteredSavedObjects . map ( ( doc ) => ( { ... doc , migrationVersion : { } , } ) ) ) ; } return results . saved_objects ; } private async _saveJobs ( jobs , applyToAllSpaces = false ) { const resp = await Promise . all ( jobs . map ( async ( job ) => { const jobId = job . id ; try { job . id = jobId ; await this . _saveJob ( job ) ; return { id : jobId , success : true } ; } catch ( { body } ) { return { id : jobId , success : false , error : body } ; } } ) ) ; if ( applyToAllSpaces === true ) { const canCreateGlobalJobs = await this . _jobSavedObjectService . canCreateGlobalJobs ( this . _request ) ; if ( canCreateGlobalJobs === true ) { await this . _jobSavedObjectService . updateJobsSpaces ( 'anomaly-detector' , jobs . map ( ( j ) => j . id ) , [ '*' ] , [ ] ) ; } } return resp ; } private async _saveJob ( job ) { return this . _mlClient . putJob ( { job_id : job . id , body : job . config } ) ; } private async _saveDatafeeds ( datafeeds ) { return await Promise . all ( datafeeds . map ( async ( datafeed ) => { try { await this . _saveDatafeed ( datafeed ) ; return { id : datafeed . id , success : true , started : false , awaitingMlNodeAllocation : false , } ; } catch ( { body } ) { return { id : datafeed . id , success : false , started : false , awaitingMlNodeAllocation : false , error : body , } ; } } ) ) ; } private async _saveDatafeed ( datafeed ) { return this . _mlClient . putDatafeed ( { datafeed_id : datafeed . id , body : datafeed . config , } , this . _authorizationHeader ) ; } private async _startDatafeeds ( datafeeds , start ? , end ? ) { const results = { } as { [ key ] : DatafeedResponse ; } ; for ( const datafeed of datafeeds ) { results [ datafeed . id ] = await this . _startDatafeed ( datafeed , start , end ) ; } return results ; } private async _startDatafeed ( datafeed , start , end ) { const result = { started : false } as DatafeedResponse ; let opened = false ; try { const { body } = await this . _mlClient . openJob ( { job_id : datafeed . config . job_id , } ) ; opened = body . opened ; } catch ( error ) { if ( error . statusCode === 409 ) { opened = true ; } else { opened = false ; result . started = false ; result . error = error . body ; } } if ( opened ) { try { const duration = { start : '0' } ; if ( start !== undefined ) { duration . start = String ( start ) ; } if ( end !== undefined ) { duration . end = String ( end ) ; } const { body : { started , node } , } = await this . _mlClient . startDatafeed < { started : boolean ; node : string ; } > ( { datafeed_id : datafeed . id , ... duration , } ) ; result . started = started ; result . awaitingMlNodeAllocation = node ?. length === 0 ; } catch ( { body } ) { result . started = false ; result . error = body ; } } return result ; } private async _updateResults ( results , saveResults ) { results . jobs . forEach ( ( j ) => { saveResults . jobs . forEach ( ( j2 ) => { if ( j . id === j2 . id ) { j . success = j2 . success ; if ( j2 . error !== undefined ) { j . error = j2 . error ; } } } ) ; } ) ; results . datafeeds . forEach ( ( d ) => { saveResults . datafeeds . forEach ( ( d2 ) => { if ( d . id === d2 . id ) { d . success = d2 . success ; d . started = d2 . started ; d . awaitingMlNodeAllocation = d2 . awaitingMlNodeAllocation ; if ( d2 . error !== undefined ) { d . error = d2 . error ; } } } ) ; } ) ; ( Object . keys ( results . kibana ) as Array < keyof DataRecognizerConfigResponse [ 'kibana' ] > ) . forEach ( ( category ) => { results . kibana [ category ] . forEach ( ( item ) => { const result = saveResults . savedObjects . find ( ( o ) => o . id === item . id ) ; if ( result !== undefined ) { item . exists = result . exists ; if ( result . error === undefined ) { item . success = true ; } else { item . success = false ; item . error = result . error ; } } } ) ; } ) ; } private _createResultsTemplate ( moduleConfig ) { const results = { } as DataRecognizerConfigResponse ; const reducedConfig = { jobs : moduleConfig . jobs , datafeeds : moduleConfig . datafeeds , kibana : moduleConfig . kibana , } ; function createResultsItems ( configItems , resultItems , index ) { resultItems [ index ] = [ ] ; configItems . forEach ( ( j ) => { resultItems [ index ] . push ( { id : j . id , success : false , } ) ; } ) ; } ( Object . keys ( reducedConfig ) as Array < keyof typeof reducedConfig > ) . forEach ( ( i ) => { if ( Array . isArray ( reducedConfig [ i ] ) ) { createResultsItems ( reducedConfig [ i ] as any [ ] , results , i ) ; } else { results [ i ] = { } as any ; Object . keys ( reducedConfig [ i ] ) . forEach ( ( k ) => { createResultsItems ( ( reducedConfig [ i ] as Module [ 'kibana' ] ) [ k ] as any [ ] , results [ i ] , k ) ; } ) ; } } ) ; return results ; } private _updateDatafeedIndices ( moduleConfig ) { const indexPatternNames = splitIndexPatternNames ( this . _indexPatternName ) ; moduleConfig . datafeeds . forEach ( ( df ) => { const newIndices = [ ] ; const currentIndices = df . config . indexes !== undefined ? df . config . indexes : df . config . indices ; currentIndices . forEach ( ( index ) => { if ( index === INDEX_PATTERN_NAME ) { newIndices . push ( ... indexPatternNames ) ; } else { newIndices . push ( index ) ; } } ) ; delete df . config . indexes ; df . config . indices = newIndices ; } ) ; } private _updateJobUrlIndexPatterns ( moduleConfig ) { if ( Array . isArray ( moduleConfig . jobs ) ) { moduleConfig . jobs . forEach ( ( job ) => { if ( job . config . custom_settings && job . config . custom_settings . custom_urls ) { job . config . custom_settings . custom_urls . forEach ( ( cUrl ) => { const url = cUrl . url_value ; if ( url . match ( INDEX_PATTERN_ID ) ) { const newUrl = url . replace ( new RegExp ( INDEX_PATTERN_ID , 'g' ) , this . _indexPatternId as string ) ; cUrl . url_value = newUrl ; } } ) ; } } ) ; } } private _doJobUrlsContainIndexPatternId ( moduleConfig ) { if ( Array . isArray ( moduleConfig . jobs ) ) { for ( const job of moduleConfig . jobs ) { if ( job . config . custom_settings && job . config . custom_settings . custom_urls ) { for ( const cUrl of job . config . custom_settings . custom_urls ) { if ( cUrl . url_value . match ( INDEX_PATTERN_ID ) ) { return true ; } } } } } return false ; } private _updateSavedObjectIndexPatterns ( moduleConfig ) { if ( moduleConfig . kibana ) { Object . keys ( moduleConfig . kibana ) . forEach ( ( category ) => { moduleConfig . kibana [ category ] ! . forEach ( ( item ) => { let jsonString = item . config . kibanaSavedObjectMeta ! . searchSourceJSON ; if ( jsonString . match ( INDEX_PATTERN_ID ) ) { jsonString = jsonString . replace ( new RegExp ( INDEX_PATTERN_ID , 'g' ) , this . _indexPatternId as string ) ; item . config . kibanaSavedObjectMeta ! . searchSourceJSON = jsonString ; } if ( category === SAVED_OBJECT_TYPES . VISUALIZATION ) { let visStateString = String ( item . config . visState ) ; if ( visStateString !== undefined && visStateString . match ( INDEX_PATTERN_NAME ) ) { visStateString = visStateString . replace ( new RegExp ( INDEX_PATTERN_NAME , 'g' ) , this . _indexPatternName ) ; item . config . visState = visStateString ; } } } ) ; } ) ; } } /**
     * Provides a time range of the last 3 months of data
     */ private async _getFallbackTimeRange ( timeField , query ? ) { const fieldsService = fieldsServiceProvider ( this . _client ) ; const timeFieldRange = await fieldsService . getTimeFieldRange ( this . _indexPatternName , timeField , query ) ; return { start : timeFieldRange . end . epoch - moment . duration ( 3 , 'months' ) . asMilliseconds ( ) , end : timeFieldRange . end . epoch , } ; } /**
     * Ensure the model memory limit for each job is not greater than
     * the max model memory setting for the cluster
     */ private async _updateModelMemoryLimits ( moduleConfig , estimateMML , start ? , end ? ) { if ( ! Array . isArray ( moduleConfig . jobs ) ) { return ; } if ( estimateMML && this . _jobsForModelMemoryEstimation . length > 0 ) { try { const firstJobTimeField = this . _jobsForModelMemoryEstimation [ 0 ] . job . config . data_description . time_field ! ; const isSameTimeFields = this . _jobsForModelMemoryEstimation . every ( ( { job } ) => job . config . data_description . time_field === firstJobTimeField ) ; if ( isSameTimeFields && ( start === undefined || end === undefined ) ) { const { start : fallbackStart , end : fallbackEnd } = await this . _getFallbackTimeRange ( firstJobTimeField , { match_all : { } } ) ; start = fallbackStart ; end = fallbackEnd ; } for ( const { job , query } of this . _jobsForModelMemoryEstimation ) { let earliestMs = start ; let latestMs = end ; if ( earliestMs === undefined || latestMs === undefined ) { const timeFieldRange = await this . _getFallbackTimeRange ( job . config . data_description . time_field ! , query ) ; earliestMs = timeFieldRange . start ; latestMs = timeFieldRange . end ; } const { modelMemoryLimit } = await this . _calculateModelMemoryLimit ( job . config . analysis_config , this . _indexPatternName , query , job . config . data_description . time_field ! , earliestMs , latestMs ) ; if ( ! job . config . analysis_limits ) { job . config . analysis_limits = { } as AnalysisLimits ; } job . config . analysis_limits . model_memory_limit = modelMemoryLimit ; } } catch ( error ) { mlLog . warn ( `Data recognizer could not estimate model memory limit ${ JSON . stringify ( error . body ) }` ) ; } } const { body : { limits } , } = await this . _mlClient . info < MlInfoResponse > ( ) ; const maxMml = limits . max_model_memory_limit ; if ( ! maxMml ) { return ; } const maxBytes = numeral ( maxMml . toUpperCase ( ) ) . value ( ) ; for ( const job of moduleConfig . jobs ) { const mml = job . config ?. analysis_limits ?. model_memory_limit ; if ( mml !== undefined ) { const mmlBytes = numeral ( mml . toUpperCase ( ) ) . value ( ) ; if ( mmlBytes > maxBytes ) { if ( ! job . config . analysis_limits ) { job . config . analysis_limits = { } as AnalysisLimits ; } job . config . analysis_limits . model_memory_limit = maxMml ; } } } } private _doSavedObjectsContainIndexPatternId ( moduleConfig ) { if ( moduleConfig . kibana ) { for ( const category of Object . keys ( moduleConfig . kibana ) ) { for ( const item of moduleConfig . kibana [ category ] ! ) { const jsonString = item . config . kibanaSavedObjectMeta ! . searchSourceJSON ; if ( jsonString . match ( INDEX_PATTERN_ID ) ) { return true ; } } } } return false ; } public applyJobConfigOverrides ( moduleConfig , jobOverrides ? , jobPrefix = '' ) { if ( jobOverrides === undefined || jobOverrides === null ) { return ; } if ( typeof jobOverrides !== 'object' ) { throw Boom . badRequest ( `Incompatible jobOverrides type (${ typeof jobOverrides }). It needs to be an object or array of objects.` ) ; } const overrides = Array . isArray ( jobOverrides ) ? jobOverrides : [ jobOverrides ] ; const { jobs } = moduleConfig ; const generalOverrides = [ ] ; const jobSpecificOverrides = [ ] ; overrides . forEach ( ( override ) => { if ( isGeneralJobOverride ( override ) ) { generalOverrides . push ( override ) ; } else { jobSpecificOverrides . push ( override ) ; } } ) ; if ( generalOverrides . some ( ( override ) => ! ! override . analysis_limits ?. model_memory_limit ) ) { this . _jobsForModelMemoryEstimation = [ ] ; } else { this . _jobsForModelMemoryEstimation = moduleConfig . jobs . filter ( ( job ) => { const override = jobSpecificOverrides . find ( ( o ) => `${ jobPrefix }${ o . job_id }` === job . id ) ; return override ?. analysis_limits ?. model_memory_limit === undefined ; } ) . map ( ( job ) => ( { job , query : moduleConfig . datafeeds . find ( ( d ) => d . config . job_id === job . id ) ?. config . query || null , } ) ) ; } function processArrayValues ( source , update ) { if ( typeof source !== 'object' || typeof update !== 'object' ) { return ; } Object . keys ( source ) . forEach ( ( key ) => { const sourceValue = source [ key ] ; const updateValue = update [ key ] ; if ( typeof sourceValue !== 'object' || sourceValue === null || typeof updateValue !== 'object' || updateValue === null ) { return ; } if ( Array . isArray ( sourceValue ) && Array . isArray ( updateValue ) ) { source [ key ] = updateValue ; } else { processArrayValues ( sourceValue , updateValue ) ; } } ) ; } generalOverrides . forEach ( ( generalOverride ) => { jobs . forEach ( ( job ) => { merge ( job . config , generalOverride ) ; processArrayValues ( job . config , generalOverride ) ; } ) ; } ) ; jobSpecificOverrides . forEach ( ( jobSpecificOverride ) => { const job = jobs . find ( ( j ) => j . id === `${ jobPrefix }${ jobSpecificOverride . job_id }` ) ; if ( job !== undefined ) { delete jobSpecificOverride . job_id ; merge ( job . config , jobSpecificOverride ) ; processArrayValues ( job . config , jobSpecificOverride ) ; } } ) ; } public applyDatafeedConfigOverrides ( moduleConfig , datafeedOverrides ? , jobPrefix = '' ) { if ( datafeedOverrides !== undefined && datafeedOverrides !== null ) { if ( typeof datafeedOverrides !== 'object' ) { throw Boom . badRequest ( `Incompatible datafeedOverrides type (${ typeof datafeedOverrides }). It needs to be an object or array of objects.` ) ; } const overrides = Array . isArray ( datafeedOverrides ) ? datafeedOverrides : [ datafeedOverrides ] ; const { datafeeds } = moduleConfig ; function overwriteObjects ( source , update ) { Object . entries ( update ) . forEach ( ( [ key , val ] ) => { if ( typeof val === 'object' ) { switch ( key ) { case 'query' : case 'aggregations' : case 'aggs' : case 'script_fields' : source [ key ] = val as any ; break ; default : break ; } } } ) ; } const generalOverrides = [ ] ; const datafeedSpecificOverrides = [ ] ; overrides . forEach ( ( o ) => { if ( o . datafeed_id === undefined && o . job_id === undefined ) { generalOverrides . push ( o ) ; } else { datafeedSpecificOverrides . push ( o ) ; } } ) ; generalOverrides . forEach ( ( o ) => { datafeeds . forEach ( ( { config } ) => { merge ( config , o ) ; overwriteObjects ( config , o ) ; } ) ; } ) ; datafeedSpecificOverrides . forEach ( ( o ) => { const tempId = String ( o . datafeed_id !== undefined ? o . datafeed_id : o . job_id ) ; const dId = prefixDatafeedId ( tempId , jobPrefix ) ; const datafeed = datafeeds . find ( ( d ) => d . id === dId ) ; if ( datafeed !== undefined ) { delete o . job_id ; delete o . datafeed_id ; merge ( datafeed . config , o ) ; overwriteObjects ( datafeed . config , o ) ; } } ) ; } } } export function dataRecognizerFactory ( client , mlClient , savedObjectsClient , dataViewsService , jobSavedObjectService , request ) { return new DataRecognizer ( client , mlClient , savedObjectsClient , dataViewsService , jobSavedObjectService , request ) ; }