Elastic Search maintenance script for periodic indices. Manages optimization/write-protection/closure/removal of indices based on age (derived from index naming convention) and status.

es_index_maintenance.js

#!/usr/bin/env node

// Configuration
var argv = require('optimist')
  .usage('Usage: $0 --prefix index_prefix [options]')
  .demand(['prefix'])
  .describe({
      host: 'Elastic Search host',
      port: 'Elastic Search port',
      secure: 'Use secure Elastic Search connection',
      prefix: 'Index name prefix',
      delim: 'Delimiter between index prefix and date/time',
      dateformat: 'Date format within index name',
      opendays: 'Number of days to keep indices open',
      keepdays: 'Number of days to keep indices prior to deletion',
      dryrun: 'Perform a dry run, only logging required actions without executing',
      skipoptimize: 'Skip index optimization',
      maxoptimize: 'Maximum number of indices to optimize in a single pass'
  })
  .default({
      host: 'localhost',
      port: 9200,
      secure: false,
      delim: '_',
      dateformat: 'YYYYMMDD',
      opendays: 45,
      keepdays: 60,
      dryrun: false,
      skipoptimize: false,
      maxoptimize: 2
  })
  .argv;

if (argv.dryrun) {
  console.log('Performing dry-run; required actions will be logged and not executed');
}

// 3rd-party libraries
var moment = require('moment');
var ElasticSearchClient = require('elasticsearchclient');

// Runtime variables
var batchComplete = 0;
var error = false;
var indexResults;
var segmentResults;
var mergeResults;

var es = new ElasticSearchClient({ host: argv.host, port: argv.port, secure: argv.secure });

// Retrieve index settings
es.getSettings(argv.prefix + argv.delim + '*')
  .on('data', function(data) {
    indexResults = JSON.parse(data);
  })
  .on('error', function(err) {
    console.log(err);
    error = true;
  })
  .on('done', function() {
    if (!error && indexResults.error) {
      console.log(indexResults.error);
      error = true;
    }
    batchComplete++;
    performMaintenance();
  })
  .exec();

// Retrieve index segmentation
es.getSegments(argv.prefix + argv.delim + '*')
  .on('data', function(data) {
    segmentResults = JSON.parse(data).indices;
  })
  .on('error', function(err) {
    console.log(err);
    error = true;
  })
  .on('done', function() {
    if (!error && segmentResults && segmentResults.error) {
      console.log(segmentResults.error);
      error = true;
    }
    batchComplete++;
    performMaintenance();
  })
  .exec();

// Retrieve index merge statistics
es.stats(argv.prefix + argv.delim + '*', { clear: true, merge: true })
  .on('data', function(data) {
    mergeResults = JSON.parse(data).indices;
  })
  .on('error', function(err) {
    console.log(err);
    error = true;
  })
  .on('done', function() {
    batchComplete++;
    performMaintenance();
  })
  .exec();

function performMaintenance() {
  // Only process once all data is collected
  if (batchComplete < 3) {
    return;
  }

  // Track number of index optimizations
  var optimizeCount = 0;

  // If a data collection error has been encountered, exit
  if (error) {
    process.exit(1);
  }

  // Iterate through indices and determine required maintenance
  for (var indexKey in indexResults) {
    var indexDate = moment(indexKey.split(argv.delim)[1], argv.dateformat);

    // Remove indices older than indexKeepDays
    if (moment.utc().diff(indexDate, 'days') > argv.keepdays) {
      console.log('Removing ' + indexKey + ' due to age > ' + argv.keepdays + ' days...');
      if (!argv.dryrun) {
        es.deleteIndex(indexKey).exec();
      }
      continue;
    }

    // Open indices newer than indexOpenDays
    // Skip further processing of other closed/recovering indices
    if (checkState(indexKey) != 'green') {
      if (moment.utc().diff(indexDate, 'days') <= argv.opendays) {
        console.log('Opening ' + indexKey + ' due to age <= ' + argv.opendays + ' days...');
        if (!argv.dryrun) {
          es.openIndex(indexKey).exec();
        }
      }
      continue;
    }

    // Skip processing of current index
    if (indexDate.format(argv.dateformat) == moment.utc().format(argv.dateformat)) {
      console.log('Skipping current index: ' + indexKey);
      continue;
    }

    // Close indices older than indexOpenDays
    if (moment.utc().diff(indexDate, 'days') > argv.opendays) {
      console.log('Closing ' + indexKey + ' due to age > ' + argv.opendays + ' days...');
      if (!argv.dryrun) {
        es.closeIndex(indexKey).on('error', function(error){ console.log(error); }).exec();
      }
      continue;
    }

    // Skip further processing of write-protected indices
    if (indexResults[indexKey].settings['index.blocks.write'] == 'true') {
      continue;
    }

    // If index is completely optimized, write-protect it; otherwise force optimize
    if (checkOptimized(indexKey)) {
      console.log('Write-protecting ' + indexKey + '...');
      if (!argv.dryrun) {
        es.updateSettings(indexKey, { 'index.blocks.write': 'true' }).exec();
      }
    } else {
      if (argv.skipoptimize) {
        continue;
      }

      if (argv.maxoptimize && optimizeCount >= argv.maxoptimize) {
        continue;
      }

      // Skip optimization if index is already merging
      if (mergeResults[indexKey].total.merges.current > 0) {
        console.log('Skipping optimization for ' + indexKey + '; already merging');
        optimizeCount++;
        continue;
      }

      console.log('Optimizing ' + indexKey + '...');
      optimizeCount++;
      if (!argv.dryrun) {
        es.optimize(indexKey, { max_num_segments: 1, wait_for_merge: false }).exec();
      }
    }
  }
}

// Check if index is online or recovering/closed
function checkState(indexKey) {
  var result = 'green';

  if (!segmentResults[indexKey]) {
    return 'red';
  }

  var shards = segmentResults[indexKey].shards;
  for (shardKey in shards) {
    var shard = shards[shardKey];
    shard.forEach(function(replica) {
      if (!replica.routing.state == 'STARTED') {
        result = 'red';
      }
    });
  }

  return result;
}

// Check if index is optimized to 1 segment/shard
function checkOptimized(indexKey) {
  var result = true;

  if (!segmentResults[indexKey]) {
    return result;
  }

  var shards = segmentResults[indexKey].shards;
  for (shardKey in shards) {
    var shard = shards[shardKey];
    shard.forEach(function(replica) {
      if (replica.num_committed_segments > 1) {
        result = false;
      }
    });
  }

  return result;
}

// vim: ft=javascript expandtab shiftwidth=2 softtabstop=2