import { IClusterClient , Logger , SavedObjectsClientContract , FakeRequest } from 'src/core/server' ; import moment from 'moment' ; import { SecurityPluginStart } from '../../../../security/server' ; import { ReindexSavedObject , ReindexStatus } from '../../../common/types' ; import { Credential , CredentialStore } from './credential_store' ; import { reindexActionsFactory } from './reindex_actions' ; import { ReindexService , reindexServiceFactory } from './reindex_service' ; import { LicensingPluginSetup } from '../../../../licensing/server' ; import { sortAndOrderReindexOperations , queuedOpHasStarted , isQueuedOp } from './op_utils' ; const POLL_INTERVAL = 30000 ; const PAUSE_WINDOW = POLL_INTERVAL * 4 ; /**
 * To avoid running the worker loop very tightly and causing a CPU bottleneck we use this
 * padding to simulate an asynchronous sleep. See the description of the tight loop below.
 */ const WORKER_PADDING_MS = 1000 ; /**
 * A singleton worker that will coordinate two polling loops:
 *   (1) A longer loop that polls for reindex operations that are in progress. If any are found, loop (2) is started.
 *   (2) A tighter loop that pushes each in progress reindex operation through ReindexService.processNextStep. If all
 *       updated reindex operations are complete, this loop will terminate.
 *
 * The worker can also be forced to start loop (2) by calling forceRefresh(). This is done when we know a new reindex
 * operation has been started.
 *
 * This worker can be ran on multiple nodes without conflicts or dropped jobs. Reindex operations are locked by the
 * ReindexService and if any operation is locked longer than the ReindexService's timeout, it is assumed to have been
 * locked by a node that is no longer running (crashed or shutdown). In this case, another node may safely acquire
 * the lock for this reindex operation.
 */ export class ReindexWorker { private static workerSingleton ? : ReindexWorker ; private continuePolling : boolean = false ; private updateOperationLoopRunning : boolean = false ; private timeout ? : NodeJS . Timeout ; private inProgressOps : ReindexSavedObject [ ] = [ ] ; private readonly reindexService : ReindexService ; private readonly log : Logger ; private readonly security : SecurityPluginStart ; public static create ( client , credentialStore , clusterClient , log , licensing , security ) { if ( ReindexWorker . workerSingleton ) { log . debug ( `More than one ReindexWorker cannot be created, returning existing worker.` ) ; } else { ReindexWorker . workerSingleton = new ReindexWorker ( client , credentialStore , clusterClient , log , licensing , security ) ; } return ReindexWorker . workerSingleton ; } private constructor ( private client , private credentialStore , private clusterClient , log , private licensing , security ) { this . log = log . get ( 'reindex_worker' ) ; this . security = security ; const callAsInternalUser = this . clusterClient . asInternalUser ; this . reindexService = reindexServiceFactory ( callAsInternalUser , reindexActionsFactory ( this . client , callAsInternalUser ) , log , this . licensing ) ; } /**
     * Begins loop (1) to begin checking for in progress reindex operations.
     */ public start = ( ) => { this . log . debug ( 'Starting worker...' ) ; this . continuePolling = true ; this . pollForOperations ( ) ; } ; /**
     * Stops the worker from processing any further reindex operations.
     */ public stop = ( ) => { this . log . debug ( 'Stopping worker...' ) ; if ( this . timeout ) { clearTimeout ( this . timeout ) ; } this . updateOperationLoopRunning = false ; this . continuePolling = false ; } ; /**
     * Should be called immediately after this server has started a new reindex operation.
     */ public forceRefresh = ( ) => { this . refresh ( ) ; } ; /**
     * Returns whether or not the given ReindexOperation is in the worker's queue.
     */ public includes = ( reindexOp ) => { return this . inProgressOps . map ( ( o ) => o . id ) . includes ( reindexOp . id ) ; } ; /**
     * Runs an async loop until all inProgress jobs are complete or failed.
     */ private startUpdateOperationLoop = async ( ) : Promise < void > => { this . updateOperationLoopRunning = true ; try { while ( this . inProgressOps . length > 0 ) { this . log . debug ( `Updating ${ this . inProgressOps . length } reindex operations` ) ; await Promise . all ( this . inProgressOps . map ( this . processNextStep ) ) ; await this . refresh ( ) ; if ( this . inProgressOps . length && this . inProgressOps . every ( ( op ) => ! this . credentialStore . get ( op ) ) ) { await new Promise ( ( resolve ) => setTimeout ( resolve , WORKER_PADDING_MS ) ) ; } } } finally { this . updateOperationLoopRunning = false ; } } ; private pollForOperations = async ( ) => { this . log . debug ( `Polling for reindex operations` ) ; await this . refresh ( ) ; if ( this . continuePolling ) { this . timeout = setTimeout ( this . pollForOperations , POLL_INTERVAL ) ; } } ; private getCredentialScopedReindexService = ( credential ) => { const fakeRequest = { headers : credential } ; const scopedClusterClient = this . clusterClient . asScoped ( fakeRequest ) ; const callAsCurrentUser = scopedClusterClient . asCurrentUser ; const actions = reindexActionsFactory ( this . client , callAsCurrentUser ) ; return reindexServiceFactory ( callAsCurrentUser , actions , this . log , this . licensing ) ; } ; private updateInProgressOps = async ( ) => { try { const inProgressOps = await this . reindexService . findAllByStatus ( ReindexStatus . inProgress ) ; const { parallel , queue } = sortAndOrderReindexOperations ( inProgressOps ) ; let [ firstOpInQueue ] = queue ; if ( firstOpInQueue && ! queuedOpHasStarted ( firstOpInQueue ) ) { this . log . debug ( `Queue detected; current length ${ queue . length }, current item ReindexOperation(id: ${ firstOpInQueue . id }, indexName: ${ firstOpInQueue . attributes . indexName })` ) ; const credential = this . credentialStore . get ( firstOpInQueue ) ; if ( credential ) { const service = this . getCredentialScopedReindexService ( credential ) ; firstOpInQueue = await service . startQueuedReindexOperation ( firstOpInQueue . attributes . indexName ) ; this . credentialStore . update ( { reindexOp : firstOpInQueue , security : this . security , credential , } ) ; } } this . inProgressOps = parallel . concat ( firstOpInQueue ? [ firstOpInQueue ] : [ ] ) ; } catch ( e ) { this . log . debug ( `Could not fetch reindex operations from Elasticsearch, ${ e . message }` ) ; this . inProgressOps = [ ] ; } } ; private refresh = async ( ) => { await this . updateInProgressOps ( ) ; if ( ! this . updateOperationLoopRunning ) { this . startUpdateOperationLoop ( ) ; } } ; private lastCheckedQueuedOpId : string | undefined ; private processNextStep = async ( reindexOp ) : Promise < void > => { const credential = this . credentialStore . get ( reindexOp ) ; if ( ! credential ) { if ( isQueuedOp ( reindexOp ) ) { if ( this . lastCheckedQueuedOpId !== reindexOp . id ) { this . lastCheckedQueuedOpId = reindexOp . id ; return ; } } const now = moment ( ) ; const updatedAt = moment ( reindexOp . updated_at ) ; if ( updatedAt < now . subtract ( PAUSE_WINDOW ) ) { await this . reindexService . pauseReindexOperation ( reindexOp . attributes . indexName ) ; return ; } else { return ; } } const service = this . getCredentialScopedReindexService ( credential ) ; reindexOp = await swallowExceptions ( service . processNextStep , this . log ) ( reindexOp ) ; this . credentialStore . update ( { reindexOp , security : this . security , credential } ) ; } ; } /**
 * Swallows any exceptions that may occur during the reindex process. This prevents any errors from
 * stopping the worker from continuing to process more jobs.
 */ const swallowExceptions = ( func , log ) => async ( reindexOp ) => { try { return await func ( reindexOp ) ; } catch ( e ) { if ( reindexOp . attributes . locked ) { log . debug ( `Skipping reindexOp with unexpired lock: ${ reindexOp . id }` ) ; } else { log . warn ( `Error when trying to process reindexOp (${ reindexOp . id }): ${ e . toString ( ) }` ) ; } return reindexOp ; } } ;