garbage-deleter.js

/*
 * This Source Code Form is subject to the terms of the Mozilla Public
 * License, v. 2.0. If a copy of the MPL was not distributed with this
 * file, You can obtain one at http://mozilla.org/MPL/2.0/.
 */

/*
 * Copyright 2020 Joyent, Inc.
 */

//
// This contains the code for GarbageDeleter objects which are used by the
// garbage-deleter program to process garbage collector instructions and delete
// the appropriate files on disk.
//

var fs = require('fs');
var path = require('path');
var util = require('util');

var assert = require('assert-plus');
var vasync = require('vasync');
var VError = require('verror').VError;

var common = require('./common');
var elapsedSince = common.elapsedSince;

var DEFAULT_CONCURRENT_DELETES = 10;
var DEFAULT_MANTA_ROOT = '/manta';

//
// These values specify limits on the number of lines and length of each of
// those lines in order to protect us against memory exhaustion.
//
// The line length value was chosen through a "thumb in the air" approach as
// something that we'd expect lines to always fit within.
//
// The maximum number of lines was chosen similarly but should always be >= the
// size that is produced by the garbage consumer programs in the
// garbage-collector zone.
//
var DEFAULT_MAX_LINE_LENGTH = 2048;
var DEFAULT_MAX_LINES = 1000;

//
// DEFAULT_MAX_RUN_WAIT is the maximum time (in ms) between runs at which point
// we'll force a run. Incoming files appearing in the directory (watched via
// fs.watch) will also trigger a run as long as it has been at least
// DEFAULT_MIN_RUN_FREQ ms since the last run.
//
var DEFAULT_MAX_RUN_WAIT = 300000; // 5m
var DEFAULT_MIN_RUN_FREQ = 1000;

var METRIC_PREFIX = 'gc_storage_';
var QUEUE_CHECK_FREQ = 60000; // ms between checks for number of files in queue

// Helpers

//
// This does basically the same thing as vasync.forEachParallel, but allows for
// a 'concurrency' parameter to limit how many are being done at once.
//
function forEachParallel(opts, callback) {
    assert.object(opts, 'opts');
    assert.optionalNumber(opts.concurrency, 'opts.concurrency');
    assert.func(opts.func, 'opts.func');
    assert.array(opts.inputs, 'opts.inputs');
    assert.func(callback, 'callback');

    var concurrency = opts.concurrency ? Math.floor(opts.concurrency) : 0;
    var error = null;
    var idx;
    var queue;
    var results = {
        ndone: 0,
        nerrors: 0,
        operations: [],
        successes: []
    };

    if (!concurrency) {
        // If they didn't want concurrency control, just give them the original
        // vasync.forEachParallel.
        vasync.forEachParallel(opts, callback);
        return;
    }

    // Tell node.js to use up to "concurrency" threads for IO so that our
    // deletes actually do happen in parallel. (Node's default is 4)
    if (concurrency > process.env.UV_THREADPOOL_SIZE) {
        process.env.UV_THREADPOOL_SIZE = concurrency;
    }

    queue = vasync.queue(opts.func, concurrency);

    queue.on('end', function() {
        callback(error, results);
    });

    function doneOne(err, result) {
        var _status;

        results.ndone++;
        if (err) {
            results.nerrors++;
            _status = 'fail';
            // Yes this overwrites with the last error seen.
            error = err;
        } else {
            results.successes.push(result);
            _status = 'ok';
        }
        results.operations.push({
            func: opts.func,
            funcname: opts.func.name || '(anon)',
            status: _status,
            err: err,
            result: result
        });
    }

    for (idx = 0; idx < opts.inputs.length; idx++) {
        queue.push(opts.inputs[idx], doneOne);
    }

    queue.close();
}

// GarbageDeleter

