Server-side create/update ingest pipelines

x-pack/plugins/ingest_pipelines/common/lib/pipeline_serialization.test.ts

@@ -20,6 +20,14 @@ describe('pipeline_serialization', () => {
                 },
               },
             ],
+            on_failure: [
+              {
+                set: {
+                  field: 'error.message',
+                  value: '{{ failure_message }}',
+                },
+              },
+            ],
           },
           pipeline2: {
             description: 'pipeline2 description',
@@ -39,6 +47,14 @@ describe('pipeline_serialization', () => {
               },
             },
           ],
+          onFailure: [
+            {
+              set: {
+                field: 'error.message',
+                value: '{{ failure_message }}',
+              },
+            },
+          ],
         },
         {
           name: 'pipeline2',

x-pack/plugins/ingest_pipelines/common/lib/pipeline_serialization.ts

@@ -9,16 +9,25 @@ import { PipelinesByName, Pipeline } from '../types';
 export function deserializePipelines(pipelinesByName: PipelinesByName): Pipeline[] {
   const pipelineNames: string[] = Object.keys(pipelinesByName);
 
-  const deserializedTemplates = pipelineNames.map((name: string) => {
-    const { description, version, processors } = pipelinesByName[name];
+  const deserializedPipelines = pipelineNames.map((name: string) => {
+    const { description, version, processors, on_failure } = pipelinesByName[name];
 
-    return {
+    const pipeline = {
       name,
       description,
       version,
       processors,
+      onFailure: on_failure,
     };
+
+    // Remove any undefined values
+    return Object.entries(pipeline).reduce((pipelineDefinition: any, [key, value]) => {
+      if (value !== undefined) {
+        pipelineDefinition[key] = value;
+      }
+      return pipelineDefinition;
+    }, {});
   });
 
-  return deserializedTemplates;
+  return deserializedPipelines;
 }

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

@@ -15,8 +15,14 @@ export interface Pipeline {
   description: string;
   version?: number;
   processors: Processor[];
+  onFailure?: Processor[];
 }
 
 export interface PipelinesByName {
-  [key: string]: Omit<Pipeline, 'name'>;
+  [key: string]: {
+    description: string;
+    version?: number;
+    processors: Processor[];
+    on_failure?: Processor[];
+  };
 }

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

@@ -0,0 +1,83 @@
+/*
+ * 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 { i18n } from '@kbn/i18n';
+import { schema } from '@kbn/config-schema';
+
+import { Pipeline } from '../../../common/types';
+import { API_BASE_PATH } from '../../../common/constants';
+import { RouteDependencies } from '../../types';
+
+const bodySchema = schema.object({
+  name: schema.string(),
+  description: schema.string(),
+  processors: schema.arrayOf(schema.recordOf(schema.string(), schema.any())),
+  version: schema.maybe(schema.number()),
+  onFailure: schema.maybe(schema.arrayOf(schema.recordOf(schema.string(), schema.any()))),
+});
+
+export const registerCreateRoute = ({
+  router,
+  license,
+  lib: { isEsError },
+}: RouteDependencies): void => {
+  router.put(
+    {
+      path: API_BASE_PATH,
+      validate: {
+        body: bodySchema,
+      },
+    },
+    license.guardApiRoute(async (ctx, req, res) => {
+      const { callAsCurrentUser } = ctx.core.elasticsearch.dataClient;
+      const pipeline = req.body as Pipeline;
+
+      const { name, description, processors, version, onFailure } = pipeline;
+
+      try {
+        // Check that a pipeline with the same name doesn't already exist
+        const pipelineByName = await callAsCurrentUser('ingest.getPipeline', { id: name });
+
+        if (pipelineByName[name]) {
+          return res.conflict({
+            body: new Error(
+              i18n.translate('xpack.ingestPipelines.createRoute.duplicatePipelineIdErrorMessage', {
+                defaultMessage: "There is already a pipeline with name '{name}'.",
+                values: {
+                  name,
+                },
+              })
+            ),
+          });
+        }
+      } catch (e) {
+        // Silently swallow error
+      }
+
+      try {
+        const response = await callAsCurrentUser('ingest.putPipeline', {
+          id: name,
+          body: {
+            description,
+            processors,
+            version,
+            on_failure: onFailure,
+          },
+        });
+
+        return res.ok({ body: response });
+      } catch (error) {
+        if (isEsError(error)) {
+          return res.customError({
+            statusCode: error.statusCode,
+            body: error,
+          });
+        }
+
+        return res.internalError({ body: error });
+      }
+    })
+  );
+};

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

@@ -5,3 +5,7 @@
  */
 
 export { registerGetRoutes } from './get';
+
+export { registerCreateRoute } from './create';
+
+export { registerUpdateRoute } from './update';

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

@@ -0,0 +1,70 @@
+/*
+ * 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 { schema } from '@kbn/config-schema';
+
+import { Pipeline } from '../../../common/types';
+import { API_BASE_PATH } from '../../../common/constants';
+import { RouteDependencies } from '../../types';
+
+const bodySchema = schema.object({
+  description: schema.string(),
+  processors: schema.arrayOf(schema.recordOf(schema.string(), schema.any())),
+  version: schema.maybe(schema.number()),
+  onFailure: schema.maybe(schema.arrayOf(schema.recordOf(schema.string(), schema.any()))),
+});
+
+const paramsSchema = schema.object({
+  name: schema.string(),
+});
+
+export const registerUpdateRoute = ({
+  router,
+  license,
+  lib: { isEsError },
+}: RouteDependencies): void => {
+  router.put(
+    {
+      path: `${API_BASE_PATH}/{name}`,
+      validate: {
+        body: bodySchema,
+        params: paramsSchema,
+      },
+    },
+    license.guardApiRoute(async (ctx, req, res) => {
+      const { callAsCurrentUser } = ctx.core.elasticsearch.dataClient;
+      const { name } = req.params as typeof paramsSchema.type;
+      const pipeline = req.body as Pipeline;
+
+      const { description, processors, version, onFailure } = pipeline;
+
+      try {
+        // Verify pipeline exists; ES will throw 404 if it doesn't
+        await callAsCurrentUser('ingest.getPipeline', { id: name });
+
+        const response = await callAsCurrentUser('ingest.putPipeline', {
+          id: name,
+          body: {
+            description,
+            processors,
+            version,
+            on_failure: onFailure,
+          },
+        });
+
+        return res.ok({ body: response });
+      } catch (error) {
+        if (isEsError(error)) {
+          return res.customError({
+            statusCode: error.statusCode,
+            body: error,
+          });
+        }
+
+        return res.internalError({ body: error });
+      }
+    })
+  );
+};

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

@@ -6,10 +6,12 @@
 
 import { RouteDependencies } from '../types';
 
-import { registerGetRoutes } from './api';
+import { registerGetRoutes, registerCreateRoute, registerUpdateRoute } from './api';
 
 export class ApiRoutes {
   setup(dependencies: RouteDependencies) {
     registerGetRoutes(dependencies);
+    registerCreateRoute(dependencies);
+    registerUpdateRoute(dependencies);
   }
 }

x-pack/test/api_integration/apis/management/index.js

@@ -12,5 +12,6 @@ export default function({ loadTestFile }) {
     loadTestFile(require.resolve('./rollup'));
     loadTestFile(require.resolve('./index_management'));
     loadTestFile(require.resolve('./index_lifecycle_management'));
+    loadTestFile(require.resolve('./ingest_pipelines'));
   });
 }

x-pack/test/api_integration/apis/management/ingest_pipelines/index.ts

@@ -0,0 +1,12 @@
+/*
+ * 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 { FtrProviderContext } from '../../../ftr_provider_context';
+
+export default function({ loadTestFile }: FtrProviderContext) {
+  describe('Ingest Node Pipelines', () => {
+    loadTestFile(require.resolve('./ingest_pipelines'));
+  });
+}