import Boom from '@hapi/boom' ; import { IScopedClusterClient } from 'kibana/server' ; import { JOB_MAP_NODE_TYPES , JobMapNodeTypes , } from '../../../common/constants/data_frame_analytics' ; import { TrainedModelConfigResponse } from '../../../common/types/trained_models' ; import { AnalyticsMapEdgeElement , AnalyticsMapReturnType , AnalyticsMapNodeElement , DataFrameAnalyticsStats , MapElements , } from '../../../common/types/data_frame_analytics' ; import { INDEX_META_DATA_CREATED_BY } from '../../../../file_upload/common' ; import { getAnalysisType } from '../../../common/util/analytics_utils' ; import { ExtendAnalyticsMapArgs , GetAnalyticsMapArgs , InitialElementsReturnType , isCompleteInitialReturnType , isAnalyticsMapEdgeElement , isAnalyticsMapNodeElement , isIndexPatternLinkReturnType , isJobDataLinkReturnType , isTransformLinkReturnType , NextLinkReturnType , } from './types' ; import type { MlClient } from '../../lib/ml_client' ; export class AnalyticsManager { private _client : IScopedClusterClient ; private _mlClient : MlClient ; private _inferenceModels : TrainedModelConfigResponse [ ] ; private _jobStats : DataFrameAnalyticsStats [ ] ; constructor ( mlClient , client ) { this . _client = client ; this . _mlClient = mlClient ; this . _inferenceModels = [ ] ; this . _jobStats = [ ] ; } public set jobStats ( stats ) { this . _jobStats = stats ; } public get jobStats ( ) { return this . _jobStats ; } public set inferenceModels ( models ) { this . _inferenceModels = models ; } public get inferenceModels ( ) { return this . _inferenceModels ; } async setInferenceModels ( ) { try { const models = await this . getAnalyticsModels ( ) ; this . inferenceModels = models ; } catch ( error ) { console . error ( 'Unable to fetch inference models' , error ) ; } } async setJobStats ( ) { try { const jobStats = await this . getAnalyticsStats ( ) ; this . jobStats = jobStats ; } catch ( error ) { console . error ( 'Unable to fetch job stats' , error ) ; } } private isDuplicateElement ( analyticsId , elements ) { let isDuplicate = false ; elements . forEach ( ( elem ) => { if ( isAnalyticsMapNodeElement ( elem ) && elem . data . label === analyticsId && elem . data . type === JOB_MAP_NODE_TYPES . ANALYTICS ) { isDuplicate = true ; } } ) ; return isDuplicate ; } private async getAnalyticsModelData ( modelId ) { const resp = await this . _mlClient . getTrainedModels ( { model_id : modelId , } ) ; const modelData = resp ?. body ?. trained_model_configs [ 0 ] ; return modelData ; } private async getAnalyticsModels ( ) { const resp = await this . _mlClient . getTrainedModels ( ) ; const models = resp ?. body ?. trained_model_configs ; return models ; } private async getAnalyticsStats ( ) { const resp = await this . _mlClient . getDataFrameAnalyticsStats < { data_frame_analytics : DataFrameAnalyticsStats [ ] ; } > ( { size : 1000 } ) ; const stats = resp ?. body ?. data_frame_analytics ; return stats ; } private async getAnalyticsData ( analyticsId ? ) { const options = analyticsId ? { id : analyticsId , } : undefined ; const resp = await this . _mlClient . getDataFrameAnalytics ( options ) ; let jobData = analyticsId ? resp ?. body ?. data_frame_analytics [ 0 ] : resp ?. body ?. data_frame_analytics ; if ( analyticsId !== undefined ) { const jobStats = this . findJobStats ( analyticsId ) ; jobData = { ... jobData , stats : { ... jobStats } } ; } else { jobData = jobData . map ( ( job ) => { const jobStats = this . findJobStats ( job . id ) ; return { ... job , stats : { ... jobStats } } ; } ) ; } return jobData ; } private async getIndexData ( index ) { const indexData = await this . _client . asInternalUser . indices . get ( { index , } ) ; return indexData ?. body ; } private async getTransformData ( transformId ) { const transform = await this . _client . asInternalUser . transform . getTransform ( { transform_id : transformId , } ) ; const transformData = transform ?. body ?. transforms [ 0 ] ; return transformData ; } private findJobModel ( analyticsId ) { return this . inferenceModels . find ( ( model ) => model . metadata ?. analytics_config ?. id === analyticsId ) ; } private findJobStats ( analyticsId ) { return this . jobStats . find ( ( js ) => js . id === analyticsId ) ; } private async getNextLink ( { id , type , } ) { try { if ( type === JOB_MAP_NODE_TYPES . INDEX ) { const indexData = await this . getIndexData ( id ) ; let isWildcardIndexPattern = false ; if ( id . includes ( '*' ) ) { isWildcardIndexPattern = true ; } const meta = indexData [ id ] ?. mappings ?. _meta ; return { isWildcardIndexPattern , isIndexPattern : true , indexData , meta } ; } else if ( type . includes ( JOB_MAP_NODE_TYPES . ANALYTICS ) ) { const jobData = await this . getAnalyticsData ( id ) ; return { jobData , isJob : true } ; } else if ( type === JOB_MAP_NODE_TYPES . TRANSFORM ) { const transformData = await this . getTransformData ( id ) ; return { transformData , isTransform : true } ; } } catch ( error ) { throw Boom . badData ( error . message ? error . message : error ) ; } } private getAnalyticsModelElements ( analyticsId ) { const analyticsModel = this . findJobModel ( analyticsId ) ; let modelElement ; let edgeElement ; if ( analyticsModel !== undefined ) { const modelId = `${ analyticsModel . model_id }-${ JOB_MAP_NODE_TYPES . TRAINED_MODEL }` ; modelElement = { data : { id : modelId , label : analyticsModel . model_id , type : JOB_MAP_NODE_TYPES . TRAINED_MODEL , } , } ; edgeElement = { data : { id : `${ analyticsId }-${ JOB_MAP_NODE_TYPES . ANALYTICS }~${ modelId }` , source : `${ analyticsId }-${ JOB_MAP_NODE_TYPES . ANALYTICS }` , target : modelId , } , } ; } return { modelElement , modelDetails : analyticsModel , edgeElement } ; } private getIndexPatternElements ( indexData , previousNodeId ) { const result = { elements : [ ] , details : { } } ; Object . keys ( indexData ) . forEach ( ( indexId ) => { const nodeId = `${ indexId }-${ JOB_MAP_NODE_TYPES . INDEX }` ; result . elements . push ( { data : { id : nodeId , label : indexId , type : JOB_MAP_NODE_TYPES . INDEX } , } ) ; result . details [ nodeId ] = indexData [ indexId ] ; result . elements . push ( { data : { id : `${ previousNodeId }~${ nodeId }` , source : nodeId , target : previousNodeId , } , } ) ; } ) ; return result ; } param modelId async getInitialElementsModelRoot ( modelId ) { const resultElements = [ ] ; const modelElements = [ ] ; const details = { } ; let data = await this . getAnalyticsModelData ( modelId ) ; const modelNodeId = `${ data . model_id }-${ JOB_MAP_NODE_TYPES . TRAINED_MODEL }` ; const sourceJobId = data ?. metadata ?. analytics_config ?. id ; let nextLinkId ; let nextType ; let previousNodeId ; modelElements . push ( { data : { id : modelNodeId , label : data . model_id , type : JOB_MAP_NODE_TYPES . TRAINED_MODEL , isRoot : true , } , } ) ; details [ modelNodeId ] = data ; if ( sourceJobId !== undefined ) { try { data = await this . getAnalyticsData ( sourceJobId ) ; nextLinkId = data ?. source ?. index [ 0 ] ; nextType = JOB_MAP_NODE_TYPES . INDEX ; previousNodeId = `${ data . id }-${ JOB_MAP_NODE_TYPES . ANALYTICS }` ; resultElements . push ( { data : { id : previousNodeId , label : data . id , type : JOB_MAP_NODE_TYPES . ANALYTICS , analysisType : getAnalysisType ( data ?. analysis ) , } , } ) ; modelElements . push ( { data : { id : `${ previousNodeId }~${ modelNodeId }` , source : previousNodeId , target : modelNodeId , } , } ) ; details [ previousNodeId ] = data ; } catch ( error ) { if ( error . statusCode !== 404 ) { throw error . body ?? error ; } } } return { data , details , resultElements , modelElements , nextLinkId , nextType , previousNodeId } ; } param jobId async getInitialElementsJobRoot ( jobId ) { const resultElements = [ ] ; const modelElements = [ ] ; const details = { } ; const data = await this . getAnalyticsData ( jobId ) ; const nextLinkId = data ?. source ?. index [ 0 ] ; const nextType = JOB_MAP_NODE_TYPES . INDEX ; const previousNodeId = `${ data . id }-${ JOB_MAP_NODE_TYPES . ANALYTICS }` ; resultElements . push ( { data : { id : previousNodeId , label : data . id , type : JOB_MAP_NODE_TYPES . ANALYTICS , analysisType : getAnalysisType ( data ?. analysis ) , isRoot : true , } , } ) ; details [ previousNodeId ] = data ; const { modelElement , modelDetails , edgeElement } = this . getAnalyticsModelElements ( jobId ) ; if ( isAnalyticsMapNodeElement ( modelElement ) ) { modelElements . push ( modelElement ) ; details [ modelElement . data . id ] = modelDetails ; } if ( isAnalyticsMapEdgeElement ( edgeElement ) ) { modelElements . push ( edgeElement ) ; } return { data , details , resultElements , modelElements , nextLinkId , nextType , previousNodeId } ; } param jobId param modelId async getAnalyticsMap ( { analyticsId , modelId , } ) { const result = { elements : [ ] , details : { } , error : null } ; const modelElements = [ ] ; const indexPatternElements = [ ] ; try { await Promise . all ( [ this . setInferenceModels ( ) , this . setJobStats ( ) ] ) ; let initialData = { } as InitialElementsReturnType ; if ( analyticsId !== undefined ) { initialData = await this . getInitialElementsJobRoot ( analyticsId ) ; } else if ( modelId !== undefined ) { initialData = await this . getInitialElementsModelRoot ( modelId ) ; } const { resultElements , details : initialDetails , modelElements : initialModelElements , } = initialData ; result . elements . push ( ... resultElements ) ; result . details = initialDetails ; modelElements . push ( ... initialModelElements ) ; if ( isCompleteInitialReturnType ( initialData ) ) { let { data , nextLinkId , nextType , previousNodeId } = initialData ; let complete = false ; let link ; let count = 0 ; let rootTransform ; let rootIndexPattern ; let modelElement ; let modelDetails ; let edgeElement ; while ( complete === false ) { count ++ ; if ( count >= 100 ) { break ; } try { link = await this . getNextLink ( { id : nextLinkId , type : nextType , } ) ; } catch ( error ) { result . error = error . message || 'Something went wrong' ; break ; } if ( isIndexPatternLinkReturnType ( link ) && link . isIndexPattern === true ) { if ( link . isWildcardIndexPattern === true ) { const { details , elements } = this . getIndexPatternElements ( link . indexData , previousNodeId ) ; indexPatternElements . push ( ... elements ) ; result . details = { ... result . details , ... details } ; complete = true ; } else { const nodeId = `${ nextLinkId }-${ JOB_MAP_NODE_TYPES . INDEX }` ; result . elements . unshift ( { data : { id : nodeId , label : nextLinkId , type : JOB_MAP_NODE_TYPES . INDEX } , } ) ; result . details [ nodeId ] = link . indexData ; } if ( link . isWildcardIndexPattern === false && ( link . meta === undefined || link . meta ?. created_by . includes ( INDEX_META_DATA_CREATED_BY ) ) ) { rootIndexPattern = nextLinkId ; complete = true ; break ; } if ( link . meta ?. created_by === 'data-frame-analytics' ) { nextLinkId = link . meta . analytics ; nextType = JOB_MAP_NODE_TYPES . ANALYTICS ; } if ( link . meta ?. created_by === JOB_MAP_NODE_TYPES . TRANSFORM ) { nextLinkId = link . meta . _transform ?. transform ; nextType = JOB_MAP_NODE_TYPES . TRANSFORM ; } } else if ( isJobDataLinkReturnType ( link ) && link . isJob === true ) { data = link . jobData ; const nodeId = `${ data . id }-${ JOB_MAP_NODE_TYPES . ANALYTICS }` ; previousNodeId = nodeId ; result . elements . unshift ( { data : { id : nodeId , label : data . id , type : JOB_MAP_NODE_TYPES . ANALYTICS , analysisType : getAnalysisType ( data ?. analysis ) , } , } ) ; result . details [ nodeId ] = data ; nextLinkId = data ?. source ?. index [ 0 ] ; nextType = JOB_MAP_NODE_TYPES . INDEX ; ( { modelElement , modelDetails , edgeElement } = this . getAnalyticsModelElements ( data . id ) ) ; if ( isAnalyticsMapNodeElement ( modelElement ) ) { modelElements . push ( modelElement ) ; result . details [ modelElement . data . id ] = modelDetails ; } if ( isAnalyticsMapEdgeElement ( edgeElement ) ) { modelElements . push ( edgeElement ) ; } } else if ( isTransformLinkReturnType ( link ) && link . isTransform === true ) { data = link . transformData ; const nodeId = `${ data . id }-${ JOB_MAP_NODE_TYPES . TRANSFORM }` ; previousNodeId = nodeId ; rootTransform = data . dest . index ; result . elements . unshift ( { data : { id : nodeId , label : data . id , type : JOB_MAP_NODE_TYPES . TRANSFORM } , } ) ; result . details [ nodeId ] = data ; nextLinkId = data ?. source ?. index [ 0 ] ; nextType = JOB_MAP_NODE_TYPES . INDEX ; } } const elemLength = result . elements . length - 1 ; for ( let i = 0 ; i < elemLength ; i ++ ) { const currentElem = result . elements [ i ] ; const nextElem = result . elements [ i + 1 ] ; if ( currentElem !== undefined && nextElem !== undefined && currentElem ?. data ?. id . includes ( '*' ) === false && nextElem ?. data ?. id . includes ( '*' ) === false ) { result . elements . push ( { data : { id : `${ currentElem . data . id }~${ nextElem . data . id }` , source : currentElem . data . id , target : nextElem . data . id , } , } ) ; } } if ( rootTransform !== undefined || rootIndexPattern !== undefined ) { const jobs = await this . getAnalyticsData ( ) ; const comparator = rootTransform !== undefined ? rootTransform : rootIndexPattern ; for ( let i = 0 ; i < jobs . length ; i ++ ) { if ( jobs [ i ] ?. source ?. index [ 0 ] === comparator && this . isDuplicateElement ( jobs [ i ] . id , result . elements ) === false ) { const nodeId = `${ jobs [ i ] . id }-${ JOB_MAP_NODE_TYPES . ANALYTICS }` ; result . elements . push ( { data : { id : nodeId , label : jobs [ i ] . id , type : JOB_MAP_NODE_TYPES . ANALYTICS , analysisType : getAnalysisType ( jobs [ i ] ?. analysis ) , } , } ) ; result . details [ nodeId ] = jobs [ i ] ; const source = `${ comparator }-${ JOB_MAP_NODE_TYPES . INDEX }` ; result . elements . push ( { data : { id : `${ source }~${ nodeId }` , source , target : nodeId , } , } ) ; ( { modelElement , modelDetails , edgeElement } = this . getAnalyticsModelElements ( jobs [ i ] . id ) ) ; if ( isAnalyticsMapNodeElement ( modelElement ) ) { modelElements . push ( modelElement ) ; result . details [ modelElement . data . id ] = modelDetails ; } if ( isAnalyticsMapEdgeElement ( edgeElement ) ) { modelElements . push ( edgeElement ) ; } } } } } result . elements . push ( ... modelElements , ... indexPatternElements ) ; return result ; } catch ( error ) { result . error = error . message || 'An error occurred fetching map' ; return result ; } } async extendAnalyticsMapForAnalyticsJob ( { analyticsId , index , } ) { const result = { elements : [ ] , details : { } , error : null } ; try { await Promise . all ( [ this . setInferenceModels ( ) , this . setJobStats ( ) ] ) ; const jobs = await this . getAnalyticsData ( ) ; let rootIndex ; let rootIndexNodeId ; if ( analyticsId !== undefined ) { const jobData = await this . getAnalyticsData ( analyticsId ) ; const currentJobNodeId = `${ jobData . id }-${ JOB_MAP_NODE_TYPES . ANALYTICS }` ; rootIndex = Array . isArray ( jobData ?. dest ?. index ) ? jobData ?. dest ?. index [ 0 ] : jobData ?. dest ?. index ; rootIndexNodeId = `${ rootIndex }-${ JOB_MAP_NODE_TYPES . INDEX }` ; const { modelElement , modelDetails , edgeElement } = this . getAnalyticsModelElements ( analyticsId ) ; if ( isAnalyticsMapNodeElement ( modelElement ) ) { result . elements . push ( modelElement ) ; result . details [ modelElement . data . id ] = modelDetails ; } if ( isAnalyticsMapEdgeElement ( edgeElement ) ) { result . elements . push ( edgeElement ) ; } const rootIndexDetails = await this . getIndexData ( rootIndex ) ; result . elements . push ( { data : { id : rootIndexNodeId , label : rootIndex , type : JOB_MAP_NODE_TYPES . INDEX , } , } ) ; result . details [ rootIndexNodeId ] = rootIndexDetails ; result . elements . push ( { data : { id : `${ currentJobNodeId }~${ rootIndexNodeId }` , source : currentJobNodeId , target : rootIndexNodeId , } , } ) ; } else { rootIndex = index ; rootIndexNodeId = `${ rootIndex }-${ JOB_MAP_NODE_TYPES . INDEX }` ; } for ( let i = 0 ; i < jobs . length ; i ++ ) { if ( jobs [ i ] ?. source ?. index [ 0 ] === rootIndex && this . isDuplicateElement ( jobs [ i ] . id , result . elements ) === false ) { const nodeId = `${ jobs [ i ] . id }-${ JOB_MAP_NODE_TYPES . ANALYTICS }` ; result . elements . push ( { data : { id : nodeId , label : jobs [ i ] . id , type : JOB_MAP_NODE_TYPES . ANALYTICS , analysisType : getAnalysisType ( jobs [ i ] ?. analysis ) , } , } ) ; result . details [ nodeId ] = jobs [ i ] ; result . elements . push ( { data : { id : `${ rootIndexNodeId }~${ nodeId }` , source : rootIndexNodeId , target : nodeId , } , } ) ; } } } catch ( error ) { result . error = error . message || 'An error occurred fetching map' ; return result ; } return result ; } }