function GarbageDeleter(opts) {
    var self = this;

    assert.object(opts, 'opts');
    assert.object(opts.log, 'opts.log');
    assert.string(opts.badInstructionDir, 'opts.badInstructionDir');
    assert.optionalNumber(opts.concurrentDeletes, 'opts.concurrentDeletes');
    assert.object(opts.config, 'opts.config');
    assert.string(opts.config.manta_storage_id, 'opts.config.manta_storage_id');
    assert.optionalString(opts.instructionDir, 'opts.instructionDir');
    assert.optionalString(opts.mantaRoot, 'opts.mantaRoot');
    assert.optionalNumber(opts.maxLineLength, 'opts.maxLineLength');
    assert.optionalNumber(opts.maxLines, 'opts.maxLines');
    assert.optionalNumber(opts.maxRunWait, 'opts.maxRunWait');
    assert.optionalObject(opts.metricsManager, 'opts.metricsManager');
    assert.optionalNumber(opts.minRunFreq, 'opts.minRunFreq');

    // Options that exist only for testing.
    assert.optionalFunc(opts._readdir, 'opts._readdir');
    assert.optionalFunc(opts._readFile, 'opts._readFile');
    assert.optionalFunc(opts._fsReadFile, 'opts._fsReadFile');
    assert.optionalFunc(opts._fsRename, 'opts._fsRename');
    assert.optionalFunc(opts._fsUnlink, 'opts._fsUnlink');
    assert.optionalFunc(opts._processFileHook, 'opts._processFileHook');

    self.log = opts.log;

    self.badInstructionDir = opts.badInstructionDir;
    self.concurrentDeletes =
        opts.concurrentDeletes || DEFAULT_CONCURRENT_DELETES;
    self.config = opts.config;
    self.instructionDir = opts.instructionDir;
    self.mantaRoot = opts.mantaRoot || DEFAULT_MANTA_ROOT;
    self.maxLineLength = opts.maxLineLength || DEFAULT_MAX_LINE_LENGTH;
    self.maxLines = opts.maxLines || DEFAULT_MAX_LINES;
    self.maxRunWait = opts.maxRunWait || DEFAULT_MAX_RUN_WAIT;
    self.metricsManager = opts.metricsManager;
    self.minRunFreq = opts.minRunFreq || DEFAULT_MIN_RUN_FREQ;
    self.storageId = opts.config.manta_storage_id;

    self.lastRun = 0;
    self.nextRunTimer = null;
    self.runningAsap = false;
    self.stopping = false;

    // Metrics
    if (self.metricsManager) {
        self.metrics = {
            deleteCountMissing: self.metricsManager.collector.counter({
                name: METRIC_PREFIX + 'delete_missing_count_total',
                help:
                    'Counter incremented every time a delete is attempted ' +
                    'for a Manta object but the object did not exist on disk.'
            }),
            deleteCountTotal: self.metricsManager.collector.counter({
                name: METRIC_PREFIX + 'delete_count_total',
                help: 'Counter incremented every time a Manta object is deleted'
            }),
            deleteErrorCount: self.metricsManager.collector.counter({
                name: METRIC_PREFIX + 'delete_error_count_total',
                help:
                    'Counter incremented every time a delete is attempted ' +
                    'for a Manta object but the delete failed.'
            }),
            deleteTimeMaxSeconds: self.metricsManager.collector.gauge({
                name: METRIC_PREFIX + 'delete_time_max_seconds',
                help:
                    'Gauge of maximum time spent in fs.unlink() deleting a ' +
                    'single Manta object (slowest delete)'
            }),
            deleteTimeMinSeconds: self.metricsManager.collector.gauge({
                name: METRIC_PREFIX + 'delete_time_min_seconds',
                help:
                    'Gauge of minimum time spent in fs.unlink() deleting a ' +
                    'single Manta object (fastest delete)'
            }),
            deleteTimeSeconds: self.metricsManager.collector.counter({
                name: METRIC_PREFIX + 'delete_time_seconds_total',
                help:
                    'Counter of total time spent in fs.unlink() deleting ' +
                    'Manta objects'
            }),
            instructionFilesBad: self.metricsManager.collector.counter({
                name: METRIC_PREFIX + 'instruction_files_bad_count_total',
                help:
                    'Counter incremented for each instruction file found to ' +
                    'be invalid'
            }),
            instructionFilesDeleted: self.metricsManager.collector.counter({
                name: METRIC_PREFIX + 'instruction_files_deleted_count_total',
                help: 'Counter incremented for each instruction file deleted'
            }),
            instructionFilesProcessed: self.metricsManager.collector.counter({
                name: METRIC_PREFIX + 'instruction_files_processed_count_total',
                help: 'Counter incremented for each instruction file processed'
            }),
            instructionLinesProcessed: self.metricsManager.collector.counter({
                name: METRIC_PREFIX + 'instruction_lines_processed_count_total',
                help: 'Counter incremented for each instruction line processed'
            }),
            instructionFilesQueued: self.metricsManager.collector.gauge({
                name: METRIC_PREFIX + 'instruction_files_queued_count',
                help: 'Gauge indicating number of instructions files in queue'
            })
        };
    } else {
        self.metrics = {};
    }
    self.addCounter('deleteErrorCount', 0);
    self.addCounter('deleteCountMissing', 0);
    self.addCounter('deleteCountTotal', 0);
    self.addCounter('deleteTimeSeconds', 0);
    self.addCounter('instructionFilesBad', 0);
    self.addCounter('instructionFilesDeleted', 0);
    self.addCounter('instructionFilesProcessed', 0);
    self.addCounter('instructionLinesProcessed', 0);
    self.setGauge('instructionFilesQueued', 0);

    // Add properties which should be modified only for testing purposes.
    self.fsCreateReadStream = opts._fsCreateReadStream || fs.createReadStream;
    self.fsReaddir = opts._fsReaddir || fs.readdir;
    self.fsReadFile = opts._fsReadFile || fs.readFile;
    self.fsRename = opts._fsRename || fs.rename;
    self.fsUnlink = opts._fsUnlink || fs.unlink;
    self.fsWatch = opts._fsWatch || fs.watch;
    if (opts._processFileHook) {
        self._processFileHook = opts._processFileHook;
    }
}

