import { errors as EsErrors } from '@elastic/elasticsearch' ; import * as Option from 'fp-ts/lib/Option' ; import { Logger , LogMeta } from '../../logging' ; import type { ElasticsearchClient } from '../../elasticsearch' ; import { getErrorMessage , getRequestDebugMeta } from '../../elasticsearch' ; import { Model , Next , stateActionMachine } from './state_action_machine' ; import { cleanup } from './migrations_state_machine_cleanup' ; import { ReindexSourceToTempTransform , ReindexSourceToTempIndexBulk , State } from './state' ; import { SavedObjectsRawDoc } from '../serialization' ; interface StateTransitionLogMeta extends LogMeta { kibana : { migrations : { state : State ; duration : number ; } ; } ; } const logStateTransition = ( logger , logMessagePrefix , prevState , currState , tookMs ) => { if ( currState . logs . length > prevState . logs . length ) { currState . logs . slice ( prevState . logs . length ) . forEach ( ( { message , level } ) => { switch ( level ) { case 'error' : return logger . error ( logMessagePrefix + message ) ; case 'warning' : return logger . warn ( logMessagePrefix + message ) ; case 'info' : return logger . info ( logMessagePrefix + message ) ; default : throw new Error ( `unexpected log level ${ level }` ) ; } } ) ; } logger . info ( logMessagePrefix + `${ prevState . controlState } -> ${ currState . controlState }. took: ${ tookMs }ms.` ) ; logger . debug < StateTransitionLogMeta > ( logMessagePrefix + `${ prevState . controlState } -> ${ currState . controlState }. took: ${ tookMs }ms.` , { kibana : { migrations : { state : currState , duration : tookMs , } , } , } ) ; } ; const logActionResponse = ( logger , logMessagePrefix , state , res ) => { logger . debug ( logMessagePrefix + `${ state . controlState } RESPONSE` , res as LogMeta ) ; } ; /**
 * A specialized migrations-specific state-action machine that:
 *  - logs messages in state.logs
 *  - logs state transitions
 *  - logs action responses
 *  - resolves if the final state is DONE
 *  - rejects if the final state is FATAL
 *  - catches and logs exceptions and then rejects with a migrations specific error
 */ export async function migrationStateActionMachine ( { initialState , logger , next , model , client , } ) { const startTime = Date . now ( ) ; const logMessagePrefix = `[${ initialState . indexPrefix }] ` ; let prevTimestamp = startTime ; let lastState ; try { const finalState = await stateActionMachine < State > ( initialState , ( state ) => next ( state ) , ( state , res ) => { lastState = state ; logActionResponse ( logger , logMessagePrefix , state , res ) ; const newState = model ( state , res ) ; const redactedNewState = { ... newState , ... { outdatedDocuments : ( ( newState as ReindexSourceToTempTransform ) . outdatedDocuments ?? [ ] ) . map ( ( doc ) => ( { _id : doc . _id , } as SavedObjectsRawDoc ) ) , } , ... { transformedDocBatches : ( ( newState as ReindexSourceToTempIndexBulk ) . transformedDocBatches ?? [ ] ) . map ( ( batches ) => batches . map ( ( doc ) => ( { _id : doc . _id } ) ) ) as [ SavedObjectsRawDoc [ ] ] , } , } ; const now = Date . now ( ) ; logStateTransition ( logger , logMessagePrefix , state , redactedNewState as State , now - prevTimestamp ) ; prevTimestamp = now ; return newState ; } ) ; const elapsedMs = Date . now ( ) - startTime ; if ( finalState . controlState === 'DONE' ) { logger . info ( logMessagePrefix + `Migration completed after ${ Math . round ( elapsedMs ) }ms` ) ; if ( finalState . sourceIndex != null && Option . isSome ( finalState . sourceIndex ) ) { return { status : 'migrated' as const , destIndex : finalState . targetIndex , sourceIndex : finalState . sourceIndex . value , elapsedMs , } ; } else { return { status : 'patched' as const , destIndex : finalState . targetIndex , elapsedMs , } ; } } else if ( finalState . controlState === 'FATAL' ) { try { await cleanup ( client , finalState ) ; } catch ( e ) { logger . warn ( 'Failed to cleanup after migrations:' , e . message ) ; } return Promise . reject ( new Error ( `Unable to complete saved object migrations for the [${ initialState . indexPrefix }] index: ` + finalState . reason ) ) ; } else { throw new Error ( 'Invalid terminating control state' ) ; } } catch ( e ) { try { await cleanup ( client , lastState ) ; } catch ( err ) { logger . warn ( 'Failed to cleanup after migrations:' , err . message ) ; } if ( e instanceof EsErrors . ResponseError ) { const req = getRequestDebugMeta ( e . meta ) ; const failedRequestMessage = `Unexpected Elasticsearch ResponseError: statusCode: ${ req . statusCode }, method: ${ req . method }, url: ${ req . url } error: ${ getErrorMessage ( e ) },` ; logger . error ( logMessagePrefix + failedRequestMessage ) ; throw new Error ( `Unable to complete saved object migrations for the [${ initialState . indexPrefix }] index. Please check the health of your Elasticsearch cluster and try again. ${ failedRequestMessage }` ) ; } else { logger . error ( e ) ; const newError = new Error ( `Unable to complete saved object migrations for the [${ initialState . indexPrefix }] index. ${ e }` ) ; newError . stack = `[${ e . stack }]` ; throw newError ; } } }