Switch Reporting to Task Manager

docs/user/reporting/script-example.asciidoc

@@ -38,8 +38,7 @@ Here is an example response for a successfully queued report:
     "created_by": "elastic",
     "payload": ..., <2>
     "timeout": 120000,
-    "max_attempts": 3,
-    "priority": 10
+    "max_attempts": 3
   }
 }
 ---------------------------------------------------------

x-pack/plugins/reporting/common/schema_utils.test.ts

@@ -0,0 +1,16 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import moment from 'moment';
+import { numberToDuration } from './schema_utils';
+
+describe('Schema Utils', () => {
+  it('numberToDuration converts a number/Duration into a Duration object', () => {
+    expect(numberToDuration(500)).toMatchInlineSnapshot(`"PT0.5S"`);
+    expect(numberToDuration(moment.duration(1, 'hour'))).toMatchInlineSnapshot(`"PT1H"`);
+  });
+});

x-pack/plugins/reporting/common/schema_utils.ts

@@ -20,6 +20,13 @@ export const durationToNumber = (value: number | moment.Duration): number => {
   return value.asMilliseconds();
 };
 
+export const numberToDuration = (value: number | moment.Duration): moment.Duration => {
+  if (typeof value === 'number') {
+    return moment.duration(value, 'milliseconds');
+  }
+  return value;
+};
+
 export const byteSizeValueToNumber = (value: number | ByteSizeValue) => {
   if (typeof value === 'number') {
     return value;

x-pack/plugins/reporting/common/types.ts

@@ -76,7 +76,6 @@ export interface ReportSource {
   started_at?: string;
   completed_at?: string;
   created_at: string;
-  priority?: number;
   process_expiration?: string;
 }
 
@@ -113,7 +112,6 @@ export interface ReportApiJSON {
   kibana_id: string;
   browser_type: string | undefined;
   created_at: string;
-  priority?: number;
   jobtype: string;
   created_by: string | false;
   timeout?: number;

x-pack/plugins/reporting/kibana.json

@@ -14,6 +14,7 @@
     "management",
     "licensing",
     "uiActions",
+    "taskManager",
     "embeddable",
     "share",
     "features"

x-pack/plugins/reporting/public/components/buttons/report_info_button.tsx

@@ -87,7 +87,6 @@ export class ReportInfoButton extends Component<Props, State> {
 
     const attempts = info.attempts ? info.attempts.toString() : NA;
     const maxAttempts = info.max_attempts ? info.max_attempts.toString() : NA;
-    const priority = info.priority ? info.priority.toString() : NA;
     const timeout = info.timeout ? info.timeout.toString() : NA;
     const warnings = info.output && info.output.warnings ? info.output.warnings.join(',') : null;
 
@@ -153,10 +152,6 @@ export class ReportInfoButton extends Component<Props, State> {
         title: 'Max Attempts',
         description: maxAttempts,
       },
-      {
-        title: 'Priority',
-        description: priority,
-      },
       {
         title: 'Timeout',
         description: timeout,

x-pack/plugins/reporting/server/config/create_config.test.ts

@@ -5,47 +5,22 @@
  * 2.0.
  */
 
-import * as Rx from 'rxjs';
 import { CoreSetup, PluginInitializerContext } from 'src/core/server';
+import { coreMock } from 'src/core/server/mocks';
 import { LevelLogger } from '../lib';
+import { createMockConfigSchema } from '../test_helpers';
 import { createConfig$ } from './create_config';
-import { ReportingConfigType } from './schema';
-
-interface KibanaServer {
-  hostname?: string;
-  port?: number;
-  protocol?: string;
-}
-
-const makeMockInitContext = (config: {
-  capture?: Partial<ReportingConfigType['capture']>;
-  encryptionKey?: string;
-  kibanaServer: Partial<ReportingConfigType['kibanaServer']>;
-}): PluginInitializerContext =>
-  ({
-    config: {
-      create: () =>
-        Rx.of({
-          ...config,
-          capture: config.capture || { browser: { chromium: { disableSandbox: false } } },
-          kibanaServer: config.kibanaServer || {},
-        }),
-    },
-  } as PluginInitializerContext);
-
-const makeMockCoreSetup = (serverInfo: KibanaServer): CoreSetup =>
-  ({ http: { getServerInfo: () => serverInfo } } as any);
 
 describe('Reporting server createConfig$', () => {
   let mockCoreSetup: CoreSetup;
   let mockInitContext: PluginInitializerContext;
   let mockLogger: LevelLogger;
 
   beforeEach(() => {
-    mockCoreSetup = makeMockCoreSetup({ hostname: 'kibanaHost', port: 5601, protocol: 'http' });
-    mockInitContext = makeMockInitContext({
-      kibanaServer: {},
-    });
+    mockCoreSetup = coreMock.createSetup();
+    mockInitContext = coreMock.createPluginInitializerContext(
+      createMockConfigSchema({ kibanaServer: {} })
+    );
     mockLogger = ({
       warn: jest.fn(),
       debug: jest.fn(),
@@ -58,14 +33,18 @@ describe('Reporting server createConfig$', () => {
   });
 
   it('creates random encryption key and default config using host, protocol, and port from server info', async () => {
+    mockInitContext = coreMock.createPluginInitializerContext({
+      ...createMockConfigSchema({ kibanaServer: {} }),
+      encryptionKey: undefined,
+    });
     const mockConfig$: any = mockInitContext.config.create();
     const result = await createConfig$(mockCoreSetup, mockConfig$, mockLogger).toPromise();
 
     expect(result.encryptionKey).toMatch(/\S{32,}/); // random 32 characters
     expect(result.kibanaServer).toMatchInlineSnapshot(`
       Object {
-        "hostname": "kibanaHost",
-        "port": 5601,
+        "hostname": "localhost",
+        "port": 80,
         "protocol": "http",
       }
     `);
@@ -76,25 +55,28 @@ describe('Reporting server createConfig$', () => {
   });
 
   it('uses the user-provided encryption key', async () => {
-    mockInitContext = makeMockInitContext({
-      encryptionKey: 'iiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiii',
-      kibanaServer: {},
-    });
+    mockInitContext = coreMock.createPluginInitializerContext(
+      createMockConfigSchema({
+        encryptionKey: 'iiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiii',
+      })
+    );
     const mockConfig$: any = mockInitContext.config.create();
     const result = await createConfig$(mockCoreSetup, mockConfig$, mockLogger).toPromise();
     expect(result.encryptionKey).toMatch('iiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiii');
     expect((mockLogger.warn as any).mock.calls.length).toBe(0);
   });
 
   it('uses the user-provided encryption key, reporting kibanaServer settings to override server info', async () => {
-    mockInitContext = makeMockInitContext({
-      encryptionKey: 'iiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiii',
-      kibanaServer: {
-        hostname: 'reportingHost',
-        port: 5677,
-        protocol: 'httpsa',
-      },
-    });
+    mockInitContext = coreMock.createPluginInitializerContext(
+      createMockConfigSchema({
+        encryptionKey: 'iiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiii',
+        kibanaServer: {
+          hostname: 'reportingHost',
+          port: 5677,
+          protocol: 'httpsa',
+        },
+      })
+    );
     const mockConfig$: any = mockInitContext.config.create();
     const result = await createConfig$(mockCoreSetup, mockConfig$, mockLogger).toPromise();
 
@@ -103,26 +85,36 @@ describe('Reporting server createConfig$', () => {
         "capture": Object {
           "browser": Object {
             "chromium": Object {
-              "disableSandbox": false,
+              "disableSandbox": true,
             },
           },
         },
+        "csv": Object {},
         "encryptionKey": "iiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiiii",
+        "index": ".reporting",
         "kibanaServer": Object {
           "hostname": "reportingHost",
           "port": 5677,
           "protocol": "httpsa",
         },
+        "queue": Object {
+          "indexInterval": "week",
+          "pollEnabled": true,
+          "pollInterval": 3000,
+          "timeout": 120000,
+        },
       }
     `);
     expect((mockLogger.warn as any).mock.calls.length).toBe(0);
   });
 
   it('uses user-provided disableSandbox: false', async () => {
-    mockInitContext = makeMockInitContext({
-      encryptionKey: '888888888888888888888888888888888',
-      capture: { browser: { chromium: { disableSandbox: false } } },
-    } as ReportingConfigType);
+    mockInitContext = coreMock.createPluginInitializerContext(
+      createMockConfigSchema({
+        encryptionKey: '888888888888888888888888888888888',
+        capture: { browser: { chromium: { disableSandbox: false } } },
+      })
+    );
     const mockConfig$: any = mockInitContext.config.create();
     const result = await createConfig$(mockCoreSetup, mockConfig$, mockLogger).toPromise();
 
@@ -131,10 +123,12 @@ describe('Reporting server createConfig$', () => {
   });
 
   it('uses user-provided disableSandbox: true', async () => {
-    mockInitContext = makeMockInitContext({
-      encryptionKey: '888888888888888888888888888888888',
-      capture: { browser: { chromium: { disableSandbox: true } } },
-    } as ReportingConfigType);
+    mockInitContext = coreMock.createPluginInitializerContext(
+      createMockConfigSchema({
+        encryptionKey: '888888888888888888888888888888888',
+        capture: { browser: { chromium: { disableSandbox: true } } },
+      })
+    );
     const mockConfig$: any = mockInitContext.config.create();
     const result = await createConfig$(mockCoreSetup, mockConfig$, mockLogger).toPromise();
 
@@ -143,9 +137,11 @@ describe('Reporting server createConfig$', () => {
   });
 
   it('provides a default for disableSandbox', async () => {
-    mockInitContext = makeMockInitContext({
-      encryptionKey: '888888888888888888888888888888888',
-    } as ReportingConfigType);
+    mockInitContext = coreMock.createPluginInitializerContext(
+      createMockConfigSchema({
+        encryptionKey: '888888888888888888888888888888888',
+      })
+    );
     const mockConfig$: any = mockInitContext.config.create();
     const result = await createConfig$(mockCoreSetup, mockConfig$, mockLogger).toPromise();
 

x-pack/plugins/reporting/server/core.ts

@@ -12,6 +12,7 @@ import {
   BasePath,
   ElasticsearchServiceSetup,
   KibanaRequest,
+  PluginInitializerContext,
   SavedObjectsClientContract,
   SavedObjectsServiceStart,
   UiSettingsServiceStart,
@@ -21,12 +22,14 @@ import { LicensingPluginSetup } from '../../licensing/server';
 import { SecurityPluginSetup } from '../../security/server';
 import { DEFAULT_SPACE_ID } from '../../spaces/common/constants';
 import { SpacesPluginSetup } from '../../spaces/server';
+import { TaskManagerSetupContract, TaskManagerStartContract } from '../../task_manager/server';
 import { ReportingConfig } from './';
 import { HeadlessChromiumDriverFactory } from './browsers/chromium/driver_factory';
+import { ReportingConfigType } from './config';
 import { checkLicense, getExportTypesRegistry, LevelLogger } from './lib';
-import { ESQueueInstance } from './lib/create_queue';
 import { screenshotsObservableFactory, ScreenshotsObservableFn } from './lib/screenshots';
 import { ReportingStore } from './lib/store';
+import { ExecuteReportTask, MonitorReportsTask, ReportTaskParams } from './lib/tasks';
 import { ReportingPluginRouter } from './types';
 
 export interface ReportingInternalSetup {
@@ -37,14 +40,15 @@ export interface ReportingInternalSetup {
   licensing: LicensingPluginSetup;
   security?: SecurityPluginSetup;
   spaces?: SpacesPluginSetup;
+  taskManager: TaskManagerSetupContract;
 }
 
 export interface ReportingInternalStart {
   browserDriverFactory: HeadlessChromiumDriverFactory;
-  esqueue: ESQueueInstance;
   store: ReportingStore;
   savedObjects: SavedObjectsServiceStart;
   uiSettings: UiSettingsServiceStart;
+  taskManager: TaskManagerStartContract;
 }
 
 export class ReportingCore {
@@ -53,24 +57,48 @@ export class ReportingCore {
   private readonly pluginSetup$ = new Rx.ReplaySubject<boolean>(); // observe async background setupDeps and config each are done
   private readonly pluginStart$ = new Rx.ReplaySubject<ReportingInternalStart>(); // observe async background startDeps
   private exportTypesRegistry = getExportTypesRegistry();
+  private executeTask: ExecuteReportTask;
+  private monitorTask: MonitorReportsTask;
   private config?: ReportingConfig;
+  private executing: Set<string>;
 
-  constructor(private logger: LevelLogger) {}
+  constructor(private logger: LevelLogger, context: PluginInitializerContext<ReportingConfigType>) {
+    const config = context.config.get<ReportingConfigType>();
+    this.executeTask = new ExecuteReportTask(this, config, this.logger);
+    this.monitorTask = new MonitorReportsTask(this, config, this.logger);
+    this.executing = new Set();
+  }
 
   /*
    * Register setupDeps
    */
   public pluginSetup(setupDeps: ReportingInternalSetup) {
     this.pluginSetup$.next(true); // trigger the observer
     this.pluginSetupDeps = setupDeps; // cache
+
+    const { executeTask, monitorTask } = this;
+    setupDeps.taskManager.registerTaskDefinitions({
+      [executeTask.TYPE]: executeTask.getTaskDefinition(),
+      [monitorTask.TYPE]: monitorTask.getTaskDefinition(),
+    });
   }
 
   /*
    * Register startDeps
    */
-  public pluginStart(startDeps: ReportingInternalStart) {
+  public async pluginStart(startDeps: ReportingInternalStart) {
     this.pluginStart$.next(startDeps); // trigger the observer
     this.pluginStartDeps = startDeps; // cache
+
+    const { taskManager } = startDeps;
+    if (this.getConfig().get('queue', 'pollEnabled')) {
+      const { executeTask, monitorTask } = this;
+      // enable this instance to generate reports and to monitor for pending reports
+      await Promise.all([executeTask.init(taskManager), monitorTask.init(taskManager)]);
+    } else {
+      // enable this instance to request other instances to generate reports
+      await this.executeTask.init(taskManager);
+    }
   }
 
   /*
@@ -151,8 +179,12 @@ export class ReportingCore {
     return this.exportTypesRegistry;
   }
 
-  public async getEsqueue() {
-    return (await this.getPluginStartDeps()).esqueue;
+  public async scheduleTask(report: ReportTaskParams) {
+    return await this.executeTask.scheduleTask(report);
+  }
+
+  public async getStore() {
+    return (await this.getPluginStartDeps()).store;
   }
 
   public async getLicenseInfo() {
@@ -239,4 +271,16 @@ export class ReportingCore {
     const savedObjectsClient = await this.getSavedObjectsClient(request);
     return await this.getUiSettingsServiceFactory(savedObjectsClient);
   }
+
+  public trackReport(reportId: string) {
+    this.executing.add(reportId);
+  }
+
+  public untrackReport(reportId: string) {
+    this.executing.delete(reportId);
+  }
+
+  public countConcurrentReports(): number {
+    return this.executing.size;
+  }
 }