[Content Connectors] Fix race condition error for task registration of Agentless connectors infra service

x-pack/platform/plugins/shared/content_connectors/server/plugin.ts

@@ -29,6 +29,7 @@ import { PLUGIN_ID } from '../common/constants';
 import { registerApiKeysRoutes } from './routes/api_keys';
 import { SearchConnectorsConfig } from './config';
 import { AgentlessConnectorDeploymentsSyncService } from './task';
+import { AgentlessConnectorsInfraServiceFactory } from './services/infra_service_factory';
 
 export class SearchConnectorsPlugin
   implements
@@ -43,6 +44,7 @@ export class SearchConnectorsPlugin
   private readonly logger: LoggerFactory;
   private readonly config: SearchConnectorsConfig;
   private agentlessConnectorDeploymentsSyncService: AgentlessConnectorDeploymentsSyncService;
+  private agentlessConnectorsInfraServiceFactory: AgentlessConnectorsInfraServiceFactory;
 
   constructor(initializerContext: PluginInitializerContext) {
     this.connectors = [];
@@ -51,10 +53,11 @@ export class SearchConnectorsPlugin
     this.agentlessConnectorDeploymentsSyncService = new AgentlessConnectorDeploymentsSyncService(
       this.logger.get()
     );
+    this.agentlessConnectorsInfraServiceFactory = new AgentlessConnectorsInfraServiceFactory();
   }
 
   public setup(
-    coreSetup: CoreSetup<SearchConnectorsPluginStartDependencies, SearchConnectorsPluginStart>,
+    coreSetup: CoreSetup<SearchConnectorsPluginStartDependencies>,
     plugins: SearchConnectorsPluginSetupDependencies
   ) {
     const http = coreSetup.http;
@@ -74,23 +77,15 @@ export class SearchConnectorsPlugin
 
     this.connectors = getConnectorTypes(http.staticAssets);
 
-    const coreStartServices = coreSetup.getStartServices();
-
     // There seems to be no way to check for agentless here
     // So we register a task, but do not execute it in `start` method
     this.logger.get().debug('Registering agentless connectors infra sync task');
 
-    coreStartServices
-      .then(([coreStart, searchConnectorsPluginStartDependencies]) => {
-        this.agentlessConnectorDeploymentsSyncService.registerInfraSyncTask(
-          plugins,
-          coreStart,
-          searchConnectorsPluginStartDependencies
-        );
-      })
-      .catch((err) => {
-        this.logger.get().error(`Error registering agentless connectors infra sync task`, err);
-      });
+    this.agentlessConnectorDeploymentsSyncService.registerInfraSyncTask(
+      coreSetup,
+      plugins,
+      this.agentlessConnectorsInfraServiceFactory
+    );
     const router = http.createRouter();
 
     // Enterprise Search Routes
@@ -121,6 +116,11 @@ export class SearchConnectorsPlugin
         .info(
           'Agentless is supported, scheduling initial agentless connectors infrastructure watcher task'
         );
+      this.agentlessConnectorsInfraServiceFactory.initialize({
+        coreStart: core,
+        plugins,
+        logger: this.logger.get(),
+      });
       this.agentlessConnectorDeploymentsSyncService
         .scheduleInfraSyncTask(this.config, plugins.taskManager)
         .catch((err) => {

x-pack/platform/plugins/shared/content_connectors/server/services/infra_service_factory.ts

@@ -0,0 +1,55 @@
+/*
+ * 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; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import type { Logger } from '@kbn/logging';
+import { CoreStart, SavedObjectsClient } from '@kbn/core/server';
+import { SearchConnectorsPluginStartDependencies } from '../types';
+import { AgentlessConnectorsInfraService } from '.';
+
+export interface AgentlessConnectorsInfraServiceContext {
+  logger: Logger;
+  coreStart: CoreStart;
+  plugins: SearchConnectorsPluginStartDependencies;
+}
+
+export class AgentlessConnectorsInfraServiceFactory {
+  private isInitialized = false;
+  private agentlessConnectorsInfraService?: AgentlessConnectorsInfraService;
+
+  public initialize({ coreStart, plugins, logger }: AgentlessConnectorsInfraServiceContext) {
+    if (this.isInitialized) {
+      throw new Error('AgentlessConnectorsInfraServiceFactory already initialized');
+    }
+    this.isInitialized = true;
+
+    const esClient = coreStart.elasticsearch.client.asInternalUser;
+    const savedObjects = coreStart.savedObjects;
+
+    const agentPolicyService = plugins.fleet.agentPolicyService;
+    const packagePolicyService = plugins.fleet.packagePolicyService;
+    const agentService = plugins.fleet.agentService;
+
+    const soClient = new SavedObjectsClient(savedObjects.createInternalRepository());
+
+    this.agentlessConnectorsInfraService = new AgentlessConnectorsInfraService(
+      soClient,
+      esClient,
+      packagePolicyService,
+      agentPolicyService,
+      agentService,
+      logger
+    );
+  }
+
+  public getAgentlessConnectorsInfraService() {
+    if (!this.isInitialized) {
+      throw new Error('AgentlessConnectorsInfraServiceFactory not initialized');
+    }
+
+    return this.agentlessConnectorsInfraService;
+  }
+}

x-pack/platform/plugins/shared/content_connectors/server/task.test.ts

@@ -14,8 +14,9 @@ import {
   PackagePolicyMetadata,
 } from './services';
 import { licensingMock } from '@kbn/licensing-plugin/server/mocks';
-import { LicensingPluginStart } from '@kbn/licensing-plugin/server';
 import { createPackagePolicyMock } from '@kbn/fleet-plugin/common/mocks';
+import { coreMock } from '@kbn/core/server/mocks';
+import { AgentlessConnectorsInfraServiceFactory } from './services/infra_service_factory';
 
 const DATE_1970 = '1970-01-01T00:00:00.000Z';
 
@@ -76,7 +77,7 @@ describe('infraSyncTaskRunner', () => {
 
   let logger: MockedLogger;
   let serviceMock: jest.Mocked<AgentlessConnectorsInfraService>;
-  let licensePluginStartMock: jest.Mocked<LicensingPluginStart>;
+  const getLicenseMock = jest.fn();
 
   const taskInstanceStub: ConcreteTaskInstance = {
     id: '',
@@ -102,6 +103,8 @@ describe('infraSyncTaskRunner', () => {
 
   const validLicenseMock = licensingMock.createLicenseMock();
   validLicenseMock.check.mockReturnValue({ state: 'valid' });
+  const { getStartServices } = coreMock.createSetup();
+  let agentlessConnectorsInfraServiceFactory: jest.Mocked<AgentlessConnectorsInfraServiceFactory>;
 
   beforeAll(async () => {
     logger = loggerMock.create();
@@ -112,9 +115,21 @@ describe('infraSyncTaskRunner', () => {
       removeDeployment: jest.fn(),
     } as unknown as jest.Mocked<AgentlessConnectorsInfraService>;
 
-    licensePluginStartMock = {
-      getLicense: jest.fn(),
-    } as unknown as jest.Mocked<LicensingPluginStart>;
+    agentlessConnectorsInfraServiceFactory = {
+      initialize: jest.fn(),
+      getAgentlessConnectorsInfraService: jest.fn().mockReturnValue(serviceMock),
+    } as unknown as jest.Mocked<AgentlessConnectorsInfraServiceFactory>;
+    const [coreStart, deps, unknown] = await getStartServices();
+    getStartServices.mockResolvedValue([
+      coreStart,
+      {
+        ...deps,
+        licensing: {
+          getLicense: getLicenseMock,
+        },
+      },
+      unknown,
+    ]);
   });
 
   beforeEach(() => {
@@ -124,9 +139,11 @@ describe('infraSyncTaskRunner', () => {
   test('Does nothing if no connectors or policies are configured', async () => {
     await infraSyncTaskRunner(
       logger,
-      serviceMock,
-      licensePluginStartMock
-    )({ taskInstance: taskInstanceStub }).run();
+      getStartServices,
+      agentlessConnectorsInfraServiceFactory
+    )({
+      taskInstance: taskInstanceStub,
+    }).run();
 
     expect(serviceMock.deployConnector).not.toBeCalled();
     expect(serviceMock.removeDeployment).not.toBeCalled();
@@ -135,12 +152,12 @@ describe('infraSyncTaskRunner', () => {
   test('Does nothing if connectors or policies requires deployment but license is not supported', async () => {
     serviceMock.getNativeConnectors.mockResolvedValue([mysqlConnector, githubConnector]);
     serviceMock.getConnectorPackagePolicies.mockResolvedValue([sharepointPackagePolicy]);
-    licensePluginStartMock.getLicense.mockResolvedValue(invalidLicenseMock);
+    getLicenseMock.mockResolvedValue(invalidLicenseMock);
 
     await infraSyncTaskRunner(
       logger,
-      serviceMock,
-      licensePluginStartMock
+      getStartServices,
+      agentlessConnectorsInfraServiceFactory
     )({ taskInstance: taskInstanceStub }).run();
 
     expect(serviceMock.deployConnector).not.toBeCalled();
@@ -160,12 +177,12 @@ describe('infraSyncTaskRunner', () => {
       githubPackagePolicy,
       sharepointPackagePolicy,
     ]);
-    licensePluginStartMock.getLicense.mockResolvedValue(validLicenseMock);
+    getLicenseMock.mockResolvedValue(validLicenseMock);
 
     await infraSyncTaskRunner(
       logger,
-      serviceMock,
-      licensePluginStartMock
+      getStartServices,
+      agentlessConnectorsInfraServiceFactory
     )({ taskInstance: taskInstanceStub }).run();
 
     expect(serviceMock.deployConnector).not.toBeCalled();
@@ -176,12 +193,12 @@ describe('infraSyncTaskRunner', () => {
   test('Deploys connectors if no policies has been created for these connectors', async () => {
     serviceMock.getNativeConnectors.mockResolvedValue([mysqlConnector, githubConnector]);
     serviceMock.getConnectorPackagePolicies.mockResolvedValue([sharepointPackagePolicy]);
-    licensePluginStartMock.getLicense.mockResolvedValue(validLicenseMock);
+    getLicenseMock.mockResolvedValue(validLicenseMock);
 
     await infraSyncTaskRunner(
       logger,
-      serviceMock,
-      licensePluginStartMock
+      getStartServices,
+      agentlessConnectorsInfraServiceFactory
     )({ taskInstance: taskInstanceStub }).run();
 
     expect(serviceMock.deployConnector).toBeCalledWith(mysqlConnector);
@@ -195,7 +212,7 @@ describe('infraSyncTaskRunner', () => {
       sharepointConnector,
     ]);
     serviceMock.getConnectorPackagePolicies.mockResolvedValue([]);
-    licensePluginStartMock.getLicense.mockResolvedValue(validLicenseMock);
+    getLicenseMock.mockResolvedValue(validLicenseMock);
     serviceMock.deployConnector.mockImplementation(async (connector) => {
       if (connector === mysqlConnector || connector === githubConnector) {
         throw new Error('Cannot deploy these connectors');
@@ -206,8 +223,8 @@ describe('infraSyncTaskRunner', () => {
 
     await infraSyncTaskRunner(
       logger,
-      serviceMock,
-      licensePluginStartMock
+      getStartServices,
+      agentlessConnectorsInfraServiceFactory
     )({ taskInstance: taskInstanceStub }).run();
 
     expect(serviceMock.deployConnector).toBeCalledWith(mysqlConnector);
@@ -222,12 +239,12 @@ describe('infraSyncTaskRunner', () => {
       githubConnector,
     ]);
     serviceMock.getConnectorPackagePolicies.mockResolvedValue([sharepointPackagePolicy]);
-    licensePluginStartMock.getLicense.mockResolvedValue(validLicenseMock);
+    getLicenseMock.mockResolvedValue(validLicenseMock);
 
     await infraSyncTaskRunner(
       logger,
-      serviceMock,
-      licensePluginStartMock
+      getStartServices,
+      agentlessConnectorsInfraServiceFactory
     )({ taskInstance: taskInstanceStub }).run();
 
     expect(serviceMock.removeDeployment).toBeCalledWith(sharepointPackagePolicy.package_policy_id);
@@ -236,12 +253,12 @@ describe('infraSyncTaskRunner', () => {
   test('Does not remove a package policy if no connectors match the policy', async () => {
     serviceMock.getNativeConnectors.mockResolvedValue([mysqlConnector, githubConnector]);
     serviceMock.getConnectorPackagePolicies.mockResolvedValue([sharepointPackagePolicy]);
-    licensePluginStartMock.getLicense.mockResolvedValue(validLicenseMock);
+    getLicenseMock.mockResolvedValue(validLicenseMock);
 
     await infraSyncTaskRunner(
       logger,
-      serviceMock,
-      licensePluginStartMock
+      getStartServices,
+      agentlessConnectorsInfraServiceFactory
     )({ taskInstance: taskInstanceStub }).run();
 
     expect(serviceMock.removeDeployment).not.toBeCalled();
@@ -258,7 +275,7 @@ describe('infraSyncTaskRunner', () => {
       mysqlPackagePolicy,
       githubPackagePolicy,
     ]);
-    licensePluginStartMock.getLicense.mockResolvedValue(validLicenseMock);
+    getLicenseMock.mockResolvedValue(validLicenseMock);
     serviceMock.removeDeployment.mockImplementation(async (policyId) => {
       if (
         policyId === sharepointPackagePolicy.package_policy_id ||
@@ -270,8 +287,8 @@ describe('infraSyncTaskRunner', () => {
 
     await infraSyncTaskRunner(
       logger,
-      serviceMock,
-      licensePluginStartMock
+      getStartServices,
+      agentlessConnectorsInfraServiceFactory
     )({ taskInstance: taskInstanceStub }).run();
 
     expect(serviceMock.removeDeployment).toBeCalledWith(sharepointPackagePolicy.package_policy_id);