Finish 1.0.0

Author: Gaurang

.gitignore

@@ -1,3 +1,6 @@
+# Custom
+config.js
+
 # Logs
 logs
 *.log

README.md

@@ -1,2 +1,19 @@
 # elasticsearch-document-transfer
-Simple script to transfer documents between two elasticsearch servers or between indices of the same elasticsearch server.
+Simple Node.js script to transfer documents between two elasticsearch servers or between indices of the same elasticsearch server.
+
+Implementation based on [consumer-producer problem][3]. 
+Since Node.js is single-threaded, `producer` and `consumer` cannot simultaneously access the `buffer`. And hence, as long as the asynchronous nature of Node.js is handled properly, there shouldn't be any issues related to concurrency.
+
+### Usage:
+1. Clone the repo.
+2. Run `npm install` in the directory.
+3. Add `config.js` using [`config-sample.js`][1]
+4. (Optional) Set appropriate values in [`options.js`][2]
+5. Run the script
+ ```
+ node index.js
+ ```
+ 
+[1]: /config-sample.js
+[2]: /options.js
+[3]: https://en.wikipedia.org/wiki/Producer–consumer_problem

config-sample.js

@@ -0,0 +1,24 @@
+//Values that will have consequences
+
+module.exports = Object.freeze({
+//Source client: from where the documents are to be fetched
+"source": {
+"host": "localhost:9200",
+
+"index": "my_source_index",
+"type": "_doc"
+},
+
+//Target client: where documents are to be transferred to
+"target": {
+"index": "my_target_index",
+"type": "_doc",
+
+ //if options.consume.byFile is not set, this will be ignored
+"filePath": "data.json",
+
+ //if options.consume.byElastic is not set, this will be ignored
+"host": "localhost:9200",
+"importMappingFromSource": true
+}
+})

elastic.js

@@ -0,0 +1,130 @@
+const elasticsearch = require('elasticsearch');
+const fs = require('fs');
+
+const config = require('./config');
+const options = require('./options');
+
+
+class Elastic{
+ constructor(){
+ this.target = {
+ doc_count: {
+ file: 0,
+ elastic: 0
+ }
+ }
+ this.batches = 0;
+ this._scroll_id = null;
+ }
+
+ async init(){
+ //configure source client
+ this.sourceClient = new elasticsearch.Client({
+ host: config.source.host,
+ log: options.elasticsearch.log_level
+ })
+ await this._checkConnection(this.sourceClient, 'source');
+
+ //configure target client if required
+ if(options.consume.byElastic){
+ this.targetClient = new elasticsearch.Client({
+ host: config.target.host,
+ log: options.elasticsearch.log_level
+ })
+ await this._checkConnection(this.targetClient, 'target');
+
+ if(config.target.importMappingFromSource){
+ await this._transferMapping(this.sourceClient, this.targetClient);
+ }
+ }
+ }
+
+ async getDocsFromSourceElastic({batchSize = 20, query = {match_all: {}}, scrollTime = '10s'}){
+ let results;
+
+ //the first time; when scroll_id isn't generated yet
+ if(!this._scroll_id){
+ results = await this.sourceClient.search({
+ scroll: options.elasticsearch.scroll_time,
+ index: config.source.index,
+ type: config.source.type,
+ body: {
+ size: batchSize,
+ query: query
+ }
+ })
+ }
+ //rest of the times
+ else{
+ results = await this.sourceClient.scroll({
+ scrollId: this._scroll_id,
+ scroll: options.elasticsearch.scroll_time
+ })
+ }
+ this.batches++;
+ this._scroll_id = results._scroll_id;
+
+ console.log('Batches done', this.batches);
+ return results
+ }
+
+ addDocsToTargetElastic(rawHitsList){
+ let body = rawHitsList.reduce((accumulator, hit) => {
+ let index = {
+ _index: config.target.index,
+ _type: config.target.type,
+ _id: ++this.target.doc_count.elastic
+ }
+ return accumulator.concat([{index}, hit._source])
+ }, []);
+
+ //console.log(body);
+ console.log('Elastic Doc Count', this.target.doc_count.elastic);
+ return this.targetClient.bulk({body});
+ }
+
+ async writeDocsToFile(rawHitsList){
+ let body = rawHitsList.map(hits => {
+ let index = {
+ _index: config.target.index,
+ _type: config.target.type,
+ _id: ++this.target.doc_count.file
+ }
+ return this._writeLineToFile(config.target.filePath, `${JSON.stringify({index})}\n${JSON.stringify(hits._source)}`);
+ })
+
+ console.log('File Doc Count', this.target.doc_count.file);
+ return Promise.all(body); 
+ }
+
+ async _checkConnection(client, clientName = ''){
+ await client.ping({requestTimeout: 3000});
+ console.log(`${clientName} client connected successfully`);
+ }
+
+ _writeLineToFile(path, line){
+ line = line + '\n';
+ return new Promise((resolve, reject) => {
+ fs.appendFile(path, line, err => err? reject(err): resolve());
+ })
+ }
+
+ async _transferMapping(source, target){
+ let mapping = await source.indices.getMapping({
+ index: config.source.index,
+ type: config.source.type
+ })
+ console.log("Mapping: ", JSON.stringify(mapping));
+
+ if(!await target.indices.exists({index: config.target.index}))
+ await target.indices.create({index: config.target.index});
+
+ return await target.indices.putMapping({
+ index: config.target.index,
+ type: config.target.type,
+ body: mapping[config.source.index].mappings
+ });
+ }
+}
+
+module.exports = {Elastic}

