[RuleRegistry plugin] Abort ResourceInstaller.install() on stop (#134635)

* Abort ResourceInstaller.install() on stop

* Fix UTs

* Fix more UTs

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

x-pack/plugins/rule_registry/server/plugin.ts

@@ -5,7 +5,8 @@
  * 2.0.
  */
 
-import {
+import { type Subject, ReplaySubject } from 'rxjs';
+import type {
   PluginInitializerContext,
   Plugin,
   CoreSetup,
@@ -15,19 +16,19 @@ import {
   IContextProvider,
 } from '@kbn/core/server';
 
-import { PluginStartContract as AlertingStart } from '@kbn/alerting-plugin/server';
-import { SecurityPluginSetup } from '@kbn/security-plugin/server';
-import { SpacesPluginStart } from '@kbn/spaces-plugin/server';
-import {
+import type { PluginStartContract as AlertingStart } from '@kbn/alerting-plugin/server';
+import type { SecurityPluginSetup } from '@kbn/security-plugin/server';
+import type { SpacesPluginStart } from '@kbn/spaces-plugin/server';
+import type {
   PluginStart as DataPluginStart,
   PluginSetup as DataPluginSetup,
 } from '@kbn/data-plugin/server';
 
-import { RuleRegistryPluginConfig } from './config';
-import { IRuleDataService, RuleDataService } from './rule_data_plugin_service';
+import type { RuleRegistryPluginConfig } from './config';
+import { type IRuleDataService, RuleDataService } from './rule_data_plugin_service';
 import { AlertsClientFactory } from './alert_data_client/alerts_client_factory';
-import { AlertsClient } from './alert_data_client/alerts_client';
-import { RacApiRequestHandlerContext, RacRequestHandlerContext } from './types';
+import type { AlertsClient } from './alert_data_client/alerts_client';
+import type { RacApiRequestHandlerContext, RacRequestHandlerContext } from './types';
 import { defineRoutes } from './routes';
 import { ruleRegistrySearchStrategyProvider, RULE_SEARCH_STRATEGY_NAME } from './search_strategy';
 
@@ -66,6 +67,7 @@ export class RuleRegistryPlugin
   private readonly alertsClientFactory: AlertsClientFactory;
   private ruleDataService: IRuleDataService | null;
   private security: SecurityPluginSetup | undefined;
+  private pluginStop$: Subject<void>;
 
   constructor(initContext: PluginInitializerContext) {
     this.config = initContext.config.get<RuleRegistryPluginConfig>();
@@ -73,6 +75,7 @@ export class RuleRegistryPlugin
     this.kibanaVersion = initContext.env.packageInfo.version;
     this.ruleDataService = null;
     this.alertsClientFactory = new AlertsClientFactory();
+    this.pluginStop$ = new ReplaySubject(1);
   }
 
   public setup(
@@ -100,6 +103,7 @@ export class RuleRegistryPlugin
         const deps = await startDependencies;
         return deps.core.elasticsearch.client.asInternalUser;
       },
+      pluginStop$: this.pluginStop$,
     });
 
     this.ruleDataService.initializeService();
@@ -171,5 +175,8 @@ export class RuleRegistryPlugin
     };
   };
 
-  public stop() {}
+  public stop() {
+    this.pluginStop$.next();
+    this.pluginStop$.complete();
+  }
 }

x-pack/plugins/rule_registry/server/rule_data_plugin_service/resource_installer.test.ts

@@ -5,6 +5,7 @@
  * 2.0.
  */
 
+import { type Subject, ReplaySubject } from 'rxjs';
 import { ResourceInstaller } from './resource_installer';
 import { loggerMock } from '@kbn/logging-mocks';
 import { AlertConsumers } from '@kbn/rule-data-utils';
