Reporting: Task Manager integration for 7.x

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

@@ -68,6 +68,7 @@ export interface ReportSource {
   };
   meta: { objectType: string; layout?: string };
   browser_type: string;
+  migration_version: string;
   max_attempts: number;
   timeout: number;
 
@@ -77,7 +78,7 @@ export interface ReportSource {
   started_at?: string;
   completed_at?: string;
   created_at: string;
-  process_expiration?: string;
+  process_expiration?: string | null; // must be set to null to clear the expiration
 }
 
 /*

x-pack/plugins/reporting/kibana.json

@@ -10,6 +10,7 @@
     "management",
     "licensing",
     "uiActions",
+    "taskManager",
     "embeddable",
     "screenshotMode",
     "share",

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

@@ -24,13 +24,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, ReportingSetup } 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 {
@@ -40,6 +41,7 @@ export interface ReportingInternalSetup {
   licensing: LicensingPluginSetup;
   security?: SecurityPluginSetup;
   spaces?: SpacesPluginSetup;
+  taskManager: TaskManagerSetupContract;
   screenshotMode: ScreenshotModePluginSetup;
   logger: LevelLogger;
 }
@@ -51,7 +53,7 @@ export interface ReportingInternalStart {
   uiSettings: UiSettingsServiceStart;
   esClient: IClusterClient;
   data: DataPluginStart;
-  esqueue: ESQueueInstance;
+  taskManager: TaskManagerStartContract;
   logger: LevelLogger;
 }
 
@@ -62,17 +64,24 @@ export class ReportingCore {
   private readonly pluginStart$ = new Rx.ReplaySubject<ReportingInternalStart>(); // observe async background startDeps
   private deprecatedAllowedRoles: string[] | false = false; // DEPRECATED. If `false`, the deprecated features have been disableed
   private exportTypesRegistry = getExportTypesRegistry();
-  private config?: ReportingConfig;
+  private executeTask: ExecuteReportTask;
+  private monitorTask: MonitorReportsTask;
+  private config?: ReportingConfig; // final config, includes dynamic values based on OS type
+  private executing: Set<string>;
 
   public getContract: () => ReportingSetup;
 
   constructor(private logger: LevelLogger, context: PluginInitializerContext<ReportingConfigType>) {
     const syncConfig = context.config.get<ReportingConfigType>();
     this.deprecatedAllowedRoles = syncConfig.roles.enabled ? syncConfig.roles.allow : false;
+    this.executeTask = new ExecuteReportTask(this, syncConfig, this.logger);
+    this.monitorTask = new MonitorReportsTask(this, syncConfig, this.logger);
 
     this.getContract = () => ({
       usesUiCapabilities: () => syncConfig.roles.enabled === false,
     });
+
+    this.executing = new Set();
   }
 
   /*
@@ -81,14 +90,25 @@ export class ReportingCore {
   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;
+    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)]);
   }
 
   /*
@@ -193,8 +213,8 @@ 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() {
@@ -296,4 +316,16 @@ export class ReportingCore {
     const startDeps = await this.getPluginStartDeps();
     return startDeps.esClient;
   }
+
+  public trackReport(reportId: string) {
+    this.executing.add(reportId);
+  }
+
+  public untrackReport(reportId: string) {
+    this.executing.delete(reportId);
+  }
+
+  public countConcurrentReports(): number {
+    return this.executing.size;
+  }
 }

x-pack/plugins/reporting/server/export_types/csv/create_job.ts

@@ -5,7 +5,6 @@
  * 2.0.
  */
 
