[Ingest Pipelines] Migrate to new ES client

x-pack/plugins/ingest_pipelines/common/types.ts

@@ -5,6 +5,8 @@
  * 2.0.
  */
 
+import { Pipeline as ESPipeline } from '@elastic/elasticsearch/api/types';
+
 export interface ESProcessorConfig {
   on_failure?: Processor[];
   ignore_failure?: boolean;
@@ -17,19 +19,10 @@ export interface Processor {
   [typeName: string]: ESProcessorConfig;
 }
 
-export interface Pipeline {
+export interface Pipeline extends ESPipeline {
   name: string;
-  description: string;
-  version?: number;
-  processors: Processor[];
-  on_failure?: Processor[];
 }
 
 export interface PipelinesByName {
-  [key: string]: {
-    description: string;
-    version?: number;
-    processors: Processor[];
-    on_failure?: Processor[];
-  };
+  [key: string]: ESPipeline;
 }

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

@@ -13,7 +13,7 @@ import { PLUGIN_ID, PLUGIN_MIN_LICENSE_TYPE } from '../common/constants';
 
 import { License } from './services';
 import { ApiRoutes } from './routes';
-import { isEsError } from './shared_imports';
+import { handleEsError } from './shared_imports';
 import { Dependencies } from './types';
 
 export class IngestPipelinesPlugin implements Plugin<void, void, any, any> {
@@ -66,7 +66,7 @@ export class IngestPipelinesPlugin implements Plugin<void, void, any, any> {
         isSecurityEnabled: () => security !== undefined && security.license.isEnabled(),
       },
       lib: {
-        isEsError,
+        handleEsError,
       },
     });
   }

x-pack/plugins/ingest_pipelines/server/routes/api/create.ts

@@ -11,8 +11,7 @@ import { schema } from '@kbn/config-schema';
 import { Pipeline } from '../../../common/types';
 import { API_BASE_PATH } from '../../../common/constants';
 import { RouteDependencies } from '../../types';
