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

//node
const fs = require('fs');
const path = require('path');
const os = require('os');
const request = require('request');

//contrib
const winston = require('winston');
const async = require('async');
const Client = require('ssh2').Client;

//mine
const config = require('../config');
const logger = new winston.Logger(config.logger.winston);
const db = require('../api/models/db');
const common = require('../api/common');
const _resource_picker = require('../api/resource').select;
const _transfer = require('../api/transfer');
const _service = require('../api/service');

db.init(function(err) {
    if(err) throw err;
    //start check loop
    check(); 
});

//set next_date incrementally longer between each checks
function set_nextdate(task) {
    task.next_date = new Date();
    if(task.start_date) {
        var start = task.start_date.getTime();
        var elapsed = task.next_date.getTime() - start;
        //logger.debug("elapsed seconds: "+elapsed);
        var next = task.next_date.getTime() + elapsed/2;
        task.next_date.setTime(next);
    } else {
        //not yet started. check again in 10 minutes (maybe resource issue?)
        //maybe I should increase the delay to something like an hour?
        task.next_date.setMinutes(task.next_date.getMinutes() + 10);
    }
}

function check() {
    //logger.info("checking..");
    db.Task.find({
        status: {$ne: "removed"}, //ignore removed tasks
        $or: [
            {next_date: {$exists: false}},
            {next_date: {$lt: new Date()}}
        ] 
    })
    
    //maybe I should do these later (only needed by requested task)
    .populate('deps', 'status resource_id')
    .populate('resource_deps') 
    .exec((err, tasks) => {
        if(err) throw err;
        async.eachSeries(tasks, (task, next) => {
            logger.info("handling task:"+task._id);
            switch(task.status) {
            case "requested":
                handle_requested(task, next); 
                break;
            case "stop_requested":
                handle_stop(task, next); 
                break;
            case "running":
                handle_running(task, next); 
                break;
            default:
                handle_housekeeping(task, next);
            }
            set_nextdate(task);
            //logger.debug("reset next_date: "+task.next_date);
            task.save();
        }, function(err) {
            if(err) logger.error(err);
            //wait a bit and recheck again
            setTimeout(check, 500);
        });
    });
}

