merged in mapper changes

.jshintrc

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

src/courier/courier.js

@@ -11,6 +11,8 @@ define(function (require) {
   var HastyRefresh = require('courier/errors').HastyRefresh;
   var nextTick = require('utils/next_tick');
 
+  var Mapper = require('courier/mapper');
+
   // map constructors to type keywords
   var sourceTypes = {
     doc: DocSource,
@@ -53,7 +55,9 @@ define(function (require) {
   // default config values
   var defaults = {
     fetchInterval: 30000,
-    docInterval: 1500
+    docInterval: 1500,
+    internalIndex: 'kibana4-int',
+    mapperCacheType: 'mappings'
   };
 
   /**
@@ -100,6 +104,12 @@ define(function (require) {
     // interval hook/fn for each type
     this._onInterval = {};
 
+    // make the mapper accessable
+    this._mapper = new Mapper(this, {
+      cacheIndex: config.internalIndex,
+      cacheType: config.mapperCacheType
+    });
+
     _.each(sourceTypes, function (fn, type) {
       var courier = this;
       // the name used outside of this module

src/courier/data_source/doc.js

@@ -36,27 +36,27 @@ define(function (require) {
       getBody.docs.push(source._flatten());
     });
 
-    return client.mget({ body: getBody }, function (err, resp) {
-      if (err) return cb(err);
+    return client.mget({ body: getBody })
+      .then(function (resp) {
+        _.each(resp.docs, function (resp, i) {
+          var ref = allRefs[i];
+          var source = ref.source;
 
-      _.each(resp.docs, function (resp, i) {
-        var ref = allRefs[i];
-        var source = ref.source;
+          if (resp.error) return source._error(new errors.DocFetchFailure(resp));
+          if (resp.found) {
+            if (ref.version === resp._version) return; // no change
+            ref.version = resp._version;
+            source._storeVersion(resp._version);
+          } else {
+            ref.version = void 0;
+            source._clearVersion();
+          }
+          source.emit('results', resp);
+        });
 
-        if (resp.error) return source._error(new errors.DocFetchFailure(resp));
-        if (resp.found) {
-          if (ref.version === resp._version) return; // no change
-          ref.version = resp._version;
-          source._storeVersion(resp._version);
-        } else {
-          ref.version = void 0;
-          source._clearVersion();
-        }
-        source.emit('results', resp);
-      });
-
-      cb(err, resp);
-    });
+        cb(void 0, resp);
+      })
+      .catch(cb);
   };
 
   /**
@@ -72,9 +72,7 @@ define(function (require) {
       /* jshint eqeqeq: false */
       return (!ref.fetchCount || ref.version != storedVersion);
     });
-    nextTick(function () {
-      cb(void 0, invalid);
-    });
+    nextTick(cb, void 0, invalid);
   };
 
   /*****

src/courier/errors.js

@@ -36,6 +36,7 @@ define(function (require) {
   };
   inherits(errors.HastyRefresh, CourierError);
 
+
   /**
    * DocFetchFailure Error - where there is an error getting a doc
    * @param {String} [msg] - An error message that will probably end up in a log.
@@ -49,6 +50,7 @@ define(function (require) {
   };
   inherits(errors.DocFetchFailure, CourierError);
 
+
   /**
    * Connection Error
    * @param {String} [msg] - An error message that will probably end up in a log.
@@ -62,5 +64,38 @@ define(function (require) {
   };
   inherits(errors.VersionConflict, CourierError);
 
+
+  /**
+   * there was a conflict storing a doc
+   * @param {String} field - the fields which contains the conflict
+   */
+  errors.MappingConflict = function MappingConflict(field) {
+    CourierError.call(this,
+      'Field ' + field + ' is defined as at least two different types in indices matching the pattern',
+      errors.MappingConflict);
+  };
+  inherits(errors.MappingConflict, CourierError);
+
+  /**
+   * a non-critical cache write to elasticseach failed
+   */
+  errors.CacheWriteFailure = function CacheWriteFailure() {
+    CourierError.call(this,
+      'A Elasticsearch cache write has failed.',
+      errors.CacheWriteFailure);
+  };
+  inherits(errors.CacheWriteFailure, CourierError);
+
+  /**
+   * when a field mapping is requested for an unknown field
+   * @param {String} name - the field name
+   */
+  errors.FieldNotFoundInCache = function FieldNotFoundInCache(name) {
+    CourierError.call(this,
+      'The ' + name + ' field was not found in the cached mappings',
+      errors.FieldNotFoundInCache);
+  };
+  inherits(errors.FieldNotFoundInCache, CourierError);
+
   return errors;
 });

src/courier/mapper.js

