- Add mapper to courier

- Add errors for mapper
- Add mapper to resolve and cache field mappings
- Add a couple tests for mapper
- Add some debug output to the kibana controller

.jshintrc

@@ -9,7 +9,7 @@
     "console": true
   },
 
-  "camelcase": true,
+  "camelcase": false,
   "white": true,
   "bitwise": false,
   "eqnull": true,

src/courier/courier.js

@@ -10,6 +10,8 @@ define(function (require) {
   var SearchSource = require('courier/data_source/search');
   var HastyRefresh = require('courier/errors').HastyRefresh;
 
+  var Mapper = require('courier/mapper.js');
+
   // map constructors to type keywords
   var sourceTypes = {
     doc: DocSource,
@@ -53,7 +55,8 @@ define(function (require) {
   // default config values
   var defaults = {
     fetchInterval: 30000,
-    docInterval: 2500
+    docInterval: 2500,
+    internalIndex: 'kibana4-int'
   };
 
   /**
@@ -99,6 +102,9 @@ define(function (require) {
     // interval hook/fn for each type
     this._onInterval = {};
 
+    // make the mapper accessable
+    this._mapper = new Mapper(this);
+
     _.each(sourceTypes, function (fn, type) {
       var courier = this;
       // the name used outside of this module

src/courier/data_source/doc.js

@@ -41,7 +41,8 @@ define(function (require) {
         var ref = allRefs[i];
         var source = ref.source;
 
-        if (resp.error) return this._error(resp);
+        //if (resp.error) return this._error(resp);
+        if (resp.error) return false;
         if (ref.version === resp._version) return; // no change
         ref.version = resp._version;
         source._storeVersion(resp._version);

src/courier/errors.js

@@ -9,4 +9,23 @@ define(function (require, module, exports) {
   HastyRefresh.prototype = new Error();
   HastyRefresh.prototype.constructor = HastyRefresh;
   exports.HastyRefresh = HastyRefresh;
+
+  // a non-critical cache write to elasticseach failed
+  function CacheWriteFailure() {
+    this.name = 'CacheWriteFailure';
+    this.message = 'A Elasticsearch cache write has failed.';
+  }
+  CacheWriteFailure.prototype = new Error();
+  CacheWriteFailure.prototype.constructor = CacheWriteFailure;
+  exports.CacheWriteFailure = CacheWriteFailure;
+
+  // when a field mapping is requested for an unknown field
+  function FieldNotFoundInCache(name) {
+    this.name = 'FieldNotFoundInCache';
+    this.message = 'The ' + name + ' field was not found in the cached mappings';
+  }
+  FieldNotFoundInCache.prototype = new Error();
+  FieldNotFoundInCache.prototype.constructor = FieldNotFoundInCache;
+  exports.FieldNotFoundInCache = FieldNotFoundInCache;
+
 });

src/courier/mapper.js

@@ -1,4 +1,7 @@
 define(function (require) {
+  var _ = require('lodash');
+  var Error = require('courier/errors');
+
   /**
    * - Resolves index patterns
    * - Fetches mappings from elasticsearch
@@ -6,21 +9,206 @@ define(function (require) {
    *
    * @class Mapper
    */
-  function Mapper(client) {
+  function Mapper(courier) {
+
+    var client = courier._getClient();
+
+    // Exclude anything wirh empty mapping except these
+    var reservedFields = {
+      '_id': { type: 'string' },
+      '_type': { type: 'string' },
+      '_index': { type: 'string' }
+    };
+
+    // Save a reference to this
+    var self = this;
+
+    // STUB Until we have another way to get the config object.
+    var config = {
+      index: 'kibana4-int'
+    };
+
+    // Store mappings we've already loaded from Elasticsearch
+    var mappings = {};
 
     /**
      * Gets an object containing all fields with their mappings
-     * @param {dataSource} [dataSource]
-     * @param {Function} [callback] A function to be executed with the results.
-     * @param {String} [type]
-     * @return {Object} A hash containing fields and their related mapping
+     * @param {dataSource} dataSource
+     * @param {Function} callback A function to be executed with the results.
      */
-    this.getFields = function (dataSource, callback, type) {
-      client.indices.getFieldMapping({index: dataSource.index}, callback);
+    this.getFields = function (dataSource, callback) {
+      if (self.getFieldsFromObject(dataSource)) {
+        // If we already have the fields in our object, use that.
+        setTimeout(callback(undefined, self.getFieldsFromObject(dataSource)), 0);
+      } else {
+        // Otherwise, try to get fields from Elasticsearch cache
+        self.getFieldsFromCache(dataSource, function (err, fields) {
+          if (err) {
+            // If we are unable to get the fields from cache, get them from mapping
+            self.getFieldsFromMapping(dataSource, function (err, fields) {
+              if (err) return courier._error(err);
+
+              // And then cache them
+              cacheFieldsToElasticsearch(config, dataSource._state.index, fields, function (err, response) {
+                if (err) return courier._error(new Error.CacheWriteError());
+              });
+
+              cacheFieldsToObject(dataSource, fields);
+              callback(err, fields);
+            });
+          } else {
+            cacheFieldsToObject(dataSource, fields);
+            callback(err, fields);
+          }
+        });
+      }
     };
 
-    this.getFieldType = function (dataSource, field, type) {
-      return field, type;
+    /**
+     * Gets an object containing the mapping for a field
+     * @param {dataSource} dataSource
+     * @param {String} field The dot notated name of a field to get the mapping for
+     * @param {Function} callback A function to be executed with the results.
+     */
+    this.getFieldMapping = function (dataSource, field, callback) {
+      self.getFields(dataSource, function (err, fields) {
+        if (_.isUndefined(fields[field])) return courier._error(new Error.FieldNotFoundInCache());
+        callback(err, fields[field]);
+      });
+    };
+
+    /**
+     * Gets an object containing the mapping for a field
+     * @param {dataSource} dataSource
+     * @param {Array} fields The dot notated names of a fields to get the mapping for
+     * @param {Function} callback A function to be executed with the results.
+     */
+    this.getFieldsMapping = function (dataSource, fields, callback) {
+      self.getFields(dataSource, function (err, fields) {
+        var _mapping = _.object(_.map(fields, function (field) {
+          if (_.isUndefined(fields[field])) return courier._error(new Error.FieldNotFoundInCache());
+          return [field, fields[field]];
+        }));
+        callback(err, _mapping);
+      });
+    };
+
+    /**
+     * Gets an object containing all fields with their mappings from kibana's cache in Elasticsearch
+     * @param {dataSource} dataSource
+     * @return {Object} An object containing fields with their mappings, or false if not found.
+     */
+    this.getFieldsFromObject = function (dataSource) {
+      return !_.isUndefined(mappings[dataSource._state.index]) ? mappings[dataSource._state.index] : false;
+    };
+
+    /**
+     * Gets an object containing all fields with their mappings from kibana's cache in Elasticsearch
+     * @param {dataSource} dataSource
+     * @param {Function} callback A function to be executed with the results.
+     */
+    this.getFieldsFromCache = function (dataSource, callback) {
+      var params = {
+        index: config.index,
+        type: 'mapping',
+        id: dataSource._state.index,
+      };
+
+      client.getSource(params, callback);
+    };
+
+    /**
+     * Gets an object containing all fields with their mappings directly from Elasticsearch
+     * @param {dataSource} dataSource
+     * @param {Function} callback A function to be executed with the results.
+     */
+    this.getFieldsFromMapping = function (dataSource, callback) {
+
+
+      var params = {
+        // TODO: Change index to be newest resolved index. Change _state.index to id().
+        index: dataSource._state.index,
+        field: '*',
+      };
+
+      // TODO: Add week/month check
+      client.indices.getFieldMapping(params, function (err, response, status) {
+        // TODO: Add error message
+
+        var fields = {};
+
+        _.each(response, function (index) {
+          _.each(index.mappings, function (type) {
+            _.each(type, function (field, name) {
+              if (_.isUndefined(field.mapping) || name[0] === '_') return;
+              fields[name] = field.mapping[_.keys(field.mapping)[0]];
+            });
+          });
+        });
+
+        // TODO if these are mapped differently this might cause problems
+        _.assign(fields, reservedFields);
+
+        callback(err, fields);
+      });
+    };
+
+    /**
+     * Stores processed mappings in Elasticsearch
+     * @param {dataSource} dataSource
+     * @param {Function} callback A function to be executed with the results.
+     */
+    var cacheFieldsToElasticsearch = function (config, index, fields, callback) {
+      client.index({
+        index : config.index,
+        type: 'mapping',
+        id : index,
+        body : fields
+      }, callback);
+    };
+
+    /**
+     * Stores processed mappings in an object
+     * @param {dataSource} dataSource
+     * @param {Function} callback A function to be executed with the results.
+     */
+    var cacheFieldsToObject = function (dataSource, fields) {
+      mappings[dataSource._state.index] = _.clone(fields);
+      return !_.isUndefined(mappings[dataSource._state.index]) ? true : false;
+    };
+
+    /**
+     * Clears mapping caches from elasticsearch and from local object
+     * @param {dataSource} dataSource
+     * @param {Function} callback A function to be executed with the results.
+     */
+    var clearCache = function (dataSource, callback) {
+      if (!_.isUndefined(mappings[dataSource._state.index])) {
+        delete mappings[dataSource._state.index];
+      }
+      client.delete({
+        index : config.index,
+        type: 'mapping',
+        id : dataSource._state.index
+      }, callback);
+    };
+
+
+    /**
+     * Sets a number of fields to be ignored in the mapping
+     * @param {dataSource} dataSource
+     * @param {Array} fields An array of fields to be ignored
+     * @param {Function} callback A function to be executed with the results.
+     */
+    var ignoreFields = function (dataSource, fields, callback) {
+      if (!_.isArray(fields)) fields = [fields];
+      var ignore = _.object(_.map(fields, function (field) {
+        return [field, {type: 'ignore'}];
+      }));
+      self.getFields(dataSource, function (err, mapping) {
+        _.assign(mapping, ignore);
+        callback(err, mapping);
+      });
     };
 
   }

src/kibana/controllers/kibana.js

@@ -17,13 +17,17 @@ define(function (require) {
       return {
         restrict: 'E',
         scope: {
-          type: '@'
+          type: '@',
+          fields: '@'
         },
-        template: '<strong style="float:left">{{count}} :&nbsp;</strong><pre>{{json}}</pre>',
+        template: 'Mappings:<br><div ng-repeat="(name,mapping) in mappedFields">{{name}} = {{mapping.type}}</div><hr>' +
+          '<strong style="float:left">{{count}} :&nbsp;</strong><pre>{{json}}</pre>',
         controller: function ($rootScope, $scope, courier) {
           $scope.count = 0;
+          $scope.mappedFields = {};
 
           var source = $rootScope.dataSource.extend()
+            .index('logstash-2014.02.14')
             .type($scope.type)
             .source({
               include: 'country'
@@ -33,8 +37,18 @@ define(function (require) {
               $scope.json = JSON.stringify(resp.hits, null, '  ');
             });
 
-          courier.mapper.getFields($rootScope.dataSource, function (data) {
-            $scope.json = data;
+          var fields = $scope.fields.split(',');
+
+
+          _.each(fields, function (field) {
+            courier._mapper.getFieldMapping(source, field, function (err, mapping) {
+              $scope.mappedFields[field] = mapping;
+            });
+          });
+
+
+          courier._mapper.getFields(source, function (err, response, status) {
+            console.log(response);
           });
 
           $scope.$watch('type', source.type);
@@ -49,10 +63,13 @@ define(function (require) {
           type: '@',
           index: '@'
         },
-        template: '<strong style="float:left">{{count}} : <button ng-click="click()">reindex</button> :&nbsp;</strong><pre>{{json}}</pre>',
+        template: '<strong style="float:left">{{count}} : <button ng-click="click()">reindex</button> :&nbsp;</strong>' +
+          '<pre>{{json}} BEER</pre>',
         controller: function (courier, $scope) {
           $scope.count = 0;
 
+          console.log(courier);
+
           var currentSource;
           $scope.click = function () {
             if (currentSource) {
@@ -69,6 +86,7 @@ define(function (require) {
               $scope.count ++;
               $scope.json = JSON.stringify(doc, null, '  ');
             });
+
         }
       };
     });

test/unit/index.html

@@ -36,8 +36,8 @@
       }
     })
     require([
-      '/specs/courier.js',
-      '/specs/data_source.js',
+      //'/specs/courier.js',
+      //'/specs/data_source.js',
       '/specs/mapper.js'
     ], function () {
       window.mochaRunner = mocha.run().on('end', function(){

test/unit/specs/mapper.js

@@ -1,45 +1,44 @@
 define(function (require) {
   var elasticsearch = require('../bower_components/elasticsearch/elasticsearch.js');
   var _ = require('lodash');
+  var sinon = require('sinon/sinon');
   var Courier = require('courier/courier');
   var DataSource = require('courier/data_source/data_source');
   var Mapper = require('courier/mapper');
+
   var client = new elasticsearch.Client({
     host: 'localhost:9200',
   });
 
-  describe('Mapper Module', function () {
+  var courier = new Courier({
+    client: client
+  });
+
+  describe('Mapper Module', function () {
+    var server, source, mapper;
+
+    beforeEach(function() {
+      source = courier.createSource('search')
+        .index('logs*')
+        .size(5);
+      mapper = new Mapper(courier);
 
-    it('provides a constructor for the Mapper class', function () {
-      var mapper = new Mapper(client);
-      expect(mapper).to.be.a(Mapper);
     });
 
-    it('has a function called getFields that returns an object', function () {
-      /*
-      var courier = new Courier({
-        client: client
-      });
+    afterEach(function () {
+    });
 
-      var dataSource = courier.createSource('search')
-        .index('_all')
-        .size(5);
+    it('provides a constructor for the Mapper class', function (done) {
+      expect(mapper).to.be.a(Mapper);
+      done();
+    });
 
-      var mapper = new Mapper(client);
-
-      var callback = function(data) {
-        console.log(data);
-      };
-
-      expect(mapper.getFields(dataSource,callback)).to.eql({
-        foo: {
-          type: 'string'
-        },
-        "foo.bar": {
-          type: 'long'
-        }
-      });
-      */
+    it('has a function called getFieldsFromMapping that calls client.indices.getFieldMapping', function (done) {
+      sinon.spy(client.indices, 'getFieldMapping');
+      mapper.getFieldsFromMapping(source,function(){});
+      expect(client.indices.getFieldMapping.called).to.be(true);
+      client.indices.getFieldMapping.restore();
+      done();
     });