GarbageDeleter.prototype.addCounter = function addCounter(counterName, value) {
    var self = this;

    // For tests, we don't want to require a full metricManager, so in that case
    // we just manually manage the values in the "metrics" object.
    if (!self.metricsManager) {
        if (!self.metrics.hasOwnProperty(counterName)) {
            self.metrics[counterName] = 0;
        }
        self.metrics[counterName] += value;
        return;
    }

    self.metrics[counterName].add(value);
};

GarbageDeleter.prototype.getCounter = function getCounter(counterName) {
    var self = this;

    if (!self.metricsManager) {
        return self.metrics[counterName];
    }

    return self.metrics[counterName].getValue();
};

GarbageDeleter.prototype.getGauge = function getGauge(gaugeName) {
    var self = this;

    if (!self.metricsManager) {
        return self.metrics[gaugeName];
    }

    return self.metrics[gaugeName].getValue();
};

GarbageDeleter.prototype.setGauge = function setGauge(gaugeName, value) {
    var self = this;

    // For tests, we don't want to require a full metricManager, so in that case
    // we just manually manage the values in the "metrics" object.
    if (!self.metricsManager) {
        self.metrics[gaugeName] = value;
        return;
    }

    self.metrics[gaugeName].set(value);
};

GarbageDeleter.prototype.readLines = function readLines(filename, callback) {
    var self = this;

    var chunks = [];
    var chunksLen = 0;
    var readStream;
    var maxLen = self.maxLines * (self.maxLineLength + 1);

    // Note: we're intentionally reading maxLen+1 here so we can catch files
    // that are too large.
    readStream = fs.createReadStream(filename, {start: 0, end: maxLen});

    readStream.on('error', function _onError(streamErr) {
        // Ensure whatever error we had, we're not going to try to keep reading.
        // The node docs don't guarantee this happens automatically on error.
        readStream.destroy();

        callback(streamErr);
    });

    readStream.on('readable', function() {
        var chunk;

        while ((chunk = readStream.read())) {
            chunksLen += Buffer.byteLength(chunk);
            chunks.push(chunk);
        }
    });

    readStream.on('close', function onClose() {
        var data = Buffer.concat(chunks, chunksLen);
        var idx;
        var lines = [];
        var strData;

        if (data.length > maxLen) {
            // File too big
            callback(
                new VError(
                    {
                        info: {
                            filename: filename,
                            size: data.length,
                            maxSize: maxLen
                        },
                        name: 'FileTooBigError'
                    },
                    util.format('File too big: %d > %d', data.length, maxLen)
                )
            );
            return;
        }

        strData = data.toString('utf8');
        lines = strData
            .trim()
            .split('\n')
            .filter(function _removeEmpty(candidate) {
                return candidate.length > 0;
            });

        if (lines.length > self.maxLines) {
            // Too many lines
            callback(
                new VError(
                    {
                        info: {
                            filename: filename,
                            lines: lines.length,
                            maxLines: self.maxLines
                        },
                        name: 'TooManyLinesError'
                    },
                    util.format(
                        'Too many lines: %d > %d',
                        lines.length,
                        self.maxLines
                    )
                )
            );
            return;
        }

        for (idx = 0; idx < lines.length; idx++) {
            if (lines[idx].length > self.maxLineLength) {
                // Line too long
                callback(
                    new VError(
                        {
                            info: {
                                filename: filename,
                                lineNum: idx + 1,
                                length: lines[idx].length,
                                maxLength: self.maxLineLength
                            },
                            name: 'LineTooLongError'
                        },
                        util.format(
                            'Line %d too long: %d > %d',
                            idx + 1,
                            lines[idx].length,
                            self.maxLineLength
                        )
                    )
                );
                return;
            }
        }

        callback(null, lines);
    });
};

