Refactor Plugins to access elasticsearch from CoreStart

src/core/server/elasticsearch/elasticsearch_service.ts

@@ -33,7 +33,12 @@ import { CoreService } from '../../types';
 import { merge } from '../../utils';
 import { CoreContext } from '../core_context';
 import { Logger } from '../logging';
-import { ClusterClient, ScopeableRequest } from './cluster_client';
+import {
+  ClusterClient,
+  ScopeableRequest,
+  IClusterClient,
+  ICustomClusterClient,
+} from './cluster_client';
 import { ElasticsearchClientConfig } from './elasticsearch_client_config';
 import { ElasticsearchConfig, ElasticsearchConfigType } from './elasticsearch_config';
 import { InternalHttpServiceSetup, GetAuthHeaders } from '../http/';
@@ -57,12 +62,14 @@ export class ElasticsearchService
   implements CoreService<InternalElasticsearchServiceSetup, ElasticsearchServiceStart> {
   private readonly log: Logger;
   private readonly config$: Observable<ElasticsearchConfig>;
-  private subscription: Subscription | undefined;
+  private subscription?: Subscription;
   private stop$ = new Subject();
   private kibanaVersion: string;
-  createClient: InternalElasticsearchServiceSetup['createClient'] | undefined;
-  dataClient: InternalElasticsearchServiceSetup['dataClient'] | undefined;
-  adminClient: InternalElasticsearchServiceSetup['adminClient'] | undefined;
+  private createClient?: (
+    type: string,
+    clientConfig?: Partial<ElasticsearchClientConfig>
+  ) => ICustomClusterClient;
+  private adminClient?: IClusterClient;
 
   constructor(private readonly coreContext: CoreContext) {
     this.kibanaVersion = coreContext.env.packageInfo.version;

src/core/server/server.ts

@@ -188,7 +188,7 @@ export class Server {
       uiSettings: uiSettingsStart,
     };
 
-    const pluginsStart = await this.plugins.start(this.coreStart!);
+    const pluginsStart = await this.plugins.start(this.coreStart);
 
     await this.legacy.start({
       core: {

src/plugins/expressions/server/legacy.ts

@@ -26,7 +26,7 @@ import { register, registryFactory, Registry, Fn } from '@kbn/interpreter/common
 
 import Boom from 'boom';
 import { schema } from '@kbn/config-schema';
-import { CoreSetup, Logger } from 'src/core/server';
+import { CoreSetup, Logger, APICaller } from 'src/core/server';
 import { ExpressionsServerSetupDependencies } from './plugin';
 import { typeSpecs, ExpressionType } from '../common';
 import { serializeProvider } from '../common';
@@ -97,7 +97,10 @@ export const createLegacyServerEndpoints = (
    * @param {*} handlers - The Canvas handlers
    * @param {*} fnCall - Describes the function being run `{ functionName, args, context }`
    */
-  async function runFunction(handlers: any, fnCall: any) {
+  async function runFunction(
+    handlers: { environment: string; elasticsearchClient: APICaller },
+    fnCall: any
+  ) {
     const { functionName, args, context } = fnCall;
     const { deserialize } = serializeProvider(registries.types.toJS());
     const fnDef = registries.serverFunctions.toJS()[functionName];
@@ -112,18 +115,14 @@ export const createLegacyServerEndpoints = (
    * results back using ND-JSON.
    */
   plugins.bfetch.addBatchProcessingRoute(`/api/interpreter/fns`, request => {
-    const scopedClient = core.elasticsearch.dataClient.asScoped(request);
-    const handlers = {
-      environment: 'server',
-      elasticsearchClient: async (
-        endpoint: string,
-        clientParams: Record<string, any> = {},
-        options?: any
-      ) => scopedClient.callAsCurrentUser(endpoint, clientParams, options),
-    };
-
     return {
       onBatchItem: async (fnCall: any) => {
+        const [coreStart] = await core.getStartServices();
+        const handlers = {
+          environment: 'server',
+          elasticsearchClient: coreStart.elasticsearch.legacy.client.asScoped(request)
+            .callAsCurrentUser,
+        };
         const result = await runFunction(handlers, fnCall);
         if (typeof result === 'undefined') {
           throw new Error(`Function ${fnCall.functionName} did not return anything.`);

x-pack/plugins/actions/server/plugin.test.ts

@@ -66,6 +66,9 @@ describe('Actions Plugin', () => {
               savedObjects: {
                 client: {},
               },
+              elasticsearch: {
+                adminClient: jest.fn(),
+              },
             },
           } as any,
           httpServerMock.createKibanaRequest(),

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

@@ -11,13 +11,13 @@ import {
   Plugin,
   CoreSetup,
   CoreStart,
-  IClusterClient,
   KibanaRequest,
   Logger,
   SharedGlobalConfig,
   RequestHandler,
   IContextProvider,
   SavedObjectsServiceStart,
+  ElasticsearchServiceStart,
 } from '../../../../src/core/server';
 
 import {
@@ -87,7 +87,6 @@ export class ActionsPlugin implements Plugin<Promise<PluginSetupContract>, Plugi
 
   private readonly logger: Logger;
   private serverBasePath?: string;
-  private adminClient?: IClusterClient;
   private taskRunnerFactory?: TaskRunnerFactory;
   private actionTypeRegistry?: ActionTypeRegistry;
   private actionExecutor?: ActionExecutor;
@@ -161,7 +160,6 @@ export class ActionsPlugin implements Plugin<Promise<PluginSetupContract>, Plugi
     this.actionTypeRegistry = actionTypeRegistry;
     this.serverBasePath = core.http.basePath.serverBasePath;
     this.actionExecutor = actionExecutor;
-    this.adminClient = core.elasticsearch.adminClient;
     this.spaces = plugins.spaces?.spacesService;
 
     registerBuiltInActionTypes({
@@ -186,7 +184,7 @@ export class ActionsPlugin implements Plugin<Promise<PluginSetupContract>, Plugi
 
     core.http.registerRouteHandlerContext(
       'actions',
-      this.createRouteHandlerContext(await this.kibanaIndex)
+      this.createRouteHandlerContext(core, await this.kibanaIndex)
     );
 
     // Routes
@@ -212,15 +210,14 @@ export class ActionsPlugin implements Plugin<Promise<PluginSetupContract>, Plugi
       actionTypeRegistry,
       taskRunnerFactory,
       kibanaIndex,
-      adminClient,
       isESOUsingEphemeralEncryptionKey,
     } = this;
 
     actionExecutor!.initialize({
       logger,
       eventLogger: this.eventLogger!,
       spaces: this.spaces,
-      getServices: this.getServicesFactory(core.savedObjects),
+      getServices: this.getServicesFactory(core.savedObjects, core.elasticsearch),
       encryptedSavedObjectsPlugin: plugins.encryptedSavedObjects,
       actionTypeRegistry: actionTypeRegistry!,
     });
@@ -253,26 +250,27 @@ export class ActionsPlugin implements Plugin<Promise<PluginSetupContract>, Plugi
           savedObjectsClient: core.savedObjects.getScopedClient(request),
           actionTypeRegistry: actionTypeRegistry!,
           defaultKibanaIndex: await kibanaIndex,
-          scopedClusterClient: adminClient!.asScoped(request),
+          scopedClusterClient: core.elasticsearch.legacy.client.asScoped(request),
         });
       },
     };
   }
 
   private getServicesFactory(
-    savedObjects: SavedObjectsServiceStart
+    savedObjects: SavedObjectsServiceStart,
+    elasticsearch: ElasticsearchServiceStart
   ): (request: KibanaRequest) => Services {
-    const { adminClient } = this;
     return request => ({
-      callCluster: adminClient!.asScoped(request).callAsCurrentUser,
+      callCluster: elasticsearch.legacy.client.asScoped(request).callAsCurrentUser,
       savedObjectsClient: savedObjects.getScopedClient(request),
     });
   }
 
   private createRouteHandlerContext = (
+    coreSetup: CoreSetup,
     defaultKibanaIndex: string
   ): IContextProvider<RequestHandler<any, any, any>, 'actions'> => {
-    const { actionTypeRegistry, adminClient, isESOUsingEphemeralEncryptionKey } = this;
+    const { actionTypeRegistry, isESOUsingEphemeralEncryptionKey } = this;
     return async function actionsRouteHandlerContext(context, request) {
       return {
         getActionsClient: () => {
@@ -285,7 +283,7 @@ export class ActionsPlugin implements Plugin<Promise<PluginSetupContract>, Plugi
             savedObjectsClient: context.core.savedObjects.client,
             actionTypeRegistry: actionTypeRegistry!,
             defaultKibanaIndex,
-            scopedClusterClient: adminClient!.asScoped(request),
+            scopedClusterClient: context.core.elasticsearch.adminClient,
           });
         },
         listTypes: actionTypeRegistry!.list.bind(actionTypeRegistry!),

x-pack/plugins/actions/server/usage/task.ts

@@ -4,7 +4,7 @@
  * you may not use this file except in compliance with the Elastic License.
  */
 
-import { Logger, CoreSetup } from 'kibana/server';
+import { Logger, CoreSetup, APICaller } from 'kibana/server';
 import moment from 'moment';
 import {
   RunContext,
@@ -62,7 +62,11 @@ async function scheduleTasks(logger: Logger, taskManager: TaskManagerStartContra
 export function telemetryTaskRunner(logger: Logger, core: CoreSetup, kibanaIndex: string) {
   return ({ taskInstance }: RunContext) => {
     const { state } = taskInstance;
-    const callCluster = core.elasticsearch.adminClient.callAsInternalUser;
+    const callCluster = (...args: Parameters<APICaller>) => {
+      return core.getStartServices().then(([{ elasticsearch: { legacy: { client } } }]) =>
+        client.callAsInternalUser(...args)
+      );
+    };
     return {
       async run() {
         return Promise.all([

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

@@ -19,7 +19,6 @@ import { TaskRunnerFactory } from './task_runner';
 import { AlertsClientFactory } from './alerts_client_factory';
 import { LicenseState } from './lib/license_state';
 import {
-  IClusterClient,
   KibanaRequest,
   Logger,
   PluginInitializerContext,
@@ -29,6 +28,7 @@ import {
   IContextProvider,
   RequestHandler,
   SharedGlobalConfig,
+  ElasticsearchServiceStart,
 } from '../../../../src/core/server';
 
 import {
@@ -83,7 +83,6 @@ export class AlertingPlugin {
   private readonly logger: Logger;
   private alertTypeRegistry?: AlertTypeRegistry;
   private readonly taskRunnerFactory: TaskRunnerFactory;
-  private adminClient?: IClusterClient;
   private serverBasePath?: string;
   private licenseState: LicenseState | null = null;
   private isESOUsingEphemeralEncryptionKey?: boolean;
@@ -107,7 +106,6 @@ export class AlertingPlugin {
   }
 
   public async setup(core: CoreSetup, plugins: AlertingPluginsSetup): Promise<PluginSetupContract> {
-    this.adminClient = core.elasticsearch.adminClient;
     this.licenseState = new LicenseState(plugins.licensing.license$);
     this.spaces = plugins.spaces?.spacesService;
     this.security = plugins.security;
@@ -204,7 +202,7 @@ export class AlertingPlugin {
 
     taskRunnerFactory.initialize({
       logger,
-      getServices: this.getServicesFactory(core.savedObjects),
+      getServices: this.getServicesFactory(core.savedObjects, core.elasticsearch),
       spaceIdToNamespace: this.spaceIdToNamespace,
       executeAction: plugins.actions.execute,
       encryptedSavedObjectsPlugin: plugins.encryptedSavedObjects,
@@ -243,11 +241,11 @@ export class AlertingPlugin {
   };
 
   private getServicesFactory(
-    savedObjects: SavedObjectsServiceStart
+    savedObjects: SavedObjectsServiceStart,
+    elasticsearch: ElasticsearchServiceStart
   ): (request: KibanaRequest) => Services {
-    const { adminClient } = this;
     return request => ({
-      callCluster: adminClient!.asScoped(request).callAsCurrentUser,
+      callCluster: elasticsearch.legacy.client.asScoped(request).callAsCurrentUser,
       savedObjectsClient: savedObjects.getScopedClient(request),
     });
   }

x-pack/plugins/alerting/server/usage/task.ts

@@ -4,7 +4,7 @@
  * you may not use this file except in compliance with the Elastic License.
  */
 
-import { Logger, CoreSetup } from 'kibana/server';
+import { Logger, CoreSetup, APICaller } from 'kibana/server';
 import moment from 'moment';
 import {
   RunContext,
@@ -65,7 +65,12 @@ async function scheduleTasks(logger: Logger, taskManager: TaskManagerStartContra
 export function telemetryTaskRunner(logger: Logger, core: CoreSetup, kibanaIndex: string) {
   return ({ taskInstance }: RunContext) => {
     const { state } = taskInstance;
-    const callCluster = core.elasticsearch.adminClient.callAsInternalUser;
+    const callCluster = (...args: Parameters<APICaller>) => {
+      return core.getStartServices().then(([{ elasticsearch: { legacy: { client } } }]) =>
+        client.callAsInternalUser(...args)
+      );
+    };
+
     return {
       async run() {
         return Promise.all([

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

@@ -3,7 +3,13 @@
  * or more contributor license agreements. Licensed under the Elastic License;
  * you may not use this file except in compliance with the Elastic License.
  */
-import { PluginInitializerContext, Plugin, CoreSetup } from 'src/core/server';
+import {
+  PluginInitializerContext,
+  Plugin,
+  CoreSetup,
+  CoreStart,
+  Logger
+} from 'src/core/server';
 import { Observable, combineLatest, AsyncSubject } from 'rxjs';
 import { map, take } from 'rxjs/operators';
 import { Server } from 'hapi';
@@ -33,6 +39,8 @@ export interface APMPluginContract {
 }
 
 export class APMPlugin implements Plugin<APMPluginContract> {
+  private currentConfig?: APMConfig;
+  private logger?: Logger;
   legacySetup$: AsyncSubject<LegacySetup>;
   constructor(private readonly initContext: PluginInitializerContext) {
     this.initContext = initContext;
@@ -49,30 +57,23 @@ export class APMPlugin implements Plugin<APMPluginContract> {
       usageCollection?: UsageCollectionSetup;
     }
   ) {
-    const logger = this.initContext.logger.get('apm');
+    const logger = (this.logger = this.initContext.logger.get('apm'));
     const config$ = this.initContext.config.create<APMXPackConfig>();
     const mergedConfig$ = combineLatest(plugins.apm_oss.config$, config$).pipe(
       map(([apmOssConfig, apmConfig]) => mergeConfigs(apmOssConfig, apmConfig))
     );
 
     this.legacySetup$.subscribe(__LEGACY => {
-      createApmApi().init(core, { config$: mergedConfig$, logger, __LEGACY });
+      createApmApi().init(core, {
+        config$: mergedConfig$,
+        logger,
+        __LEGACY
+      });
     });
 
-    const currentConfig = await mergedConfig$.pipe(take(1)).toPromise();
-
-    // create agent configuration index without blocking setup lifecycle
-    createApmAgentConfigurationIndex({
-      esClient: core.elasticsearch.dataClient,
-      config: currentConfig,
-      logger
-    });
-    // create custom action index without blocking setup lifecycle
-    createApmCustomLinkIndex({
-      esClient: core.elasticsearch.dataClient,
-      config: currentConfig,
-      logger
-    });
+    const currentConfig = (this.currentConfig = await mergedConfig$
+      .pipe(take(1))
+      .toPromise());
 
     plugins.home.tutorials.registerTutorial(
       tutorialProvider({
@@ -115,6 +116,23 @@ export class APMPlugin implements Plugin<APMPluginContract> {
     };
   }
 
-  public start() {}
+  public start(core: CoreStart) {
+    if (this.currentConfig == null || this.logger == null) {
+      throw new Error('APMPlugin needs to be setup before calling start()');
+    }
+
+    // create agent configuration index without blocking setup lifecycle
+    createApmAgentConfigurationIndex({
+      esClient: core.elasticsearch.legacy.client,
+      config: this.currentConfig,
+      logger: this.logger
+    });
+    // create custom action index without blocking setup lifecycle
+    createApmCustomLinkIndex({
+      esClient: core.elasticsearch.legacy.client,
+      config: this.currentConfig,
+      logger: this.logger
+    });
+  }
   public stop() {}
 }

x-pack/plugins/event_log/server/es/cluster_client_adapter.test.ts

@@ -19,7 +19,7 @@ beforeEach(() => {
   clusterClient = elasticsearchServiceMock.createClusterClient();
   clusterClientAdapter = new ClusterClientAdapter({
     logger,
-    clusterClient,
+    clusterClientPromise: new Promise(resolve => resolve(clusterClient)),
   });
 });