support modern and legacy elasticsearch clients

configMap.js

@@ -0,0 +1,42 @@
+const _ = require('lodash');
+
+/**
+ * This module is responsible for mapping betwen the legacy `npm elasticsearch`
+ * config format and the modern `npm @elastic/elasticsearch` format.
+ * 
+ * legacy: https://www.elastic.co/guide/en/elasticsearch/client/elasticsearch-js/16.x/config-options.html
+ * modern: https://www.elastic.co/docs/reference/elasticsearch/clients/javascript/basic-config#maxretries
+ */
+
+const modernToLegacy = (modern) => {
+  const legacy = {};
+
+  // modern config allows for *either* the key 'node' (string) or 'nodes' ([]string)
+  const nodes = _.has(modern, 'nodes') ? _.get(modern, 'nodes') : [_.get(modern, 'node', {})];
+  _.set(legacy, 'hosts', nodes);
+
+  // map basic configuration options
+  _.set(legacy, 'maxRetries', _.get(modern, 'maxRetries', undefined));
+  _.set(legacy, 'requestTimeout', _.get(modern, 'requestTimeout', undefined));
+
+  return legacy;
+};
+
+const legacyToModern = (legacy) => {
+  const modern = {};
+
+  // legacy config allows for *either* the key 'host' (string) or 'hosts' ([]string|[]Object)
+  const hosts = _.has(legacy, 'hosts') ? _.get(legacy, 'hosts') : [_.get(legacy, 'host', {})];
+  _.set(modern, 'nodes', hosts.map(host => {
+    if (_.isObject(host)) { host = `${host.protocol}://${host.host}:${host.port}`; }
+    return host;
+  }));
+
+  // map basic configuration options
+  _.set(modern, 'maxRetries', _.get(legacy, 'maxRetries', undefined));
+  _.set(modern, 'requestTimeout', _.get(legacy, 'requestTimeout', undefined));
+
+  return modern;
+};
+
+module.exports = { modernToLegacy, legacyToModern };

index.js

@@ -2,4 +2,13 @@ if (process.env.NODE_ENV !== 'test') {
   require('./src/configValidation').validate(require('pelias-config').generate());
 }
 
-module.exports = require('./src/sink');
+module.exports = {
+  v1: {
+    client: require('./src/client'),
+    createWriteStream: require('./src/sink')
+  },
+  v2: {
+    client: require('./v2/client'),
+    createWriteStream: require('./v2/sink')
+  }
+};

package.json

@@ -36,8 +36,10 @@
     "tape": "^5.0.0"
   },
   "dependencies": {
+    "@elastic/elasticsearch": "^9.0.2",
     "@hapi/joi": "^16.0.0",
     "elasticsearch": "^16.0.0",
+    "lodash": "^4.17.21",
     "pelias-config": "^6.0.0",
     "pelias-logger": "^1.2.1",
     "through2": "^3.0.0"

src/client.js

@@ -1,6 +1,20 @@
+const _ = require('lodash');
 const elasticsearch = require('elasticsearch');
-const settings = require('pelias-config').generate();
+const config = require('pelias-config').generate();
+const { modernToLegacy } = require('../configMap');
 
 module.exports = function(){
-  return new elasticsearch.Client( settings.esclient || {} );
+
+  // use legacy config
+  if (_.has(config, 'esclient')) {
+    return new elasticsearch.Client(config.esclient);
+  }
+
+  // check for modern config
+  if (_.has(config, 'elasticsearch.client')) {
+    return new elasticsearch.Client(modernToLegacy(config.elasticsearch.client));
+  }
+
+  // no config specified
+  return new elasticsearch.Client({});
 };

test/index.js

@@ -6,13 +6,15 @@ module.exports.tests = {};
 
 module.exports.tests.interface = function(test) {
   test('configValidation not throwing error should return a function', function(t) {
-    const factory = proxyquire('../index', {
+    const index = proxyquire('../index', {
       './src/configValidation': {
         validate: () => {}
       }
     });
 
-    t.equal(typeof factory, 'function', 'stream factory');
+    t.equal(typeof index, 'object', 'exports');
+    t.equal(typeof index.v1.client, 'function', 'client factory');
+    t.equal(typeof index.v1.createWriteStream, 'function', 'stream factory');
     t.end();
 
   });
@@ -25,7 +27,7 @@ module.exports.tests.interface = function(test) {
       './src/configValidation': {
         validate: () => {}
       }
-    })(opts);
+    }).v1.createWriteStream(opts);
 
     t.equal(typeof stream, 'object', 'valid stream');
     t.equal(typeof stream._read, 'function', 'valid readable');
@@ -49,8 +51,7 @@ module.exports.tests.invalidConfig = function(test) {
             throw Error('config is not valid');
           }
         }
-      });
-
+      }).v1.createWriteStream();
     }, /config is not valid/);
 
     process.env.NODE_ENV = env;

test/stream.js

@@ -10,7 +10,7 @@ module.exports.tests.functional_example = function(test, common) {
     './src/configValidation': {
       validate: () => {}
     }
-  });
+  }).v1.createWriteStream;
 
   test('functional example', function(t) {
 

v2/client.js

@@ -0,0 +1,20 @@
+const _ = require('lodash');
+const { Client } = require('@elastic/elasticsearch');
+const config = require('pelias-config').generate();
+const { legacyToModern } = require('../configMap');
+
+module.exports = function(){
+
+  // use modern config
+  if (_.has(config, 'elasticsearch.client')) {
+    return new Client(config.elasticsearch.client);
+  }
+
+  // check for legacy config
+  if (_.has(config, 'esclient')) {
+    return new Client(legacyToModern(config.esclient));
+  }
+
+  // no config specified
+  return new Client({});
+};

v2/sink.js

@@ -0,0 +1,18 @@
+const through = require('through2');
+
+function streamFactory( opts ){
+  opts = opts || {};
+  if( !opts.client ){ opts.client = require('./client')(); }
+
+  // passthrough stream
+  const stream = through.obj();
+
+  stream.bulk = opts.client.helpers.bulk({
+    datasource: stream,
+    onDocument (doc) { return doc; }
+  });
+
+  return stream;
+}
+
+module.exports = streamFactory;