Add pipeline support to Ingest API

src/plugins/elasticsearch/index.js

@@ -29,7 +29,7 @@ module.exports = function ({ Plugin }) {
           key: string()
         }).default(),
         apiVersion: string().default('2.0'),
-        engineVersion: string().valid('^2.1.0').default('^2.1.0')
+        engineVersion: Joi.string().valid('^3.0.0').default('^3.0.0')
       }).default();
     },
 

src/plugins/kibana/server/lib/__tests__/convert_pattern_and_template_name.js

@@ -1,29 +1,29 @@
-const {templateToPattern, patternToTemplate} = require('../convert_pattern_and_template_name');
+const {ingestToPattern, patternToIngest} = require('../convert_pattern_and_ingest_name');
 const expect = require('expect.js');
 
 describe('convertPatternAndTemplateName', function () {
 
-  describe('templateToPattern', function () {
+  describe('ingestToPattern', function () {
 
     it('should convert an index template\'s name to its matching index pattern\'s title', function () {
-      expect(templateToPattern('kibana-logstash-*')).to.be('logstash-*');
+      expect(ingestToPattern('kibana-logstash-*')).to.be('logstash-*');
     });
 
     it('should throw an error if the template name isn\'t a valid kibana namespaced name', function () {
-      expect(templateToPattern).withArgs('logstash-*').to.throwException('not a valid kibana namespaced template name');
-      expect(templateToPattern).withArgs('').to.throwException(/not a valid kibana namespaced template name/);
+      expect(ingestToPattern).withArgs('logstash-*').to.throwException('not a valid kibana namespaced template name');
+      expect(ingestToPattern).withArgs('').to.throwException(/not a valid kibana namespaced template name/);
     });
 
   });
 
-  describe('patternToTemplate', function () {
+  describe('patternToIngest', function () {
 
     it('should convert an index pattern\'s title to its matching index template\'s name', function () {
-      expect(patternToTemplate('logstash-*')).to.be('kibana-logstash-*');
+      expect(patternToIngest('logstash-*')).to.be('kibana-logstash-*');
     });
 
     it('should throw an error if the pattern is empty', function () {
-      expect(patternToTemplate).withArgs('').to.throwException(/pattern must not be empty/);
+      expect(patternToIngest).withArgs('').to.throwException(/pattern must not be empty/);
     });
 
   });

src/plugins/kibana/server/lib/convert_pattern_and_template_name.js → src/plugins/kibana/server/lib/convert_pattern_and_ingest_name.js

@@ -4,15 +4,15 @@
 // This module provides utility functions for easily converting between template and pattern names.
 
 module.exports = {
-  templateToPattern: (templateName) => {
+  ingestToPattern: (templateName) => {
     if (templateName.indexOf('kibana-') === -1) {
       throw new Error('not a valid kibana namespaced template name');
     }
 
     return templateName.slice(templateName.indexOf('-') + 1);
   },
 
-  patternToTemplate: (patternName) => {
+  patternToIngest: (patternName) => {
     if (patternName === '') {
       throw new Error('pattern must not be empty');
     }

src/plugins/kibana/server/lib/handle_es_error.js

@@ -18,7 +18,7 @@ module.exports = function handleESError(error) {
     return Boom.forbidden(error);
   } else if (error instanceof esErrors.NotFound) {
     return Boom.notFound(error);
-  } else if (error instanceof esErrors.BadRequest || error instanceof TypeError) {
+  } else if (error instanceof esErrors.BadRequest) {
     return Boom.badRequest(error);
   } else {
     return error;

src/plugins/kibana/server/lib/schemas/resources/ingest_config_schema.js

@@ -0,0 +1,8 @@
+const Joi = require('joi');
+const indexPatternSchema = require('./index_pattern_schema');
+const pipelineSchema = require('./pipeline_schema');
+
+module.exports = Joi.object({
+  index_pattern: indexPatternSchema.required(),
+  pipeline: pipelineSchema.required()
+});

src/plugins/kibana/server/lib/schemas/resources/pipeline_schema.js

@@ -0,0 +1,3 @@
+const Joi = require('joi');
+
+module.exports = Joi.array().items(Joi.object());

src/plugins/kibana/server/routes/api/ingest/register_delete.js

@@ -1,6 +1,6 @@
 const Promise = require('bluebird');
 const handleESError = require('../../../lib/handle_es_error');
-const {templateToPattern, patternToTemplate} = require('../../../lib/convert_pattern_and_template_name');
+const {ingestToPattern, patternToIngest} = require('../../../lib/convert_pattern_and_ingest_name');
 
 module.exports = function registerDelete(server) {
   server.route({
@@ -16,7 +16,11 @@ module.exports = function registerDelete(server) {
 
       Promise.all([
         callWithRequest(req, 'delete', deletePatternParams),
-        callWithRequest(req, 'indices.deleteTemplate', {name: patternToTemplate(req.params.id), ignore: [404]})
+        callWithRequest(req, 'indices.deleteTemplate', {name: patternToIngest(req.params.id), ignore: [404]}),
+        callWithRequest(req, 'transport.request', {
+          path: `_ingest/pipeline/${patternToIngest(req.params.id)}`,
+          method: 'DELETE'
+        })
       ])
       .then(
         function (pattern) {

src/plugins/kibana/server/routes/api/ingest/register_post.js

@@ -1,26 +1,72 @@
 const Boom = require('boom');
 const _ = require('lodash');
-const {templateToPattern, patternToTemplate} = require('../../../lib/convert_pattern_and_template_name');
-const indexPatternSchema = require('../../../lib/schemas/resources/index_pattern_schema');
+const {ingestToPattern, patternToIngest} = require('../../../lib/convert_pattern_and_ingest_name');
+const ingestConfigSchema = require('../../../lib/schemas/resources/ingest_config_schema');
 const handleESError = require('../../../lib/handle_es_error');
 const { keysToCamelCaseShallow } = require('../../../lib/case_conversion');
 const createMappingsFromPatternFields = require('../../../lib/create_mappings_from_pattern_fields');
 const initDefaultFieldProps = require('../../../lib/init_default_field_props');
 
+function patternRollback(rootError, indexPatternId, boundCallWithRequest) {
+  const deleteParams = {
+    index: '.kibana',
+    type: 'index-pattern',
+    id: indexPatternId
+  };
+
+  return boundCallWithRequest('delete', deleteParams)
+  .then(
+    () => {
+      throw rootError;
+    },
+    (patternDeletionError) => {
+      throw new Error(
+        `index-pattern ${indexPatternId} created successfully but index template or pipeline
+                creation failed. Failed to rollback index-pattern creation, must delete manually.
+                ${patternDeletionError.toString()}
+                ${rootError.toString()}`
+      );
+    }
+  );
+}
+
+function templateRollback(rootError, templateName, boundCallWithRequest) {
+  const deleteParams = {
+    name: templateName
+  };
+
+  return boundCallWithRequest('indices.deleteTemplate', deleteParams)
+    .then(
+      () => {
+        throw rootError;
+      },
+      (templateDeletionError) => {
+        throw new Error(
+          `index template ${templateName} created successfully but pipeline
+                creation failed. Failed to rollback template creation, must delete manually.
+                ${templateDeletionError.toString()}
+                ${rootError.toString()}`
+        );
+      }
+    );
+}
+
+
 module.exports = function registerPost(server) {
   server.route({
     path: '/api/kibana/ingest',
     method: 'POST',
     config: {
       validate: {
-        payload: indexPatternSchema
+        payload: ingestConfigSchema
       }
     },
     handler: function (req, reply) {
-      const callWithRequest = server.plugins.elasticsearch.callWithRequest;
+      const boundCallWithRequest = _.partial(server.plugins.elasticsearch.callWithRequest, req);
       const requestDocument = _.cloneDeep(req.payload);
-      const indexPatternId = requestDocument.id;
-      const indexPattern = keysToCamelCaseShallow(requestDocument);
+      const indexPattern = keysToCamelCaseShallow(requestDocument.index_pattern);
+      const indexPatternId = indexPattern.id;
+      const ingestConfigName = patternToIngest(indexPatternId);
       delete indexPattern.id;
 
       const mappings = createMappingsFromPatternFields(indexPattern.fields);
@@ -29,70 +75,69 @@ module.exports = function registerPost(server) {
       indexPattern.fields = JSON.stringify(indexPattern.fields);
       indexPattern.fieldFormatMap = JSON.stringify(indexPattern.fieldFormatMap);
 
-      return callWithRequest(req, 'indices.exists', {index: indexPatternId})
-      .then((matchingIndices) => {
-        if (matchingIndices) {
-          throw Boom.conflict('Cannot create an index pattern via this API if existing indices already match the pattern');
-        }
+      const pipeline = {
+        processors: requestDocument.pipeline
+      };
 
-        const patternCreateParams = {
-          index: '.kibana',
-          type: 'index-pattern',
-          id: indexPatternId,
-          body: indexPattern
-        };
+      // Set up call with request params
+      const patternCreateParams = {
+        index: '.kibana',
+        type: 'index-pattern',
+        id: indexPatternId,
+        body: indexPattern
+      };
 
-        return callWithRequest(req, 'create', patternCreateParams)
-        .then((patternResponse) => {
-          const templateParams = {
-            order: 0,
-            create: true,
-            name: patternToTemplate(indexPatternId),
-            body: {
-              template: indexPatternId,
-              mappings: {
-                _default_: {
-                  dynamic_templates: [{
-                    string_fields: {
-                      match: '*',
-                      match_mapping_type: 'string',
-                      mapping: {
-                        type: 'string',
-                        index: 'analyzed',
-                        omit_norms: true,
-                        fielddata: {format: 'disabled'},
-                        fields: {
-                          raw: {type: 'string', index: 'not_analyzed', doc_values: true, ignore_above: 256}
-                        }
-                      }
+      const templateParams = {
+        order: 0,
+        create: true,
+        name: ingestConfigName,
+        body: {
+          template: indexPatternId,
+          mappings: {
+            _default_: {
+              dynamic_templates: [{
+                string_fields: {
+                  match: '*',
+                  match_mapping_type: 'string',
+                  mapping: {
+                    type: 'string',
+                    index: 'analyzed',
+                    omit_norms: true,
+                    fielddata: {format: 'disabled'},
+                    fields: {
+                      raw: {type: 'string', index: 'not_analyzed', doc_values: true, ignore_above: 256}
                     }
-                  }],
-                  properties: mappings
+                  }
                 }
-              }
+              }],
+              properties: mappings
             }
-          };
+          }
+        }
+      };
 
-          return callWithRequest(req, 'indices.putTemplate', templateParams)
-          .catch((templateError) => {
-            const deleteParams = {
-              index: '.kibana',
-              type: 'index-pattern',
-              id: indexPatternId
-            };
+      const pipelineParams = {
+        path: `_ingest/pipeline/${ingestConfigName}`,
+        method: 'PUT',
+        body: pipeline
+      };
 
-            return callWithRequest(req, 'delete', deleteParams)
-            .then(() => {
-              throw templateError;
-            }, (patternDeletionError) => {
-              throw new Error(
-                `index-pattern ${indexPatternId} created successfully but index template
-                creation failed. Failed to rollback index-pattern creation, must delete manually.
-                ${patternDeletionError.toString()}
-                ${templateError.toString()}`
-              );
-            });
-          });
+
+      return boundCallWithRequest('indices.exists', {index: indexPatternId})
+      .then((matchingIndices) => {
+        if (matchingIndices) {
+          throw Boom.conflict('Cannot create an index pattern via this API if existing indices already match the pattern');
+        }
+
+        return boundCallWithRequest('create', patternCreateParams)
+        .then(() => {
+          return boundCallWithRequest('indices.putTemplate', templateParams)
+          .catch((templateError) => {return patternRollback(templateError, indexPatternId, boundCallWithRequest);});
+        })
+        .then(() => {
+          return boundCallWithRequest('transport.request', pipelineParams)
+          .catch((pipelineError) => {return templateRollback(pipelineError, ingestConfigName, boundCallWithRequest);})
+          .catch((templateRollbackError) => {return patternRollback(templateRollbackError, indexPatternId, boundCallWithRequest);});
         });
       })
       .then(

test/unit/api/ingest/_post.js

@@ -24,20 +24,20 @@ define(function (require) {
             .expect(400),
 
           request.post('/kibana/ingest')
-            .send(_.set(createTestData(), 'title', false))
+            .send(_.set(createTestData(), 'index_pattern.title', false))
             .expect(400),
 
           request.post('/kibana/ingest')
-            .send(_.set(createTestData(), 'fields', {}))
+            .send(_.set(createTestData(), 'index_pattern.fields', {}))
             .expect(400),
 
           request.post('/kibana/ingest')
-            .send(_.set(createTestData(), 'fields', []))
+            .send(_.set(createTestData(), 'index_pattern.fields', []))
             .expect(400),
 
           // Fields must have a name and type
           request.post('/kibana/ingest')
-            .send(_.set(createTestData(), 'fields', [{count: 0}]))
+            .send(_.set(createTestData(), 'index_pattern.fields', [{count: 0}]))
             .expect(400)
         ]);
       });
@@ -144,22 +144,22 @@ define(function (require) {
 
       bdd.it('should return 409 conflict when the pattern matches existing indices',
         function existingIndicesConflict() {
-          var pattern = createTestData();
-          pattern.id = pattern.title = '.kib*';
+          var ingestConfig = createTestData();
+          ingestConfig.index_pattern.id = ingestConfig.index_pattern.title = '.kib*';
 
           return request.post('/kibana/ingest')
-            .send(pattern)
+            .send(ingestConfig)
             .expect(409);
         });
 
       bdd.it('should enforce snake_case in the request body', function () {
-        var pattern = createTestData();
-        pattern = _.mapKeys(pattern, function (value, key) {
+        var ingestConfig = createTestData();
+        ingestConfig.index_pattern = _.mapKeys(ingestConfig.index_pattern, function (value, key) {
           return _.camelCase(key);
         });
 
         return request.post('/kibana/ingest')
-          .send(pattern)
+          .send(ingestConfig)
           .expect(400);
       });