Do not await for getOptInStatus() on stop() (#134735)

* Do not await for getOptInStatus() on stop()

* Fix incorrect comment

* Make sure we call analytics.optIn() during setup()

* Remove unused import

* [CI] Auto-commit changed files from 'node scripts/precommit_hook.js --ref HEAD~1..HEAD --fix'

* Add simple comment

* Revert switchMap => exhaustMap to prevent memory leaks; move takeUntil after exhaustMap

Co-authored-by: kibanamachine <42973632+kibanamachine@users.noreply.github.com>

src/plugins/telemetry/server/plugin.test.ts

@@ -14,6 +14,26 @@ import { TelemetryPlugin } from './plugin';
 
 describe('TelemetryPlugin', () => {
   describe('setup', () => {
+    describe('when initial config does not allow changing opt in status', () => {
+      it('calls analytics optIn', () => {
+        const initializerContext = coreMock.createPluginInitializerContext({
+          optIn: true,
+          allowChangingOptInStatus: false,
+        });
+        const coreSetupMock = coreMock.createSetup();
+
+        new TelemetryPlugin(initializerContext).setup(coreSetupMock, {
+          usageCollection: usageCollectionPluginMock.createSetupContract(),
+          telemetryCollectionManager: telemetryCollectionManagerPluginMock.createSetupContract(),
+        });
+
+        expect(coreSetupMock.analytics.optIn).toHaveBeenCalledTimes(1);
+        expect(coreSetupMock.analytics.optIn).toHaveBeenCalledWith({
+          global: { enabled: true },
+        });
+      });
+    });
+
     describe('EBT shipper registration', () => {
       it('registers the Server telemetry shipper', () => {
         const initializerContext = coreMock.createPluginInitializerContext();

src/plugins/telemetry/server/plugin.ts

@@ -7,16 +7,20 @@
  */
 
 import { URL } from 'url';
-import type { Observable } from 'rxjs';
 import {
-  BehaviorSubject,
+  type Observable,
+  startWith,
   firstValueFrom,
   ReplaySubject,
   exhaustMap,
   timer,
   distinctUntilChanged,
   filter,
+  takeUntil,
+  tap,
+  shareReplay,
 } from 'rxjs';
+
 import { ElasticV3ServerShipper } from '@kbn/analytics-shippers-elastic-v3-server';
 
 import type { UsageCollectionSetup } from '@kbn/usage-collection-plugin/server';
@@ -34,6 +38,7 @@ import type {
 } from '@kbn/core/server';
 import type { SecurityPluginStart } from '@kbn/security-plugin/server';
 import { SavedObjectsClient } from '@kbn/core/server';
+
 import { registerRoutes } from './routes';
 import { registerCollection } from './telemetry_collection';
 import {
@@ -86,10 +91,10 @@ export class TelemetryPlugin implements Plugin<TelemetryPluginSetup, TelemetryPl
   private readonly currentKibanaVersion: string;
   private readonly initialConfig: TelemetryConfigType;
   private readonly config$: Observable<TelemetryConfigType>;
-  private readonly isOptedIn$ = new BehaviorSubject<boolean | undefined>(undefined);
+  private readonly isOptedIn$: Observable<boolean>;
+  private isOptedIn?: boolean;
   private readonly isDev: boolean;
   private readonly fetcherTask: FetcherTask;
-  private optInPromise?: Promise<boolean | undefined>;
   /**
    * @private Used to mark the completion of the old UI Settings migration
    */
@@ -105,19 +110,7 @@ export class TelemetryPlugin implements Plugin<TelemetryPluginSetup, TelemetryPl
    */
   private savedObjectsInternalClient$ = new ReplaySubject<SavedObjectsClient>(1);
 
-  /**
-   * Poll for the opt-in status and update the `isOptedIn$` subject.
-   * @private
-   */
-  private readonly optInPollerSubscription = timer(0, OPT_IN_POLL_INTERVAL_MS)
-    .pipe(
-      exhaustMap(() => {
-        this.optInPromise = this.getOptInStatus();
-        return this.optInPromise;
-      }),
-      distinctUntilChanged()
-    )
-    .subscribe((isOptedIn) => this.isOptedIn$.next(isOptedIn));
+  private pluginStop$ = new ReplaySubject<void>(1);
 
   private security?: SecurityPluginStart;
 
@@ -134,17 +127,26 @@ export class TelemetryPlugin implements Plugin<TelemetryPluginSetup, TelemetryPl
 
     // If the opt-in selection cannot be changed, set it as early as possible.
     const { optIn, allowChangingOptInStatus } = this.initialConfig;
-    if (allowChangingOptInStatus === false) {
-      this.isOptedIn$.next(optIn);
-    }
+    this.isOptedIn = allowChangingOptInStatus === false ? optIn : undefined;
+
+    // Poll for the opt-in status
+    this.isOptedIn$ = timer(0, OPT_IN_POLL_INTERVAL_MS).pipe(
+      exhaustMap(() => this.getOptInStatus()),
+      takeUntil(this.pluginStop$),
+      startWith(this.isOptedIn),
+      filter((isOptedIn): isOptedIn is boolean => typeof isOptedIn === 'boolean'),
+      distinctUntilChanged(),
+      tap((optedIn) => (this.isOptedIn = optedIn)),
+      shareReplay(1)
+    );
   }
 
   public setup(
     { analytics, http, savedObjects }: CoreSetup,
     { usageCollection, telemetryCollectionManager }: TelemetryPluginsDepsSetup
   ): TelemetryPluginSetup {
-    if (this.isOptedIn$.value !== undefined) {
-      analytics.optIn({ global: { enabled: this.isOptedIn$.value } });
+    if (this.isOptedIn !== undefined) {
+      analytics.optIn({ global: { enabled: this.isOptedIn } });
     }
 
     const currentKibanaVersion = this.currentKibanaVersion;
@@ -193,9 +195,7 @@ export class TelemetryPlugin implements Plugin<TelemetryPluginSetup, TelemetryPl
   ) {
     const { analytics, savedObjects } = core;
 
-    this.isOptedIn$
-      .pipe(filter((isOptedIn): isOptedIn is boolean => typeof isOptedIn === 'boolean'))
-      .subscribe((isOptedIn) => analytics.optIn({ global: { enabled: isOptedIn } }));
+    this.isOptedIn$.subscribe((enabled) => analytics.optIn({ global: { enabled } }));
 
     const savedObjectsInternalRepository = savedObjects.createInternalRepository();
     this.savedObjectsInternalRepository = savedObjectsInternalRepository;
@@ -206,19 +206,22 @@ export class TelemetryPlugin implements Plugin<TelemetryPluginSetup, TelemetryPl
     this.startFetcher(core, telemetryCollectionManager);
 
     return {
-      getIsOptedIn: async () => this.isOptedIn$.value === true,
+      getIsOptedIn: async () => this.isOptedIn === true,
     };
   }
 
-  public async stop() {
-    this.optInPollerSubscription.unsubscribe();
-    this.isOptedIn$.complete();
+  public stop() {
+    this.pluginStop$.next();
+    this.pluginStop$.complete();
+    this.savedObjectsInternalClient$.complete();
     this.fetcherTask.stop();
-    if (this.optInPromise) await this.optInPromise;
   }
 
   private async getOptInStatus(): Promise<boolean | undefined> {
-    const internalRepositoryClient = await firstValueFrom(this.savedObjectsInternalClient$);
+    const internalRepositoryClient = await firstValueFrom(this.savedObjectsInternalClient$, {
+      defaultValue: undefined,
+    });
+    if (!internalRepositoryClient) return;
 
     let telemetrySavedObject: TelemetrySavedObject | undefined;
     try {