forked from elastic/kibana
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Reporting: register a single ESQueue worker, simultaneous poll for a…
…ll export types (elastic#32839) * Reporting: register a single ESQueue worker, simultaneous poll for all export types * more typescript * PLUGIN_ID constant * move down log / internal state * fix tests * jest test for createWorker * assert arguments to queue.registerWorker * logic move * make ts ignore specific * minor reversion to fix some esqueue worker tests
- Loading branch information
Showing
10 changed files
with
231 additions
and
73 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,99 @@ | ||
| /* | ||
| * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one | ||
| * or more contributor license agreements. Licensed under the Elastic License; | ||
| * you may not use this file except in compliance with the Elastic License. | ||
| */ | ||
|
|
||
| import * as sinon from 'sinon'; | ||
| import { KbnServer } from '../../types'; | ||
| import { createWorkerFactory } from './create_worker'; | ||
| // @ts-ignore | ||
| import { Esqueue } from './esqueue'; | ||
| // @ts-ignore | ||
| import { ClientMock } from './esqueue/__tests__/fixtures/elasticsearch'; | ||
|
|
||
| const configGetStub = sinon.stub(); | ||
| configGetStub.withArgs('xpack.reporting.queue').returns({ | ||
| pollInterval: 3300, | ||
| pollIntervalErrorMultiplier: 10, | ||
| }); | ||
| configGetStub.withArgs('server.name').returns('test-server-123'); | ||
| configGetStub.withArgs('server.uuid').returns('g9ymiujthvy6v8yrh7567g6fwzgzftzfr'); | ||
|
|
||
| const executeJobFactoryStub = sinon.stub(); | ||
|
|
||
| const getMockServer = ( | ||
| exportTypes: any[] = [{ executeJobFactory: executeJobFactoryStub }] | ||
| ): Partial<KbnServer> => ({ | ||
| log: sinon.stub(), | ||
| expose: sinon.stub(), | ||
| config: () => ({ get: configGetStub }), | ||
| plugins: { reporting: { exportTypesRegistry: { getAll: () => exportTypes } } }, | ||
| }); | ||
|
|
||
| describe('Create Worker', () => { | ||
| let queue: Esqueue; | ||
| let client: ClientMock; | ||
|
|
||
| beforeEach(() => { | ||
| client = new ClientMock(); | ||
| queue = new Esqueue('reporting-queue', { client }); | ||
| executeJobFactoryStub.reset(); | ||
| }); | ||
|
|
||
| test('Creates a single Esqueue worker for Reporting', async () => { | ||
| const createWorker = createWorkerFactory(getMockServer()); | ||
| const registerWorkerSpy = sinon.spy(queue, 'registerWorker'); | ||
|
|
||
| createWorker(queue); | ||
|
|
||
| sinon.assert.callCount(executeJobFactoryStub, 1); | ||
| sinon.assert.callCount(registerWorkerSpy, 1); | ||
|
|
||
| const { firstCall } = registerWorkerSpy; | ||
| const [workerName, workerFn, workerOpts] = firstCall.args; | ||
|
|
||
| expect(workerName).toBe('reporting'); | ||
| expect(workerFn).toMatchInlineSnapshot(`[Function]`); | ||
| expect(workerOpts).toMatchInlineSnapshot(` | ||
| Object { | ||
| "interval": 3300, | ||
| "intervalErrorMultiplier": 10, | ||
| "kibanaId": "g9ymiujthvy6v8yrh7567g6fwzgzftzfr", | ||
| "kibanaName": "test-server-123", | ||
| } | ||
| `); | ||
| }); | ||
|
|
||
| test('Creates a single Esqueue worker for Reporting, even if there are multiple export types', async () => { | ||
| const createWorker = createWorkerFactory( | ||
| getMockServer([ | ||
| { executeJobFactory: executeJobFactoryStub }, | ||
| { executeJobFactory: executeJobFactoryStub }, | ||
| { executeJobFactory: executeJobFactoryStub }, | ||
| { executeJobFactory: executeJobFactoryStub }, | ||
| { executeJobFactory: executeJobFactoryStub }, | ||
| ]) | ||
| ); | ||
| const registerWorkerSpy = sinon.spy(queue, 'registerWorker'); | ||
|
|
||
| createWorker(queue); | ||
|
|
||
| sinon.assert.callCount(executeJobFactoryStub, 5); | ||
| sinon.assert.callCount(registerWorkerSpy, 1); | ||
|
|
||
| const { firstCall } = registerWorkerSpy; | ||
| const [workerName, workerFn, workerOpts] = firstCall.args; | ||
|
|
||
| expect(workerName).toBe('reporting'); | ||
| expect(workerFn).toMatchInlineSnapshot(`[Function]`); | ||
| expect(workerOpts).toMatchInlineSnapshot(` | ||
| Object { | ||
| "interval": 3300, | ||
| "intervalErrorMultiplier": 10, | ||
| "kibanaId": "g9ymiujthvy6v8yrh7567g6fwzgzftzfr", | ||
| "kibanaName": "test-server-123", | ||
| } | ||
| `); | ||
| }); | ||
| }); |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,70 @@ | ||
| /* | ||
| * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one | ||
| * or more contributor license agreements. Licensed under the Elastic License; | ||
| * you may not use this file except in compliance with the Elastic License. | ||
| */ | ||
|
|
||
| // @ts-ignore | ||
| import { PLUGIN_ID } from '../../common/constants'; | ||
| import { | ||
| ESQueueInstance, | ||
| ESQueueWorkerExecuteFn, | ||
| ExportType, | ||
| JobDoc, | ||
| JobSource, | ||
| KbnServer, | ||
| } from '../../types'; | ||
| // @ts-ignore untyped dependency | ||
| import { events as esqueueEvents } from './esqueue'; | ||
| // @ts-ignore untyped dependency | ||
| import { LevelLogger } from './level_logger'; | ||
| // @ts-ignore untyped dependency | ||
| import { oncePerServer } from './once_per_server'; | ||
|
|
||
| function createWorkerFn(server: KbnServer) { | ||
| const config = server.config(); | ||
| const queueConfig = config.get('xpack.reporting.queue'); | ||
| const kibanaName = config.get('server.name'); | ||
| const kibanaId = config.get('server.uuid'); | ||
| const exportTypesRegistry = server.plugins.reporting.exportTypesRegistry; | ||
| const logger = LevelLogger.createForServer(server, [PLUGIN_ID, 'queue', 'worker']); | ||
|
|
||
| // Once more document types are added, this will need to be passed in | ||
| return function createWorker(queue: ESQueueInstance) { | ||
| // export type / execute job map | ||
| const jobExectors: Map<string, ESQueueWorkerExecuteFn> = new Map(); | ||
|
|
||
| for (const exportType of exportTypesRegistry.getAll() as ExportType[]) { | ||
| const executeJob = exportType.executeJobFactory(server); | ||
| jobExectors.set(exportType.jobType, executeJob); | ||
| } | ||
|
|
||
| const workerFn = (job: JobSource, jobdoc: JobDoc, cancellationToken: any) => { | ||
| // pass the work to the jobExecutor | ||
| const jobExecutor = jobExectors.get(job._source.jobtype); | ||
| if (!jobExecutor) { | ||
| throw new Error(`Unable to find a job executor for the claimed job: [${job._id}]`); | ||
| } | ||
| return jobExecutor(jobdoc, cancellationToken); | ||
| }; | ||
| const workerOptions = { | ||
| kibanaName, | ||
| kibanaId, | ||
| interval: queueConfig.pollInterval, | ||
| intervalErrorMultiplier: queueConfig.pollIntervalErrorMultiplier, | ||
| }; | ||
| const worker = queue.registerWorker(PLUGIN_ID, workerFn, workerOptions); | ||
|
|
||
| worker.on(esqueueEvents.EVENT_WORKER_COMPLETE, (res: any) => { | ||
| logger.debug(`Worker completed: (${res.job.id})`); | ||
| }); | ||
| worker.on(esqueueEvents.EVENT_WORKER_JOB_EXECUTION_ERROR, (res: any) => { | ||
| logger.debug(`Worker error: (${res.job.id})`); | ||
| }); | ||
| worker.on(esqueueEvents.EVENT_WORKER_JOB_TIMEOUT, (res: any) => { | ||
| logger.debug(`Job timeout exceeded: (${res.job.id})`); | ||
| }); | ||
| }; | ||
| } | ||
|
|
||
| export const createWorkerFactory = oncePerServer(createWorkerFn); |
This file was deleted.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.