[ResponseOps][Reporting] Send email with report attachment for scheduled reporting task (merging into feature branch)

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

@@ -63,6 +63,7 @@ import {
 import type { ReportingPluginRouter } from './types';
 import { EventTracker } from './usage';
 import { SCHEDULED_REPORT_SAVED_OBJECT_TYPE } from './saved_objects';
+import { EmailNotificationService } from './services/notifications/email_notification_service';
 import { API_PRIVILEGES } from './features';
 
 export interface ReportingInternalSetup {
@@ -181,12 +182,16 @@ export class ReportingCore {
       et.start({ ...startDeps });
     });
 
-    const { taskManager } = startDeps;
+    const { taskManager, notifications } = startDeps;
+    const emailNotificationService = new EmailNotificationService({
+      notifications,
+    });
+
     const { runSingleReportTask, runScheduledReportTask } = this;
     // enable this instance to generate reports
     await Promise.all([
       runSingleReportTask.init(taskManager),
-      runScheduledReportTask.init(taskManager),
+      runScheduledReportTask.init(taskManager, emailNotificationService),
     ]);
   }
 

x-pack/platform/plugins/private/reporting/server/lib/store/store.test.ts

@@ -400,6 +400,48 @@ describe('ReportingStore', () => {
     `);
   });
 
+  it('setReportWarning sets the status of a saved report to warning', async () => {
+    const store = new ReportingStore(mockCore, mockLogger);
+    const report = new SavedReport({
+      _id: 'id-of-processing',
+      _index: '.reporting-test-index-12345',
+      _seq_no: 42,
+      _primary_term: 10002,
+      jobtype: 'test-report',
+      created_by: 'created_by_test_string',
+      max_attempts: 50,
+      payload: {
+        title: 'test report',
+        headers: 'rp_test_headers',
+        objectType: 'testOt',
+        browserTimezone: 'ABC',
+        version: '7.14.0',
+      },
+      timeout: 30000,
+    });
+
+    await store.setReportWarning(report, {
+      output: { warnings: ['warning1'] },
+      warning: 'warning2',
+    } as any);
+
+    const [[updateCall]] = mockEsClient.update.mock.calls;
+
+    const response = (updateCall as estypes.UpdateRequest)?.doc as Report;
+    expect(response.migration_version).toBe(`7.14.0`);
+    expect(response.status).toBe(`completed_with_warnings`);
+    expect(response.output).toMatchInlineSnapshot(`
+      Object {
+        "warnings": Array [
+          "warning1",
+          "warning2",
+        ],
+      }
+    `);
+    expect(updateCall.if_seq_no).toBe(42);
+    expect(updateCall.if_primary_term).toBe(10002);
+  });
+
   describe('start', () => {
     class TestReportingStore extends ReportingStore {
       constructor(...args: ConstructorParameters<typeof ReportingStore>) {

x-pack/platform/plugins/private/reporting/server/lib/store/store.ts

@@ -53,6 +53,11 @@ export type ReportCompletedFields = Required<{
   output: Omit<ReportOutput, 'content'> | null;
 }>;
 
+export interface ReportWarningFields {
+  output: Omit<ReportOutput, 'content'>;
+  warning: string;
+}
+
 /*
  * When searching for long-pending reports, we get a subset of fields
  */
@@ -337,4 +342,31 @@ export class ReportingStore {
 
     return body;
   }
+
+  public async setReportWarning(
+    report: SavedReport,
+    warningInfo: ReportWarningFields
+  ): Promise<UpdateResponse<ReportDocument>> {
+    const { output, warning } = warningInfo;
+    const warnings: string[] = output.warnings ?? [];
+    warnings.push(warning);
+    const doc = sourceDoc({
+      output: {
+        ...output,
+        warnings,
+      },
+      status: JOB_STATUS.WARNINGS,
+    } as ReportSource);
+
+    let body: UpdateResponse<ReportDocument>;
+    try {
+      const client = await this.getClient();
+      body = await client.update<unknown, unknown, ReportDocument>(esDocForUpdate(report, doc));
+    } catch (err) {
+      this.logError(`Error in updating status to warning! Report: ${jobDebugMessage(report)}`, err, report); // prettier-ignore
+      throw err;
+    }
+
+    return body;
+  }
 }

x-pack/platform/plugins/private/reporting/server/lib/tasks/run_report.ts

@@ -11,7 +11,7 @@ import { timeout } from 'rxjs';
 import { Writable } from 'stream';
 import type { FakeRawRequest, Headers } from '@kbn/core-http-server';
 import { UpdateResponse } from '@elastic/elasticsearch/lib/api/types';
-import type { KibanaRequest, Logger } from '@kbn/core/server';
+import type { KibanaRequest, Logger, SavedObject } from '@kbn/core/server';
 import {
   CancellationToken,
   KibanaShuttingDownError,
@@ -44,9 +44,11 @@ import { ReportTaskParams, ReportingTask, ReportingTaskStatus, TIME_BETWEEN_ATTE
 import type { ReportingCore } from '../..';
 import { EventTracker } from '../../usage';
 import { Report, SavedReport } from '../store';
-import type { ReportFailedFields } from '../store/store';
+import type { ReportFailedFields, ReportWarningFields } from '../store/store';
 import { errorLogger } from './error_logger';
 import { finishedWithNoPendingCallbacks, getContentStream } from '../content_stream';
+import { EmailNotificationService } from '../../services/notifications/email_notification_service';
+import { ScheduledReportType } from '../../types';
 
 type CompletedReportOutput = Omit<ReportOutput, 'content'>;
 
@@ -91,6 +93,7 @@ export interface PrepareJobResults {
   jobId: string;
   report?: SavedReport;
   task?: ReportTaskParams;
+  reportSO?: SavedObject<ScheduledReportType>;
 }
 
 type ReportTaskParamsType = Record<string, any>;
@@ -106,6 +109,7 @@ export abstract class RunReportTask<TaskParams extends ReportTaskParamsType>
   protected kibanaName?: string;
   protected exportTypesRegistry: ExportTypesRegistry;
   protected eventTracker?: EventTracker;
+  protected emailNotificationService?: EmailNotificationService;
 
   constructor(protected readonly opts: ConstructorOpts) {
     this.logger = opts.logger.get('runTask');
@@ -127,13 +131,27 @@ export abstract class RunReportTask<TaskParams extends ReportTaskParamsType>
 
   protected abstract getMaxAttempts(): number | undefined;
 
+  protected abstract notify(
+    report: SavedReport,
+    taskInstance: ConcreteTaskInstance,
+    output: TaskRunResult,
+    byteSize: number,
+    reportSO?: SavedObject<ScheduledReportType>,
+    spaceId?: string
+  ): Promise<void>;
+
   // Public methods
-  public async init(taskManager: TaskManagerStartContract) {
+  public async init(
+    taskManager: TaskManagerStartContract,
+    emailNotificationService?: EmailNotificationService
+  ) {
     this.taskManagerStart = taskManager;
 
     const { uuid, name } = this.opts.reporting.getServerInfo();
     this.kibanaId = uuid;
     this.kibanaName = name;
+
+    this.emailNotificationService = emailNotificationService;
   }
 
   public getStatus() {
@@ -171,6 +189,11 @@ export abstract class RunReportTask<TaskParams extends ReportTaskParamsType>
     return exportType.jobContentEncoding;
   }
 
+  protected getJobContentExtension(jobType: string) {
+    const exportType = this.exportTypesRegistry.getByJobType(jobType);
+    return exportType.jobContentExtension;
+  }
+
   protected async failJob(
     report: SavedReport,
     error?: ReportingError
@@ -227,6 +250,24 @@ export abstract class RunReportTask<TaskParams extends ReportTaskParamsType>
     return await store.setReportError(report, doc);
   }
 
+  protected async saveExecutionWarning(
+    report: SavedReport,
+    output: CompletedReportOutput,
+    message: string
+  ): Promise<UpdateResponse<ReportDocument>> {
+    const logger = this.logger.get(report._id);
+    logger.warn(message);
+
+    // update the report in the store
+    const store = await this.opts.reporting.getStore();
+    const doc: ReportWarningFields = {
+      output,
+      warning: message,
+    };
+
+    return await store.setReportWarning(report, doc);
+  }
+
   protected formatOutput(output: CompletedReportOutput | ReportingError): ReportOutput {
     const docOutput = {} as ReportOutput;
     const unknownMime = null;
@@ -440,6 +481,7 @@ export abstract class RunReportTask<TaskParams extends ReportTaskParamsType>
             jobId: jId,
             report: preparedReport,
             task,
+            reportSO,
           } = await this.prepareJob(taskInstance);
           jobId = jId;
           report = preparedReport;
@@ -517,18 +559,28 @@ export abstract class RunReportTask<TaskParams extends ReportTaskParamsType>
             report._seq_no = stream.getSeqNo()!;
             report._primary_term = stream.getPrimaryTerm()!;
 
+            const byteSize = stream.bytesWritten;
             eventLog.logExecutionComplete({
               ...(output.metrics ?? {}),
-              byteSize: stream.bytesWritten,
+              byteSize,
             });
 
             if (output) {
-              logger.debug(`Job output size: ${stream.bytesWritten} bytes.`);
+              logger.debug(`Job output size: ${byteSize} bytes.`);
               // Update the job status to "completed"
               report = await this.completeJob(report, {
                 ...output,
-                size: stream.bytesWritten,
+                size: byteSize,
               });
+
+              await this.notify(
+                report,
+                taskInstance,
+                output,
+                byteSize,
+                reportSO,
+                task.payload.spaceId
+              );
             }
 
             // untrack the report for concurrency awareness