function handle_housekeeping(task, cb) {
    //.. if to remove, iterate through all resources that this task has run or synchornized to)
    //logger.debug("handling book keeping "+task._id);

    async.series([
        //remove task dir?
        function(next) {
            var need_remove = false;
            
            //check for early remove specified by user
            var now = new Date();
            if(task.remove_date && task.remove_date < now) {
                logger.info("remove_date is set and task is passed the date");
                need_remove = true;
            }
            
            //check for max life time
            var maxage = new Date();
            maxage.setDate(now.getDate() - 25); //25 days max (TODO - use resource's configured data)
            if(task.request_date < maxage) {
                logger.info("this task was requested more than specified days ago..");
                need_remove = true;
            }
        
            if(!need_remove) return next();

            //if job is still running/running_sync, then I need to first stop it
            if(task.status == "running") {
                task.status = "stop_requested";
                task.status_msg = "";
                task.save(next);
                return;
            }
            if(task.status == "running_sync") {
                logger.debug("TODO - can't remove task that's still running_sync - maybe stuck?");
                return next();
            }

            logger.info("need to remove this task. resource_ids.length:"+task.resource_ids.length);
            async.forEach(task.resource_ids, function(resource_id, next_resource) {
                db.Resource.findById(resource_id, function(err, resource) {
                    if(resource.status != "ok") {
                        logger.info("couldn't fail taskdir from resource_id:"+resource._id.toString()+" because resource status is not ok");
                        return next_resource();
                    }
                    common.get_ssh_connection(resource, function(err, conn) {
                        if(err) return next_resource(err);
                        var taskdir = common.gettaskdir(task.instance_id, task._id, resource);
                        if(!taskdir || taskdir.length < 10) return next_resource("taskdir looks odd.. bailing");
                        logger.info("rm -rf "+taskdir);
                        conn.exec("rm -rf "+taskdir, function(err, stream) {
                            if(err) return next_resource(err);
                            stream.on('close', function(code, signal) {
                                if(code) return next_resource("Failed to remove taskdir "+taskdir);
                                else next_resource();
                            })
                            .on('data', function(data) {
                                logger.info(data.toString());
                            }).stderr.on('data', function(data) {
                                logger.error(data.toString());
                            });
                        });<commit_msg>handled a case where resource status is not yet reported<commit_after>#!/usr/bin/node
'use strict';

//node
const fs = require('fs');
const path = require('path');
const os = require('os');
const request = require('request');

//contrib
const winston = require('winston');
const async = require('async');
const Client = require('ssh2').Client;

//mine
const config = require('../config');
const logger = new winston.Logger(config.logger.winston);
const db = require('../api/models/db');
const common = require('../api/common');
const _resource_picker = require('../api/resource').select;
const _transfer = require('../api/transfer');
const _service = require('../api/service');

db.init(function(err) {
    if(err) throw err;
    //start check loop
    check(); 
});

//set next_date incrementally longer between each checks
function set_nextdate(task) {
    task.next_date = new Date();
    if(task.start_date) {
        var start = task.start_date.getTime();
        var elapsed = task.next_date.getTime() - start;
        //logger.debug("elapsed seconds: "+elapsed);
        var next = task.next_date.getTime() + elapsed/2;
        task.next_date.setTime(next);
    } else {
        //not yet started. check again in 10 minutes (maybe resource issue?)
        //maybe I should increase the delay to something like an hour?
        task.next_date.setMinutes(task.next_date.getMinutes() + 10);
    }
}

function check() {
    //logger.info("checking..");
    db.Task.find({
        status: {$ne: "removed"}, //ignore removed tasks
        $or: [
            {next_date: {$exists: false}},
            {next_date: {$lt: new Date()}}
        ] 
    })
    
    //maybe I should do these later (only needed by requested task)
    .populate('deps', 'status resource_id')
    .populate('resource_deps') 
    .exec((err, tasks) => {
        if(err) throw err;
        async.eachSeries(tasks, (task, next) => {
            logger.info("handling task:"+task._id);
            switch(task.status) {
            case "requested":
                handle_requested(task, next); 
                break;
            case "stop_requested":
                handle_stop(task, next); 
                break;
            case "running":
                handle_running(task, next); 
                break;
            default:
                handle_housekeeping(task, next);
            }
            set_nextdate(task);
            //logger.debug("reset next_date: "+task.next_date);
            task.save();
        }, function(err) {
            if(err) logger.error(err);
            //wait a bit and recheck again
            setTimeout(check, 500);
        });
    });
}

function handle_housekeeping(task, cb) {
    //.. if to remove, iterate through all resources that this task has run or synchornized to)
    //logger.debug("handling book keeping "+task._id);

    async.series([
        //remove task dir?
        function(next) {
            var need_remove = false;
            
            //check for early remove specified by user
            var now = new Date();
            if(task.remove_date && task.remove_date < now) {
                logger.info("remove_date is set and task is passed the date");
                need_remove = true;
            }
            
            //check for max life time
            var maxage = new Date();
            maxage.setDate(now.getDate() - 25); //25 days max (TODO - use resource's configured data)
            if(task.request_date < maxage) {
                logger.info("this task was requested more than specified days ago..");
                need_remove = true;
            }
        
            if(!need_remove) return next();

            //if job is still running/running_sync, then I need to first stop it
            if(task.status == "running") {
                task.status = "stop_requested";
                task.status_msg = "";
                task.save(next);
                return;
            }
            if(task.status == "running_sync") {
                logger.debug("TODO - can't remove task that's still running_sync - maybe stuck?");
                return next();
            }

            logger.info("need to remove this task. resource_ids.length:"+task.resource_ids.length);
            async.forEach(task.resource_ids, function(resource_id, next_resource) {
                db.Resource.findById(resource_id, function(err, resource) {
                    if(!resource.status || resource.status != "ok") {
                        logger.info("couldn't test taskdir on resource_id:"+resource._id.toString()+" because resource status is not ok");
                        return next_resource();
                    }
                    common.get_ssh_connection(resource, function(err, conn) {
                        if(err) return next_resource(err);
                        var taskdir = common.gettaskdir(task.instance_id, task._id, resource);
                        if(!taskdir || taskdir.length < 10) return next_resource("taskdir looks odd.. bailing");
                        logger.info("rm -rf "+taskdir);
                        conn.exec("rm -rf "+taskdir, function(err, stream) {
                            if(err) return next_resource(err);
                            stream.on('close', function(code, signal) {
                                if(code) return next_resource("Failed to remove taskdir "+taskdir);
                                else next_resource();
                            })
                            .on('data', function(data) {
                                logger.info(data.toString());
                            }).stderr.on('data', function(data) {
                                logger.error(data.toString());
                            });
                        });
{'fraction_non_alphanumeric': 0.12381860747420445, 'fraction_numerical': 0.002601231249458077, 'mean_word_length': 1.6967500584521862, 'pattern_counts': {'":': 6, '<': 9, '<?xml version=': 0, '>': 7, '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': '3572837', 'n_tokens_mistral': 3101, 'n_tokens_neox': 2970, 'n_words': 780}