@@ -1,4 +1,8 @@
 define(function (require) {
+  var _ = require('lodash');
+  var Error = require('courier/errors');
+  var nextTick = require('utils/next_tick');
+
   /**
    * - Resolves index patterns
    * - Fetches mappings from elasticsearch
@@ -6,21 +10,202 @@ define(function (require) {
    *
    * @class Mapper
    */
-  function Mapper(client) {
+  function Mapper(courier, config) {
+
+    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;
+
+    // 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.
+        nextTick(callback, void 0, self.getFieldsFromObject(dataSource));
+      } 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(field));
+        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(field));
+          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.cacheIndex,
+        type: config.cacheType,
+        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;
+              if (!_.isUndefined(fields[name]) && fields[name] !== field.mapping[_.keys(field.mapping)[0]])
+                return courier._error(new Error.MappingConflict());
+              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.cacheIndex,
+        type: config.cacheType,
+        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.
+     */
+    this.clearCache = function (dataSource, callback) {
+      if (!_.isUndefined(mappings[dataSource._state.index])) {
+        delete mappings[dataSource._state.index];
+      }
+      client.delete({
+        index: config.cacheIndex,
+        type: config.cacheType,
+        id: dataSource._state.index
+      }, callback);
+    };
+
+
+    /**
+     * Sets a number of fields to be ignored in the mapping. Not sure this is a good idea?
+     * @param {dataSource} dataSource
+     * @param {Array} fields An array of fields to be ignored
+     * @param {Function} callback A function to be executed with the results.
+     */
+    this.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/courier/tests/directives.js

@@ -8,7 +8,7 @@ define(function (require) {
         restrict: 'E',
         template: 'My favorite number is {{favoriteNum}} <button ng-click="click()">New Favorite</button>',
         controller: function ($scope, config) {
-          config.bind($scope, 'favoriteNum', {
+          config.$bind($scope, 'favoriteNum', {
             default: 0
           });
 
@@ -69,5 +69,46 @@ define(function (require) {
             });
         }
       };
+    })
+    .directive('mappingTest', function () {
+      return {
+        restrict: 'E',
+        scope: {
+          type: '@',
+          fields: '@'
+        },
+        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 = courier.rootSearchSource.extend()
+            .index('logstash-*')
+            .type($scope.type)
+            .source({
+              include: 'geo'
+            })
+            .$scope($scope)
+            .on('results', function (resp) {
+              $scope.count ++;
+              $scope.json = JSON.stringify(resp.hits, null, '  ');
+            });
+
+          var fields = $scope.fields.split(',');
+
+          fields.forEach(function (field) {
+            courier._mapper.getFieldMapping(source, field, function (err, mapping) {
+              $scope.$apply(function () {
+                $scope.mappedFields[field] = mapping;
+              });
+            });
+          });
+
+          courier._mapper.getFields(source, function (err, response, status) {
+            console.log(response);
+          });
+        }
+      };
     });
 });

src/courier/tests/mapper.html

@@ -0,0 +1 @@
+<mapping-test type="apache" fields="extension,response,request"></mapping-test>

src/kibana/controllers/kibana.js

