<commit_before>/* eslint no-underscore-dangle: ['error', { 'allow': ['_index', '_type'] }] */

const Promise = require('promise');
const debug = require('debug')('winston:elasticsearch');
const retry = require('retry');

const BulkWriter = function BulkWriter(transport, client, options) {
  this.transport = transport;
  this.client = client;
  this.options = options;
  this.interval = options.interval || 5000;
  this.healthCheckTimeout = options.healthCheckTimeout || '30s';
  this.healthCheckWaitForStatus = options.healthCheckWaitForStatus || 'yellow';
  this.healthCheckWaitForNodes = options.healthCheckWaitForNodes || '>=1';
  this.waitForActiveShards = options.waitForActiveShards || '1';
  this.pipeline = options.pipeline;
  this.retryLimit = options.retryLimit || 400;

  this.bulk = []; // bulk to be flushed
  this.running = false;
  this.timer = false;
  debug('created', this);
};

BulkWriter.prototype.start = function start() {
  this.checkEsConnection(this.retryLimit);
  debug('started');
};

BulkWriter.prototype.stop = function stop() {
  this.running = false;
  if (!this.timer) {
    return;
  }
  clearTimeout(this.timer);
  this.timer = null;
  debug('stopped');
};

BulkWriter.prototype.schedule = function schedule() {
  const thiz = this;
  this.timer = setTimeout(() => {
    thiz.tick();
  }, this.interval);
};

BulkWriter.prototype.tick = function tick() {
  debug('tick');
  const thiz = this;
  if (!this.running) {
    return;
  }
  this.flush()
    .then(() => {
    // Emulate finally with last .then()
    })
    .then(() => {
    // finally()
      thiz.schedule();
    });
};

BulkWriter.prototype.flush = function flush() {
  // write bulk to elasticsearch
  if (this.bulk.length === 0) {
    debug('nothing to flush');
    return new Promise((resolve) => {
      // pause the buffering process when there's no more bulk to flush
      // thus allowing the process to be terminated
      this.running = false;
      return resolve();
    });
  }
  const bulk = this.bulk.concat();
  this.bulk = [];
  const body = [];
  // eslint-disable-next-line object-curly-newline
  bulk.forEach(({ index, doc, attempts }) => {
    body.push(
      {
        [this.options.dataStream ? 'create' : 'index']: {
          _index: index,
          pipeline: this.pipeline
        },
        attempts
      },
      doc
    );
  });
  debug('bulk writer is going to write', body);
  return this.write(body);
};

BulkWriter.prototype.append = function append(index, doc) {
  if (this.options.buffering === true) {
    if (
      typeof this.options.bufferLimit === 'number'
      && this.bulk.length >= this.options.bufferLimit
    ) {
      debug('message discarded because buffer limit exceeded');
      // @todo: i guess we can use callback to notify caller
      return;
    }
    this.bulk.unshift({
      index,
      doc,
      attempts: 0
    });
    // resume the buffering process
    if (!this.running) {
      this.running = true;
      this.tick();
    }
  } else {
    this.write([
      { [this.options.dataStream ? 'create' : 'index']: { _index: index, pipeline: this.pipeline } },
      doc
    ]);
  }
};