//
// This function is called for each line of each otherwise valid instruction
// file. It is responsible for parsing the line, dispatching the delete and
// updating the related metrics.
//
GarbageDeleter.prototype.processInstruction = function processInstruction(
    opts,
    callback
) {
    var self = this;

    assert.object(opts, 'opts');
    assert.string(opts.filename, 'opts.filename');
    assert.string(opts.instructionLine, 'opts.instructionLine');

    var beginDelete;
    var deleteFile;
    var fields;
    var filename = opts.filename;
    var instructionLine = opts.instructionLine;

    fields = instructionLine.split(/\t/);

    self.log.trace(
        {
            fields: fields,
            line: instructionLine
        },
        'Split line into fields.'
    );

    if (fields.length !== 5) {
        self.log.error(
            {
                fields: fields,
                line: instructionLine
            },
            'Instruction line contains bad number of fields.'
        );

        callback(
            new VError(
                {
                    info: {
                        fields: fields,
                        filename: filename,
                        line: instructionLine
                    },
                    name: 'InvalidNumberOfFieldsError'
                },
                'File "' +
                    path.basename(filename) +
                    '" has invalid number' +
                    ' of fields ' +
                    fields.length +
                    ' !== 5'
            )
        );
        return;
    }

    self.addCounter('instructionLinesProcessed', 1);

    //
    // We have 2 kinds of instructions. The first kind (we'll call v1 here) is
    // what garbage-dir-consumer and buckets bits before MANTA-4591 create
    // which look like:
    //
    // 2.stor.cascadia.joyent.us       2873b2b0-7f90-477a-b1ae-4f8d4841dd53    afd7e79a-488d-e6eb-a315-df32712f3d16    2.moray.cascadia.joyent.us      10809
    //
    // The second kind (v2) is what garbage-buckets-consumer produces and looks
    // like:
    //
    // 2.stor.cascadia.joyent.us       DELETEv2    /v2/2873b2b0-7f90-477a-b1ae-4f8d4841dd53/9cc202e7-9579-4e9f-90b6-264876cb9601/41/41a608ed-a607-4a14-9b45-16b2d41b91e9,2e54144ba487ae25d03a3caba233da71  2.buckets-mdapi.cascadia.joyent.us  12
    //
    // In both cases the 1st field is the storageId of the storage zone that
    // should handle this file (used to sanity check and ensure we don't run
    // files intended for another storage zone).
    //
    // The 4th and 5th fields are also the same between the two formats. These
    // indicate the shard from which the record was consumed, and the number of
    // bytes in the object respectively.
    //
    // The difference between the two is in the 2nd and 3rd fields.
    //
    // For v1:
    //
    //  - the 2nd field is the creatorId
    //  - the 3rd field is the objectId
    //
    // For v2:
    //
    //  - the 2nd field is "DELETEv2" indicating this is a v2 record
    //  - the 3rd field is an absolute path which will be prefixed with '/manta'
    //    and then represents the path to the object on the storage zone.
    //

    try {
        assert.equal(fields[0], self.storageId); // storageId
        if (fields[1] === 'DELETEv2') {
            assert.string(fields[2]); // path under /manta

            // The path should start with a '/' and contain only:
            //
            //  * 'v' (in the case of v2)
            //  * hex characters
            //  * '/'
            //  * ',' (for the hash separator)
            //
            // I disagree with eslint that the escaped '/' is useless here, so:
            //
            // eslint-disable-next-line no-useless-escape
            assert.ok(fields[2].match(/^\/[0-9a-fv,-\/]+$/));
        } else {
            assert.uuid(fields[1]); // creatorUuid
            assert.uuid(fields[2]); // objectUuid
        }
        // fields[3] is the metadata shardId, not relevant here.
        assert.string(fields[4]);
        assert.ok(fields[4].match(/^[0-9]+$/)); // size
    } catch (e) {
        callback(
            new VError(
                {
                    cause: e,
                    info: {
                        fields: fields,
                        filename: filename
                    },
                    name: 'InvalidInstructionError'
                },
                'Invalid instruction in ' +
                    path.basename(filename) +
                    ': ' +
                    JSON.stringify(fields)
            )
        );
        return;
    }

    if (fields[1] === 'DELETEv2') {
        deleteFile = path.join(self.mantaRoot, fields[2]);
    } else {
        deleteFile = path.join(self.mantaRoot, fields[1], fields[2]);
    }

    //
    // This is the only record in the log in the normal case of each file
    // deleted. We log this for auditing purposes so we can be confident that
    // any file we've deleted we also generated at least one log entry for.
    //
    self.log.info({filename: deleteFile}, 'Deleting file.');

    beginDelete = process.hrtime();
    self.fsUnlink(deleteFile, function _unlinkMantaFile(unlinkErr) {
        var curval;
        var elapsed = elapsedSince(beginDelete);

        self.log.trace(
            {
                elapsed: elapsed,
                err: unlinkErr,
                filename: deleteFile
            },
            'Deleted one object.'
        );

        self.addCounter('deleteCountTotal', 1);
        self.addCounter('deleteTimeSeconds', elapsed);
        curval = self.getGauge('deleteTimeMaxSeconds');
        if (curval === undefined || elapsed > curval) {
            self.setGauge('deleteTimeMaxSeconds', elapsed);
        }
        curval = self.getGauge('deleteTimeMinSeconds');
        if (curval === undefined || elapsed < curval) {
            self.setGauge('deleteTimeMinSeconds', elapsed);
        }

        if (unlinkErr) {
            if (unlinkErr.code === 'ENOENT') {
                self.log.debug(
                    {
                        filename: deleteFile
                    },
                    'File did not exist. Skipping.'
                );
                self.addCounter('deleteCountMissing', 1);
            } else {
                callback(
                    new VError(
                        {
                            cause: unlinkErr,
                            info: {
                                filename: deleteFile
                            },
                            name: 'UnlinkFileError'
                        },
                        'Failed to delete file.'
                    )
                );
                self.addCounter('deleteErrorCount', 1);
                return;
            }
        } else {
            self.addCounter('instructionFilesDeleted', 1);
        }

        callback();
    });
};

