[Watcher] Refactor to use client from RequestHandlerContext (#57834)

* 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>

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/call_with_request_factory.ts

@@ -1,28 +0,0 @@
-/*
- * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
- * or more contributor license agreements. Licensed under the Elastic License;
- * you may not use this file except in compliance with the Elastic License.
- */
-
-import { ElasticsearchServiceSetup } from 'kibana/server';
-import { once } from 'lodash';
-import { elasticsearchJsPlugin } from './elasticsearch_js_plugin';
-
-const callWithRequest = once((elasticsearchService: ElasticsearchServiceSetup) => {
-  const config = { plugins: [elasticsearchJsPlugin] };
-  return elasticsearchService.createClient('watcher', config);
-});
-
-export const callWithRequestFactory = (
-  elasticsearchService: ElasticsearchServiceSetup,
-  request: any
-) => {
-  return (...args: any[]) => {
-    return (
-      callWithRequest(elasticsearchService)
-        .asScoped(request)
-        // @ts-ignore
-        .callAsCurrentUser(...args)
-    );
-  };
-};

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 } });
+    })
   );
 }

x-pack/plugins/watcher/server/routes/api/register_list_fields_route.ts

@@ -5,15 +5,18 @@
  */
 
 import { schema } from '@kbn/config-schema';
-import { RequestHandler } from 'kibana/server';
-import { callWithRequestFactory } from '../../lib/call_with_request_factory';
+import { IScopedClusterClient } from 'kibana/server';
 import { isEsError } from '../../lib/is_es_error';
 // @ts-ignore
 import { Fields } from '../../models/fields/index';
 import { licensePreRoutingFactory } from '../../lib/license_pre_routing_factory';
 import { RouteDependencies } from '../../types';
 
