import type * as estypes from '@elastic/elasticsearch/lib/api/typesWithBodyKey' ; import { i18n } from '@kbn/i18n' ; import { IScopedClusterClient } from 'kibana/server' ; import { JOB_STATE , DATAFEED_STATE } from '../../../common/constants/states' ; import { fillResultsWithTimeouts , isRequestTimeout } from './error_utils' ; import { Datafeed , DatafeedStats } from '../../../common/types/anomaly_detection_jobs' ; import type { MlClient } from '../../lib/ml_client' ; export interface MlDatafeedsResponse { datafeeds : Datafeed [ ] ; count : number ; } export interface MlDatafeedsStatsResponse { datafeeds : DatafeedStats [ ] ; count : number ; } interface Results { [ id ] : { started ? : estypes . MlStartDatafeedResponse [ 'started' ] ; stopped ? : estypes . MlStopDatafeedResponse [ 'stopped' ] ; error ? : any ; } ; } export type DatafeedsService = ReturnType < typeof datafeedsProvider > ; export function datafeedsProvider ( client , mlClient ) { async function forceStartDatafeeds ( datafeedIds , start ? , end ? ) { const jobIds = await getJobIdsByDatafeedId ( ) ; const doStartsCalled = datafeedIds . reduce ( ( acc , cur ) => { acc [ cur ] = false ; return acc ; } , { } as { [ id ] : boolean ; } ) ; const results = { } ; async function doStart ( datafeedId ) { if ( doStartsCalled [ datafeedId ] === false ) { doStartsCalled [ datafeedId ] = true ; try { await startDatafeed ( datafeedId , start , end ) ; return { started : true } ; } catch ( { body } ) { return { started : false , error : body } ; } } else { return { started : true } ; } } for ( const datafeedId of datafeedIds ) { const jobId = jobIds [ datafeedId ] ; if ( jobId !== undefined ) { try { if ( await openJob ( jobId ) ) { results [ datafeedId ] = await doStart ( datafeedId ) ; } } catch ( error ) { if ( isRequestTimeout ( error ) ) { results [ datafeedId ] = await doStart ( datafeedId ) ; return fillResultsWithTimeouts ( results , datafeedId , datafeedIds , JOB_STATE . OPENED ) ; } results [ datafeedId ] = { started : false , error : error . body } ; } } else { results [ datafeedId ] = { started : false , error : i18n . translate ( 'xpack.ml.models.jobService.jobHasNoDatafeedErrorMessage' , { defaultMessage : 'Job has no datafeed' , } ) , } ; } } return results ; } async function openJob ( jobId ) { let opened = false ; try { const { body } = await mlClient . openJob ( { job_id : jobId } ) ; opened = body . opened ; } catch ( error ) { if ( error . statusCode === 409 ) { opened = true ; } else { throw error ; } } return opened ; } async function startDatafeed ( datafeedId , start ? , end ? ) { return mlClient . startDatafeed ( { datafeed_id : datafeedId , body : { start : start !== undefined ? String ( start ) : undefined , end : end !== undefined ? String ( end ) : undefined , } , } ) ; } async function stopDatafeeds ( datafeedIds ) { const results = { } ; for ( const datafeedId of datafeedIds ) { try { const { body } = await mlClient . stopDatafeed ( { datafeed_id : datafeedId , } ) ; results [ datafeedId ] = { stopped : body . stopped } ; } catch ( error ) { if ( isRequestTimeout ( error ) ) { return fillResultsWithTimeouts ( results , datafeedId , datafeedIds , DATAFEED_STATE . STOPPED ) ; } else { results [ datafeedId ] = { stopped : false , error : error . body , } ; } } } return results ; } async function forceDeleteDatafeed ( datafeedId ) { const { body } = await mlClient . deleteDatafeed < { acknowledged : boolean ; } > ( { datafeed_id : datafeedId , force : true , } ) ; return body ; } async function getDatafeedIdsByJobId ( ) { const { body : { datafeeds } , } = await mlClient . getDatafeeds < MlDatafeedsResponse > ( ) ; return datafeeds . reduce ( ( acc , cur ) => { acc [ cur . job_id ] = cur . datafeed_id ; return acc ; } , { } as { [ id ] : string ; } ) ; } async function getJobIdsByDatafeedId ( ) { const { body : { datafeeds } , } = await mlClient . getDatafeeds < MlDatafeedsResponse > ( ) ; return datafeeds . reduce ( ( acc , cur ) => { acc [ cur . datafeed_id ] = cur . job_id ; return acc ; } , { } as { [ id ] : string ; } ) ; } async function getDatafeedByJobId ( jobId , excludeGenerated ? ) ; async function getDatafeedByJobId ( jobId , excludeGenerated ? ) ; async function getDatafeedByJobId ( jobId , excludeGenerated ? ) { const jobIds = Array . isArray ( jobId ) ? jobId : [ jobId ] ; async function findDatafeed ( ) { const { body : { datafeeds } , } = await mlClient . getDatafeeds ( excludeGenerated ? { exclude_generated : true } : { } ) ; if ( typeof jobId === 'string' ) { return datafeeds . find ( ( v ) => v . job_id === jobId ) ; } if ( Array . isArray ( jobId ) ) { return datafeeds . filter ( ( v ) => jobIds . includes ( v . job_id ) ) ; } } const assumedDefaultDatafeedId = jobIds . map ( ( v ) => `datafeed-${ v }` ) . join ( ',' ) ; try { const { body : { datafeeds : datafeedsResults } , } = await mlClient . getDatafeeds ( { datafeed_id : assumedDefaultDatafeedId , ... ( excludeGenerated ? { exclude_generated : true } : { } ) , } ) ; if ( Array . isArray ( datafeedsResults ) ) { const result = datafeedsResults . filter ( ( d ) => jobIds . includes ( d . job_id ) ) ; if ( typeof jobId === 'string' ) { if ( datafeedsResults . length === 1 && datafeedsResults [ 0 ] . job_id === jobId ) { return datafeedsResults [ 0 ] ; } else { return await findDatafeed ( ) ; } } if ( result . length === jobIds . length ) { return datafeedsResults ; } else { return await findDatafeed ( ) ; } } else { return await findDatafeed ( ) ; } } catch ( e ) { return await findDatafeed ( ) ; } } return { forceStartDatafeeds , stopDatafeeds , forceDeleteDatafeed , getDatafeedIdsByJobId , getJobIdsByDatafeedId , getDatafeedByJobId , } ; }