Add server FTRs

packages/elastic-analytics/src/analytics_client/analytics_client.test.ts

@@ -8,14 +8,16 @@
 
 // eslint-disable-next-line max-classes-per-file
 import type { Observable } from 'rxjs';
-import { Subject } from 'rxjs';
+import { BehaviorSubject, Subject } from 'rxjs';
 import type { MockedLogger } from '@kbn/logging-mocks';
 import { loggerMock } from '@kbn/logging-mocks';
 import { AnalyticsClient } from './analytics_client';
 import { take, toArray } from 'rxjs/operators';
 import { shippersMock } from '../shippers/mocks';
 import type { EventContext, TelemetryCounter } from '../events';
 
+const delay = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));
+
 describe('AnalyticsClient', () => {
   let analyticsClient: AnalyticsClient;
   let logger: MockedLogger;
@@ -307,9 +309,12 @@ describe('AnalyticsClient', () => {
       expect(optIn).toHaveBeenCalledWith(true);
     });
 
-    test('Spreads the context updates to the shipper', () => {
+    test('Spreads the context updates to the shipper (only after opt-in)', async () => {
       const extendContextMock = jest.fn();
       analyticsClient.registerShipper(MockedShipper, { extendContextMock });
+      expect(extendContextMock).toHaveBeenCalledTimes(0); // Not until we have opt-in
+      analyticsClient.optIn({ global: { enabled: true } });
+      await delay(10);
       expect(extendContextMock).toHaveBeenCalledWith({}); // The initial context
 
       const context$ = new Subject<{ a_field: boolean }>();
@@ -329,11 +334,22 @@ describe('AnalyticsClient', () => {
       expect(extendContextMock).toHaveBeenCalledWith({ a_field: true }); // After update
     });
 
-    test('Handles errors in the shipper', () => {
+    test('Does not spread the context if opt-in === false', async () => {
+      const extendContextMock = jest.fn();
+      analyticsClient.registerShipper(MockedShipper, { extendContextMock });
+      expect(extendContextMock).toHaveBeenCalledTimes(0); // Not until we have opt-in
+      analyticsClient.optIn({ global: { enabled: false } });
+      await delay(10);
+      expect(extendContextMock).toHaveBeenCalledTimes(0); // Not until we have opt-in
+    });
+
+    test('Handles errors in the shipper', async () => {
       const extendContextMock = jest.fn().mockImplementation(() => {
         throw new Error('Something went terribly wrong');
       });
       analyticsClient.registerShipper(MockedShipper, { extendContextMock });
+      analyticsClient.optIn({ global: { enabled: true } });
+      await delay(10);
       expect(extendContextMock).toHaveBeenCalledWith({}); // The initial context
       expect(logger.warn).toHaveBeenCalledWith(
         `Shipper "${MockedShipper.shipperName}" failed to extend the context`,
@@ -372,6 +388,50 @@ describe('AnalyticsClient', () => {
       ]);
     });
 
+    test('It does not break if context emits `undefined`', async () => {
+      const context$ = new Subject<{ a_field: boolean }>();
+      analyticsClient.registerContextProvider({
+        schema: {
+          a_field: {
+            type: 'boolean',
+            _meta: {
+              description: 'a_field description',
+            },
+          },
+        },
+        context$,
+      });
+
+      const globalContextPromise = globalContext$.pipe(take(3), toArray()).toPromise();
+      context$.next();
+      context$.next(undefined);
+      await expect(globalContextPromise).resolves.toEqual([
+        {}, // Original empty state
+        {},
+        {},
+      ]);
+    });
+
+    test('It does not break for BehaviourSubjects (emitting as soon as they connect)', async () => {
+      const context$ = new BehaviorSubject<{ a_field: boolean }>({ a_field: true });
+      analyticsClient.registerContextProvider({
+        schema: {
+          a_field: {
+            type: 'boolean',
+            _meta: {
+              description: 'a_field description',
+            },
+          },
+        },
+        context$,
+      });
+
+      const globalContextPromise = globalContext$.pipe(take(1), toArray()).toPromise();
+      await expect(globalContextPromise).resolves.toEqual([
+        { a_field: true }, // No original empty state
+      ]);
+    });
+
     test('Merges all the contexts together', async () => {
       const contextA$ = new Subject<{ a_field: boolean }>();
       analyticsClient.registerContextProvider({
@@ -481,10 +541,11 @@ describe('AnalyticsClient', () => {
         context$,
       });
 
-      // The size of the registry grows
+      const globalContextPromise = globalContext$.pipe(take(4), toArray()).toPromise();
+      context$.next({ a_field: true });
+      // The size of the registry grows on the first emission
       expect(contextProvidersRegistry.size).toBe(1);
 
-      const globalContextPromise = globalContext$.pipe(take(3), toArray()).toPromise();
       context$.next({ a_field: true });
       // Still in the registry
       expect(contextProvidersRegistry.size).toBe(1);
@@ -494,6 +555,7 @@ describe('AnalyticsClient', () => {
       await expect(globalContextPromise).resolves.toEqual([
         {}, // Original empty state
         { a_field: true },
+        { a_field: true },
         {},
       ]);
     });
@@ -687,6 +749,7 @@ describe('AnalyticsClient', () => {
       const reportEventsMock = jest.fn();
       analyticsClient.registerShipper(MockedShipper1, { reportEventsMock });
       analyticsClient.optIn({ global: { enabled: true } });
+      await delay(10);
 
       expect(reportEventsMock).toHaveBeenCalledTimes(2);
       expect(reportEventsMock).toHaveBeenNthCalledWith(1, [
@@ -810,6 +873,7 @@ describe('AnalyticsClient', () => {
         global: { enabled: true },
         event_types: { ['event-type-a']: { enabled: false } },
       });
+      await delay(10);
 
       expect(reportEventsMock).toHaveBeenCalledTimes(1);
       expect(reportEventsMock).toHaveBeenNthCalledWith(1, [
@@ -875,6 +939,7 @@ describe('AnalyticsClient', () => {
           ['event-type-a']: { enabled: true, shippers: { [MockedShipper2.shipperName]: false } },
         },
       });
+      await delay(10);
 
       expect(reportEventsMock1).toHaveBeenCalledTimes(2);
       expect(reportEventsMock1).toHaveBeenNthCalledWith(1, [
@@ -970,6 +1035,7 @@ describe('AnalyticsClient', () => {
           ['event-type-a']: { enabled: true },
         },
       });
+      await delay(10);
 
       expect(reportEventsMock1).toHaveBeenCalledTimes(2);
       expect(reportEventsMock1).toHaveBeenNthCalledWith(1, [

packages/elastic-analytics/src/analytics_client/analytics_client.ts

@@ -6,18 +6,20 @@
  * Side Public License, v 1.
  */
 
-import type { Observable, Subscription } from 'rxjs';
+import type { Observable } from 'rxjs';
 import { BehaviorSubject, Subject, combineLatest, from } from 'rxjs';
 import {
   buffer,
   bufferCount,
   concatMap,
+  delay,
   filter,
   groupBy,
   map,
   mergeMap,
   share,
   shareReplay,
+  skipWhile,
   takeUntil,
   tap,
 } from 'rxjs/operators';
@@ -54,23 +56,26 @@ export class AnalyticsClient implements IAnalyticsClient {
   private readonly shipperRegistered$ = new Subject<void>();
   private readonly eventTypeRegistry = new Map<EventType, EventTypeOpts<unknown>>();
   private readonly context$ = new BehaviorSubject<Partial<EventContext>>({});
-  private readonly contextWithReplay$ = this.context$.pipe(shareReplay(1));
   private readonly contextProvidersRegistry = new Map<
     Observable<Partial<EventContext>>,
-    { subscription: Subscription; context: Partial<EventContext> }
+    Partial<EventContext>
   >();
   private readonly optInConfig$ = new BehaviorSubject<OptInConfig | undefined>(undefined);
   private readonly optInConfigWithReplay$ = this.optInConfig$.pipe(
     filter((optInConfig): optInConfig is OptInConfig => typeof optInConfig !== 'undefined'),
     shareReplay(1)
   );
+  private readonly contextWithReplay$ = this.context$.pipe(
+    skipWhile(() => !this.optInConfig$.value), // Do not forward the context events until we have an optInConfig value
+    shareReplay(1)
+  );
 
   constructor(private readonly initContext: AnalyticsClientInitContext) {
     this.telemetryCounter$ = this.internalTelemetryCounter$.pipe(share()); // Using `share` so we can have multiple subscribers
 
     // Observer that will emit when both events occur: the OptInConfig is set + a shipper has been registered
     const configReceivedAndShipperReceivedObserver$ = combineLatest([
-      this.optInConfig$.pipe(filter((optInConfig) => typeof optInConfig !== 'undefined')),
+      this.optInConfigWithReplay$,
       this.shipperRegistered$,
     ]);
 
@@ -83,6 +88,20 @@ export class AnalyticsClient implements IAnalyticsClient {
         // Accumulate the events until we can send them
         buffer(configReceivedAndShipperReceivedObserver$),
 
+        // Minimal delay only to make this chain async and let the optIn operation to complete first.
+        delay(0),
+
+        // Re-emit the context to make sure all the shippers got it (only if opted-in)
+        tap(() => {
+          if (this.optInConfig$.value?.global.enabled) {
+            this.context$.next(this.context$.value);
+          }
+        }),
+
+        // Minimal delay only to make this chain async and let
+        // the context update operation to complete first.
+        delay(0),
+
         // Flatten the array of events
         concatMap((events) => from(events)),
 
@@ -181,7 +200,8 @@ export class AnalyticsClient implements IAnalyticsClient {
   };
 
   public registerContextProvider = <Context>(contextProviderOpts: ContextProviderOpts<Context>) => {
-    const subscription = contextProviderOpts.context$
+    // const contextProviderID = `context_provider_${Date.now()}`;
+    contextProviderOpts.context$
       .pipe(
         tap((ctx) => {
           if (this.initContext.isDev) {
@@ -191,7 +211,9 @@ export class AnalyticsClient implements IAnalyticsClient {
       )
       .subscribe({
         next: (ctx) => {
-          this.contextProvidersRegistry.get(contextProviderOpts.context$)!.context = ctx;
+          // We store each context linked to the context provider so they can increase and reduce
+          // the number of fields they report without having left-overs in the global context.
+          this.contextProvidersRegistry.set(contextProviderOpts.context$, ctx);
 
           // For every context change, we rebuild the global context.
           // It's better to do it here than to rebuild it for every reportEvent.
@@ -203,10 +225,6 @@ export class AnalyticsClient implements IAnalyticsClient {
           this.updateGlobalContext();
         },
       });
-
-    // We store each context linked to the context provider so they can increase and reduce
-    // the number of fields they report without having left-overs in the global context.
-    this.contextProvidersRegistry.set(contextProviderOpts.context$, { subscription, context: {} });
   };
 
   public registerShipper = <Shipper extends IShipper, ShipperConfig>(
@@ -320,7 +338,7 @@ export class AnalyticsClient implements IAnalyticsClient {
    */
   private updateGlobalContext() {
     this.context$.next(
-      [...this.contextProvidersRegistry.values()].reduce((acc, { context }) => {
+      [...this.contextProvidersRegistry.values()].reduce((acc, context) => {
         return {
           ...acc,
           ...context,

src/core/server/analytics/index.ts

@@ -15,6 +15,8 @@ export type {
 
 export type {
   AnalyticsClient,
+  Event,
+  EventContext,
   EventType,
   EventTypeOpts,
   IShipper,

src/core/server/index.ts

@@ -452,6 +452,8 @@ export type {
   AnalyticsServicePreboot,
   AnalyticsServiceStart,
   AnalyticsClient,
+  Event,
+  EventContext,
   EventType,
   EventTypeOpts,
   IShipper,

test/scripts/test/server_integration.sh

@@ -19,3 +19,10 @@ checks-reporter-with-killswitch "Status Integration Tests" \
     --config test/server_integration/http/platform/config.status.ts \
     --bail \
     --debug \
+
+# Tests that must be run against source in order to build test plugins
+checks-reporter-with-killswitch "Analytics Integration Tests" \
+  node scripts/functional_tests \
+    --config test/server_integration/analytics/config.ts \
+    --bail \
+    --debug \

test/server_integration/__fixtures__/plugins/analytics_plugin_a/kibana.json

@@ -0,0 +1,8 @@
+{
+  "id": "analyticsPluginA",
+  "owner": { "name": "Core", "githubTeam": "kibana-core" },
+  "version": "0.0.1",
+  "kibanaVersion": "kibana",
+  "server": true,
+  "ui": false
+}

test/server_integration/__fixtures__/plugins/analytics_plugin_a/package.json

@@ -0,0 +1,14 @@
+{
+  "name": "analytics_plugin_a",
+  "version": "1.0.0",
+  "main": "target/test/server_integration/__fixtures__/plugins/analytics_plugin_a",
+  "kibana": {
+    "version": "kibana",
+    "templateVersion": "1.0.0"
+  },
+  "license": "SSPL-1.0 OR Elastic License 2.0",
+  "scripts": {
+    "kbn": "node ../../../../../../scripts/kbn.js",
+    "build": "rm -rf './target' && ../../../../../../node_modules/.bin/tsc"
+  }
+}

test/server_integration/__fixtures__/plugins/analytics_plugin_a/server/custom_shipper.ts

@@ -0,0 +1,42 @@
+/*
+ * 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 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+import { Subject } from 'rxjs';
+import type { IShipper, Event, EventContext, TelemetryCounter } from 'src/core/server';
+
+export interface Action {
+  action: string;
+  meta: any;
+}
+
+export class CustomShipper implements IShipper {
+  public static shipperName = 'FTR-shipper';
+
+  public telemetryCounter$ = new Subject<TelemetryCounter>();
+
+  constructor(private readonly actions$: Subject<Action>) {}
+
+  public reportEvents(events: Event[]) {
+    this.actions$.next({ action: 'reportEvents', meta: events });
+    events.forEach((event) => {
+      this.telemetryCounter$.next({
+        type: 'succeed',
+        event_type: event.event_type,
+        code: '200',
+        count: 1,
+        source: 'random_value',
+      });
+    });
+  }
+  optIn(isOptedIn: boolean) {
+    this.actions$.next({ action: 'optIn', meta: isOptedIn });
+  }
+  extendContext(newContext: EventContext) {
+    this.actions$.next({ action: 'extendContext', meta: newContext });
+  }
+}

test/server_integration/__fixtures__/plugins/analytics_plugin_a/server/index.ts

@@ -0,0 +1,11 @@
+/*
+ * 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 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+import { AnalyticsPluginAPlugin } from './plugin';
+
+export const plugin = () => new AnalyticsPluginAPlugin();