import moment from 'moment' ; import { SavedObjectsFindResponse , SavedObjectsClientContract , ElasticsearchClient , } from 'src/core/server' ; import { REINDEX_OP_TYPE , ReindexOperation , ReindexOptions , ReindexSavedObject , ReindexStatus , ReindexStep , } from '../../../common/types' ; import { versionService } from '../version' ; import { generateNewIndexName } from './index_settings' ; import { FlatSettings , FlatSettingsWithTypeName } from './types' ; export const LOCK_WINDOW = moment . duration ( 90 , 'seconds' ) ; /**
 * A collection of utility functions pulled out out of the ReindexService to make testing simpler.
 * This is NOT intended to be used by any other code.
 */ export interface ReindexActions { param indexName param opts createReindexOp ( indexName , opts ? ) : Promise < ReindexSavedObject > ; param reindexOp deleteReindexOp ( reindexOp ) : void ; param reindexOp param attrs updateReindexOp ( reindexOp , attrs ? ) : Promise < ReindexSavedObject > ; param func runWhileLocked ( reindexOp , func ) : Promise < ReindexSavedObject > ; param indexName findReindexOperations ( indexName ) : Promise < SavedObjectsFindResponse < ReindexOperation > > ; /**
     * Returns an array of all reindex operations that have a status.
     */ findAllByStatus ( status ) : Promise < ReindexSavedObject [ ] > ; param indexName getFlatSettings ( indexName , withTypeName ? ) : Promise < FlatSettings | FlatSettingsWithTypeName | null > ; } export const reindexActionsFactory = ( client , esClient ) : ReindexActions => { const isLocked = ( reindexOp ) => { if ( reindexOp . attributes . locked ) { const now = moment ( ) ; const lockedTime = moment ( reindexOp . attributes . locked ) ; if ( now . subtract ( LOCK_WINDOW ) < lockedTime ) { return true ; } } return false ; } ; const acquireLock = async ( reindexOp ) => { if ( isLocked ( reindexOp ) ) { throw new Error ( `Another Kibana process is currently modifying this reindex operation.` ) ; } return client . update < ReindexOperation > ( REINDEX_OP_TYPE , reindexOp . id , { ... reindexOp . attributes , locked : moment ( ) . format ( ) } , { version : reindexOp . version } ) as Promise < ReindexSavedObject > ; } ; const releaseLock = ( reindexOp ) => { return client . update < ReindexOperation > ( REINDEX_OP_TYPE , reindexOp . id , { ... reindexOp . attributes , locked : null } , { version : reindexOp . version } ) as Promise < ReindexSavedObject > ; } ; return { async createReindexOp ( indexName , opts ? ) { return client . create < ReindexOperation > ( REINDEX_OP_TYPE , { indexName , newIndexName : generateNewIndexName ( indexName ) , status : ReindexStatus . inProgress , lastCompletedStep : ReindexStep . created , locked : null , reindexTaskId : null , reindexTaskPercComplete : null , errorMessage : null , runningReindexCount : null , reindexOptions : opts , } ) ; } , deleteReindexOp ( reindexOp ) { return client . delete ( REINDEX_OP_TYPE , reindexOp . id ) ; } , async updateReindexOp ( reindexOp , attrs = { } ) { if ( ! isLocked ( reindexOp ) ) { throw new Error ( `ReindexOperation must be locked before updating.` ) ; } const newAttrs = { ... reindexOp . attributes , locked : moment ( ) . format ( ) , ... attrs } ; return client . update < ReindexOperation > ( REINDEX_OP_TYPE , reindexOp . id , newAttrs , { version : reindexOp . version , } ) as Promise < ReindexSavedObject > ; } , async runWhileLocked ( reindexOp , func ) { reindexOp = await acquireLock ( reindexOp ) ; try { reindexOp = await func ( reindexOp ) ; } finally { reindexOp = await releaseLock ( reindexOp ) ; } return reindexOp ; } , findReindexOperations ( indexName ) { return client . find < ReindexOperation > ( { type : REINDEX_OP_TYPE , search : `"${ indexName }"` , searchFields : [ 'indexName' ] , } ) ; } , async findAllByStatus ( status ) { const firstPage = await client . find < ReindexOperation > ( { type : REINDEX_OP_TYPE , search : status . toString ( ) , searchFields : [ 'status' ] , } ) ; if ( firstPage . total === firstPage . saved_objects . length ) { return firstPage . saved_objects ; } let allOps = firstPage . saved_objects ; let page = firstPage . page + 1 ; while ( allOps . length < firstPage . total ) { const nextPage = await client . find < ReindexOperation > ( { type : REINDEX_OP_TYPE , search : status . toString ( ) , searchFields : [ 'status' ] , page , } ) ; allOps = [ ... allOps , ... nextPage . saved_objects ] ; page ++ ; } return allOps ; } , async getFlatSettings ( indexName ) { let flatSettings ; if ( versionService . getMajorVersion ( ) === 7 ) { flatSettings = await esClient . indices . get < { [ indexName ] : FlatSettingsWithTypeName ; } > ( { index : indexName , flat_settings : true , include_type_name : true , } ) ; } else { flatSettings = await esClient . indices . get < { [ indexName ] : FlatSettings ; } > ( { index : indexName , flat_settings : true , } ) ; } if ( ! flatSettings . body [ indexName ] ) { return null ; } return flatSettings . body [ indexName ] ; } , } ; } ;