import { errors } from '@elastic/elasticsearch' ; import { i18n } from '@kbn/i18n' ; import type { TransportResult } from '@elastic/elasticsearch' ; import { schema } from '@kbn/config-schema' ; import { IScopedClusterClient , SavedObjectsClientContract } from 'kibana/server' ; import { API_BASE_PATH } from '../../common/constants' ; import { MlOperation , ML_UPGRADE_OP_TYPE } from '../../common/types' ; import { versionCheckHandlerWrapper } from '../lib/es_version_precheck' ; import { RouteDependencies } from '../types' ; const findMlOperation = async ( savedObjectsClient , snapshotId ) => { return savedObjectsClient . find < MlOperation > ( { type : ML_UPGRADE_OP_TYPE , search : `"${ snapshotId }"` , searchFields : [ 'snapshotId' ] , } ) ; } ; const createMlOperation = async ( savedObjectsClient , attributes ) => { const foundSnapshots = await findMlOperation ( savedObjectsClient , attributes . snapshotId ) ; if ( foundSnapshots ?. total > 0 ) { throw new Error ( `A ML operation is already in progress for snapshot: ${ attributes . snapshotId }` ) ; } return savedObjectsClient . create < MlOperation > ( ML_UPGRADE_OP_TYPE , attributes ) ; } ; const deleteMlOperation = ( savedObjectsClient , id ) => { return savedObjectsClient . delete ( ML_UPGRADE_OP_TYPE , id ) ; } ; const verifySnapshotUpgrade = async ( esClient , snapshot ) : Promise < { isSuccessful : boolean ; error ? : errors . ResponseError ; } > => { const { snapshotId , jobId } = snapshot ; try { const { body : deprecations } = await esClient . asCurrentUser . migration . deprecations ( ) ; const mlSnapshotDeprecations = deprecations . ml_settings . filter ( ( deprecation ) => { return /[Mm]odel snapshot/ . test ( deprecation . message ) ; } ) ; if ( typeof mlSnapshotDeprecations === 'undefined' || mlSnapshotDeprecations . length === 0 ) { return { isSuccessful : true , } ; } const isSuccessful = Boolean ( mlSnapshotDeprecations . find ( ( snapshotDeprecation ) => { const regex = /(?<=\[).*?(?=\])/g ; const matches = snapshotDeprecation . message . match ( regex ) ; if ( matches ?. length === 3 ) { return matches [ 0 ] === snapshotId && matches [ 1 ] === jobId ? false : true ; } return false ; } ) ) ; return { isSuccessful , } ; } catch ( e ) { return { isSuccessful : false , error : e , } ; } } ; interface ModelSnapshotUpgradeStatus { model_snapshot_upgrades : Array < { state : 'saving_new_state' | 'loading_old_state' | 'stopped' | 'failed' ; } > ; } const getModelSnapshotUpgradeStatus = async ( esClient , jobId , snapshotId ) => { try { const { body } = ( await esClient . asCurrentUser . transport . request ( { method : 'GET' , path : `/_ml/anomaly_detectors/${ jobId }/model_snapshots/${ snapshotId }/_upgrade/_stats` , } ) ) as TransportResult < ModelSnapshotUpgradeStatus > ; return body && body . model_snapshot_upgrades [ 0 ] ; } catch ( err ) { if ( err . statusCode !== 404 ) { throw err ; } } } ; export function registerMlSnapshotRoutes ( { router , log , lib : { handleEsError } , } ) { router . post ( { path : `${ API_BASE_PATH }/ml_snapshots` , validate : { body : schema . object ( { snapshotId : schema . string ( ) , jobId : schema . string ( ) , } ) , } , } , versionCheckHandlerWrapper ( async ( { core : { savedObjects : { client : savedObjectsClient } , elasticsearch : { client : esClient } , } , } , request , response ) => { try { const { snapshotId , jobId } = request . body ; const { body } = await esClient . asCurrentUser . ml . upgradeJobSnapshot ( { job_id : jobId , snapshot_id : snapshotId , } ) ; const snapshotInfo = { nodeId : body . node , snapshotId , jobId , } ; if ( body . completed !== true ) { await createMlOperation ( savedObjectsClient , snapshotInfo ) ; } return response . ok ( { body : { ... snapshotInfo , status : body . completed === true ? 'complete' : 'in_progress' , } , } ) ; } catch ( error ) { return handleEsError ( { error , response } ) ; } } ) ) ; router . get ( { path : `${ API_BASE_PATH }/ml_snapshots/{jobId}/{snapshotId}` , validate : { params : schema . object ( { snapshotId : schema . string ( ) , jobId : schema . string ( ) , } ) , } , } , versionCheckHandlerWrapper ( async ( { core : { savedObjects : { client : savedObjectsClient } , elasticsearch : { client : esClient } , } , } , request , response ) => { try { const { snapshotId , jobId } = request . params ; await esClient . asCurrentUser . ml . getModelSnapshots ( { job_id : jobId , snapshot_id : snapshotId , } ) ; const foundSnapshots = await findMlOperation ( savedObjectsClient , snapshotId ) ; if ( typeof foundSnapshots === 'undefined' || foundSnapshots . total === 0 ) { return response . ok ( { body : { snapshotId , jobId , nodeId : undefined , status : 'idle' , } , } ) ; } const upgradeStatus = await getModelSnapshotUpgradeStatus ( esClient , jobId , snapshotId ) ; const snapshotOp = foundSnapshots . saved_objects [ 0 ] ; const snapshotInfo = { ... snapshotOp . attributes , } ; if ( upgradeStatus ) { if ( upgradeStatus . state === 'loading_old_state' || upgradeStatus . state === 'saving_new_state' ) { return response . ok ( { body : { ... snapshotInfo , status : 'in_progress' , } , } ) ; } else if ( upgradeStatus . state === 'failed' ) { return response . customError ( { statusCode : 500 , body : { message : i18n . translate ( 'xpack.upgradeAssistant.ml_snapshots.modelSnapshotUpgradeFailed' , { defaultMessage : 'The upgrade process for this model snapshot failed. Check the Elasticsearch logs for more details.' , } ) , } , } ) ; } else { const { isSuccessful : isSnapshotDeprecationResolved , error : upgradeSnapshotError } = await verifySnapshotUpgrade ( esClient , { snapshotId , jobId , } ) ; await deleteMlOperation ( savedObjectsClient , snapshotOp . id ) ; if ( isSnapshotDeprecationResolved ) { return response . ok ( { body : { ... snapshotInfo , status : 'complete' , } , } ) ; } return response . customError ( { statusCode : upgradeSnapshotError ? upgradeSnapshotError . statusCode ! : 500 , body : { message : upgradeSnapshotError ?. body ?. error ?. reason || 'The upgrade process for this model snapshot stopped yet the snapshot is not upgraded. Check the Elasticsearch logs for more details.' , } , } ) ; } } else { const { isSuccessful : isSnapshotDeprecationResolved , error : upgradeSnapshotError } = await verifySnapshotUpgrade ( esClient , { snapshotId , jobId , } ) ; await deleteMlOperation ( savedObjectsClient , snapshotOp . id ) ; if ( isSnapshotDeprecationResolved ) { return response . ok ( { body : { ... snapshotInfo , status : 'complete' , } , } ) ; } log . error ( `Failed to determine status of the ML model upgrade, upgradeStatus is not defined and snapshot upgrade is not completed. snapshotId=${ snapshotId } and jobId=${ jobId }` ) ; return response . customError ( { statusCode : upgradeSnapshotError ? upgradeSnapshotError . statusCode ! : 500 , body : { message : upgradeSnapshotError ?. body ?. error ?. reason || 'The upgrade process for this model snapshot completed yet the snapshot is not upgraded. Check the Elasticsearch logs for more details.' , } , } ) ; } } catch ( e ) { return handleEsError ( { error : e , response } ) ; } } ) ) ; router . get ( { path : `${ API_BASE_PATH }/ml_upgrade_mode` , validate : false , } , versionCheckHandlerWrapper ( async ( { core : { elasticsearch : { client : esClient } , } , } , request , response ) => { try { const { body : mlInfo } = await esClient . asCurrentUser . ml . info ( ) ; return response . ok ( { body : { mlUpgradeModeEnabled : mlInfo . upgrade_mode , } , } ) ; } catch ( e ) { return handleEsError ( { error : e , response } ) ; } } ) ) ; router . delete ( { path : `${ API_BASE_PATH }/ml_snapshots/{jobId}/{snapshotId}` , validate : { params : schema . object ( { snapshotId : schema . string ( ) , jobId : schema . string ( ) , } ) , } , } , versionCheckHandlerWrapper ( async ( { core : { elasticsearch : { client } , } , } , request , response ) => { try { const { snapshotId , jobId } = request . params ; const { body : deleteSnapshotResponse } = await client . asCurrentUser . ml . deleteModelSnapshot ( { job_id : jobId , snapshot_id : snapshotId , } ) ; return response . ok ( { body : deleteSnapshotResponse , } ) ; } catch ( e ) { return handleEsError ( { error : e , response } ) ; } } ) ) ; }