add support for on_failure parameter

Author: alisonelizabeth

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

@@ -13,8 +13,9 @@ import { RouteDependencies } from '../../types';
 const bodySchema = schema.object({
   name: schema.string(),
   description: schema.string(),
-  processors: schema.arrayOf(schema.any()), // todo fix
+  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 = ({
@@ -33,7 +34,7 @@ export const registerCreateRoute = ({
       const { callAsCurrentUser } = ctx.core.elasticsearch.dataClient;
       const pipeline = req.body as Pipeline;
 
-      const { name, description, processors, version } = pipeline;
+      const { name, description, processors, version, onFailure } = pipeline;
 
       try {
         // Check that a pipeline with the same name doesn't already exist
@@ -62,6 +63,7 @@ export const registerCreateRoute = ({
             description,
             processors,
             version,
+            on_failure: onFailure,
           },
         });
 

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

@@ -11,8 +11,9 @@ import { RouteDependencies } from '../../types';
 
 const bodySchema = schema.object({
   description: schema.string(),
-  processors: schema.arrayOf(schema.any()), // todo fix
+  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({
@@ -37,7 +38,7 @@ export const registerUpdateRoute = ({
       const { name } = req.params as typeof paramsSchema.type;
       const pipeline = req.body as Pipeline;
 
-      const { description, processors, version } = pipeline;
+      const { description, processors, version, onFailure } = pipeline;
 
       try {
         // Verify pipeline exists; ES will throw 404 if it doesn't
@@ -49,6 +50,7 @@ export const registerUpdateRoute = ({
             description,
             processors,
             version,
+            on_failure: onFailure,
           },
         });
 

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

@@ -35,6 +35,14 @@ export default function({ getService }: FtrProviderContext) {
                 },
               },
             ],
+            onFailure: [
+              {
+                set: {
+                  field: 'error.message',
+                  value: '{{ failure_message }}',
+                },
+              },
+            ],
             version: 1,
           })
           .expect(200);