-import { pipelineSchema } from './pipeline_schema';
-import { isObjectWithKeys } from './shared';
+import { pipelineSchema } from './shared';
 
 const bodySchema = schema.object({
   name: schema.string(),
@@ -22,7 +21,7 @@ const bodySchema = schema.object({
 export const registerCreateRoute = ({
   router,
   license,
-  lib: { isEsError },
+  lib: { handleEsError },
 }: RouteDependencies): void => {
   router.post(
     {
@@ -32,15 +31,17 @@ export const registerCreateRoute = ({
       },
     },
     license.guardApiRoute(async (ctx, req, res) => {
-      const { callAsCurrentUser } = ctx.core.elasticsearch.legacy.client;
+      const { client: clusterClient } = ctx.core.elasticsearch;
       const pipeline = req.body as Pipeline;
 
       // eslint-disable-next-line @typescript-eslint/naming-convention
       const { name, description, processors, version, on_failure } = pipeline;
 
       try {
         // Check that a pipeline with the same name doesn't already exist
-        const pipelineByName = await callAsCurrentUser('ingest.getPipeline', { id: name });
+        const { body: pipelineByName } = await clusterClient.asCurrentUser.ingest.getPipeline({
+          id: name,
+        });
 
         if (pipelineByName[name]) {
           return res.conflict({
@@ -59,7 +60,7 @@ export const registerCreateRoute = ({
       }
 
       try {
-        const response = await callAsCurrentUser('ingest.putPipeline', {
+        const { body: response } = await clusterClient.asCurrentUser.ingest.putPipeline({
           id: name,
           body: {
             description,
@@ -71,19 +72,7 @@ export const registerCreateRoute = ({
 
         return res.ok({ body: response });
       } catch (error) {
-        if (isEsError(error)) {
-          return res.customError({
-            statusCode: error.statusCode,
-            body: isObjectWithKeys(error.body)
-              ? {
-                  message: error.message,
-                  attributes: error.body,
-                }
-              : error,
-          });
-        }
-
-        throw error;
+        return handleEsError({ error, response: res });
       }
     })
   );

x-pack/plugins/ingest_pipelines/server/routes/api/delete.ts

@@ -23,7 +23,7 @@ export const registerDeleteRoute = ({ router, license }: RouteDependencies): voi
       },
     },
     license.guardApiRoute(async (ctx, req, res) => {
-      const { callAsCurrentUser } = ctx.core.elasticsearch.legacy.client;
+      const { client: clusterClient } = ctx.core.elasticsearch;
       const { names } = req.params;
       const pipelineNames = names.split(',');
 
@@ -34,7 +34,8 @@ export const registerDeleteRoute = ({ router, license }: RouteDependencies): voi
 
       await Promise.all(
         pipelineNames.map((pipelineName) => {
-          return callAsCurrentUser('ingest.deletePipeline', { id: pipelineName })
+          return clusterClient.asCurrentUser.ingest
+            .deletePipeline({ id: pipelineName })
             .then(() => response.itemsDeleted.push(pipelineName))
             .catch((e) =>
               response.errors.push({

x-pack/plugins/ingest_pipelines/server/routes/api/documents.ts

@@ -18,7 +18,7 @@ const paramsSchema = schema.object({
 export const registerDocumentsRoute = ({
   router,
   license,
-  lib: { isEsError },
+  lib: { handleEsError },
 }: RouteDependencies): void => {
   router.get(
     {
@@ -28,11 +28,11 @@ export const registerDocumentsRoute = ({
       },
     },
     license.guardApiRoute(async (ctx, req, res) => {
-      const { callAsCurrentUser } = ctx.core.elasticsearch.legacy.client;
+      const { client: clusterClient } = ctx.core.elasticsearch;
       const { index, id } = req.params;
 
       try {
-        const document = await callAsCurrentUser('get', { index, id });
+        const { body: document } = await clusterClient.asCurrentUser.get({ index, id });
 
         const { _id, _index, _source } = document;
 
@@ -44,14 +44,7 @@ export const registerDocumentsRoute = ({
           },
         });
       } catch (error) {
-        if (isEsError(error)) {
-          return res.customError({
-            statusCode: error.statusCode,
-            body: error,
-          });
-        }
-
-        throw error;
+        return handleEsError({ error, response: res });
       }
     })
   );

x-pack/plugins/ingest_pipelines/server/routes/api/get.ts

@@ -18,33 +18,22 @@ const paramsSchema = schema.object({
 export const registerGetRoutes = ({
   router,
   license,
-  lib: { isEsError },
+  lib: { handleEsError },
 }: RouteDependencies): void => {
   // Get all pipelines
   router.get(
     { path: API_BASE_PATH, validate: false },
     license.guardApiRoute(async (ctx, req, res) => {
-      const { callAsCurrentUser } = ctx.core.elasticsearch.legacy.client;
+      const { client: clusterClient } = ctx.core.elasticsearch;
 
       try {
-        const pipelines = await callAsCurrentUser('ingest.getPipeline');
+        const { body: pipelines } = await clusterClient.asCurrentUser.ingest.getPipeline({
+          id: '_all',
+        });
 
         return res.ok({ body: deserializePipelines(pipelines) });
       } catch (error) {
-        if (isEsError(error)) {
-          // ES returns 404 when there are no pipelines
-          // Instead, we return an empty array and 200 status back to the client
-          if (error.status === 404) {
-            return res.ok({ body: [] });
-          }
-
-          return res.customError({
-            statusCode: error.statusCode,
-            body: error,
-          });
-        }
-
-        throw error;
+        return handleEsError({ error, response: res });
       }
     })
   );
@@ -58,27 +47,22 @@ export const registerGetRoutes = ({
       },
     },
     license.guardApiRoute(async (ctx, req, res) => {
-      const { callAsCurrentUser } = ctx.core.elasticsearch.legacy.client;
+      const { client: clusterClient } = ctx.core.elasticsearch;
       const { name } = req.params;
 
       try {
-        const pipeline = await callAsCurrentUser('ingest.getPipeline', { id: name });
+        const { body: pipelines } = await clusterClient.asCurrentUser.ingest.getPipeline({
+          id: name,
+        });
 
         return res.ok({
           body: {
-            ...pipeline[name],
+            ...pipelines[name],
             name,
           },
         });
       } catch (error) {
-        if (isEsError(error)) {
-          return res.customError({
-            statusCode: error.statusCode,
-            body: error,
-          });
-        }
-
-        throw error;
+        return handleEsError({ error, response: res });
       }
     })
   );

x-pack/plugins/ingest_pipelines/server/routes/api/privileges.ts

@@ -36,24 +36,13 @@ export const registerPrivilegesRoute = ({ license, router, config }: RouteDepend
         return res.ok({ body: privilegesResult });
       }
 
-      const {
-        core: {
-          elasticsearch: {
-            legacy: { client },
-          },
-        },
-      } = ctx;
+      const { client: clusterClient } = ctx.core.elasticsearch;
 
-      const { has_all_requested: hasAllPrivileges, cluster } = await client.callAsCurrentUser(
-        'transport.request',
-        {
-          path: '/_security/user/_has_privileges',
-          method: 'POST',
-          body: {
-            cluster: APP_CLUSTER_REQUIRED_PRIVILEGES,
-          },
-        }
-      );
+      const {
+        body: { has_all_requested: hasAllPrivileges, cluster },
+      } = await clusterClient.asCurrentUser.security.hasPrivileges({
+        body: { cluster: APP_CLUSTER_REQUIRED_PRIVILEGES },
+      });
 
       if (!hasAllPrivileges) {
         privilegesResult.missingPrivileges.cluster = extractMissingPrivileges(cluster);

x-pack/plugins/ingest_pipelines/server/routes/api/shared/index.ts

@@ -5,4 +5,4 @@
  * 2.0.
  */
 
-export { isObjectWithKeys } from './is_object_with_keys';
+export { pipelineSchema } from './pipeline_schema';

x-pack/plugins/ingest_pipelines/server/routes/api/simulate.ts

@@ -4,12 +4,12 @@
  * 2.0; you may not use this file except in compliance with the Elastic License
  * 2.0.
  */
-
+import { SimulatePipelineDocument } from '@elastic/elasticsearch/api/types';
 import { schema } from '@kbn/config-schema';
 
 import { API_BASE_PATH } from '../../../common/constants';
 import { RouteDependencies } from '../../types';
-import { pipelineSchema } from './pipeline_schema';
+import { pipelineSchema } from './shared';
 
 const bodySchema = schema.object({
   pipeline: schema.object(pipelineSchema),
@@ -20,7 +20,7 @@ const bodySchema = schema.object({
 export const registerSimulateRoute = ({
   router,
   license,
-  lib: { isEsError },
+  lib: { handleEsError },
 }: RouteDependencies): void => {
   router.post(
     {
@@ -30,29 +30,22 @@ export const registerSimulateRoute = ({
       },
     },
     license.guardApiRoute(async (ctx, req, res) => {
-      const { callAsCurrentUser } = ctx.core.elasticsearch.legacy.client;
+      const { client: clusterClient } = ctx.core.elasticsearch;
 
       const { pipeline, documents, verbose } = req.body;
 
       try {
-        const response = await callAsCurrentUser('ingest.simulate', {
+        const { body: response } = await clusterClient.asCurrentUser.ingest.simulate({
           verbose,
           body: {
             pipeline,
-            docs: documents,
+            docs: documents as SimulatePipelineDocument[],
           },
         });
 
         return res.ok({ body: response });
       } catch (error) {
-        if (isEsError(error)) {
-          return res.customError({
-            statusCode: error.statusCode,
-            body: error,
-          });
-        }
-
-        throw error;
+        return handleEsError({ error, response: res });
       }
     })
   );