Swap Q to bluebird for promises, allow skipping indexing by resolving with 'undefined' and fix error

Author: jwarkentin

bin/elasticsearch-reindex.js

@@ -19,11 +19,13 @@ cli
 .option('-b, --bulk [value]', 'bulk size for a thread', 100)
 .option('-q, --query_size [value]', 'query size for scroll', 100)
 .option('-s, --scroll [value]', 'default 1m', '1m')
+.option('-i, --sniff_cluster [value]', 'sniff the rest of the cluster upon initial connection and connection errors', true)
 .option('-o, --request_timeout [value]', 'default 60000', 60000)
 .option('-l, --log_path [value]', 'default ./reindex.log', './reindex.log')
 .option('-r, --trace', 'default false', false)
 .option('-n, --max_docs [value]', 'default -1 unlimited', -1)
-.option('-v, --api_ver [value]', 'default 1.5', '1.5')
+.option('--from_ver [value]', 'default 1.5', '1.5')
+.option('--to_ver [value]', 'default 1.5', '1.5')
 .option('-p, --parent [value]', 'if set, uses this field as parent field', '')
 .option('-m, --promise [value]', 'if set indexes expecting promises, default: false', false)
 .option('-z, --compress [value]', 'if set, requests compression of data in transit', false)
@@ -32,6 +34,16 @@ cli
 .option('-e, --region [value]', 'AWS region', false)
 .parse(process.argv);
 
+for (var key in cli) {
+  if (cli.hasOwnProperty(key)) {
+    if (cli[key] === 'false') {
+      cli[key] = false;
+    } else if (cli[key] === 'true') {
+      cli[key] = true;
+    }
+  }
+}
+
 var logger = bunyan.createLogger({
   src: true,
   name: "elasticsearch-reindex",
@@ -147,7 +159,7 @@ if (cluster.isMaster) {
     shard_name = cluster.worker.id;
   }
 
-  function createClient(uri) {
+  function createClient(uri, apiVersion) {
     if (!/\w+:\/\//.test(uri)) {
       uri = 'http://' + uri;
     }
@@ -162,10 +174,10 @@ if (cluster.isMaster) {
 
     var config = {
       requestTimeout: cli.request_timeout,
-      apiVersion: cli.api_ver,
+      apiVersion: apiVersion,
       suggestCompression: cli.compress,
-      sniffOnStart: true,
-      sniffOnConnectionFault: true
+      sniffOnStart: cli.sniff_cluster,
+      sniffOnConnectionFault: cli.sniff_cluster
     };
 
     if (cli.access_key && cli.secret_key && cli.region && /\.amazonaws\./.test(uri)) {
@@ -187,8 +199,8 @@ if (cluster.isMaster) {
     throw new Error('"from" and "to" parameters are required');
   }
 
-  var from = createClient(cli.from);
-      to = createClient(cli.to),
+  var from = createClient(cli.from, cli.from_ver);
+      to = createClient(cli.to, cli.to_ver),
       processed_total = 0,
       processed_failed = 0;
 
@@ -226,6 +238,9 @@ if (cluster.isMaster) {
 
   from.client.search(scan_options, function scroll_fetch(err, res) {
     if (err) {
+      if (err.message instanceof Error) {
+        err = err.message;
+      }
       logger.fatal(err);
       if (err.message.indexOf('parse') > -1) {
         throw new Error("Scroll body parsing error, query_size param is possibly too high.");

lib/indexer.js

@@ -4,7 +4,7 @@ var async         = require('async'),
     _             = require('underscore'),
     EventEmitter  = require('events').EventEmitter,
     inherits      = require('util').inherits,
-    Q             = require('q');
+    Promise       = require('bluebird');
 
 function Indexer () {
   EventEmitter.call(this);
@@ -90,16 +90,20 @@ Indexer.prototype.indexPromise = function(docs, options, cb) {
     var sent = false;
 
     // map each indexer to return a promise
-    return Q.all(chunk.map(function(item) {
-      return Q.when(options.indexer(item, options, options.client));
+    return Promise.all(chunk.map(function(item) {
+      return options.indexer(item, options, options.client);
     })).catch(function (err) {
       self.emit('error', err);
       throw cb(sent = err);
     })
 
     // Once all of the data resolves index it
     .then(function (bulk_data) {
-      bulk_data = _.flatten(bulk_data);
+      bulk_data = _.flatten(bulk_data).filter(function(x) { return x; });
+
+      if (!bulk_data.length) {
+        return {};
+      }
 
       return options.client.bulk({
         body: bulk_data

package.json

@@ -26,15 +26,15 @@
   "homepage": "https://github.com/garbin/elasticsearch-reindex",
   "dependencies": {
     "async": "^1.5.0",
+    "bluebird": "^3.3.4",
     "bunyan": "^1.2.0",
     "commander": "^2.4.0",
-    "elasticsearch": "^9.0.2",
+    "elasticsearch": "^11.0.0",
     "event-emitter": "^0.3.1",
     "http-aws-es": "^1.1.3",
     "moment": "^2.8.3",
     "pace": "0.0.4",
     "progress": "^1.1.8",
-    "q": "^1.4.1",
     "underscore": "^1.7.0"
   }
 }