-import { CSV_JOB_TYPE_DEPRECATED } from '../../../common/constants';
 import { cryptoFactory } from '../../lib';
 import { CreateJobFn, CreateJobFnFactory } from '../../types';
 import {
@@ -16,9 +15,7 @@ import {
 
 export const createJobFnFactory: CreateJobFnFactory<
   CreateJobFn<JobParamsDeprecatedCSV, TaskPayloadDeprecatedCSV>
-> = function createJobFactoryFn(reporting, parentLogger) {
-  const logger = parentLogger.clone([CSV_JOB_TYPE_DEPRECATED, 'create-job']);
-
+> = function createJobFactoryFn(reporting, logger) {
   const config = reporting.getConfig();
   const crypto = cryptoFactory(config.get('encryptionKey'));
 

x-pack/plugins/reporting/server/export_types/csv/execute_job.test.ts

@@ -676,8 +676,7 @@ describe('CSV Execute Job', function () {
     });
   });
 
-  // FLAKY: https://github.com/elastic/kibana/issues/43069
-  describe.skip('cancellation', function () {
+  describe('cancellation', function () {
     const scrollId = getRandomScrollId();
 
     beforeEach(function () {
@@ -709,7 +708,7 @@ describe('CSV Execute Job', function () {
         cancellationToken
       );
 
-      await delay(100);
+      await delay(250);
 
       expect(mockEsClient.search).toHaveBeenCalled();
       expect(mockEsClient.scroll).toHaveBeenCalled();

x-pack/plugins/reporting/server/export_types/csv/execute_job.ts

@@ -5,7 +5,7 @@
  * 2.0.
  */
 
-import { CONTENT_TYPE_CSV, CSV_JOB_TYPE_DEPRECATED } from '../../../common/constants';
+import { CONTENT_TYPE_CSV } from '../../../common/constants';
 import { RunTaskFn, RunTaskFnFactory } from '../../types';
 import { decryptJobHeaders } from '../common';
 import { createGenerateCsv } from './generate_csv';
@@ -18,7 +18,7 @@ export const runTaskFnFactory: RunTaskFnFactory<
 
   return async function runTask(jobId, job, cancellationToken) {
     const elasticsearch = await reporting.getEsClient();
-    const logger = parentLogger.clone([CSV_JOB_TYPE_DEPRECATED, 'execute-job', jobId]);
+    const logger = parentLogger.clone([jobId]);
     const generateCsv = createGenerateCsv(logger);
 
     const encryptionKey = config.get('encryptionKey');

x-pack/plugins/reporting/server/export_types/png/create_job/index.ts

@@ -5,16 +5,14 @@
  * 2.0.
  */
 
-import { PNG_JOB_TYPE } from '../../../../common/constants';
 import { cryptoFactory } from '../../../lib';
 import { CreateJobFn, CreateJobFnFactory } from '../../../types';
 import { validateUrls } from '../../common';
 import { JobParamsPNG, TaskPayloadPNG } from '../types';
 
 export const createJobFnFactory: CreateJobFnFactory<
   CreateJobFn<JobParamsPNG, TaskPayloadPNG>
-> = function createJobFactoryFn(reporting, parentLogger) {
-  const logger = parentLogger.clone([PNG_JOB_TYPE, 'execute-job']);
+> = function createJobFactoryFn(reporting, logger) {
   const config = reporting.getConfig();
   const crypto = cryptoFactory(config.get('encryptionKey'));
 

x-pack/plugins/reporting/server/export_types/printable_pdf/create_job/index.ts

@@ -6,7 +6,6 @@
  */
 
 import { KibanaRequest, RequestHandlerContext } from 'src/core/server';
-import { PDF_JOB_TYPE } from '../../../../common/constants';
 import { cryptoFactory } from '../../../lib';
 import { CreateJobFn, CreateJobFnFactory } from '../../../types';
 import { validateUrls } from '../../common';
@@ -16,10 +15,9 @@ import { compatibilityShimFactory } from './compatibility_shim';
 
 export const createJobFnFactory: CreateJobFnFactory<
   CreateJobFn<JobParamsPDF, TaskPayloadPDF>
-> = function createJobFactoryFn(reporting, parentLogger) {
+> = function createJobFactoryFn(reporting, logger) {
   const config = reporting.getConfig();
   const crypto = cryptoFactory(config.get('encryptionKey'));
-  const logger = parentLogger.clone([PDF_JOB_TYPE, 'create-job']);
   const compatibilityShim = compatibilityShimFactory(logger);
 
   // 7.x and below only