<commit_before>const _          = require('lodash');
const Promise    = require('bluebird');
const HttpStatus = require('http-status');
const config     = require('../../config/index');

const log = config.log;

const MAX_FLUSH_RETRY         = 5;
const MIN_RETRY_WAIT_MS       = 2000;
const MAX_RETRY_WAIT_MS       = 7000;
const BULK_REQUEST_TIMEOUT_MS = 3600000;

/**
 * Transfer constructor
 *
 * @param sourceEs
 * @param destEs
 * @returns {Transfer}
 * @constructor
 */
const Transfer = function (sourceEs, destEs) {
  const self         = this;
  let updateCallback = null;

  let flushRetryCount = 0;
  const bulkQueue     = [];
  let mutatorsByType  = {};
  let queueSummary    = {
    tick:        0,
    transferred: 0,
    scrolled:    0,
    errors:      0
  };

  self.source = sourceEs;
  self.dest = destEs;

  /**
   * Flushes any queued bulk inserts
   *
   * @returns {Promise.<TResult>}
   */
  const flushQueue = () => {
    if (bulkQueue.length > 0) {
      const bulkBody = [];
      while (bulkQueue.length > 0) {
        bulkBody.push(bulkQueue.shift());
      }

      queueSummary.tick = 0;

      return self.dest.bulk({refresh: true, body: bulkBody, requestTimeout: BULK_REQUEST_TIMEOUT_MS})
      .then((results) => {
        log.trace('response: %s', JSON.stringify(results, null, config.jsonIndent));

        if (results.errors && results.errors > 0) {
          return self.handleBulkErrors(results, bulkBody);
        } else {
          queueSummary.transferred += results.items.length;
          queueSummary.tick = results.items.length;

          if (_.isFunction(updateCallback)) {
            updateCallback(queueSummary);
          }

          log.debug('flush complete: ', queueSummary);
          return Promise.resolve();
        }
      });
    } else {
      return Promise.resolve();
    }
  };

  /**
   * Scan and scroll to get data from specific index and type in source ES.
   *
   * Apply any relevant mutations to the data.
   *
   * Queue it for sending to destination.
   *
   * Optionally provide query to restrict the data to be retrieved.
   * @param targetIndex
   * @param targetType
   * @param flushSize
   */
  self.transferData = (targetIndex, targetType, flushSize) => {
    queueSummary = {
      tick:        0,
      transferred: 0,
      scrolled:    0,
      errors:      0
    };

    if (!_.isString(targetIndex) || targetIndex.length === 0) {
      throw new Error('targetIndex must be string with length');
    } else if (!_.isString(targetType) || targetType.length === 0) {
      throw new Error('targetType must be string with length');
    }

    flushRetryCount = 0;

    const scrollAndGetData = (response) => {
      const documents = [];
      response.hits.hits.forEach((hit) => {
        documents.push(hit);
        queueSummary.scrolled++;
      });

      return putData(self.mutate(documents, 'data'), flushSize)
      .then(() => {
        if (response.hits.total !== queueSummary.scrolled) {
          return self.source.scroll({
            scroll_id: response._scroll_id,
            scroll:    '1h'
          })
          .then((inner_response) => {
            log.debug('scrolling: ', queueSummary);
            return scrollAndGetData(inner_response);
          });
        } else {
          return flushQueue()
          .then(() => {
            log.debug('transfer complete: ', queueSummary);
            return queueSummary.transferred;
          });
        }
      });
    };
    return self.source.search({
      index:  targetIndex,
      type:   targetType,
      scroll: '5m',
      size:   flushSize
    })
    .then(scrollAndGetData)
    .catch((error) => {
      log.error('Error during search: ', error);
      return Promise.reject(error);<commit_msg>retries on ES search errors<commit_after>const _          = require('lodash');
const Promise    = require('bluebird');
const HttpStatus = require('http-status');
const config     = require('../../config/index');

const log = config.log;

const MAX_FLUSH_RETRY         = 5;
const MIN_RETRY_WAIT_MS       = 2000;
const MAX_RETRY_WAIT_MS       = 7000;
const BULK_REQUEST_TIMEOUT_MS = 3600000;

/**
 * Transfer constructor
 *
 * @param sourceEs
 * @param destEs
 * @returns {Transfer}
 * @constructor
 */
const Transfer = function (sourceEs, destEs) {
  const self         = this;
  let updateCallback = null;

  let flushRetryCount = 0;
  const bulkQueue     = [];
  let mutatorsByType  = {};
  let queueSummary    = {
    tick:        0,
    transferred: 0,
    scrolled:    0,
    errors:      0
  };

  self.source = sourceEs;
  self.dest = destEs;

  /**
   * Flushes any queued bulk inserts
   *
   * @returns {Promise.<TResult>}
   */
  const flushQueue = () => {
    if (bulkQueue.length > 0) {
      const bulkBody = [];
      while (bulkQueue.length > 0) {
        bulkBody.push(bulkQueue.shift());
      }

      queueSummary.tick = 0;

      return self.dest.bulk({refresh: true, body: bulkBody, requestTimeout: BULK_REQUEST_TIMEOUT_MS})
      .then((results) => {
        log.trace('response: %s', JSON.stringify(results, null, config.jsonIndent));

        if (results.errors && results.errors > 0) {
          return self.handleBulkErrors(results, bulkBody);
        } else {
          queueSummary.transferred += results.items.length;
          queueSummary.tick = results.items.length;

          if (_.isFunction(updateCallback)) {
            updateCallback(queueSummary);
          }

          log.debug('flush complete: ', queueSummary);
          return Promise.resolve();
        }
      });
    } else {
      return Promise.resolve();
    }
  };

  self.scroll = (response, retries = 0) => 
    self.source.scroll({
      scroll_id: response._scroll_id,
      scroll:    '1h'
    }).catch(() => retries > 3 ? Promise.reject(new Error(`can't scroll: ${response._scroll_id}`)) : self.scroll(response, retries + 1));

  self.search = (request, retries = 0) =>
    self.source.search(request)
      .catch(() => retries > 3 ? Promise.reject(new Error(`can't search: ${request}`)) : self.search(request, retries + 1));


  /**
   * Scan and scroll to get data from specific index and type in source ES.
   *
   * Apply any relevant mutations to the data.
   *
   * Queue it for sending to destination.
   *
   * Optionally provide query to restrict the data to be retrieved.
   * @param targetIndex
   * @param targetType
   * @param flushSize
   */
  self.transferData = (targetIndex, targetType, flushSize) => {
    queueSummary = {
      tick:        0,
      transferred: 0,
      scrolled:    0,
      errors:      0
    };

    if (!_.isString(targetIndex) || targetIndex.length === 0) {
      throw new Error('targetIndex must be string with length');
    } else if (!_.isString(targetType) || targetType.length === 0) {
      throw new Error('targetType must be string with length');
    }

    flushRetryCount = 0;

    const scrollAndGetData = (response) => {
      const documents = [];
      response.hits.hits.forEach((hit) => {
        documents.push(hit);
        queueSummary.scrolled++;
      });

      return putData(self.mutate(documents, 'data'), flushSize)
      .then(() => {
        if (response.hits.total !== queueSummary.scrolled) {
          return self.scroll(response)
          .then((inner_response) => {
            log.debug('scrolling: ', queueSummary);
            return scrollAndGetData(inner_response);
          });
        } else {
          return flushQueue()
          .then(() => {
            log.debug('transfer complete: ', queueSummary);
            return queueSummary.transferred;
          });
        }
      });
    };
    return self.search({
      index:  targetIndex,
      type:   targetType,
      scroll: '5m',
      size:   flushSize
    })
    .then(scrollAndGetData)
    .catch((error) => {
      log.error('Error during search: ', error);
      return Promise.reject(error);
{'fraction_non_alphanumeric': 0.12995705986360193, 'fraction_numerical': 0.009345794392523364, 'mean_word_length': 2.265567010309278, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 35, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1745548', 'n_tokens_mistral': 2405, 'n_tokens_neox': 2322, 'n_words': 553}