[NP] Remove observables from es internal contract (#54556) (#54581)

* request context always uses the latest es client

* update integration tests

Co-authored-by: Elastic Machine <elasticmachine@users.noreply.github.com>

Co-authored-by: Elastic Machine <elasticmachine@users.noreply.github.com>

Author: mshustov

src/core/server/elasticsearch/elasticsearch_service.mock.ts

@@ -74,8 +74,6 @@ const createInternalSetupContractMock = () => {
     legacy: {
       config$: new BehaviorSubject({} as ElasticsearchConfig),
     },
-    adminClient$: new BehaviorSubject(createClusterClientMock()),
-    dataClient$: new BehaviorSubject(createClusterClientMock()),
   };
   setupContract.adminClient.asScoped.mockReturnValue(createScopedClusterClientMock());
   setupContract.dataClient.asScoped.mockReturnValue(createScopedClusterClientMock());

src/core/server/elasticsearch/elasticsearch_service.test.ts

@@ -21,7 +21,7 @@ import { first } from 'rxjs/operators';
 
 import { MockClusterClient } from './elasticsearch_service.test.mocks';
 
-import { BehaviorSubject, combineLatest } from 'rxjs';
+import { BehaviorSubject } from 'rxjs';
 import { Env } from '../config';
 import { getEnvOptions } from '../config/__mocks__/env';
 import { CoreContext } from '../core_context';
@@ -91,44 +91,6 @@ describe('#setup', () => {
     expect(mockDataClusterClientInstance.callAsInternalUser).toHaveBeenCalledTimes(1);
   });
 
-  it('returns data and admin client observables as a part of the contract', async () => {
-    const mockAdminClusterClientInstance = { close: jest.fn() };
-    const mockDataClusterClientInstance = { close: jest.fn() };
-    MockClusterClient.mockImplementationOnce(
-      () => mockAdminClusterClientInstance
-    ).mockImplementationOnce(() => mockDataClusterClientInstance);
-
-    const setupContract = await elasticsearchService.setup(deps);
-
-    const [esConfig, adminClient, dataClient] = await combineLatest(
-      setupContract.legacy.config$,
-      setupContract.adminClient$,
-      setupContract.dataClient$
-    )
-      .pipe(first())
-      .toPromise();
-
-    expect(adminClient).toBe(mockAdminClusterClientInstance);
-    expect(dataClient).toBe(mockDataClusterClientInstance);
-
-    expect(MockClusterClient).toHaveBeenCalledTimes(2);
-    expect(MockClusterClient).toHaveBeenNthCalledWith(
-      1,
-      esConfig,
-      expect.objectContaining({ context: ['elasticsearch', 'admin'] }),
-      undefined
-    );
-    expect(MockClusterClient).toHaveBeenNthCalledWith(
-      2,
-      esConfig,
-      expect.objectContaining({ context: ['elasticsearch', 'data'] }),
-      expect.any(Function)
-    );
-
-    expect(mockAdminClusterClientInstance.close).not.toHaveBeenCalled();
-    expect(mockDataClusterClientInstance.close).not.toHaveBeenCalled();
-  });
-
   describe('#createClient', () => {
     it('allows to specify config properties', async () => {
       const setupContract = await elasticsearchService.setup(deps);

src/core/server/elasticsearch/elasticsearch_service.ts

@@ -152,8 +152,6 @@ export class ElasticsearchService implements CoreService<InternalElasticsearchSe
     return {
       legacy: { config$: clients$.pipe(map(clients => clients.config)) },
 
-      adminClient$,
-      dataClient$,
       adminClient,
       dataClient,
 

src/core/server/elasticsearch/types.ts

@@ -77,7 +77,4 @@ export interface InternalElasticsearchServiceSetup extends ElasticsearchServiceS
   readonly legacy: {
     readonly config$: Observable<ElasticsearchConfig>;
   };
-
-  readonly adminClient$: Observable<IClusterClient>;
-  readonly dataClient$: Observable<IClusterClient>;
 }

src/core/server/http/integration_tests/core_service.test.mocks.ts

@@ -16,8 +16,11 @@
  * specific language governing permissions and limitations
  * under the License.
  */
+import { elasticsearchServiceMock } from '../../elasticsearch/elasticsearch_service.mock';
 
 export const clusterClientMock = jest.fn();
 jest.doMock('../../elasticsearch/scoped_cluster_client', () => ({
-  ScopedClusterClient: clusterClientMock,
+  ScopedClusterClient: clusterClientMock.mockImplementation(function() {
+    return elasticsearchServiceMock.createScopedClusterClient();
+  }),
 }));

src/core/server/http/integration_tests/core_services.test.ts

@@ -133,7 +133,7 @@ describe('http service', () => {
         const { http } = await root.setup();
         const { registerAuth } = http;
 
-        await registerAuth((req, res, toolkit) => {
+        registerAuth((req, res, toolkit) => {
           return toolkit.authenticated({ responseHeaders: authResponseHeader });
         });
 
@@ -157,7 +157,7 @@ describe('http service', () => {
         const { http } = await root.setup();
         const { registerAuth } = http;
 
-        await registerAuth((req, res, toolkit) => {
+        registerAuth((req, res, toolkit) => {
           return toolkit.authenticated({ responseHeaders: authResponseHeader });
         });
 
@@ -222,12 +222,15 @@ describe('http service', () => {
       const { http } = await root.setup();
       const { registerAuth, createRouter } = http;
 
-      await registerAuth((req, res, toolkit) =>
-        toolkit.authenticated({ requestHeaders: authHeaders })
-      );
+      registerAuth((req, res, toolkit) => toolkit.authenticated({ requestHeaders: authHeaders }));
 
       const router = createRouter('/new-platform');
-      router.get({ path: '/', validate: false }, (context, req, res) => res.ok());
+      router.get({ path: '/', validate: false }, async (context, req, res) => {
+        // it forces client initialization since the core creates them lazily.
+        await context.core.elasticsearch.adminClient.callAsCurrentUser('ping');
+        await context.core.elasticsearch.dataClient.callAsCurrentUser('ping');
+        return res.ok();
+      });
 
       await root.start();
 
@@ -247,7 +250,12 @@ describe('http service', () => {
       const { createRouter } = http;
 
       const router = createRouter('/new-platform');
-      router.get({ path: '/', validate: false }, (context, req, res) => res.ok());
+      router.get({ path: '/', validate: false }, async (context, req, res) => {
+        // it forces client initialization since the core creates them lazily.
+        await context.core.elasticsearch.adminClient.callAsCurrentUser('ping');
+        await context.core.elasticsearch.dataClient.callAsCurrentUser('ping');
+        return res.ok();
+      });
 
       await root.start();
 

src/core/server/server.ts

@@ -17,7 +17,6 @@
  * under the License.
  */
 
-import { take } from 'rxjs/operators';
 import { Type } from '@kbn/config-schema';
 
 import {
@@ -216,9 +215,6 @@ export class Server {
       coreId,
       'core',
       async (context, req, res): Promise<RequestHandlerContext['core']> => {
-        // it consumes elasticsearch observables to provide the same client throughout the context lifetime.
-        const adminClient = await coreSetup.elasticsearch.adminClient$.pipe(take(1)).toPromise();
-        const dataClient = await coreSetup.elasticsearch.dataClient$.pipe(take(1)).toPromise();
         const savedObjectsClient = coreSetup.savedObjects.getScopedClient(req);
         const uiSettingsClient = coreSetup.uiSettings.asScopedToClient(savedObjectsClient);
 
@@ -230,8 +226,8 @@ export class Server {
             client: savedObjectsClient,
           },
           elasticsearch: {
-            adminClient: adminClient.asScoped(req),
-            dataClient: dataClient.asScoped(req),
+            adminClient: coreSetup.elasticsearch.adminClient.asScoped(req),
+            dataClient: coreSetup.elasticsearch.dataClient.asScoped(req),
           },
           uiSettings: {
             client: uiSettingsClient,