-function fetchFields(callWithRequest: any, indexes: string[]) {
+const bodySchema = schema.object({
+  indexes: schema.arrayOf(schema.string()),
+});
+
+function fetchFields(dataClient: IScopedClusterClient, indexes: string[]) {
   const params = {
     index: indexes,
     fields: ['*'],
@@ -22,44 +25,39 @@ function fetchFields(callWithRequest: any, indexes: string[]) {
     ignore: 404,
   };
 
-  return callWithRequest('fieldCaps', params);
+  return dataClient.callAsCurrentUser('fieldCaps', params);
 }
 
 export function registerListFieldsRoute(deps: RouteDependencies) {
-  const handler: RequestHandler<any, any, any> = async (ctx, request, response) => {
-    const callWithRequest = callWithRequestFactory(deps.elasticsearchService, request);
-    const { indexes } = request.body;
-
-    try {
-      const fieldsResponse = await fetchFields(callWithRequest, indexes);
-      const json = fieldsResponse.status === 404 ? { fields: [] } : fieldsResponse;
-      const fields = Fields.fromUpstreamJson(json);
-      return response.ok({ body: fields.downstreamJson });
-    } catch (e) {
-      // Case: Error from Elasticsearch JS client
-      if (isEsError(e)) {
-        return response.customError({
-          statusCode: e.statusCode,
-          body: {
-            message: e.message,
-          },
-        });
-      }
-
-      // Case: default
-      return response.internalError({ body: e });
-    }
-  };
-
   deps.router.post(
     {
       path: '/api/watcher/fields',
       validate: {
-        body: schema.object({
-          indexes: schema.arrayOf(schema.string()),
-        }),
+        body: bodySchema,
       },
     },
-    licensePreRoutingFactory(deps, handler)
+    licensePreRoutingFactory(deps, async (ctx, request, response) => {
+      const { indexes } = request.body;
+
+      try {
+        const fieldsResponse = await fetchFields(ctx.watcher!.client, indexes);
+        const json = fieldsResponse.status === 404 ? { fields: [] } : fieldsResponse;
+        const fields = Fields.fromUpstreamJson(json);
+        return response.ok({ body: fields.downstreamJson });
+      } catch (e) {
+        // Case: Error from Elasticsearch JS client
+        if (isEsError(e)) {
+          return response.customError({
+            statusCode: e.statusCode,
+            body: {
+              message: e.message,
+            },
+          });
+        }
+
+        // Case: default
+        return response.internalError({ body: e });
+      }
+    })
   );
 }

x-pack/plugins/watcher/server/routes/api/register_load_history_route.ts

@@ -6,8 +6,7 @@
 
 import { schema } from '@kbn/config-schema';
 import { get } from 'lodash';
-import { RequestHandler } from 'kibana/server';
-import { callWithRequestFactory } from '../../lib/call_with_request_factory';
+import { IScopedClusterClient } from 'kibana/server';
 import { isEsError } from '../../lib/is_es_error';
 import { INDEX_NAMES } from '../../../common/constants';
 import { RouteDependencies } from '../../types';
@@ -15,8 +14,12 @@ import { licensePreRoutingFactory } from '../../lib/license_pre_routing_factory'
 // @ts-ignore
 import { WatchHistoryItem } from '../../models/watch_history_item/index';
 
-function fetchHistoryItem(callWithRequest: any, watchHistoryItemId: string) {
-  return callWithRequest('search', {
+const paramsSchema = schema.object({
+  id: schema.string(),
+});
+
+function fetchHistoryItem(dataClient: IScopedClusterClient, watchHistoryItemId: string) {
+  return dataClient.callAsCurrentUser('search', {
     index: INDEX_NAMES.WATCHER_HISTORY,
     body: {
       query: {
@@ -29,49 +32,44 @@ function fetchHistoryItem(callWithRequest: any, watchHistoryItemId: string) {
 }
 
 export function registerLoadHistoryRoute(deps: RouteDependencies) {
-  const handler: RequestHandler<any, any, any> = async (ctx, request, response) => {
-    const callWithRequest = callWithRequestFactory(deps.elasticsearchService, request);
-    const id = request.params.id;
-
-    try {
-      const responseFromES = await fetchHistoryItem(callWithRequest, id);
-      const hit = get(responseFromES, 'hits.hits[0]');
-      if (!hit) {
-        return response.notFound({ body: `Watch History Item with id = ${id} not found` });
-      }
-      const watchHistoryItemJson = get(hit, '_source');
-      const watchId = get(hit, '_source.watch_id');
-      const json = {
-        id,
-        watchId,
-        watchHistoryItemJson,
-        includeDetails: true,
-      };
-
-      const watchHistoryItem = WatchHistoryItem.fromUpstreamJson(json);
-      return response.ok({
-        body: { watchHistoryItem: watchHistoryItem.downstreamJson },
-      });
-    } 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.get(
     {
       path: '/api/watcher/history/{id}',
       validate: {
-        params: schema.object({
-          id: schema.string(),
-        }),
+        params: paramsSchema,
       },
     },
-    licensePreRoutingFactory(deps, handler)
+    licensePreRoutingFactory(deps, async (ctx, request, response) => {
+      const id = request.params.id;
+
+      try {
+        const responseFromES = await fetchHistoryItem(ctx.watcher!.client, id);
+        const hit = get(responseFromES, 'hits.hits[0]');
+        if (!hit) {
+          return response.notFound({ body: `Watch History Item with id = ${id} not found` });
+        }
+        const watchHistoryItemJson = get(hit, '_source');
+        const watchId = get(hit, '_source.watch_id');
+        const json = {
+          id,
+          watchId,
+          watchHistoryItemJson,
+          includeDetails: true,
+        };
+
+        const watchHistoryItem = WatchHistoryItem.fromUpstreamJson(json);
+        return response.ok({
+          body: { watchHistoryItem: watchHistoryItem.downstreamJson },
+        });
+      } 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/settings/register_load_route.ts

@@ -4,14 +4,14 @@
  * you may not use this file except in compliance with the Elastic License.
  */
 
-import { IClusterClient, RequestHandler } from 'kibana/server';
+import { IScopedClusterClient } from 'kibana/server';
 import { isEsError } from '../../../lib/is_es_error';
 // @ts-ignore
 import { Settings } from '../../../models/settings/index';
 import { RouteDependencies } from '../../../types';
 import { licensePreRoutingFactory } from '../../../lib/license_pre_routing_factory';
 
-function fetchClusterSettings(client: IClusterClient) {
+function fetchClusterSettings(client: IScopedClusterClient) {
   return client.callAsInternalUser('cluster.getSettings', {
     includeDefaults: true,
     filterPath: '**.xpack.notification',
@@ -19,25 +19,24 @@ function fetchClusterSettings(client: IClusterClient) {
 }
 
 export function registerLoadRoute(deps: RouteDependencies) {
-  const handler: RequestHandler<any, any, any> = async (ctx, request, response) => {
-    try {
-      const settings = await fetchClusterSettings(deps.elasticsearch);
-      return response.ok({ body: Settings.fromUpstreamJson(settings).downstreamJson });
-    } 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.get(
     {
       path: '/api/watcher/settings',
       validate: false,
     },
-    licensePreRoutingFactory(deps, handler)
+    licensePreRoutingFactory(deps, async (ctx, request, response) => {
+      try {
+        const settings = await fetchClusterSettings(ctx.watcher!.client);
+        return response.ok({ body: Settings.fromUpstreamJson(settings).downstreamJson });
+      } 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/watch/action/register_acknowledge_route.ts

@@ -6,60 +6,58 @@
 
 import { schema } from '@kbn/config-schema';
 import { get } from 'lodash';
-import { RequestHandler } from 'kibana/server';
-import { callWithRequestFactory } from '../../../../lib/call_with_request_factory';
+import { IScopedClusterClient } from 'kibana/server';
 import { isEsError } from '../../../../lib/is_es_error';
 // @ts-ignore
 import { WatchStatus } from '../../../../models/watch_status/index';
 import { RouteDependencies } from '../../../../types';
 import { licensePreRoutingFactory } from '../../../../lib/license_pre_routing_factory';
 
-function acknowledgeAction(callWithRequest: any, watchId: string, actionId: string) {
-  return callWithRequest('watcher.ackWatch', {
+const paramsSchema = schema.object({
+  watchId: schema.string(),
+  actionId: schema.string(),
+});
+
+function acknowledgeAction(dataClient: IScopedClusterClient, watchId: string, actionId: string) {
+  return dataClient.callAsCurrentUser('watcher.ackWatch', {
     id: watchId,
     action: actionId,
   });
 }
 
 export function registerAcknowledgeRoute(deps: RouteDependencies) {
-  const handler: RequestHandler<any, any, any> = async (ctx, request, response) => {
-    const callWithRequest = callWithRequestFactory(deps.elasticsearchService, request);
-    const { watchId, actionId } = request.params;
-
-    try {
-      const hit = await acknowledgeAction(callWithRequest, watchId, actionId);
-      const watchStatusJson = get(hit, 'status');
-      const json = {
-        id: watchId,
-        watchStatusJson,
-      };
-
-      const watchStatus = WatchStatus.fromUpstreamJson(json);
-      return response.ok({
-        body: { watchStatus: watchStatus.downstreamJson },
-      });
-    } catch (e) {
-      // Case: Error from Elasticsearch JS client
-      if (isEsError(e)) {
-        const body = e.statusCode === 404 ? `Watch with id = ${watchId} not found` : e;
-        return response.customError({ statusCode: e.statusCode, body });
-      }
-
-      // Case: default
-      return response.internalError({ body: e });
-    }
-  };
-
   deps.router.put(
     {
       path: '/api/watcher/watch/{watchId}/action/{actionId}/acknowledge',
       validate: {
-        params: schema.object({
-          watchId: schema.string(),
-          actionId: schema.string(),
-        }),
+        params: paramsSchema,
       },
     },
-    licensePreRoutingFactory(deps, handler)
+    licensePreRoutingFactory(deps, async (ctx, request, response) => {
+      const { watchId, actionId } = request.params;
+
+      try {
+        const hit = await acknowledgeAction(ctx.watcher!.client, watchId, actionId);
+        const watchStatusJson = get(hit, 'status');
+        const json = {
+          id: watchId,
+          watchStatusJson,
+        };
+
+        const watchStatus = WatchStatus.fromUpstreamJson(json);
+        return response.ok({
+          body: { watchStatus: watchStatus.downstreamJson },
+        });
+      } catch (e) {
+        // Case: Error from Elasticsearch JS client
+        if (isEsError(e)) {
+          const body = e.statusCode === 404 ? `Watch with id = ${watchId} not found` : e;
+          return response.customError({ statusCode: e.statusCode, body });
+        }
+
+        // Case: default
+        return response.internalError({ body: e });
+      }
+    })
   );
 }

x-pack/plugins/watcher/server/routes/api/watch/register_activate_route.ts

@@ -5,62 +5,59 @@
  */
 
 import { schema } from '@kbn/config-schema';
-import { RequestHandler } from 'kibana/server';
+import { IScopedClusterClient } from 'kibana/server';
 import { get } 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';
 // @ts-ignore
 import { WatchStatus } from '../../../models/watch_status/index';
 
-function activateWatch(callWithRequest: any, watchId: string) {
-  return callWithRequest('watcher.activateWatch', {
+function activateWatch(dataClient: IScopedClusterClient, watchId: string) {
+  return dataClient.callAsCurrentUser('watcher.activateWatch', {
     id: watchId,
   });
 }
 
+const paramsSchema = schema.object({
+  watchId: schema.string(),
+});
+
 export function registerActivateRoute(deps: RouteDependencies) {
-  const handler: RequestHandler<any, any, any> = async (ctx, request, response) => {
-    const callWithRequest = callWithRequestFactory(deps.elasticsearchService, request);
-
-    const { watchId } = request.params;
-
-    try {
-      const hit = await activateWatch(callWithRequest, watchId);
-      const watchStatusJson = get(hit, 'status');
-      const json = {
-        id: watchId,
-        watchStatusJson,
-      };
-
-      const watchStatus = WatchStatus.fromUpstreamJson(json);
-      return response.ok({
-        body: {
-          watchStatus: watchStatus.downstreamJson,
-        },
-      });
-    } catch (e) {
-      // Case: Error from Elasticsearch JS client
-      if (isEsError(e)) {
-        const body = e.statusCode === 404 ? `Watch with id = ${watchId} not found` : e;
-        return response.customError({ statusCode: e.statusCode, body });
-      }
-
-      // Case: default
-      return response.internalError({ body: e });
-    }
-  };
-
   deps.router.put(
     {
       path: '/api/watcher/watch/{watchId}/activate',
       validate: {
-        params: schema.object({
-          watchId: schema.string(),
-        }),
+        params: paramsSchema,
       },
     },
-    licensePreRoutingFactory(deps, handler)
+    licensePreRoutingFactory(deps, async (ctx, request, response) => {
+      const { watchId } = request.params;
+
+      try {
+        const hit = await activateWatch(ctx.watcher!.client, watchId);
+        const watchStatusJson = get(hit, 'status');
+        const json = {
+          id: watchId,
+          watchStatusJson,
+        };
+
+        const watchStatus = WatchStatus.fromUpstreamJson(json);
+        return response.ok({
+          body: {
+            watchStatus: watchStatus.downstreamJson,
+          },
+        });
+      } catch (e) {
+        // Case: Error from Elasticsearch JS client
+        if (isEsError(e)) {
+          const body = e.statusCode === 404 ? `Watch with id = ${watchId} not found` : e;
+          return response.customError({ statusCode: e.statusCode, body });
+        }
+
+        // Case: default
+        return response.internalError({ body: e });
+      }
+    })
   );
 }

x-pack/plugins/watcher/server/routes/api/watch/register_deactivate_route.ts

@@ -3,63 +3,61 @@
  * or more contributor license agreements. Licensed under the Elastic License;
  * you may not use this file except in compliance with the Elastic License.
  */
+
 import { schema } from '@kbn/config-schema';
-import { RequestHandler } from 'kibana/server';
+import { IScopedClusterClient } from 'kibana/server';
 import { get } 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';
 // @ts-ignore
 import { WatchStatus } from '../../../models/watch_status/index';
 
-function deactivateWatch(callWithRequest: any, watchId: string) {
-  return callWithRequest('watcher.deactivateWatch', {
+const paramsSchema = schema.object({
+  watchId: schema.string(),
+});
+
+function deactivateWatch(dataClient: IScopedClusterClient, watchId: string) {
+  return dataClient.callAsCurrentUser('watcher.deactivateWatch', {
     id: watchId,
   });
 }
 
 export function registerDeactivateRoute(deps: RouteDependencies) {
-  const handler: RequestHandler<any, any, any> = async (ctx, request, response) => {
-    const callWithRequest = callWithRequestFactory(deps.elasticsearchService, request);
-
-    const { watchId } = request.params;
-
-    try {
-      const hit = await deactivateWatch(callWithRequest, watchId);
-      const watchStatusJson = get(hit, 'status');
-      const json = {
-        id: watchId,
-        watchStatusJson,
-      };
-
-      const watchStatus = WatchStatus.fromUpstreamJson(json);
-      return response.ok({
-        body: {
-          watchStatus: watchStatus.downstreamJson,
-        },
-      });
-    } catch (e) {
-      // Case: Error from Elasticsearch JS client
-      if (isEsError(e)) {
-        const body = e.statusCode === 404 ? `Watch with id = ${watchId} not found` : e;
-        return response.customError({ statusCode: e.statusCode, body });
-      }
-
-      // Case: default
-      return response.internalError({ body: e });
-    }
-  };
-
   deps.router.put(
     {
       path: '/api/watcher/watch/{watchId}/deactivate',
       validate: {
-        params: schema.object({
-          watchId: schema.string(),
-        }),
+        params: paramsSchema,
       },
     },
-    licensePreRoutingFactory(deps, handler)
+    licensePreRoutingFactory(deps, async (ctx, request, response) => {
+      const { watchId } = request.params;
+
+      try {
+        const hit = await deactivateWatch(ctx.watcher!.client, watchId);
+        const watchStatusJson = get(hit, 'status');
+        const json = {
+          id: watchId,
+          watchStatusJson,
+        };
+
+        const watchStatus = WatchStatus.fromUpstreamJson(json);
+        return response.ok({
+          body: {
+            watchStatus: watchStatus.downstreamJson,
+          },
+        });
+      } catch (e) {
+        // Case: Error from Elasticsearch JS client
+        if (isEsError(e)) {
+          const body = e.statusCode === 404 ? `Watch with id = ${watchId} not found` : e;
+          return response.customError({ statusCode: e.statusCode, body });
+        }
+
+        // Case: default
+        return response.internalError({ body: e });
+      }
+    })
   );
 }

x-pack/plugins/watcher/server/routes/api/watch/register_delete_route.ts

@@ -5,49 +5,46 @@
  */
 
 import { schema } from '@kbn/config-schema';
-import { RequestHandler } from 'kibana/server';
-import { callWithRequestFactory } from '../../../lib/call_with_request_factory';
+import { IScopedClusterClient } from 'kibana/server';
 import { isEsError } from '../../../lib/is_es_error';
 import { RouteDependencies } from '../../../types';
 import { licensePreRoutingFactory } from '../../../lib/license_pre_routing_factory';
 
-function deleteWatch(callWithRequest: any, watchId: string) {
-  return callWithRequest('watcher.deleteWatch', {
+const paramsSchema = schema.object({
+  watchId: schema.string(),
+});
+
+function deleteWatch(dataClient: IScopedClusterClient, watchId: string) {
+  return dataClient.callAsCurrentUser('watcher.deleteWatch', {
     id: watchId,
   });
 }
 
 export function registerDeleteRoute(deps: RouteDependencies) {
-  const handler: RequestHandler<any, any, any> = async (ctx, request, response) => {
-    const callWithRequest = callWithRequestFactory(deps.elasticsearchService, request);
-
-    const { watchId } = request.params;
-
-    try {
-      return response.ok({
-        body: await deleteWatch(callWithRequest, watchId),
-      });
-    } catch (e) {
-      // Case: Error from Elasticsearch JS client
-      if (isEsError(e)) {
-        const body = e.statusCode === 404 ? `Watch with id = ${watchId} not found` : e;
-        return response.customError({ statusCode: e.statusCode, body });
-      }
-
-      // Case: default
-      return response.internalError({ body: e });
-    }
-  };
-
   deps.router.delete(
     {
       path: '/api/watcher/watch/{watchId}',
       validate: {
-        params: schema.object({
-          watchId: schema.string(),
-        }),
+        params: paramsSchema,
       },
     },
-    licensePreRoutingFactory(deps, handler)
+    licensePreRoutingFactory(deps, async (ctx, request, response) => {
+      const { watchId } = request.params;
+
+      try {
+        return response.ok({
+          body: await deleteWatch(ctx.watcher!.client, watchId),
+        });
+      } catch (e) {
+        // Case: Error from Elasticsearch JS client
+        if (isEsError(e)) {
+          const body = e.statusCode === 404 ? `Watch with id = ${watchId} not found` : e;
+          return response.customError({ statusCode: e.statusCode, body });
+        }
+
+        // Case: default
+        return response.internalError({ body: e });
+      }
+    })
   );
 }

x-pack/plugins/watcher/server/routes/api/watch/register_execute_route.ts

@@ -5,9 +5,8 @@
  */
 
 import { schema } from '@kbn/config-schema';
-import { RequestHandler } from 'kibana/server';
+import { IScopedClusterClient } from 'kibana/server';
 import { get } from 'lodash';
-import { callWithRequestFactory } from '../../../lib/call_with_request_factory';
 import { isEsError } from '../../../lib/is_es_error';
 import { licensePreRoutingFactory } from '../../../lib/license_pre_routing_factory';
 
@@ -19,60 +18,63 @@ import { Watch } from '../../../models/watch/index';
 // @ts-ignore
 import { WatchHistoryItem } from '../../../models/watch_history_item/index';
 
-function executeWatch(callWithRequest: any, executeDetails: any, watchJson: any) {
+const bodySchema = schema.object({
+  executeDetails: schema.object({}, { allowUnknowns: true }),
+  watch: schema.object({}, { allowUnknowns: true }),
+});
+
+function executeWatch(dataClient: IScopedClusterClient, executeDetails: any, watchJson: any) {
   const body = executeDetails;
   body.watch = watchJson;
 
-  return callWithRequest('watcher.executeWatch', {
+  return dataClient.callAsCurrentUser('watcher.executeWatch', {
     body,
   });
 }
 
 export function registerExecuteRoute(deps: RouteDependencies) {
-  const handler: RequestHandler<any, any, any> = async (ctx, request, response) => {
-    const callWithRequest = callWithRequestFactory(deps.elasticsearchService, request);
-    const executeDetails = ExecuteDetails.fromDownstreamJson(request.body.executeDetails);
-    const watch = Watch.fromDownstreamJson(request.body.watch);
-
-    try {
-      const hit = await executeWatch(callWithRequest, executeDetails.upstreamJson, watch.watchJson);
-      const id = get(hit, '_id');
-      const watchHistoryItemJson = get(hit, 'watch_record');
-      const watchId = get(hit, 'watch_record.watch_id');
-      const json = {
-        id,
-        watchId,
-        watchHistoryItemJson,
-        includeDetails: true,
-      };
-
-      const watchHistoryItem = WatchHistoryItem.fromUpstreamJson(json);
-      return response.ok({
-        body: {
-          watchHistoryItem: watchHistoryItem.downstreamJson,
-        },
-      });
-    } 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.put(
     {
       path: '/api/watcher/watch/execute',
       validate: {
-        body: schema.object({
-          executeDetails: schema.object({}, { allowUnknowns: true }),
-          watch: schema.object({}, { allowUnknowns: true }),
-        }),
+        body: bodySchema,
       },
     },
-    licensePreRoutingFactory(deps, handler)
+    licensePreRoutingFactory(deps, async (ctx, request, response) => {
+      const executeDetails = ExecuteDetails.fromDownstreamJson(request.body.executeDetails);
+      const watch = Watch.fromDownstreamJson(request.body.watch);
+
+      try {
+        const hit = await executeWatch(
+          ctx.watcher!.client,
+          executeDetails.upstreamJson,
+          watch.watchJson
+        );
+        const id = get(hit, '_id');
+        const watchHistoryItemJson = get(hit, 'watch_record');
+        const watchId = get(hit, 'watch_record.watch_id');
+        const json = {
+          id,
+          watchId,
+          watchHistoryItemJson,
+          includeDetails: true,
+        };
+
+        const watchHistoryItem = WatchHistoryItem.fromUpstreamJson(json);
+        return response.ok({
+          body: {
+            watchHistoryItem: watchHistoryItem.downstreamJson,
+          },
+        });
+      } 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/watch/register_history_route.ts

@@ -5,9 +5,8 @@
  */
 
 import { schema } from '@kbn/config-schema';
-import { RequestHandler } from 'kibana/server';
+import { IScopedClusterClient } from 'kibana/server';
 import { get } from 'lodash';
-import { callWithRequestFactory } from '../../../lib/call_with_request_factory';
 import { fetchAllFromScroll } from '../../../lib/fetch_all_from_scroll';
 import { INDEX_NAMES, ES_SCROLL_SETTINGS } from '../../../../common/constants';
 import { isEsError } from '../../../lib/is_es_error';
@@ -16,7 +15,15 @@ import { licensePreRoutingFactory } from '../../../lib/license_pre_routing_facto
 // @ts-ignore
 import { WatchHistoryItem } from '../../../models/watch_history_item/index';
 
-function fetchHistoryItems(callWithRequest: any, watchId: any, startTime: any) {
+const paramsSchema = schema.object({
+  watchId: schema.string(),
+});
+
+const querySchema = schema.object({
+  startTime: schema.string(),
+});
+
+function fetchHistoryItems(dataClient: IScopedClusterClient, watchId: any, startTime: any) {
   const params: any = {
     index: INDEX_NAMES.WATCHER_HISTORY,
     scroll: ES_SCROLL_SETTINGS.KEEPALIVE,
@@ -37,61 +44,57 @@ function fetchHistoryItems(callWithRequest: any, watchId: any, startTime: any) {
     params.body.query.bool.must.push(timeRangeQuery);
   }
 
-  return callWithRequest('search', params).then((response: any) =>
-    fetchAllFromScroll(response, callWithRequest)
-  );
+  return dataClient
+    .callAsCurrentUser('search', params)
+    .then((response: any) => fetchAllFromScroll(response, dataClient));
 }
 
 export function registerHistoryRoute(deps: RouteDependencies) {
-  const handler: RequestHandler<any, any, any> = async (ctx, request, response) => {
-    const callWithRequest = callWithRequestFactory(deps.elasticsearchService, request);
-    const { watchId } = request.params;
-    const { startTime } = request.query;
-
-    try {
-      const hits = await fetchHistoryItems(callWithRequest, watchId, startTime);
-      const watchHistoryItems = hits.map((hit: any) => {
-        const id = get(hit, '_id');
-        const watchHistoryItemJson = get(hit, '_source');
-
-        const opts = { includeDetails: false };
-        return WatchHistoryItem.fromUpstreamJson(
-          {
-            id,
-            watchId,
-            watchHistoryItemJson,
-          },
-          opts
-        );
-      });
-
-      return response.ok({
-        body: {
-          watchHistoryItems: watchHistoryItems.map(
-            (watchHistoryItem: any) => watchHistoryItem.downstreamJson
-          ),
-        },
-      });
-    } 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.get(
     {
       path: '/api/watcher/watch/{watchId}/history',
       validate: {
-        params: schema.object({
-          watchId: schema.string(),
-        }),
+        params: paramsSchema,
+        query: querySchema,
       },
     },
-    licensePreRoutingFactory(deps, handler)
+    licensePreRoutingFactory(deps, async (ctx, request, response) => {
+      const { watchId } = request.params;
+      const { startTime } = request.query;
+
+      try {
+        const hits = await fetchHistoryItems(ctx.watcher!.client, watchId, startTime);
+        const watchHistoryItems = hits.map((hit: any) => {
+          const id = get(hit, '_id');
+          const watchHistoryItemJson = get(hit, '_source');
+
+          const opts = { includeDetails: false };
+          return WatchHistoryItem.fromUpstreamJson(
+            {
+              id,
+              watchId,
+              watchHistoryItemJson,
+            },
+            opts
+          );
+        });
+
+        return response.ok({
+          body: {
+            watchHistoryItems: watchHistoryItems.map(
+              (watchHistoryItem: any) => watchHistoryItem.downstreamJson
+            ),
+          },
+        });
+      } 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/watch/register_load_route.ts

@@ -5,65 +5,63 @@
  */
 
 import { schema } from '@kbn/config-schema';
-import { RequestHandler } from 'kibana/server';
+import { IScopedClusterClient } from 'kibana/server';
 import { get } from 'lodash';
-import { callWithRequestFactory } from '../../../lib/call_with_request_factory';
 import { isEsError } from '../../../lib/is_es_error';
 import { licensePreRoutingFactory } from '../../../lib/license_pre_routing_factory';
 // @ts-ignore
 import { Watch } from '../../../models/watch/index';
 import { RouteDependencies } from '../../../types';
 
-function fetchWatch(callWithRequest: any, watchId: string) {
-  return callWithRequest('watcher.getWatch', {
+const paramsSchema = schema.object({
+  id: schema.string(),
+});
+
+function fetchWatch(dataClient: IScopedClusterClient, watchId: string) {
+  return dataClient.callAsCurrentUser('watcher.getWatch', {
     id: watchId,
   });
 }
 
 export function registerLoadRoute(deps: RouteDependencies) {
-  const handler: RequestHandler<any, any, any> = async (ctx, request, response) => {
-    const callWithRequest = callWithRequestFactory(deps.elasticsearchService, request);
-
-    const id = request.params.id;
-
-    try {
-      const hit = await fetchWatch(callWithRequest, id);
-      const watchJson = get(hit, 'watch');
-      const watchStatusJson = get(hit, 'status');
-      const json = {
-        id,
-        watchJson,
-        watchStatusJson,
-      };
-
-      const watch = Watch.fromUpstreamJson(json, {
-        throwExceptions: {
-          Action: false,
-        },
-      });
-      return response.ok({
-        body: { watch: watch.downstreamJson },
-      });
-    } catch (e) {
-      // Case: Error from Elasticsearch JS client
-      if (isEsError(e)) {
-        const body = e.statusCode === 404 ? `Watch with id = ${id} not found` : e;
-        return response.customError({ statusCode: e.statusCode, body });
-      }
-
-      // Case: default
-      return response.internalError({ body: e });
-    }
-  };
   deps.router.get(
     {
       path: '/api/watcher/watch/{id}',
       validate: {
-        params: schema.object({
-          id: schema.string(),
-        }),
+        params: paramsSchema,
       },
     },
-    licensePreRoutingFactory(deps, handler)
+    licensePreRoutingFactory(deps, async (ctx, request, response) => {
+      const id = request.params.id;
+
+      try {
+        const hit = await fetchWatch(ctx.watcher!.client, id);
+        const watchJson = get(hit, 'watch');
+        const watchStatusJson = get(hit, 'status');
+        const json = {
+          id,
+          watchJson,
+          watchStatusJson,
+        };
+
+        const watch = Watch.fromUpstreamJson(json, {
+          throwExceptions: {
+            Action: false,
+          },
+        });
+        return response.ok({
+          body: { watch: watch.downstreamJson },
+        });
+      } catch (e) {
+        // Case: Error from Elasticsearch JS client
+        if (isEsError(e)) {
+          const body = e.statusCode === 404 ? `Watch with id = ${id} not found` : e;
+          return response.customError({ statusCode: e.statusCode, body });
+        }
+
+        // Case: default
+        return response.internalError({ body: e });
+      }
+    })
   );
 }

x-pack/plugins/watcher/server/routes/api/watch/register_save_route.ts

@@ -5,98 +5,104 @@
  */
 
 import { schema } from '@kbn/config-schema';
-import { RequestHandler } from 'kibana/server';
+import { IScopedClusterClient } from 'kibana/server';
 import { i18n } from '@kbn/i18n';
 import { WATCH_TYPES } from '../../../../common/constants';
 import { serializeJsonWatch, serializeThresholdWatch } from '../../../../common/lib/serialization';
-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';
 
-function fetchWatch(callWithRequest: any, watchId: string) {
-  return callWithRequest('watcher.getWatch', {
+const paramsSchema = schema.object({
+  id: schema.string(),
+});
+
+const bodySchema = schema.object(
+  {
+    type: schema.string(),
+    isNew: schema.boolean(),
+  },
+  { allowUnknowns: true }
+);
+
+function fetchWatch(dataClient: IScopedClusterClient, watchId: string) {
+  return dataClient.callAsCurrentUser('watcher.getWatch', {
     id: watchId,
   });
 }
 
-function saveWatch(callWithRequest: any, id: string, body: any) {
-  return callWithRequest('watcher.putWatch', {
+function saveWatch(dataClient: IScopedClusterClient, id: string, body: any) {
+  return dataClient.callAsCurrentUser('watcher.putWatch', {
     id,
     body,
   });
 }
 
 export function registerSaveRoute(deps: RouteDependencies) {
-  const handler: RequestHandler<any, any, any> = async (ctx, request, response) => {
-    const callWithRequest = callWithRequestFactory(deps.elasticsearchService, request);
-    const { id } = request.params;
-    const { type, isNew, ...watchConfig } = request.body;
-
-    // For new watches, verify watch with the same ID doesn't already exist
-    if (isNew) {
-      try {
-        const existingWatch = await fetchWatch(callWithRequest, id);
-        if (existingWatch.found) {
-          return response.conflict({
-            body: {
-              message: i18n.translate('xpack.watcher.saveRoute.duplicateWatchIdErrorMessage', {
-                defaultMessage: "There is already a watch with ID '{watchId}'.",
-                values: {
-                  watchId: id,
-                },
-              }),
-            },
-          });
-        }
-      } catch (e) {
-        const es404 = isEsError(e) && e.statusCode === 404;
-        if (!es404) {
-          return response.internalError({ body: e });
-        }
-        // Else continue...
-      }
-    }
-
-    let serializedWatch;
-
-    switch (type) {
-      case WATCH_TYPES.JSON:
-        const { name, watch } = watchConfig;
-        serializedWatch = serializeJsonWatch(name, watch);
-        break;
-
-      case WATCH_TYPES.THRESHOLD:
-        serializedWatch = serializeThresholdWatch(watchConfig);
-        break;
-    }
-
-    try {
-      // Create new watch
-      return response.ok({
-        body: await saveWatch(callWithRequest, id, serializedWatch),
-      });
-    } 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.put(
     {
       path: '/api/watcher/watch/{id}',
       validate: {
-        params: schema.object({
-          id: schema.string(),
-        }),
-        body: schema.object({}, { allowUnknowns: true }),
+        params: paramsSchema,
+        body: bodySchema,
       },
     },
-    licensePreRoutingFactory(deps, handler)
+    licensePreRoutingFactory(deps, async (ctx, request, response) => {
+      const { id } = request.params;
+      const { type, isNew, ...watchConfig } = request.body;
+
+      // For new watches, verify watch with the same ID doesn't already exist
+      if (isNew) {
+        try {
+          const existingWatch = await fetchWatch(ctx.watcher!.client, id);
+          if (existingWatch.found) {
+            return response.conflict({
+              body: {
+                message: i18n.translate('xpack.watcher.saveRoute.duplicateWatchIdErrorMessage', {
+                  defaultMessage: "There is already a watch with ID '{watchId}'.",
+                  values: {
+                    watchId: id,
+                  },
+                }),
+              },
+            });
+          }
+        } catch (e) {
+          const es404 = isEsError(e) && e.statusCode === 404;
+          if (!es404) {
+            return response.internalError({ body: e });
+          }
+          // Else continue...
+        }
+      }
+
+      let serializedWatch;
+
+      switch (type) {
+        case WATCH_TYPES.JSON:
+          const { name, watch } = watchConfig as any;
+          serializedWatch = serializeJsonWatch(name, watch);
+          break;
+
+        case WATCH_TYPES.THRESHOLD:
+          serializedWatch = serializeThresholdWatch(watchConfig);
+          break;
+      }
+
+      try {
+        // Create new watch
+        return response.ok({
+          body: await saveWatch(ctx.watcher!.client, id, serializedWatch),
+        });
+      } 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/watch/register_visualize_route.ts

@@ -5,8 +5,7 @@
  */
 
 import { schema } from '@kbn/config-schema';
-import { RequestHandler } from 'kibana/server';
-import { callWithRequestFactory } from '../../../lib/call_with_request_factory';
+import { IScopedClusterClient } from 'kibana/server';
 import { isEsError } from '../../../lib/is_es_error';
 import { RouteDependencies } from '../../../types';
 import { licensePreRoutingFactory } from '../../../lib/license_pre_routing_factory';
@@ -16,7 +15,12 @@ import { Watch } from '../../../models/watch/index';
 // @ts-ignore
 import { VisualizeOptions } from '../../../models/visualize_options/index';
 
-function fetchVisualizeData(callWithRequest: any, index: any, body: any) {
+const bodySchema = schema.object({
+  watch: schema.object({}, { allowUnknowns: true }),
+  options: schema.object({}, { allowUnknowns: true }),
+});
+
+function fetchVisualizeData(dataClient: IScopedClusterClient, index: any, body: any) {
   const params = {
     index,
     body,
@@ -25,46 +29,40 @@ function fetchVisualizeData(callWithRequest: any, index: any, body: any) {
     ignore: [404],
   };
 
-  return callWithRequest('search', params);
+  return dataClient.callAsCurrentUser('search', params);
 }
 
 export function registerVisualizeRoute(deps: RouteDependencies) {
-  const handler: RequestHandler<any, any, any> = async (ctx, request, response) => {
-    const callWithRequest = callWithRequestFactory(deps.elasticsearchService, request);
-    const watch = Watch.fromDownstreamJson(request.body.watch);
-    const options = VisualizeOptions.fromDownstreamJson(request.body.options);
-    const body = watch.getVisualizeQuery(options);
-
-    try {
-      const hits = await fetchVisualizeData(callWithRequest, watch.index, body);
-      const visualizeData = watch.formatVisualizeData(hits);
-
-      return response.ok({
-        body: {
-          visualizeData,
-        },
-      });
-    } 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/watch/visualize',
       validate: {
-        body: schema.object({
-          watch: schema.object({}, { allowUnknowns: true }),
-          options: schema.object({}, { allowUnknowns: true }),
-        }),
+        body: bodySchema,
       },
     },
-    licensePreRoutingFactory(deps, handler)
+    licensePreRoutingFactory(deps, async (ctx, request, response) => {
+      const watch = Watch.fromDownstreamJson(request.body.watch);
+      const options = VisualizeOptions.fromDownstreamJson(request.body.options);
+      const body = watch.getVisualizeQuery(options);
+
+      try {
+        const hits = await fetchVisualizeData(ctx.watcher!.client, watch.index, body);
+        const visualizeData = watch.formatVisualizeData(hits);
+
+        return response.ok({
+          body: {
+            visualizeData,
+          },
+        });
+      } 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/watches/register_delete_route.ts

@@ -5,16 +5,20 @@
  */
 
 import { schema } from '@kbn/config-schema';
-import { RequestHandler } from 'kibana/server';
-import { callWithRequestFactory } from '../../../lib/call_with_request_factory';
+import { IScopedClusterClient } from 'kibana/server';
 import { RouteDependencies } from '../../../types';
 import { licensePreRoutingFactory } from '../../../lib/license_pre_routing_factory';
 
-function deleteWatches(callWithRequest: any, watchIds: string[]) {
+const bodySchema = schema.object({
+  watchIds: schema.arrayOf(schema.string()),
+});
+
+function deleteWatches(dataClient: IScopedClusterClient, watchIds: string[]) {
   const deletePromises = watchIds.map(watchId => {
-    return callWithRequest('watcher.deleteWatch', {
-      id: watchId,
-    })
+    return dataClient
+      .callAsCurrentUser('watcher.deleteWatch', {
+        id: watchId,
+      })
       .then((success: Array<{ _id: string }>) => ({ success }))
       .catch((error: Array<{ _id: string }>) => ({ error }));
   });
@@ -22,7 +26,7 @@ function deleteWatches(callWithRequest: any, watchIds: string[]) {
   return Promise.all(deletePromises).then(results => {
     const errors: Error[] = [];
     const successes: boolean[] = [];
-    results.forEach(({ success, error }) => {
+    results.forEach(({ success, error }: { success?: any; error?: any }) => {
       if (success) {
         successes.push(success._id);
       } else if (error) {
@@ -38,26 +42,20 @@ function deleteWatches(callWithRequest: any, watchIds: string[]) {
 }
 
 export function registerDeleteRoute(deps: RouteDependencies) {
-  const handler: RequestHandler<any, any, any> = async (ctx, request, response) => {
-    const callWithRequest = callWithRequestFactory(deps.elasticsearchService, request);
-
-    try {
-      const results = await deleteWatches(callWithRequest, request.body.watchIds);
-      return response.ok({ body: { results } });
-    } catch (e) {
-      return response.internalError({ body: e });
-    }
-  };
-
   deps.router.post(
     {
       path: '/api/watcher/watches/delete',
       validate: {
-        body: schema.object({
-          watchIds: schema.arrayOf(schema.string()),
-        }),
+        body: bodySchema,
       },
     },
-    licensePreRoutingFactory(deps, handler)
+    licensePreRoutingFactory(deps, async (ctx, request, response) => {
+      try {
+        const results = await deleteWatches(ctx.watcher!.client, request.body.watchIds);
+        return response.ok({ body: { results } });
+      } catch (e) {
+        return response.internalError({ body: e });
+      }
+    })
   );
 }

x-pack/plugins/watcher/server/routes/api/watches/register_list_route.ts

@@ -4,9 +4,8 @@
  * you may not use this file except in compliance with the Elastic License.
  */
 
-import { RequestHandler } from 'kibana/server';
+import { IScopedClusterClient } from 'kibana/server';
 import { get } from 'lodash';
-import { callWithRequestFactory } from '../../../lib/call_with_request_factory';
 import { fetchAllFromScroll } from '../../../lib/fetch_all_from_scroll';
 import { INDEX_NAMES, ES_SCROLL_SETTINGS } from '../../../../common/constants';
 import { isEsError } from '../../../lib/is_es_error';
@@ -15,7 +14,7 @@ import { licensePreRoutingFactory } from '../../../lib/license_pre_routing_facto
 // @ts-ignore
 import { Watch } from '../../../models/watch/index';
 
-function fetchWatches(callWithRequest: any) {
+function fetchWatches(dataClient: IScopedClusterClient) {
   const params = {
     index: INDEX_NAMES.WATCHES,
     scroll: ES_SCROLL_SETTINGS.KEEPALIVE,
@@ -25,62 +24,58 @@ function fetchWatches(callWithRequest: any) {
     ignore: [404],
   };
 
-  return callWithRequest('search', params).then((response: any) =>
-    fetchAllFromScroll(response, callWithRequest)
-  );
+  return dataClient
+    .callAsCurrentUser('search', params)
+    .then((response: any) => fetchAllFromScroll(response, dataClient));
 }
 
 export function registerListRoute(deps: RouteDependencies) {
-  const handler: RequestHandler<any, any, any> = async (ctx, request, response) => {
-    const callWithRequest = callWithRequestFactory(deps.elasticsearchService, request);
-
-    try {
-      const hits = await fetchWatches(callWithRequest);
-      const watches = hits.map((hit: any) => {
-        const id = get(hit, '_id');
-        const watchJson = get(hit, '_source');
-        const watchStatusJson = get(hit, '_source.status');
-
-        return Watch.fromUpstreamJson(
-          {
-            id,
-            watchJson,
-            watchStatusJson,
-          },
-          {
-            throwExceptions: {
-              Action: false,
-            },
-          }
-        );
-      });
-
-      return response.ok({
-        body: {
-          watches: watches.map((watch: any) => watch.downstreamJson),
-        },
-      });
-    } catch (e) {
-      // Case: Error from Elasticsearch JS client
-      if (isEsError(e)) {
-        return response.customError({
-          statusCode: e.statusCode,
-          body: {
-            message: e.message,
-          },
-        });
-      }
-
-      // Case: default
-      return response.internalError({ body: e });
-    }
-  };
-
   deps.router.get(
     {
       path: '/api/watcher/watches',
       validate: false,
     },
-    licensePreRoutingFactory(deps, handler)
+    licensePreRoutingFactory(deps, async (ctx, request, response) => {
+      try {
+        const hits = await fetchWatches(ctx.watcher!.client);
+        const watches = hits.map((hit: any) => {
+          const id = get(hit, '_id');
+          const watchJson = get(hit, '_source');
+          const watchStatusJson = get(hit, '_source.status');
+
+          return Watch.fromUpstreamJson(
+            {
+              id,
+              watchJson,
+              watchStatusJson,
+            },
+            {
+              throwExceptions: {
+                Action: false,
+              },
+            }
+          );
+        });
+
+        return response.ok({
+          body: {
+            watches: watches.map((watch: any) => watch.downstreamJson),
+          },
+        });
+      } catch (e) {
+        // Case: Error from Elasticsearch JS client
+        if (isEsError(e)) {
+          return response.customError({
+            statusCode: e.statusCode,
+            body: {
+              message: e.message,
+            },
+          });
+        }
+
+        // Case: default
+        return response.internalError({ body: e });
+      }
+    })
   );
 }

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

@@ -4,7 +4,7 @@
  * you may not use this file except in compliance with the Elastic License.
  */
 
-import { IRouter, ElasticsearchServiceSetup, IClusterClient } from 'kibana/server';
+import { IRouter } from 'kibana/server';
 import { LicensingPluginSetup } from '../../licensing/server';
 
 import { XPackMainPlugin } from '../../../legacy/plugins/xpack_main/server/xpack_main';
@@ -24,8 +24,6 @@ export interface ServerShim {
 export interface RouteDependencies {
   router: IRouter;
   getLicenseStatus: () => LicenseStatus;
-  elasticsearchService: ElasticsearchServiceSetup;
-  elasticsearch: IClusterClient;
 }
 
 export interface LicenseStatus {