GarbageDeleter.prototype.processFile = function processFile(
    instrFile,
    callback
) {
    var self = this;

    var badFilename = path.join(self.badInstructionDir, instrFile);
    var beginning = process.hrtime();
    var filename = path.join(self.instructionDir, instrFile);
    var lineCount = 0;

    // We use this function so that we can add a hook for tests to be able to
    // know the result from each file that was processed.
    function _doneProcessing(err) {
        self.addCounter('instructionFilesProcessed', 1);

        // On any error, we move the instruction file to the bad_instructions
        // directory.
        if (err) {
            self.addCounter('instructionFilesBad', 1);

            self.log.warn(
                {
                    err: err,
                    errInfo: VError.info(err),
                    filename: filename
                },
                'Failed to process file, moving to bad_instructions.'
            );

            self.fsRename(filename, badFilename, function _onRename(renameErr) {
                if (renameErr) {
                    self.log.error(
                        {
                            err: renameErr,
                            srcFilename: filename,
                            targFilename: badFilename
                        },
                        'Failed to rename file to bad_instructions.'
                    );
                }

                if (self._processFileHook) {
                    self._processFileHook({
                        err: err,
                        filename: instrFile,
                        lineCount: lineCount
                    });
                }

                // We don't return an error here because we don't want the
                // failure to remove one invalid instruction file to prevent
                // other files from being processed.
                callback(null, lineCount);
            });
        } else {
            if (self._processFileHook) {
                self._processFileHook({
                    err: err,
                    filename: instrFile,
                    lineCount: lineCount
                });
            }

            callback(null, lineCount);
        }
    }

    if (!filename.match(/\.instruction$/)) {
        self.log.warn(
            {
                filename: filename
            },
            'Ignoring non-instruction file.'
        );

        _doneProcessing(
            new VError(
                {
                    info: {
                        filename: filename
                    },
                    name: 'MissingInstructionSuffixError'
                },
                'Filename missing .instruction suffix.'
            )
        );

        return;
    }

    self.log.debug({filename: filename}, 'Processing file.');

    self.readLines(filename, function _onReadFile(err, lines) {
        if (err) {
            self.log.error({err: err}, 'Error reading lines.');
            _doneProcessing(err);
            return;
        }

        lineCount = lines.length;

        if (lineCount === 0) {
            _doneProcessing(
                new VError(
                    {
                        info: {
                            filename: filename
                        },
                        name: 'EmptyFileError'
                    },
                    'Instruction file is empty.'
                )
            );
            return;
        }

        if (lineCount > self.maxLines) {
            _doneProcessing(
                new VError(
                    {
                        info: {
                            filename: filename,
                            lines: lineCount,
                            maxLines: self.maxLines
                        },
                        name: 'TooManyLinesError'
                    },
                    'Instruction file contains too many lines.'
                )
            );
            return;
        }

        // At this point we know the file has > 0 and < self.maxLines lines of
        // instructions. So we'll process up to self.concurrentDeletes of the
        // lines at a time.

        forEachParallel(
            {
                concurrency: self.concurrentDeletes,
                func: function _runInstructions(line, cb) {
                    self.processInstruction(
                        {
                            filename: filename,
                            instructionLine: line
                        },
                        cb
                    );
                },
                inputs: lines
            },
            function _ranInstructions(parallelErr, results) {
                self.log.info(
                    {
                        elapsed: elapsedSince(beginning),
                        filename: filename,
                        lines: lineCount
                    },
                    'Ran instructions.'
                );

                if (parallelErr) {
                    _doneProcessing(parallelErr);
                    return;
                }

                // No error, so delete the instruction file.
                self.fsUnlink(filename, function _onUnlinkInstructionFile(e) {
                    if (e) {
                        if (e.code !== 'ENOENT') {
                            _doneProcessing(e);
                            return;
                        }
                        self.log.debug(
                            'Went to delete "%s" but did not exist',
                            filename
                        );
                    }
                    _doneProcessing();
                });
            }
        );
    });
};

