<commit_before>#! /usr/bin/env node
'use strict';

var util = require('util');

var async = require('async');
var mongoose = require('mongoose');
var ObjectId = require('mongoose').Types.ObjectId;
var moment = require('moment');
var DisqEventEmitter = require('disque-eventemitter');

var config = require('../lib/config');
var constants = require('../lib/constants');
var Job = require('../models/job');
var Run = require('../models/run');
// Don't remove.  Loading this causes logger to start
var logger = require('../lib/logger');
// Don't remove.  Loading this causes status to start
var status = require('../lib/disqstatus');
var iostatus = require('../lib/iostatus');

mongoose.connect(config.db);
var debug = require('debug')('tilloo:scheduler');

var _loadedJobs = {};

Job.loadAllJobs(function(err, jobs) {
    debug('loading jobs');
    jobs.forEach(function(job) {
        debug('found job %s', job.name);
        if(job.schedule && job.schedule.trim()!=='') {
            debug('setting up cron %s for %s', job.schedule, job.name);
            _loadedJobs[job._id] = job;
            job.startCron();
        }
    });
    console.info('Scheduler started connected to %s', config.db);
});


// Garbage collector to clean up orphaned jobs
// Looks any run in a idle or busy state with an
// updatedAt that is more than 5 minutes old and
// marks it as failed
debug('zombie garbage collection interval: %d minutes', config.scheduler.zombieFrequency);
setInterval(function() {
    debug('garbage collecting zombie runs');
    Run.find({
        $and: [{updatedAt: {$lte: moment().subtract(config.scheduler.zombieAge, 'minutes').toDate()}},
                {$or: [{status: constants.JOBSTATUS.BUSY}, {status: constants.JOBSTATUS.IDLE}]}
            ]},
        function(err, zombieRuns) {
            async.eachLimit(zombieRuns, 5, function(zombieRun, done) {
                debug('settting status to fail runId: %s', zombieRun._id);
                zombieRun.status = constants.JOBSTATUS.FAIL;
                zombieRun.save(done);
            }, function(err) {
                if(err) {
                    console.error(err);
                }
            });
        }
    );
}, config.scheduler.zombieFrequency * 60000);

// Used so scheduler is notified of changes and can add/remove/change jobs
console.info('Listening to %s:%d queue: %s', config.disque.host, config.disque.port, constants.QUEUES.SCHEDULER);
var ee = new DisqEventEmitter(config.disque, constants.QUEUES.SCHEDULER);
ee.on('job', function(job, done) {
    // Immediately ack job to remove from queue.  If we have a failure
    // we don't want to deal with job again, best to just restart the scheduler
    ee.ack(job, function(err) {
        if(err) {
            console.error(err);
        }
<commit_msg>Update job lastStatus when killing a zombie<commit_after>#! /usr/bin/env node
'use strict';

var util = require('util');

var async = require('async');
var mongoose = require('mongoose');
var ObjectId = require('mongoose').Types.ObjectId;
var moment = require('moment');
var DisqEventEmitter = require('disque-eventemitter');

var config = require('../lib/config');
var constants = require('../lib/constants');
var Job = require('../models/job');
var Run = require('../models/run');
// Don't remove.  Loading this causes logger to start
var logger = require('../lib/logger');
// Don't remove.  Loading this causes status to start
var status = require('../lib/disqstatus');
var iostatus = require('../lib/iostatus');

mongoose.connect(config.db);
var debug = require('debug')('tilloo:scheduler');

var _loadedJobs = {};

Job.loadAllJobs(function(err, jobs) {
    debug('loading jobs');
    jobs.forEach(function(job) {
        debug('found job %s', job.name);
        if(job.schedule && job.schedule.trim()!=='') {
            debug('setting up cron %s for %s', job.schedule, job.name);
            _loadedJobs[job._id] = job;
            job.startCron();
        }
    });
    console.info('Scheduler started connected to %s', config.db);
});


// Garbage collector to clean up orphaned jobs
// Looks any run in a idle or busy state with an
// updatedAt that is more than 5 minutes old and
// marks it as failed
debug('zombie garbage collection interval: %d minutes', config.scheduler.zombieFrequency);
setInterval(function() {
    debug('garbage collecting zombie runs');
    Run.find({
        $and: [{updatedAt: {$lte: moment().subtract(config.scheduler.zombieAge, 'minutes').toDate()}},
                {$or: [{status: constants.JOBSTATUS.BUSY}, {status: constants.JOBSTATUS.IDLE}]}
            ]},
        function(err, zombieRuns) {
            async.eachLimit(zombieRuns, 5, function(zombieRun, done) {
                debug('settting status to fail runId: %s', zombieRun._id);
                async.parallel([
                    function(done) {
                        zombieRun.status = constants.JOBSTATUS.FAIL;
                        zombieRun.save(done);
                    },
                    function(done) {
                        Job.findByJobId(zombieRun.jobId, function(err, job) {
                            if(err) {
                                console.error('Unable to find jobId: ' + jobId);
                                done();
                            }
                            else {
                                job.lastStatus = constants.JOBSTATUS.FAIL;
                                job.save(done);
                            }
                        });
                    }
                ], done);
            }, function(err) {
                if(err) {
                    console.error(err);
                }
            });
        }
    );
}, config.scheduler.zombieFrequency * 60000);

// Used so scheduler is notified of changes and can add/remove/change jobs
console.info('Listening to %s:%d queue: %s', config.disque.host, config.disque.port, constants.QUEUES.SCHEDULER);
var ee = new DisqEventEmitter(config.disque, constants.QUEUES.SCHEDULER);
ee.on('job', function(job, done) {
    // Immediately ack job to remove from queue.  If we have a failure
    // we don't want to deal with job again, best to just restart the scheduler
    ee.ack(job, function(err) {
        if(err) {
            console.error(err);
        }

{'fraction_non_alphanumeric': 0.14044854461587403, 'fraction_numerical': 0.0022268172419277876, 'mean_word_length': 2.33935209771641, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '3574619', 'n_tokens_mistral': 1876, 'n_tokens_neox': 1810, 'n_words': 476}