index.js

@@ -0,0 +1,84 @@
+const Elastic = require('./elastic').Elastic;
+const events = require('events');
+
+const options = require('./options');
+
+const elastic = new Elastic();
+const eventEmitter = new events.EventEmitter();
+
+let buffer = [];
+let EOS = false; //End Of Scroll/Search/document Stream
+
+const EVENT_CONSUME_READY = 'BUFFER_CONSUME_READY';
+const EVENT_PRODUCE_READY = 'BUFFER_PRODUCE_READY';
+
+//to push documents to target
+eventEmitter.on(EVENT_CONSUME_READY, async () => {
+ try{
+ //while buffer is larger than the batch_size
+ //keep consuming it
+ while(buffer.length > options.consume.batch_size){
+ let batch = buffer.splice(0, options.consume.batch_size);
+
+ if(options.consume.byFile)
+ await elastic.writeDocsToFile(batch);
+ 
+ if(options.consume.byElastic)
+ await elastic.addDocsToTargetElastic(batch);
+ }
+
+ console.log('consumeSkip');
+
+ //if buffer is depleted but we know more documents are on the way,
+ //we wait for some time and continue when the buffer fills up
+ if(!EOS)
+ setTimeout(() => eventEmitter.emit(EVENT_CONSUME_READY), options.consume.timeout);
+
+ //else, we just wind up our work with the rest of the buffer
+ else{
+ if(options.consume.byFile)
+ await elastic.writeDocsToFile(buffer);
+ 
+ if(options.consume.byElastic)
+ await elastic.addDocsToTargetElastic(buffer);
+ }
+ }
+ catch(err){console.error(err)}
+});
+
+//to get documents from source elastic
+eventEmitter.on(EVENT_PRODUCE_READY, async () => {
+ let response;
+ try{
+ //till we don't overflow our buffer
+ //we keep adding documents to it
+ while(buffer.length < options.buffer_size){
+ response = await elastic.getDocsFromSourceElastic(options.produce.batch_size);
+ 
+ //once we stop getting new documents
+ //we signal that we've reached End Of Scroll (EOS)
+ if(response.hits.hits.length > 0)
+ buffer = buffer.concat(response.hits.hits);
+ else{
+ EOS = true;
+ break;
+ }
+ }
+
+ //if we overflow the buffer
+ //we wait some time and try again later
+ console.log('produceSkip');
+ if(!EOS)
+ setTimeout(() => eventEmitter.emit(EVENT_PRODUCE_READY), options.produce.timeout);
+ }
+ catch(err){console.error(err)}
+});
+
+(async function() {
+ try{
+ await elastic.init();
+ eventEmitter.emit(EVENT_PRODUCE_READY);
+ eventEmitter.emit(EVENT_CONSUME_READY);
+ }
+ catch(err){console.error(err)}
+})();

options.js

@@ -0,0 +1,22 @@
+//Values you can play around with
+
+module.exports = Object.freeze({
+ "elasticsearch": {
+ "log_level": "error",
+ "scroll_time": "10s"
+ },
+
+ "buffer_size": 1000,
+
+ "produce": {
+ "batch_size": 100,
+ "timeout": 1000 //in ms
+ },
+
+ "consume": {
+ "batch_size": 50,
+ "timeout": 1000, //in ms
+ "byFile": true,
+ "byElastic": true
+ }
+})