GarbageDeleter.prototype.run = function run() {
    var self = this;

    var beginning = process.hrtime();

    if (self.stopping) {
        self.log.trace('Not running Deleter, stopping in progress.');
        return;
    }

    self.log.trace('Running Deleter.');

    self.fsReaddir(self.instructionDir, function _onReaddir(err, files) {
        if (err) {
            // We'll run again and there's not much we can do here.
            self.log.error({err: err}, 'Failed to read instruction dir.');
            return;
        }

        // We serially process files, but process the instructions inside the
        // files in parallel with a concurrency limit. This prevents us from
        // having to worry about tuning multiple knobs.
        vasync.forEachPipeline(
            {
                func: self.processFile.bind(self),
                inputs: files
            },
            function _processedInstructions(e, results) {
                var elapsed = elapsedSince(beginning);
                var numLines;

                // This sums all the "successes" which means the value passed to
                // the callback of self.processFile() which in this case is the
                // number of lines processed from each file.
                numLines = results.successes.reduce(function _add(a, b) {
                    return a + b;
                }, 0);

                self.log.info(
                    {
                        elapsed: elapsed,
                        err: e,
                        files: files.length,
                        lines: numLines
                    },
                    'Processed instructions.'
                );

                //
                // Since we just ran now, we're going to set the next run up so that
                // we make sure that we don't wait longer than maxRunWait ms between
                // runs.
                //
                if (self.nextRunTimer !== null) {
                    clearTimeout(self.nextRunTimer);
                }

                // If we're stopping, we don't want to run again.
                if (self.stopping) {
                    return;
                }

                self.nextRunTimer = setTimeout(
                    self.run.bind(self),
                    self.maxRunWait
                );
                self.runningAsap = false;
            }
        );
    });
};

