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