BulkWriter.prototype.write = function write(body) {
  const thiz = this;
  const operation = [thiz.options.dataStream ? 'create' : 'index'];
  debug('writing to ES');
  return this.client
    .bulk({
      body,
      waitForActiveShards: this.waitForActiveShards,
      timeout: this.interval + 'ms',
    })
    .then((response) => {
      const res = response.body;
      if (res && res.errors && res.items) {
        const err = new Error('Elasticsearch error');
        res.items.forEach((item, itemIndex) => {
          const bodyData = body[itemIndex * 2 + 1];
          const opKey = Object.keys(item)[0];
          if (item[opKey] && item[opKey].error) {
            debug('elasticsearch indexing error', item[opKey].error);
            thiz.options.internalLogger('elasticsearch indexing error', item[opKey].error, bodyData);
            err.indexError = item[opKey].error;
            err.causedBy = bodyData;
          }
        });
        throw err;
      }
    })
    .catch((e) => {
    // rollback this.bulk array
      const newBody = [];
      body.forEach((chunk, index, chunks) => {
        const { attempts, created } = chunk;
        if (!created && attempts < thiz.retryLimit) {
          newBody.push({
            index: chunk[operation]._index,
            doc: chunks[index + 1],
            attempts: attempts + 1,<commit_msg>fix: elasticsearch 8 support<commit_after>/* eslint no-underscore-dangle: ['error', { 'allow': ['_index', '_type'] }] */

const Promise = require('promise');
const debug = require('debug')('winston:elasticsearch');
const retry = require('retry');

const BulkWriter = function BulkWriter(transport, client, options) {
  this.transport = transport;
  this.client = client;
  this.options = options;
  this.interval = options.interval || 5000;
  this.healthCheckTimeout = options.healthCheckTimeout || '30s';
  this.healthCheckWaitForStatus = options.healthCheckWaitForStatus || 'yellow';
  this.healthCheckWaitForNodes = options.healthCheckWaitForNodes || '>=1';
  this.waitForActiveShards = options.waitForActiveShards || '1';
  this.pipeline = options.pipeline;
  this.retryLimit = options.retryLimit || 400;

  this.bulk = []; // bulk to be flushed
  this.running = false;
  this.timer = false;
  debug('created', this);
};

BulkWriter.prototype.start = function start() {
  this.checkEsConnection(this.retryLimit);
  debug('started');
};

BulkWriter.prototype.stop = function stop() {
  this.running = false;
  if (!this.timer) {
    return;
  }
  clearTimeout(this.timer);
  this.timer = null;
  debug('stopped');
};

BulkWriter.prototype.schedule = function schedule() {
  const thiz = this;
  this.timer = setTimeout(() => {
    thiz.tick();
  }, this.interval);
};

BulkWriter.prototype.tick = function tick() {
  debug('tick');
  const thiz = this;
  if (!this.running) {
    return;
  }
  this.flush()
    .then(() => {
    // Emulate finally with last .then()
    })
    .then(() => {
    // finally()
      thiz.schedule();
    });
};

BulkWriter.prototype.flush = function flush() {
  // write bulk to elasticsearch
  if (this.bulk.length === 0) {
    debug('nothing to flush');
    return new Promise((resolve) => {
      // pause the buffering process when there's no more bulk to flush
      // thus allowing the process to be terminated
      this.running = false;
      return resolve();
    });
  }
  const bulk = this.bulk.concat();
  this.bulk = [];
  const body = [];
  // eslint-disable-next-line object-curly-newline
  bulk.forEach(({ index, doc, attempts }) => {
    body.push(
      {
        [this.options.dataStream ? 'create' : 'index']: {
          _index: index,
          pipeline: this.pipeline
        },
        attempts
      },
      doc
    );
  });
  debug('bulk writer is going to write', body);
  return this.write(body);
};

BulkWriter.prototype.append = function append(index, doc) {
  if (this.options.buffering === true) {
    if (
      typeof this.options.bufferLimit === 'number'
      && this.bulk.length >= this.options.bufferLimit
    ) {
      debug('message discarded because buffer limit exceeded');
      // @todo: i guess we can use callback to notify caller
      return;
    }
    this.bulk.unshift({
      index,
      doc,
      attempts: 0
    });
    // resume the buffering process
    if (!this.running) {
      this.running = true;
      this.tick();
    }
  } else {
    this.write([
      { [this.options.dataStream ? 'create' : 'index']: { _index: index, pipeline: this.pipeline } },
      doc
    ]);
  }
};

BulkWriter.prototype.write = function write(body) {
  const thiz = this;
  const operation = [thiz.options.dataStream ? 'create' : 'index'];
  debug('writing to ES');
  return this.client
    .bulk({
      body,
      wait_for_active_shards: this.waitForActiveShards,
      timeout: this.interval + 'ms',
    })
    .then((response) => {
      const res = response.body;
      if (res && res.errors && res.items) {
        const err = new Error('Elasticsearch error');
        res.items.forEach((item, itemIndex) => {
          const bodyData = body[itemIndex * 2 + 1];
          const opKey = Object.keys(item)[0];
          if (item[opKey] && item[opKey].error) {
            debug('elasticsearch indexing error', item[opKey].error);
            thiz.options.internalLogger('elasticsearch indexing error', item[opKey].error, bodyData);
            err.indexError = item[opKey].error;
            err.causedBy = bodyData;
          }
        });
        throw err;
      }
    })
    .catch((e) => {
    // rollback this.bulk array
      const newBody = [];
      body.forEach((chunk, index, chunks) => {
        const { attempts, created } = chunk;
        if (!created && attempts < thiz.retryLimit) {
          newBody.push({
            index: chunk[operation]._index,
            doc: chunks[index + 1],
            attempts: attempts + 1,
{'fraction_non_alphanumeric': 0.14882535460992907, 'fraction_numerical': 0.004100177304964539, 'mean_word_length': 2.6449919224555734, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 25, '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': '1623067', 'n_tokens_mistral': 2784, 'n_tokens_neox': 2764, 'n_words': 640}