@@ -6,7 +6,8 @@ define(function (require) {
   require('services/config');
   require('services/courier');
 
-  angular.module('kibana/controllers')
+  angular
+    .module('kibana/controllers')
     .controller('kibana', function ($scope, courier) {
       setTimeout(function () {
         courier.start();

src/kibana/main.js

@@ -48,11 +48,14 @@ define(function (require) {
       .when('/config-test', {
         templateUrl: 'courier/tests/config.html',
       })
+      .when('/mapper-test', {
+        templateUrl: 'courier/tests/mapper.html',
+      })
       .when('/courier-test', {
         templateUrl: 'courier/tests/index.html',
       })
       .otherwise({
-        redirectTo: ''
+        redirectTo: '/'
       });
   });
 
@@ -61,8 +64,8 @@ define(function (require) {
 
     // load the elasticsearch service
     require([
+      'courier/tests/directives',
       'controllers/kibana',
-      'courier/test_directives',
       'constants/base'
     ], function () {
       // bootstrap the app

src/kibana/partials/index.html

@@ -1,4 +1,5 @@
 <ul>
   <li><a ng-href="#/courier-test">Courier Test</a></li>
   <li><a ng-href="#/config-test">Config Test</a></li>
+  <li><a ng-href="#/mapper-test">Mapper Test</a></li>
 </ul>

src/kibana/services/config.js

@@ -10,6 +10,7 @@ define(function (require) {
   // share doc and val cache between apps
   var doc;
   var vals = {};
+
   module.service('config', function ($q, $rootScope, courier, kbnVersion) {
     var watchers = {};
     var unwatchers = [];
@@ -21,8 +22,9 @@ define(function (require) {
         .id(kbnVersion);
     } else {
       // clean up after previous app
-      doc.removeAllListeners('results');
-      doc.courier(courier);
+      doc
+        .removeAllListeners('results')
+        .courier(courier);
     }
 
     doc.on('results', function (resp) {
@@ -128,9 +130,7 @@ define(function (require) {
     function _notify(fns, cur, prev) {
       if ($rootScope.$$phase) {
         // reschedule
-        nextTick(function () {
-          _notify(fns, cur, prev);
-        });
+        nextTick(_notify, fns, cur, prev);
         return;
       }
 

src/kibana/utils/async_modules.js

@@ -1,8 +1,9 @@
 define(function (require) {
   var _ = require('lodash');
 
-  // TODO: this will probably fail to work when we have multiple apps. Might need to propogate
-  // registrations to multiple providers
+  /* TODO: this will probably fail to work when we have multiple apps.
+   * Might need to propogate registrations to multiple providers
+   */
   function enable(app) {
     // keep a reference to each module defined before boot, so that
     // after boot it can define new features. Also serves as a flag.

src/kibana/utils/next_tick.js

@@ -15,13 +15,24 @@ define(function () {
         ev.stopPropagation();
         if (queue.length > 0) {
           var fn = queue.shift();
-          fn();
+          if (typeof fn === 'function') {
+            fn();
+          } else {
+            // partial args were supplied
+            var args = fn;
+            fn = args.shift();
+            fn.apply(null, args);
+          }
         }
       }
     }, true);
 
     return function nextTick(fn) {
-      queue.push(fn);
+      if (arguments.length > 1) {
+        queue.push([fn].concat([].slice.call(arguments, 1)));
+      } else {
+        queue.push(fn);
+      }
       window.postMessage('process-tick', '*');
     };
   }

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,47 +1,134 @@
 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
+  });
 
-    it('provides a constructor for the Mapper class', function () {
-      var mapper = new Mapper(client);
-      expect(mapper).to.be.a(Mapper);
-    });
+  describe('Mapper', function () {
+    var server, source, mapper;
 
-    it('has a function called getFields that returns an object', function () {
-      /*
-      var courier = new Courier({
-        client: client
-      });
-
-      var dataSource = courier.createSource('search')
-        .index('_all')
+    beforeEach(function() {
+      source = courier.createSource('search')
+        .index('valid')
         .size(5);
+      mapper = new Mapper(courier);
 
-      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'
+      // Stub out a mini mapping response.
+      sinon.stub(client.indices, 'getFieldMapping',function (params, callback) {
+        if(params.index === 'valid') {
+          setTimeout(callback(undefined,{
+            "test": {
+              "mappings": {
+                "testType": {
+                  "foo.bar": {
+                    "full_name": "foo.bar",
+                      "mapping": {
+                        "bar": {
+                          "type": "string"
+                        }}}}}}}
+          ),0);
+        } else {
+          setTimeout(callback('Error: Not Found',undefined));
         }
       });
-      */
+
+      sinon.stub(client, 'getSource', function (params, callback) {
+        if(params.id === 'valid') {
+          setTimeout(callback(undefined,{"foo.bar": {"type": "string"}}),0);
+        } else {
+          setTimeout(callback('Error: Not Found',undefined),0);
+        }
+      });
+
+      sinon.stub(client, 'delete', function (params, callback) {callback(undefined,true);});
     });
 
+    afterEach(function () {
+      client.indices.getFieldMapping.restore();
+      client.getSource.restore();
+      client.delete.restore();
+    });
+
+    it('provides a constructor for the Mapper class', function (done) {
+      var mapper = new Mapper(courier);
+      expect(mapper).to.be.a(Mapper);
+      done();
+    });
+
+    it('has getFieldsFromMapping function that returns a mapping', function (done) {
+      mapper.getFieldsFromMapping(source,function (err, mapping) {
+        expect(client.indices.getFieldMapping.called).to.be(true);
+        expect(mapping['foo.bar'].type).to.be('string');
+        done();
+      });
+    });
+
+    it('has getFieldsFromCache that returns an error for uncached indices', function (done) {
+      source = courier.createSource('search')
+        .index('invalid')
+        .size(5);
+
+      mapper.getFieldsFromCache(source,function (err, mapping) {
+        expect(client.getSource.called).to.be(true);
+        expect(err).to.be('Error: Not Found');
+        done();
+      });
+    });
+
+    it('has getFieldsFromCache that returns a mapping', function (done) {
+      mapper.getFieldsFromCache(source,function (err, mapping) {
+        expect(client.getSource.called).to.be(true);
+        expect(mapping['foo.bar'].type).to.be('string');
+        done();
+      });
+    });
+
+    it('has a getFieldsFromObject function', function (done) {
+      expect(mapper.getFieldsFromObject).to.be.a('function');
+      done();
+    });
+
+    it('has a getFields that returns a mapping from cache', function (done) {
+      mapper.getFields(source, function (err, mapping) {
+        expect(client.getSource.called).to.be(true);
+        expect(client.indices.getFieldMapping.called).to.be(false);
+        expect(mapping['foo.bar'].type).to.be('string');
+        done();
+      });
+    });
+
+    it('has getFields that throws an error for invalid indices', function (done) {
+      source = courier.createSource('search')
+        .index('invalid')
+        .size(5);
+      try {
+        mapper.getFields(source, function (err, mapping) {
+          // Should not be called
+          expect('the callback').to.be('not executed');
+          done();
+        });
+      } catch (e) {
+        expect(true).to.be(true);
+        done();
+      }
+    });
+
+    it('has a clearCache that calls client.delete', function (done) {
+      mapper.clearCache(source, function () {
+        expect(client.delete.called).to.be(true);
+        done();
+      });
+    });
 
   });