@@ -19,6 +20,17 @@ import {
 } from '../../common/assets';
 
 describe('resourceInstaller', () => {
+  let pluginStop$: Subject<void>;
+
+  beforeEach(() => {
+    pluginStop$ = new ReplaySubject(1);
+  });
+
+  afterEach(() => {
+    pluginStop$.next();
+    pluginStop$.complete();
+  });
+
   describe('if write is disabled', () => {
     it('should not install common resources', async () => {
       const mockClusterClient = elasticsearchServiceMock.createElasticsearchClient();
@@ -29,6 +41,7 @@ describe('resourceInstaller', () => {
         disabledRegistrationContexts: [],
         getResourceName: jest.fn(),
         getClusterClient,
+        pluginStop$,
       });
       installer.installCommonResources();
       expect(getClusterClient).not.toHaveBeenCalled();
@@ -44,6 +57,7 @@ describe('resourceInstaller', () => {
         disabledRegistrationContexts: [],
         getResourceName: jest.fn(),
         getClusterClient,
+        pluginStop$,
       });
       const indexOptions = {
         feature: AlertConsumers.LOGS,
@@ -78,6 +92,7 @@ describe('resourceInstaller', () => {
         disabledRegistrationContexts: [],
         getResourceName: getResourceNameMock,
         getClusterClient,
+        pluginStop$,
       });
 
       await installer.installCommonResources();
@@ -102,6 +117,7 @@ describe('resourceInstaller', () => {
         disabledRegistrationContexts: [],
         getResourceName: jest.fn(),
         getClusterClient,
+        pluginStop$,
       });
 
       const indexOptions = {

x-pack/plugins/rule_registry/server/rule_data_plugin_service/resource_installer.ts

@@ -5,12 +5,13 @@
  * 2.0.
  */
 
+import { firstValueFrom, type Observable } from 'rxjs';
 import { get, isEmpty } from 'lodash';
 import type * as estypes from '@elastic/elasticsearch/lib/api/typesWithBodyKey';
 
-import { ElasticsearchClient, Logger } from '@kbn/core/server';
+import type { ElasticsearchClient, Logger } from '@kbn/core/server';
 
-import { PublicMethodsOf } from '@kbn/utility-types';
+import type { PublicMethodsOf } from '@kbn/utility-types';
 import {
   DEFAULT_ILM_POLICY_ID,
   ECS_COMPONENT_TEMPLATE_NAME,
@@ -20,7 +21,7 @@ import { technicalComponentTemplate } from '../../common/assets/component_templa
 import { ecsComponentTemplate } from '../../common/assets/component_templates/ecs_component_template';
 import { defaultLifecyclePolicy } from '../../common/assets/lifecycle_policies/default_lifecycle_policy';
 
-import { IndexInfo } from './index_info';
+import type { IndexInfo } from './index_info';
 
 const INSTALLATION_TIMEOUT = 20 * 60 * 1000; // 20 minutes
 
@@ -30,6 +31,7 @@ interface ConstructorOptions {
   logger: Logger;
   isWriteEnabled: boolean;
   disabledRegistrationContexts: string[];
+  pluginStop$: Observable<void>;
 }
 
 export type IResourceInstaller = PublicMethodsOf<ResourceInstaller>;
@@ -41,6 +43,7 @@ export class ResourceInstaller {
     installer: () => Promise<void>
   ): Promise<void> {
     try {
+      let timeoutId: NodeJS.Timeout;
       const installResources = async (): Promise<void> => {
         const { logger, isWriteEnabled } = this.options;
         if (!isWriteEnabled) {
@@ -51,14 +54,24 @@ export class ResourceInstaller {
         logger.info(`Installing ${resources}`);
         await installer();
         logger.info(`Installed ${resources}`);
+
+        if (timeoutId) {
+          clearTimeout(timeoutId);
+        }
       };
 
       const throwTimeoutException = (): Promise<void> => {
         return new Promise((resolve, reject) => {
-          setTimeout(() => {
+          timeoutId = setTimeout(() => {
             const msg = `Timeout: it took more than ${INSTALLATION_TIMEOUT}ms`;
             reject(new Error(msg));
           }, INSTALLATION_TIMEOUT);
+
+          firstValueFrom(this.options.pluginStop$).then(() => {
+            clearTimeout(timeoutId);
+            const msg = 'Server is stopping; must stop all async operations';
+            reject(new Error(msg));
+          });
         });
       };
 

x-pack/plugins/rule_registry/server/rule_data_plugin_service/rule_data_plugin_service.test.ts

@@ -5,6 +5,7 @@
  * 2.0.
  */
 
+import { type Subject, ReplaySubject } from 'rxjs';
 import { loggerMock } from '@kbn/logging-mocks';
 import { RuleDataService } from './rule_data_plugin_service';
 import { elasticsearchServiceMock } from '@kbn/core/server/mocks';
@@ -18,8 +19,16 @@ jest.mock('../rule_data_client/rule_data_client', () => ({
 }));
 
 describe('ruleDataPluginService', () => {
+  let pluginStop$: Subject<void>;
+
   beforeEach(() => {
     jest.resetAllMocks();
+    pluginStop$ = new ReplaySubject(1);
+  });
+
+  afterEach(() => {
+    pluginStop$.next();
+    pluginStop$.complete();
   });
 
   describe('isRegistrationContextDisabled', () => {
@@ -34,6 +43,7 @@ describe('ruleDataPluginService', () => {
         isWriteEnabled: true,
         disabledRegistrationContexts: ['observability.logs'],
         isWriterCacheEnabled: true,
+        pluginStop$,
       });
       expect(ruleDataService.isRegistrationContextDisabled('observability.logs')).toBe(true);
     });
@@ -49,6 +59,7 @@ describe('ruleDataPluginService', () => {
         isWriteEnabled: true,
         disabledRegistrationContexts: ['observability.logs'],
         isWriterCacheEnabled: true,
+        pluginStop$,
       });
       expect(ruleDataService.isRegistrationContextDisabled('observability.apm')).toBe(false);
     });
@@ -66,6 +77,7 @@ describe('ruleDataPluginService', () => {
         isWriteEnabled: true,
         disabledRegistrationContexts: ['observability.logs'],
         isWriterCacheEnabled: true,
+        pluginStop$,
       });
 
       expect(ruleDataService.isWriteEnabled('observability.logs')).toBe(false);
@@ -84,6 +96,7 @@ describe('ruleDataPluginService', () => {
         isWriteEnabled: true,
         disabledRegistrationContexts: ['observability.logs'],
         isWriterCacheEnabled: true,
+        pluginStop$,
       });
       const indexOptions = {
         feature: AlertConsumers.LOGS,

x-pack/plugins/rule_registry/server/rule_data_plugin_service/rule_data_plugin_service.ts

@@ -5,16 +5,17 @@
  * 2.0.
  */
 
-import { Either, isLeft, left, right } from 'fp-ts/lib/Either';
-import { ValidFeatureId } from '@kbn/rule-data-utils';
+import type { Observable } from 'rxjs';
+import { type Either, isLeft, left, right } from 'fp-ts/lib/Either';
+import type { ValidFeatureId } from '@kbn/rule-data-utils';
 
-import { ElasticsearchClient, Logger } from '@kbn/core/server';
+import type { ElasticsearchClient, Logger } from '@kbn/core/server';
 
 import { INDEX_PREFIX } from '../config';
-import { IRuleDataClient, RuleDataClient, WaitResult } from '../rule_data_client';
+import { type IRuleDataClient, RuleDataClient, WaitResult } from '../rule_data_client';
 import { IndexInfo } from './index_info';
-import { Dataset, IndexOptions } from './index_options';
-import { IResourceInstaller, ResourceInstaller } from './resource_installer';
+import type { Dataset, IndexOptions } from './index_options';
+import { type IResourceInstaller, ResourceInstaller } from './resource_installer';
 import { joinWithDash } from './utils';
 
 /**
@@ -90,6 +91,7 @@ interface ConstructorOptions {
   isWriteEnabled: boolean;
   isWriterCacheEnabled: boolean;
   disabledRegistrationContexts: string[];
+  pluginStop$: Observable<void>;
 }
 
 export class RuleDataService implements IRuleDataService {
@@ -110,6 +112,7 @@ export class RuleDataService implements IRuleDataService {
       logger: options.logger,
       disabledRegistrationContexts: options.disabledRegistrationContexts,
       isWriteEnabled: options.isWriteEnabled,
+      pluginStop$: options.pluginStop$,
     });
 
     this.installCommonResources = Promise.resolve(right('ok'));

x-pack/test/rule_registry/spaces_only/tests/trial/lifecycle_executor.ts

@@ -5,6 +5,7 @@
  * 2.0.
  */
 
+import { type Subject, ReplaySubject } from 'rxjs';
 import type { ElasticsearchClient, Logger, LogMeta } from '@kbn/core/server';
 import sinon from 'sinon';
 import expect from '@kbn/expect';
@@ -58,9 +59,13 @@ export default function createLifecycleExecutorApiTest({ getService }: FtrProvid
   // FAILING ES PROMOTION: https://github.com/elastic/kibana/issues/125851
   describe.skip('createLifecycleExecutor', () => {
     let ruleDataClient: IRuleDataClient;
+    let pluginStop$: Subject<void>;
+
     before(async () => {
       // First we need to setup the data service. This happens within the
       // Rule Registry plugin as part of the server side setup phase.
+      pluginStop$ = new ReplaySubject(1);
+
       const ruleDataService = new RuleDataService({
         getClusterClient,
         logger,
@@ -68,6 +73,7 @@ export default function createLifecycleExecutorApiTest({ getService }: FtrProvid
         isWriteEnabled: true,
         isWriterCacheEnabled: false,
         disabledRegistrationContexts: [] as string[],
+        pluginStop$,
       });
 
       // This initializes the service. This happens immediately after the creation
@@ -102,6 +108,8 @@ export default function createLifecycleExecutorApiTest({ getService }: FtrProvid
 
     after(async () => {
       cleanupRegistryIndices(getService, ruleDataClient);
+      pluginStop$.next();
+      pluginStop$.complete();
     });
 
     it('should work with object fields', async () => {