GarbageDeleter.prototype.start = function start(callback) {
    var self = this;

    vasync.pipeline(
        {
            funcs: [
                function _setupWatcher(_, cb) {
                    self.fsWatcher = self.fsWatch(
                        self.instructionDir,
                        function _onEvent() {
                            self.log.trace(
                                'Saw event on "%s".',
                                self.instructionDir
                            );
                            self.runAsap();
                        }
                    );
                    cb();
                },
                function _startQueueMonitor(_, cb) {
                    self.countQueue();
                    cb();
                },
                function _startFirstRun(_, cb) {
                    self.runAsap();
                    cb();
                }
            ]
        },
        function _started(err) {
            self.log.trace({err: err}, 'Started.');
            if (callback) {
                callback();
            }
        }
    );
};

GarbageDeleter.prototype.stop = function stop(callback) {
    var self = this;

    assert.optionalFunc(callback, 'callback');

    self.stopping = true;

    self.log.trace('Clearing next run timer.');
    clearTimeout(self.nextRunTimer);

    if (self.fsWatcher) {
        self.log.trace('Stopping fsWatcher.');
        self.fsWatcher.close();
        self.fsWatcher = null;
    }

    self.runningAsap = false;

    self.log.trace('Clearing queue counter timer.');
    clearTimeout(self.queueCounterTimer);

    self.log.trace('Stopped.');

    if (callback) {
        callback();
    }
};

GarbageDeleter.prototype.runAsap = function runAsap() {
    var self = this;

    var nextRun;
    var now;

    self.log.trace(
        {
            runningAsap: self.runningAsap
        },
        'Will run again ASAP.'
    );

    if (self.runningAsap) {
        // We're already going to run asap, nothing further to do.
        return;
    }
    self.runningAsap = true;

    now = new Date().getTime();

    if (now - self.lastRun >= self.minRunFreq) {
        // It has been long enough, so we can just run immediately.
        setImmediate(self.run.bind(self));
    } else {
        // It hasn't been long enough, so we want to schedule the run for the
        // future.
        if (self.nextRunTimer !== null) {
            clearTimeout(self.nextRunTimer);
        }
        nextRun = now - (self.minRunFreq + self.lastRun);
        self.log.trace('setTimeout(self.run, %d)', nextRun);
        self.nextRunTimer = setTimeout(self.run.bind(self), nextRun);
    }
};

//
// This simply counts how many instructions files are in the instruction dir,
// updates the metric, and then schedules the next check.
//
GarbageDeleter.prototype.countQueue = function countQueue() {
    var self = this;

    self.fsReaddir(self.instructionDir, function _onReaddir(err, files) {
        if (!err) {
            self.setGauge('instructionFilesQueued', files.length);
        } else {
            self.log.warn(
                {
                    dir: self.instructionDir,
                    err: err
                },
                'Failed to read instruction dir to count files.'
            );
        }

        if (self.stopping) {
            return;
        }

        // schedule next check
        self.queueCounterTimer = setTimeout(
            self.countQueue.bind(self),
            QUEUE_CHECK_FREQ
        );
    });
};

module.exports = GarbageDeleter;