[Watcher] Refactor to use client from RequestHandlerContext (elasti…

…c#57834) (elastic#58116)

* Remove elasticsearch setup service from route deps

Removed the callWithRequestFactory entirely. This setup was introducing a pattern where
route handlers were not pulling the ES client fromt the route handler context. For this
refactor we need to extend the route handler context with watcher specific client actions
and so we also extend RequestHandlerContext globally.

In this commit we also update the types for params, query and body schema on each route to avoid
using any everwhere.

* Add generic types to license wrapper

Adding <P, Q, B> to the license wrapper made it a transparent
wrapper from a type perspective so we can remove the need to
eplicitly set RequestHandler<P, Q, B> on the handler.

Also cleaned up a variable name "response" -> "searchResults"

Also removed elasticsearch from the RouteDependencies type.

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

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

x-pack/plugins/watcher/server/index.ts

@@ -6,4 +6,6 @@
 import { PluginInitializerContext } from 'kibana/server';
 import { WatcherServerPlugin } from './plugin';
 
+export { WatcherContext } from './plugin';
+
 export const plugin = (ctx: PluginInitializerContext) => new WatcherServerPlugin(ctx);

x-pack/plugins/watcher/server/lib/fetch_all_from_scroll/fetch_all_from_scroll.ts

@@ -4,24 +4,31 @@
  * you may not use this file except in compliance with the Elastic License.
  */
 
+import { IScopedClusterClient } from 'kibana/server';
 import { get } from 'lodash';
 import { ES_SCROLL_SETTINGS } from '../../../common/constants';
 
-export function fetchAllFromScroll(response: any, callWithRequest: any, hits: any[] = []) {
-  const newHits = get(response, 'hits.hits', []);
-  const scrollId = get(response, '_scroll_id');
+export function fetchAllFromScroll(
+  searchResuls: any,
+  dataClient: IScopedClusterClient,
+  hits: any[] = []
+): Promise<any> {
+  const newHits = get(searchResuls, 'hits.hits', []);
+  const scrollId = get(searchResuls, '_scroll_id');
 
   if (newHits.length > 0) {
     hits.push(...newHits);
 
-    return callWithRequest('scroll', {
-      body: {
-        scroll: ES_SCROLL_SETTINGS.KEEPALIVE,
-        scroll_id: scrollId,
-      },
-    }).then((innerResponse: any) => {
-      return fetchAllFromScroll(innerResponse, callWithRequest, hits);
-    });
+    return dataClient
+      .callAsCurrentUser('scroll', {
+        body: {
+          scroll: ES_SCROLL_SETTINGS.KEEPALIVE,
+          scroll_id: scrollId,
+        },
+      })
+      .then((innerResponse: any) => {
+        return fetchAllFromScroll(innerResponse, dataClient, hits);
+      });
   }
 
   return Promise.resolve(hits);

x-pack/plugins/watcher/server/lib/license_pre_routing_factory/license_pre_routing_factory.ts

@@ -12,13 +12,13 @@ import {
 } from 'kibana/server';
 import { RouteDependencies } from '../../types';
 
-export const licensePreRoutingFactory = (
+export const licensePreRoutingFactory = <P, Q, B>(
   { getLicenseStatus }: RouteDependencies,
-  handler: RequestHandler
+  handler: RequestHandler<P, Q, B>
 ) => {
   return function licenseCheck(
     ctx: RequestHandlerContext,
-    request: KibanaRequest,
+    request: KibanaRequest<P, Q, B>,
     response: KibanaResponseFactory
   ) {
     const licenseStatus = getLicenseStatus();

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

@@ -3,7 +3,20 @@
  * or more contributor license agreements. Licensed under the Elastic License;
  * you may not use this file except in compliance with the Elastic License.
  */
-import { CoreSetup, Logger, Plugin, PluginInitializerContext } from 'kibana/server';
+
+declare module 'kibana/server' {
+  interface RequestHandlerContext {
+    watcher?: WatcherContext;
+  }
+}
+
+import {
+  CoreSetup,
+  IScopedClusterClient,
+  Logger,
+  Plugin,
+  PluginInitializerContext,
+} from 'kibana/server';
 import { PLUGIN } from '../common/constants';
 import { Dependencies, LicenseStatus, RouteDependencies } from './types';
 import { LICENSE_CHECK_STATE } from '../../licensing/server';
@@ -15,6 +28,11 @@ import { registerWatchesRoutes } from './routes/api/watches';
 import { registerWatchRoutes } from './routes/api/watch';
 import { registerListFieldsRoute } from './routes/api/register_list_fields_route';
 import { registerLoadHistoryRoute } from './routes/api/register_load_history_route';
+import { elasticsearchJsPlugin } from './lib/elasticsearch_js_plugin';
+
+export interface WatcherContext {
+  client: IScopedClusterClient;
+}
 
 export class WatcherServerPlugin implements Plugin<void, void, any, any> {
   log: Logger;
@@ -31,15 +49,20 @@ export class WatcherServerPlugin implements Plugin<void, void, any, any> {
     { http, elasticsearch: elasticsearchService }: CoreSetup,
     { licensing }: Dependencies
   ) {
-    const elasticsearch = await elasticsearchService.adminClient;
     const router = http.createRouter();
     const routeDependencies: RouteDependencies = {
-      elasticsearch,
-      elasticsearchService,
       router,
       getLicenseStatus: () => this.licenseStatus,
     };
 
+    const config = { plugins: [elasticsearchJsPlugin] };
+    const watcherESClient = elasticsearchService.createClient('watcher', config);
+    http.registerRouteHandlerContext('watcher', (ctx, request) => {
+      return {
+        client: watcherESClient.asScoped(request),
+      };
+    });
+
     registerListFieldsRoute(routeDependencies);
     registerLoadHistoryRoute(routeDependencies);
     registerIndicesRoutes(routeDependencies);

x-pack/plugins/watcher/server/routes/api/indices/register_get_route.ts

@@ -5,13 +5,14 @@
  */
 
 import { schema } from '@kbn/config-schema';
-import { RequestHandler } from 'kibana/server';
+import { IScopedClusterClient } from 'kibana/server';
 import { reduce, size } from 'lodash';
-import { callWithRequestFactory } from '../../../lib/call_with_request_factory';
 import { isEsError } from '../../../lib/is_es_error';
 import { RouteDependencies } from '../../../types';
 import { licensePreRoutingFactory } from '../../../lib/license_pre_routing_factory';
 
+const bodySchema = schema.object({ pattern: schema.string() }, { allowUnknowns: true });
+
 function getIndexNamesFromAliasesResponse(json: Record<string, any>) {
   return reduce(
     json,
@@ -26,67 +27,66 @@ function getIndexNamesFromAliasesResponse(json: Record<string, any>) {
   );
 }
 
-function getIndices(callWithRequest: any, pattern: string, limit = 10) {
-  return callWithRequest('indices.getAlias', {
-    index: pattern,
-    ignore: [404],
-  }).then((aliasResult: any) => {
-    if (aliasResult.status !== 404) {
-      const indicesFromAliasResponse = getIndexNamesFromAliasesResponse(aliasResult);
-      return indicesFromAliasResponse.slice(0, limit);
-    }
-
-    const params = {
+function getIndices(dataClient: IScopedClusterClient, pattern: string, limit = 10) {
+  return dataClient
+    .callAsCurrentUser('indices.getAlias', {
       index: pattern,
       ignore: [404],
-      body: {
-        size: 0, // no hits
-        aggs: {
-          indices: {
-            terms: {
-              field: '_index',
-              size: limit,
+    })
+    .then((aliasResult: any) => {
+      if (aliasResult.status !== 404) {
+        const indicesFromAliasResponse = getIndexNamesFromAliasesResponse(aliasResult);
+        return indicesFromAliasResponse.slice(0, limit);
+      }
+
+      const params = {
+        index: pattern,
+        ignore: [404],
+        body: {
+          size: 0, // no hits
+          aggs: {
+            indices: {
+              terms: {
+                field: '_index',
+                size: limit,
+              },
             },
           },
         },
-      },
-    };
+      };
 
-    return callWithRequest('search', params).then((response: any) => {
-      if (response.status === 404 || !response.aggregations) {
-        return [];
-      }
-      return response.aggregations.indices.buckets.map((bucket: any) => bucket.key);
+      return dataClient.callAsCurrentUser('search', params).then((response: any) => {
+        if (response.status === 404 || !response.aggregations) {
+          return [];
+        }
+        return response.aggregations.indices.buckets.map((bucket: any) => bucket.key);
+      });
     });
-  });
 }
 
 export function registerGetRoute(deps: RouteDependencies) {
-  const handler: RequestHandler<any, any, any> = async (ctx, request, response) => {
-    const callWithRequest = callWithRequestFactory(deps.elasticsearchService, request);
-    const { pattern } = request.body;
-
-    try {
-      const indices = await getIndices(callWithRequest, pattern);
-      return response.ok({ body: { indices } });
-    } catch (e) {
-      // Case: Error from Elasticsearch JS client
-      if (isEsError(e)) {
-        return response.customError({ statusCode: e.statusCode, body: e });
-      }
-
-      // Case: default
-      return response.internalError({ body: e });
-    }
-  };
-
   deps.router.post(
     {
       path: '/api/watcher/indices',
       validate: {
-        body: schema.object({}, { allowUnknowns: true }),
+        body: bodySchema,
       },
     },
-    licensePreRoutingFactory(deps, handler)
+    licensePreRoutingFactory(deps, async (ctx, request, response) => {
+      const { pattern } = request.body;
+
+      try {
+        const indices = await getIndices(ctx.watcher!.client, pattern);
+        return response.ok({ body: { indices } });
+      } catch (e) {
+        // Case: Error from Elasticsearch JS client
+        if (isEsError(e)) {
+          return response.customError({ statusCode: e.statusCode, body: e });
+        }
+
+        // Case: default
+        return response.internalError({ body: e });
+      }
+    })
   );
 }

x-pack/plugins/watcher/server/routes/api/license/register_refresh_route.ts

@@ -4,7 +4,6 @@
  * you may not use this file except in compliance with the Elastic License.
  */
 
-import { RequestHandler } from 'kibana/server';
 import { RouteDependencies } from '../../../types';
 import { licensePreRoutingFactory } from '../../../lib/license_pre_routing_factory';
 /*
@@ -13,16 +12,15 @@ it needs to make a round-trip to the kibana server. This refresh endpoint is pro
 for when the client needs to check the license, but doesn't need to pull data from the
 server for any reason, i.e., when adding a new watch.
 */
-export function registerRefreshRoute(deps: RouteDependencies) {
-  const handler: RequestHandler<any, any, any> = (ctx, request, response) => {
-    return response.ok({ body: { success: true } });
-  };
 
+export function registerRefreshRoute(deps: RouteDependencies) {
   deps.router.get(
     {
       path: '/api/watcher/license/refresh',
       validate: false,
     },
-    licensePreRoutingFactory(deps, handler)
+    licensePreRoutingFactory(deps, (ctx, request, response) => {
+      return response.ok({ body: { success: true } });
+    })
   );
 }