Pulse poc errors channel with notifications

src/core/public/core_system.ts

@@ -26,6 +26,7 @@ import { ChromeService } from './chrome';
 import { FatalErrorsService, FatalErrorsSetup } from './fatal_errors';
 import { HttpService } from './http';
 import { I18nService } from './i18n';
+import { PulseService } from './pulse';
 import {
   InjectedMetadataParams,
   InjectedMetadataService,
@@ -100,6 +101,7 @@ export class CoreSystem {
   private readonly rendering: RenderingService;
   private readonly context: ContextService;
   private readonly integrations: IntegrationsService;
+  private readonly pulse: PulseService;
 
   private readonly rootDomElement: HTMLElement;
   private readonly coreContext: CoreContext;
@@ -137,6 +139,7 @@ export class CoreSystem {
     this.rendering = new RenderingService();
     this.application = new ApplicationService();
     this.integrations = new IntegrationsService();
+    this.pulse = new PulseService();
 
     this.coreContext = { coreId: Symbol('core'), env: injectedMetadata.env };
 
@@ -162,6 +165,7 @@ export class CoreSystem {
       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 pluginDependencies = this.plugins.getOpaqueIds();
       const context = this.context.setup({
@@ -184,6 +188,7 @@ export class CoreSystem {
         injectedMetadata,
         notifications,
         uiSettings,
+        pulse,
       };
 
       // Services that do not expose contracts at setup
@@ -216,6 +221,7 @@ export class CoreSystem {
       const i18n = await this.i18n.start();
       const application = await this.application.start({ http, injectedMetadata });
       await this.integrations.start({ uiSettings });
+      const pulse = await this.pulse.start();
 
       const coreUiTargetDomElement = document.createElement('div');
       coreUiTargetDomElement.id = 'kibana-body';
@@ -271,6 +277,7 @@ export class CoreSystem {
         notifications,
         overlays,
         uiSettings,
+        pulse,
       };
 
       const plugins = await this.plugins.start(core);

src/core/public/index.ts

@@ -66,6 +66,7 @@ import { UiSettingsState, IUiSettingsClient } from './ui_settings';
 import { ApplicationSetup, Capabilities, ApplicationStart } from './application';
 import { DocLinksStart } from './doc_links';
 import { SavedObjectsStart } from './saved_objects';
+import { PulseServiceSetup, PulseServiceStart } from './pulse';
 export { PackageInfo, EnvironmentMode } from '../server/types';
 import {
   IContextContainer,
@@ -177,6 +178,7 @@ export interface CoreSetup<TPluginsStart extends object = object> {
   notifications: NotificationsSetup;
   /** {@link IUiSettingsClient} */
   uiSettings: IUiSettingsClient;
+  pulse: PulseServiceSetup;
   /**
    * exposed temporarily until https://github.com/elastic/kibana/issues/41990 done
    * use *only* to retrieve config values. There is no way to set injected values
@@ -219,6 +221,7 @@ export interface CoreStart {
   i18n: I18nStart;
   /** {@link NotificationsStart} */
   notifications: NotificationsStart;
+  pulse: PulseServiceStart;
   /** {@link OverlayStart} */
   overlays: OverlayStart;
   /** {@link IUiSettingsClient} */
@@ -306,4 +309,6 @@ export {
   PluginOpaqueId,
   IUiSettingsClient,
   UiSettingsState,
+  PulseServiceSetup,
+  PulseServiceStart,
 };

src/core/public/plugins/plugin_context.ts

@@ -104,6 +104,7 @@ export function createPluginSetupContext<
     http: deps.http,
     notifications: deps.notifications,
     uiSettings: deps.uiSettings,
+    pulse: deps.pulse,
     injectedMetadata: {
       getInjectedVar: deps.injectedMetadata.getInjectedVar,
     },
@@ -147,6 +148,7 @@ export function createPluginStartContext<
     overlays: deps.overlays,
     uiSettings: deps.uiSettings,
     savedObjects: deps.savedObjects,
+    pulse: deps.pulse,
     injectedMetadata: {
       getInjectedVar: deps.injectedMetadata.getInjectedVar,
     },

src/core/public/pulse/channel.ts

@@ -0,0 +1,21 @@
+/*
+ * 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.
+ */
+
+// eslint-disable-next-line @kbn/eslint/no-restricted-paths
+export { PulseInstruction, PulseChannel } from '../../server/pulse/channel';

src/core/public/pulse/collectors/default.ts

@@ -0,0 +1,22 @@
+/*
+ * 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.
+ */
+
+export async function getRecords() {
+  return [{}];
+}

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

@@ -0,0 +1,22 @@
+/*
+ * 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.
+ */
+
+export async function getRecords() {
+  return [{}];
+}

src/core/public/pulse/collectors/notifications.ts

@@ -0,0 +1,22 @@
+/*
+ * 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.
+ */
+
+export async function getRecords() {
+  return [{}];
+}

src/core/public/pulse/index.ts

@@ -0,0 +1,153 @@
+/*
+ * 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 { Subject } from 'rxjs';
+
+// eslint-disable-next-line @kbn/eslint/no-restricted-paths
+import { InstructionsResponse } from '../../server/pulse';
+// eslint-disable-next-line @kbn/eslint/no-restricted-paths
+import { PulseChannel, PulseInstruction } from './channel';
+// eslint-disable-next-line @kbn/eslint/no-restricted-paths
+import { Fetcher, sendPulse } from '../../server/pulse/send_pulse';
+
+export interface PulseServiceSetup {
+  getChannel: (id: string) => PulseChannel;
+}
+
+// eslint-disable-next-line @typescript-eslint/no-empty-interface
+export interface PulseServiceStart {}
+
+const channelNames = ['default', 'notifications', 'errors'];
+
+export class PulseService {
+  private retriableErrors = 0;
+  private readonly channels: Map<string, PulseChannel>;
+  private readonly instructions: Map<string, Subject<any>> = new Map();
+
+  constructor() {
+    this.channels = new Map(
+      channelNames.map((id): [string, PulseChannel] => {
+        const instructions$ = new Subject<PulseInstruction>();
+        this.instructions.set(id, instructions$);
+        const channel = new PulseChannel({ id, instructions$ });
+        return [channel.id, channel];
+      })
+    );
+  }
+
+  public async setup(): Promise<PulseServiceSetup> {
+    // poll for instructions every second for this deployment
+    setInterval(() => {
+      // eslint-disable-next-line no-console
+      this.loadInstructions().catch(err => console.error(err.stack));
+    }, 10000);
+
+    // eslint-disable-next-line no-console
+    console.log('Will attempt first telemetry collection in 5 seconds...');
+    setTimeout(() => {
+      setInterval(() => {
+        // eslint-disable-next-line no-console
+        this.sendTelemetry().catch(err => console.error(err.stack));
+      }, 5000);
+    }, 5000);
+
+    return {
+      getChannel: (id: string) => {
+        const channel = this.channels.get(id);
+        if (!channel) {
+          throw new Error(`Unknown channel: ${id}`);
+        }
+        return channel;
+      },
+    };
+  }
+
+  private async sendTelemetry() {
+    const fetcher: Fetcher<Response> = async (url, channels) => {
+      return await fetch(url, {
+        method: 'post',
+
+        headers: {
+          'content-type': 'application/json',
+          'kbn-xsrf': 'true',
+        },
+        body: JSON.stringify({
+          channels,
+        }),
+      });
+    };
+
+    return await sendPulse(this.channels, fetcher);
+  }
+
+  private async loadInstructions() {
+    const url = 'http://localhost:5601/api/pulse_poc/instructions/123';
+    let response: any;
+    try {
+      response = await fetch(url);
+    } catch (err) {
+      if (!err.message.includes('ECONNREFUSED')) {
+        throw err;
+      }
+      this.handleRetriableError();
+      return;
+    }
+    if (response.status === 503) {
+      this.handleRetriableError();
+      return;
+    }
+
+    if (response.status !== 200) {
+      const responseBody = await response.text();
+      throw new Error(`${response.status}: ${responseBody}`);
+    }
+
+    const responseBody: InstructionsResponse = await response.json();
+
+    responseBody.channels.forEach(channel => {
+      const instructions$ = this.instructions.get(channel.id);
+      if (!instructions$) {
+        throw new Error(
+          `Channel (${channel.id}) from service has no corresponding channel handler in client`
+        );
+      }
+
+      channel.instructions.forEach(instruction => instructions$.next(instruction));
+    });
+  }
+
+  private handleRetriableError() {
+    this.retriableErrors++;
+    if (this.retriableErrors === 1) {
+      // eslint-disable-next-line no-console
+      console.warn(
+        'Kibana is not yet available at http://localhost:5601/api, will continue to check for the next 120 seconds...'
+      );
+    } else if (this.retriableErrors > 120) {
+      this.retriableErrors = 0;
+    }
+  }
+
+  async start(): Promise<PulseServiceStart> {
+    return {};
+  }
+  public stop() {
+    // nothing to do here currently
+  }
+}

src/core/server/index.ts

@@ -47,6 +47,7 @@ import { IUiSettingsClient, UiSettingsServiceSetup, UiSettingsServiceStart } fro
 import { SavedObjectsClientContract } from './saved_objects/types';
 import { SavedObjectsServiceSetup, SavedObjectsServiceStart } from './saved_objects';
 import { CapabilitiesSetup, CapabilitiesStart } from './capabilities';
+import { PulseServiceSetup } from './pulse';
 import { UuidServiceSetup } from './uuid';
 
 export { bootstrap } from './bootstrap';
@@ -274,6 +275,7 @@ export interface CoreSetup {
   /** {@link UiSettingsServiceSetup} */
   uiSettings: UiSettingsServiceSetup;
   /** {@link UuidServiceSetup} */
+  pulse: PulseServiceSetup;
   uuid: UuidServiceSetup;
 }
 
@@ -298,5 +300,6 @@ export {
   PluginsServiceSetup,
   PluginsServiceStart,
   PluginOpaqueId,
+  PulseServiceSetup,
   UuidServiceSetup,
 };

src/core/server/legacy/legacy_service.ts

@@ -281,6 +281,7 @@ export class LegacyService implements CoreService {
         dataClient$: setupDeps.core.elasticsearch.dataClient$,
         createClient: setupDeps.core.elasticsearch.createClient,
       },
+      pulse: setupDeps.core.pulse,
       http: {
         createCookieSessionStorageFactory: setupDeps.core.http.createCookieSessionStorageFactory,
         registerRouteHandlerContext: setupDeps.core.http.registerRouteHandlerContext.bind(

src/core/server/plugins/plugin_context.ts

@@ -137,6 +137,7 @@ export function createPluginSetupContext<TPlugin, TPluginDependencies>(
   plugin: PluginWrapper<TPlugin, TPluginDependencies>
 ): CoreSetup {
   return {
+    pulse: deps.pulse,
     capabilities: {
       registerProvider: deps.capabilities.registerProvider,
       registerSwitcher: deps.capabilities.registerSwitcher,