Pulse errors implementation in toasts

src/core/public/core_system.ts

@@ -164,9 +164,9 @@ export class CoreSystem {
       await this.integrations.setup();
       const http = this.http.setup({ injectedMetadata, fatalErrors: this.fatalErrorsSetup });
       const uiSettings = this.uiSettings.setup({ http, injectedMetadata });
-      const notifications = this.notifications.setup({ uiSettings });
-      const pulse = await this.pulse.setup();
 
+      const pulse = await this.pulse.setup();
+      const notifications = this.notifications.setup({ uiSettings, pulse });
       const pluginDependencies = this.plugins.getOpaqueIds();
       const context = this.context.setup({
         // We inject a fake "legacy plugin" with dependencies on every plugin so that legacy plugins:

src/core/public/notifications/notifications_service.ts

@@ -19,14 +19,18 @@
 
 import { i18n } from '@kbn/i18n';
 
-import { Subscription } from 'rxjs';
+import { Subscription, Observable } from 'rxjs';
 import { I18nStart } from '../i18n';
 import { ToastsService, ToastsSetup, ToastsStart } from './toasts';
 import { IUiSettingsClient } from '../ui_settings';
 import { OverlayStart } from '../overlays';
+import { PulseServiceSetup, PulseService } from '../pulse';
+import { errorChannelPayloads } from '../pulse/mock_data/errors';
+import { PulseErrorInstructionValue, PulseInstruction } from '../pulse/channel';
 
 interface SetupDeps {
   uiSettings: IUiSettingsClient;
+  pulse: PulseServiceSetup;
 }
 
 interface StartDeps {
@@ -41,13 +45,19 @@ export class NotificationsService {
   private uiSettingsErrorSubscription?: Subscription;
   private targetDomElement?: HTMLElement;
 
+  private pulse: PulseService;
+  private instructionsSubscription?: Subscription;
+
   constructor() {
     this.toasts = new ToastsService();
+    this.pulse = new PulseService();
   }
 
-  public setup({ uiSettings }: SetupDeps): NotificationsSetup {
-    const notificationSetup = { toasts: this.toasts.setup({ uiSettings }) };
-
+  public setup({ uiSettings, pulse }: SetupDeps): NotificationsSetup {
+    const notificationSetup = {
+      toasts: this.toasts.setup({ uiSettings }),
+      pulse: this.pulse.setup(),
+    };
     this.uiSettingsErrorSubscription = uiSettings.getUpdateErrors$().subscribe((error: Error) => {
       notificationSetup.toasts.addDanger({
         title: i18n.translate('core.notifications.unableUpdateUISettingNotificationMessageTitle', {
@@ -57,6 +67,24 @@ export class NotificationsService {
       });
     });
 
+    errorChannelPayloads.forEach((element: PulseErrorInstructionValue) => {
+      pulse.getChannel('errors').sendPulse(element);
+    });
+    this.instructionsSubscription = pulse
+      .getChannel('errors')
+      .instructions$()
+      .subscribe(instructions => {
+        if (instructions && instructions.length) {
+          instructions.forEach(instruction => {
+            notificationSetup.toasts.addError(new Error(JSON.stringify(instruction)), {
+              // @ts-ignore-next-line
+              title: `Error:${instruction.hash}`,
+              toastMessage: 'The error has been reported to Pulse',
+            });
+          });
+        }
+      });
+
     return notificationSetup;
   }
 
@@ -72,6 +100,7 @@ export class NotificationsService {
 
   public stop() {
     this.toasts.stop();
+    this.pulse.stop();
 
     if (this.targetDomElement) {
       this.targetDomElement.textContent = '';
@@ -80,13 +109,18 @@ export class NotificationsService {
     if (this.uiSettingsErrorSubscription) {
       this.uiSettingsErrorSubscription.unsubscribe();
     }
+
+    if (this.instructionsSubscription) {
+      this.instructionsSubscription.unsubscribe();
+    }
   }
 }
 
 /** @public */
 export interface NotificationsSetup {
   /** {@link ToastsSetup} */
   toasts: ToastsSetup;
+  pulse: Promise<PulseServiceSetup>;
 }
 
 /** @public */

src/core/public/pulse/channel.ts

@@ -22,7 +22,7 @@ import { ChannelConfig } from 'src/core/server/pulse/channel';
 import { PulseClient } from './client_wrappers/pulse';
 
 // eslint-disable-next-line @kbn/eslint/no-restricted-paths
-export { PulseInstruction } from '../../server/pulse/channel';
+export { PulseInstruction, PulseErrorInstructionValue } from '../../server/pulse/channel';
 
 export class PulseChannel<Payload = any> {
   private readonly pulseClient: PulseClient;
@@ -32,8 +32,9 @@ export class PulseChannel<Payload = any> {
   }
 
   public async getRecords() {
-    return this.pulseClient.getRecords(this.id);
+    return await this.pulseClient.getRecords(this.id);
   }
+
   public get id() {
     return this.config.id;
   }

src/core/public/pulse/client_wrappers/pulse.ts

@@ -18,7 +18,6 @@
  */
 
 export class PulseClient {
-
   constructor() {}
 
   public async putRecord(channel: string, doc: any) {

src/core/public/pulse/index.ts

@@ -48,7 +48,7 @@ export class PulseService {
   constructor() {
     this.channels = new Map(
       channelNames.map((id): [string, PulseChannel] => {
-        const instructions$ = new Subject<PulseInstruction>();
+        const instructions$ = new Subject<PulseInstruction[]>();
         this.instructions$.set(id, instructions$);
         const channel = new PulseChannel({ id, instructions$, logger });
         return [channel.id, channel];

src/core/public/pulse/mock_data/errors.ts

@@ -0,0 +1,57 @@
+/*
+ * Licensed to Elasticsearch B.V. under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import moment from 'moment';
+// eslint-disable-next-line @kbn/eslint/no-restricted-paths
+import { PulseErrorInstructionValue } from '../channel';
+
+export const errorChannelPayloads: PulseErrorInstructionValue[] = [
+  {
+    channel_id: 'errors',
+    deployment_id: '123',
+    message: 'Error: [resource_already_exists_exception]',
+    hash: 'index [pulse-poc-raw-default/1234567890] already exists',
+    status: 'new',
+    currentKibanaVersion: 'v7.x',
+    timestamp: moment().toDate(),
+  },
+  {
+    channel_id: 'errors',
+    deployment_id: '123',
+    message: 'Error: [resource_already_exists_exception]',
+    hash: 'index [pulse-poc-raw-default/1QJURO2GRfqpFfuOp12rIg] already exists',
+    status: 'seen',
+    currentKibanaVersion: 'v7.x',
+    timestamp: moment()
+      .subtract(1, 'days')
+      .toDate(),
+  },
+  {
+    channel_id: 'errors',
+    deployment_id: '123',
+    message: '[TypeError]: Component failed to mount',
+    hash: 'generic:arbitraryError 1QJURO2GRfqpFfuOp12rIg',
+    status: 'seen',
+    fixedVersion: 'v7.5.2',
+    currentKibanaVersion: 'v7.x',
+    timestamp: moment()
+      .subtract(60, 'seconds')
+      .toDate(),
+  },
+];

src/core/server/pulse/channel.ts

@@ -23,27 +23,37 @@ import { PulseCollectorConstructor } from './types';
 import { Logger } from '../logging';
 
 import { IPulseElasticsearchClient } from './client_wrappers/types';
-
+// I'll probably need to extend the PulseInstruction to declare the value types for an Error Instruction
+export interface PulseErrorInstructionValue {
+  timestamp: Date;
+  message: string;
+  hash: string; // just use the i18n strings as they are unique
+  status: 'new' | 'seen';
+  currentKibanaVersion: string;
+  channel_id: string;
+  deployment_id: string;
+  fixedVersion?: string;
+}
 export interface PulseInstruction {
   owner: string;
   id: string;
-  value: unknown;
+  value: PulseErrorInstructionValue | unknown;
 }
 
-export interface ChannelConfig {
+export interface ChannelConfig<I = PulseInstruction> {
   id: string;
-  instructions$: Subject<PulseInstruction>;
+  instructions$: Subject<I[]>;
   logger: Logger;
 }
 export interface ChannelSetupContext {
   elasticsearch: IPulseElasticsearchClient;
   // savedObjects: SavedObjectsServiceSetup;
 }
 
-export class PulseChannel<Payload = any, Rec = Payload> {
+export class PulseChannel<I = PulseInstruction> {
   private readonly collector: any;
 
-  constructor(private readonly config: ChannelConfig) {
+  constructor(private readonly config: ChannelConfig<I>) {
     const Collector: PulseCollectorConstructor = require(`${__dirname}/collectors/${this.id}`)
       .Collector;
     this.collector = new Collector(this.config.logger);

src/core/server/pulse/client_wrappers/elasticsearch.ts

@@ -46,7 +46,9 @@ export class PulseElasticsearchClient implements IPulseElasticsearchClient {
   }
 
   public async index(channel: string, doc: any) {
-    const id = uuid.v4();
+    // if the document has an id, use that, otherwise general a unique id.
+    const providedDocumentId = doc._id || doc.hash || null;
+    const id = providedDocumentId ? providedDocumentId : uuid.v4();
     await this.elasticsearch!.callAsInternalUser('index', {
       index: this.buildIndex(channel),
       id: `${id}`,

src/core/server/pulse/collectors/errors.ts

@@ -24,8 +24,16 @@
 import moment from 'moment';
 import { PulseCollector, CollectorSetupContext } from '../types';
 
-export interface Payload {
-  errorId: string;
+export type Payload = Exclude<ErrorInstruction, 'channel_id' | 'deployment_id' | 'timestamp'>;
+export interface ErrorInstruction {
+  channel_id: string;
+  currentKibanaVersion?: string;
+  deployment_id: string;
+  hash: string;
+  fixedVersion?: string;
+  message: string;
+  status: 'new' | 'seen';
+  timestamp: Date;
 }
 
 export class Collector extends PulseCollector<Payload> {
@@ -36,12 +44,39 @@ export class Collector extends PulseCollector<Payload> {
     if (this.elasticsearch?.createIndexIfNotExist) {
       const mappings = {
         properties: {
-          timestamp: {
-            type: 'date',
+          channel_id: {
+            type: 'keyword',
+          },
+          currentKibanaVersion: {
+            type: 'text',
+            fields: {
+              keyword: {
+                type: 'keyword',
+                ignore_above: 256,
+              },
+            },
           },
-          errorId: {
+          deployment_id: {
             type: 'keyword',
           },
+          hash: {
+            type: 'text',
+            fields: {
+              keyword: {
+                type: 'keyword',
+                ignore_above: 256,
+              },
+            },
+          },
+          id: {
+            type: 'text',
+            fields: {
+              keyword: {
+                type: 'keyword',
+                ignore_above: 256,
+              },
+            },
+          },
           message: {
             type: 'text',
             fields: {
@@ -50,24 +85,47 @@ export class Collector extends PulseCollector<Payload> {
               },
             },
           },
+          status: {
+            type: 'keyword',
+          },
+          timestamp: {
+            type: 'date',
+          },
+          fixedVersion: {
+            type: 'keyword',
+          },
         },
       };
       await this.elasticsearch!.createIndexIfNotExist(this.channelName, mappings);
     }
   }
-  public async putRecord(originalPayload: Payload) {
-    const payload = { timestamp: moment.utc().toISOString(), ...originalPayload };
-    if (this.elasticsearch) await this.elasticsearch.index(this.channelName, payload);
+  public async putRecord(originalPayload: Payload | Payload[]) {
+    const payloads = Array.isArray(originalPayload) ? originalPayload : [originalPayload];
+    payloads.forEach(async payload => {
+      if (!payload.hash) {
+        throw Error(`error payload does not contain hash: ${JSON.stringify(payload)}.`);
+      }
+      if (this.elasticsearch) {
+        await this.elasticsearch.index(this.channelName, {
+          ...payload,
+          channel_id: 'errors',
+          deployment_id: '123',
+          id: payload.hash,
+          timestamp: moment(),
+        });
+      }
+    });
   }
 
   public async getRecords() {
     if (this.elasticsearch) {
       const results = await this.elasticsearch.search(this.channelName, {
         bool: {
+          should: [{ match: { status: 'new' } }],
           filter: {
             range: {
               timestamp: {
-                gte: 'now-10s',
+                gte: 'now-20s',
                 lte: 'now',
               },
             },

src/core/server/pulse/index.ts

@@ -28,7 +28,7 @@ import { Logger } from '../logging';
 import { ElasticsearchServiceSetup, IClusterClient } from '../elasticsearch';
 import { PulseChannel, PulseInstruction } from './channel';
 import { sendUsageFrom, sendPulse, Fetcher } from './send_pulse';
-import { SavedObjectsServiceSetup } from '../saved_objects';
+// import { SavedObjectsServiceSetup } from '../saved_objects';
 import { InternalHttpServiceSetup } from '../http';
 import { PulseElasticsearchClient } from './client_wrappers/elasticsearch';
 import { registerPulseRoutes } from './routes';
@@ -39,7 +39,7 @@ export interface InternalPulseService {
 
 export interface PulseSetupDeps {
   elasticsearch: ElasticsearchServiceSetup;
-  savedObjects: SavedObjectsServiceSetup;
+  // savedObjects: SavedObjectsServiceSetup;
   http: InternalHttpServiceSetup;
 }
 
@@ -72,7 +72,7 @@ export class PulseService {
     this.log = coreContext.logger.get('pulse-service');
     this.channels = new Map(
       channelNames.map((id): [string, PulseChannel] => {
-        const instructions$ = new Subject<PulseInstruction>();
+        const instructions$ = new Subject<PulseInstruction[]>();
         this.instructions$.set(id, instructions$);
         const channel = new PulseChannel({ id, instructions$, logger: this.log });
         return [channel.id, channel];