import { ElasticsearchClient , Logger } from 'src/core/server' ; import { first } from 'rxjs/operators' ; import { LicensingPluginSetup } from '../../../../licensing/server' ; import { ReindexSavedObject , ReindexStatus , ReindexStep , ReindexWarning , } from '../../../common/types' ; import { esIndicesStateCheck } from '../es_indices_state_check' ; import { generateNewIndexName , getReindexWarnings , sourceNameForIndex , transformFlatSettings , } from './index_settings' ; import { ReindexActions } from './reindex_actions' ; import { error } from './error' ; export interface ReindexService { param indexName hasRequiredPrivileges ( indexName ) : Promise < boolean > ; param indexName detectReindexWarnings ( indexName ) : Promise < ReindexWarning [ ] | undefined > ; param indexName param opts createReindexOperation ( indexName , opts ? ) : Promise < ReindexSavedObject > ; param status findAllByStatus ( status ) : Promise < ReindexSavedObject [ ] > ; param indexName findReindexOperation ( indexName ) : Promise < ReindexSavedObject | null > ; param indexNames cleanupReindexOperations ( indexNames ) : Promise < void > | null ; param reindexOp processNextStep ( reindexOp ) : Promise < ReindexSavedObject > ; param indexName pauseReindexOperation ( indexName ) : Promise < ReindexSavedObject > ; param indexName param opts As with  createReindexOperation we support this setting.
      resumeReindexOperation ( indexName , opts ? ) : Promise < ReindexSavedObject > ; remark param indexName startQueuedReindexOperation ( indexName ) : Promise < ReindexSavedObject > ; param indexName cancelReindexing ( indexName ) : Promise < ReindexSavedObject > ; getIndexAliases ( indexName ) : any ; } export const reindexServiceFactory = ( esClient , actions , log , licensing ) : ReindexService => { const cleanupChanges = async ( reindexOp ) => { if ( reindexOp . attributes . lastCompletedStep === ReindexStep . reindexStarted ) { await esClient . tasks . cancel ( { task_id : reindexOp . attributes . reindexTaskId ?? undefined , } ) . catch ( ( ) => undefined ) ; } if ( reindexOp . attributes . lastCompletedStep >= ReindexStep . readonly ) { await esClient . indices . putSettings ( { index : reindexOp . attributes . indexName , body : { blocks : { write : false } } , } ) ; } if ( reindexOp . attributes . lastCompletedStep >= ReindexStep . newIndexCreated && reindexOp . attributes . lastCompletedStep < ReindexStep . aliasCreated ) { await esClient . indices . delete ( { index : reindexOp . attributes . newIndexName , } ) ; } return reindexOp ; } ; param reindexOp const setReadonly = async ( reindexOp ) => { const { indexName } = reindexOp . attributes ; const { body : putReadonly } = await esClient . indices . putSettings ( { index : indexName , body : { blocks : { write : true } } , } ) ; if ( ! putReadonly . acknowledged ) { throw new Error ( `Index could not be set to readonly.` ) ; } return actions . updateReindexOp ( reindexOp , { lastCompletedStep : ReindexStep . readonly } ) ; } ; param reindexOp const createNewIndex = async ( reindexOp ) => { const { indexName , newIndexName } = reindexOp . attributes ; const flatSettings = await actions . getFlatSettings ( indexName ) ; if ( ! flatSettings ) { throw error . indexNotFound ( `Index ${ indexName } does not exist.` ) ; } const { settings , mappings } = transformFlatSettings ( flatSettings ) ; const { body : createIndex } = await esClient . indices . create ( { index : newIndexName , body : { settings , mappings , } , } ) ; if ( ! createIndex . acknowledged ) { throw error . cannotCreateIndex ( `Index could not be created: ${ newIndexName }` ) ; } return actions . updateReindexOp ( reindexOp , { lastCompletedStep : ReindexStep . newIndexCreated , } ) ; } ; param reindexOp const startReindexing = async ( reindexOp ) => { const { indexName , reindexOptions } = reindexOp . attributes ; const indicesState = await esIndicesStateCheck ( esClient , [ indexName ] ) ; const shouldOpenAndClose = indicesState [ indexName ] === 'closed' ; if ( shouldOpenAndClose ) { log . debug ( `Detected closed index ${ indexName }, opening...` ) ; await esClient . indices . open ( { index : indexName } ) ; } const flatSettings = await actions . getFlatSettings ( indexName ) ; if ( ! flatSettings ) { throw error . indexNotFound ( `Index ${ indexName } does not exist.` ) ; } const { body : startReindexResponse } = await esClient . reindex ( { refresh : true , wait_for_completion : false , body : { source : { index : indexName } , dest : { index : reindexOp . attributes . newIndexName } , } , } ) ; return actions . updateReindexOp ( reindexOp , { lastCompletedStep : ReindexStep . reindexStarted , reindexTaskId : startReindexResponse . task === undefined ? startReindexResponse . task : String ( startReindexResponse . task ) , reindexTaskPercComplete : 0 , reindexOptions : { ... ( reindexOptions ?? { } ) , openAndClose : shouldOpenAndClose , } , } ) ; } ; param reindexOp const updateReindexStatus = async ( reindexOp ) => { const taskId = reindexOp . attributes . reindexTaskId ! ; const { body : taskResponse } = await esClient . tasks . get ( { task_id : taskId , wait_for_completion : false , } ) ; if ( ! taskResponse . completed ) { const perc = taskResponse . task . status ! . created / taskResponse . task . status ! . total ; return actions . updateReindexOp ( reindexOp , { reindexTaskPercComplete : perc , } ) ; } else if ( taskResponse . task . status ?. canceled === 'by user request' ) { reindexOp = await actions . updateReindexOp ( reindexOp , { status : ReindexStatus . cancelled , } ) ; reindexOp = await cleanupChanges ( reindexOp ) ; } else { const { body : { count } , } = await esClient . count ( { index : reindexOp . attributes . indexName } ) ; if ( taskResponse . task . status ! . created < count ) { throw error . reindexTaskFailed ( `Reindexing failed: ${ JSON . stringify ( taskResponse ) }` ) ; } reindexOp = await actions . updateReindexOp ( reindexOp , { lastCompletedStep : ReindexStep . reindexCompleted , reindexTaskPercComplete : 1 , } ) ; } const { body : deleteTaskResp } = await esClient . delete ( { index : '.tasks' , id : taskId , } ) ; if ( deleteTaskResp . result !== 'deleted' ) { throw error . reindexTaskCannotBeDeleted ( `Could not delete reindexing task ${ taskId }` ) ; } return reindexOp ; } ; const getIndexAliases = async ( indexName ) => { const { body : response } = await esClient . indices . getAlias ( { index : indexName , } ) ; return response [ indexName ] ?. aliases ?? { } ; } ; param reindexOp const switchAlias = async ( reindexOp ) => { const { indexName , newIndexName , reindexOptions } = reindexOp . attributes ; const existingAliases = await getIndexAliases ( indexName ) ; const extraAliases = Object . keys ( existingAliases ) . map ( ( aliasName ) => ( { add : { index : newIndexName , alias : aliasName , ... existingAliases [ aliasName ] } , } ) ) ; const { body : aliasResponse } = await esClient . indices . updateAliases ( { body : { actions : [ { add : { index : newIndexName , alias : indexName } } , { remove_index : { index : indexName } } , ... extraAliases , ] , } , } ) ; if ( ! aliasResponse . acknowledged ) { throw error . cannotCreateIndex ( `Index aliases could not be created.` ) ; } if ( reindexOptions ?. openAndClose === true ) { await esClient . indices . close ( { index : indexName } ) ; } return actions . updateReindexOp ( reindexOp , { lastCompletedStep : ReindexStep . aliasCreated , } ) ; } ; return { async hasRequiredPrivileges ( indexName ) { /**
             * To avoid a circular dependency on Security we use a work around
             * here to detect whether Security is available and enabled
             * (i.e., via the licensing plugin). This enables Security to use
             * functionality exposed through Upgrade Assistant.
             */ const license = await licensing . license$ . pipe ( first ( ) ) . toPromise ( ) ; const securityFeature = license . getFeature ( 'security' ) ; if ( ! securityFeature || ! ( securityFeature . isAvailable && securityFeature . isEnabled ) ) { return true ; } const names = [ indexName , generateNewIndexName ( indexName ) ] ; const sourceName = sourceNameForIndex ( indexName ) ; if ( sourceName !== indexName ) { names . push ( sourceName ) ; } const body = { cluster : [ 'manage' ] , index : [ { names , allow_restricted_indices : true , privileges : [ 'all' ] , } , { names : [ '.tasks' ] , privileges : [ 'read' , 'delete' ] , } , ] , } as any ; const { body : resp } = await esClient . security . hasPrivileges ( { body , } ) ; return resp . has_all_requested ; } , async detectReindexWarnings ( indexName ) { const flatSettings = await actions . getFlatSettings ( indexName ) ; if ( ! flatSettings ) { return undefined ; } else { return [ { warningType : 'replaceIndexWithAlias' , } , ... getReindexWarnings ( flatSettings ) , ] ; } } , async createReindexOperation ( indexName , opts ? ) { const { body : indexExists } = await esClient . indices . exists ( { index : indexName } ) ; if ( ! indexExists ) { throw error . indexNotFound ( `Index ${ indexName } does not exist in this cluster.` ) ; } const existingReindexOps = await actions . findReindexOperations ( indexName ) ; if ( existingReindexOps . total !== 0 ) { const existingOp = existingReindexOps . saved_objects [ 0 ] ; if ( existingOp . attributes . status === ReindexStatus . failed || existingOp . attributes . status === ReindexStatus . cancelled ) { await actions . deleteReindexOp ( existingOp ) ; } else { throw error . reindexAlreadyInProgress ( `A reindex operation already in-progress for ${ indexName }` ) ; } } return actions . createReindexOp ( indexName , opts ?. enqueue ? { queueSettings : { queuedAt : Date . now ( ) } } : undefined ) ; } , async findReindexOperation ( indexName ) { const findResponse = await actions . findReindexOperations ( indexName ) ; if ( findResponse . total === 0 ) { return null ; } else if ( findResponse . total > 1 ) { throw error . multipleReindexJobsFound ( `More than one reindex operation found for ${ indexName }` ) ; } return findResponse . saved_objects [ 0 ] ; } , async cleanupReindexOperations ( indexNames ) { const performCleanup = async ( indexName ) => { const existingReindexOps = await actions . findReindexOperations ( indexName ) ; if ( existingReindexOps && existingReindexOps . total !== 0 ) { const existingOp = existingReindexOps . saved_objects [ 0 ] ; if ( existingOp . attributes . status === ReindexStatus . completed ) { await actions . deleteReindexOp ( existingOp ) ; } } } ; await Promise . all ( indexNames . map ( performCleanup ) ) ; } , findAllByStatus : actions . findAllByStatus , async processNextStep ( reindexOp ) { return actions . runWhileLocked ( reindexOp , async ( lockedReindexOp ) => { try { switch ( lockedReindexOp . attributes . lastCompletedStep ) { case ReindexStep . created : lockedReindexOp = await setReadonly ( lockedReindexOp ) ; break ; case ReindexStep . readonly : lockedReindexOp = await createNewIndex ( lockedReindexOp ) ; break ; case ReindexStep . newIndexCreated : lockedReindexOp = await startReindexing ( lockedReindexOp ) ; break ; case ReindexStep . reindexStarted : lockedReindexOp = await updateReindexStatus ( lockedReindexOp ) ; break ; case ReindexStep . reindexCompleted : lockedReindexOp = await switchAlias ( lockedReindexOp ) ; break ; case ReindexStep . aliasCreated : lockedReindexOp = await actions . updateReindexOp ( lockedReindexOp , { status : ReindexStatus . completed , } ) ; break ; default : break ; } } catch ( e ) { log . error ( `Reindexing step failed: ${ e instanceof Error ? e . stack : e . toString ( ) }` ) ; lockedReindexOp = await actions . updateReindexOp ( lockedReindexOp , { status : ReindexStatus . failed , errorMessage : e . toString ( ) , } ) ; lockedReindexOp = await cleanupChanges ( lockedReindexOp ) . catch ( ( err ) => lockedReindexOp ) ; } return lockedReindexOp ; } ) ; } , async pauseReindexOperation ( indexName ) { const reindexOp = await this . findReindexOperation ( indexName ) ; if ( ! reindexOp ) { throw new Error ( `No reindex operation found for index ${ indexName }` ) ; } return actions . runWhileLocked ( reindexOp , async ( op ) => { if ( op . attributes . status === ReindexStatus . paused ) { return reindexOp ; } else if ( op . attributes . status !== ReindexStatus . inProgress ) { throw new Error ( `Reindex operation must be inProgress in order to be paused.` ) ; } return actions . updateReindexOp ( op , { status : ReindexStatus . paused } ) ; } ) ; } , async resumeReindexOperation ( indexName , opts ? ) { const reindexOp = await this . findReindexOperation ( indexName ) ; if ( ! reindexOp ) { throw new Error ( `No reindex operation found for index ${ indexName }` ) ; } return actions . runWhileLocked ( reindexOp , async ( op ) => { if ( op . attributes . status === ReindexStatus . inProgress ) { return reindexOp ; } else if ( op . attributes . status !== ReindexStatus . paused ) { throw new Error ( `Reindex operation must be paused in order to be resumed.` ) ; } const queueSettings = opts ?. enqueue ? { queuedAt : Date . now ( ) } : undefined ; return actions . updateReindexOp ( op , { status : ReindexStatus . inProgress , reindexOptions : queueSettings ? { queueSettings } : undefined , } ) ; } ) ; } , async startQueuedReindexOperation ( indexName ) { const reindexOp = await this . findReindexOperation ( indexName ) ; if ( ! reindexOp ) { throw error . indexNotFound ( `No reindex operation found for index ${ indexName }` ) ; } if ( ! reindexOp . attributes . reindexOptions ?. queueSettings ) { throw error . reindexIsNotInQueue ( `Reindex operation ${ indexName } is not in the queue.` ) ; } return actions . runWhileLocked ( reindexOp , async ( lockedReindexOp ) => { const { reindexOptions } = lockedReindexOp . attributes ; reindexOptions ! . queueSettings ! . startedAt = Date . now ( ) ; return actions . updateReindexOp ( lockedReindexOp , { reindexOptions , } ) ; } ) ; } , async cancelReindexing ( indexName ) { const reindexOp = await this . findReindexOperation ( indexName ) ; if ( ! reindexOp ) { throw error . indexNotFound ( `No reindex operation found for index ${ indexName }` ) ; } else if ( reindexOp . attributes . status !== ReindexStatus . inProgress ) { throw error . reindexCannotBeCancelled ( `Reindex operation is not in progress` ) ; } else if ( reindexOp . attributes . lastCompletedStep !== ReindexStep . reindexStarted ) { throw error . reindexCannotBeCancelled ( `Reindex operation is not currently waiting for reindex task to complete` ) ; } const { body : resp } = await esClient . tasks . cancel ( { task_id : reindexOp . attributes . reindexTaskId ! , } ) ; if ( resp . node_failures && resp . node_failures . length > 0 ) { throw error . reindexCannotBeCancelled ( `Could not cancel reindex.` ) ; } return reindexOp ; } , getIndexAliases , } ; } ;