<commit_before>/*    Copyright 2016 Rottiesoft LLC 
 *
 *    This program is free software: you can redistribute it and/or  modify
 *    it under the terms of the GNU Affero General Public License, version 3,
 *    as published by the Free Software Foundation.
 *
 *    This program is distributed in the hope that it will be useful,
 *    but WITHOUT ANY WARRANTY; without even the implied warranty of
 *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *    GNU Affero General Public License for more details.
 *
 *    You should have received a copy of the GNU Affero General Public License
 *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */
'use strict';
var log;
var ip = require('ip');
var os = require('os');
var network = require('network');
var stats = require('stats-lite');

var redis = require("redis");
var rclient = redis.createClient();

var SysManager = require('./SysManager.js');
var sysManager = new SysManager('info');
var DNSManager = require('./DNSManager.js');
var dnsManager = new DNSManager('info');
var bone = require("../lib/Bone.js");

rclient.on("error", function (err) {
    console.log("Redis(alarm) Error " + err);
});

var async = require('async');
var instance = null;

var maxflow = 10000;

class FlowGraph {
    constructor(name,flowarray) {
         if (flowarray) {
             this.flowarray = flowarray;
         } else {
             this.flowarray = [];
         }
         this.name = name;
    }


    flowarraySorted(recent) {
        if (recent == true) {
             this.flowarray.sort(function (a, b) {
                 return Number(b[1]) - Number(a[1]);
             })
             return this.flowarray;
        } else {
             this.flowarray.sort(function (a, b) {
                 return Number(a[1]) - Number(b[1]);
             })
             return this.flowarray;
        }
    }


    addFlow(flow) {
         if (flow.flows == null) {
             let flowStart = Math.ceil(Number(flow.__ts) - Number(flow.du));
             let flowEnd = Math.ceil(Number(flow.__ts));
             if (flow.__ts==null) {
                 flowStart = Math.ceil(Number(flow._ts) - Number(flow.du));
                 flowEnd = Math.ceil(Number(flow._ts));
             }
             let ob = Number(flow.ob);
             let rb = Number(flow.rb);
                
             this.addRawFlow(flowStart,flowEnd,ob,rb,flow.ct);
         } else {
             //console.log("$$$ before ",flow.flows);
             for (let i in flow.flows) {
                 let f = flow.flows[i];
                 this.addRawFlow(f[0],f[1],f[2],f[3],1);
             }
             //console.log("$$$ after",this.flowarray);
         }
    }

    addRawFlow(flowStart, flowEnd, ob,rb,ct) {
         let insertindex = 0;

         for (let i in this.flowarray) {
             let e = this.flowarray[i];
             if (flowStart < e[0]) {
                 break;
             }
             if (flowStart<e[1]) {
                 flowStart = e[0];
                 break;
             }
             insertindex = Number(i)+Number(1);
         }

         let removed = Number(0);
         for (let i = insertindex; i <this.flowarray.length;i++) {
             let e = this.flowarray[Number(i)];
             if (e[1]<flowEnd) {
                 ob += e[2];
                 rb += e[3];
                 ct += e[4];
                 removed++;
                 continue;
             } else if (e[1]>=flowEnd) {
                 ob += e[2];
                 rb += e[3];
                 ct += e[4];
                 flowEnd = e[1];
                 removed++;
                 break;
             }
         }

         this.flowarray.splice(insertindex,removed, [flowStart,flowEnd, ob,rb,ct]);
    //     console.log("insertindex",insertindex,"removed",removed,this.flowarray,"<=end");

    }

}

module.exports = class FlowManager {
    constructor(loglevel) {
        if (instance == null) {
            let cache = {};
            instance = this;
            log = require("./logger.js")("FlowManager", loglevel);
        }
        return instance;
    }

    // stats are 'hour', 'day'
    // stats:hour:ip_address score=bytes key=_ts-_ts%3600
    // ip = 0.0.0.0 is system
    recordStats(ip,type,ts,inBytes,outBytes,callback) {
        let period = 3600;
        if (type === "day") {
            period = 60*60*24;
        } else if (type ==="month") {
            period = 60*60*24*30;
        }
        let inkey = "stats:"+type+":in:"+ip;
        let outkey = "stats:"+type+":out:"+ip;
        let subkey = ts-ts%period;
        rclient.zincrby(inkey,Number(inBytes),subkey,(err,data)=>{
            rclient.zincrby(outkey,Number(outBytes),subkey,(err,data)=>{
                if (callback) {
                    callback(err);
                } 
            }); 
        }); 
    }

    getStats(iplist,type,from,to,callback) {
        let outdb = {};
        let indb = {};
        let inbytes = 0;
        let outbytes = 0;
        async.eachLimit(iplist, 1, (ip, cb) => {
            let inkey = "stats:"+type+":in:"+ip;
            let outkey = "stats:"+type+":out:"+ip;
            rclient.zscan(inkey,0,(err,data)=>{
                if (data && data.length==2) {
                    let array = data[1];
                    for (let i=0;i<array.length;i++) {
                        let clock = Number(array[i]);
                        let bytes = Number(array[i+1]);
                        i++;
                        if (clock<from) {
                            continue;
                        }
                        if (to!=-1 &&  clock>to) {
                            continue;
                        }
                        
                        if (indb[clock]) {
                            indb[clock] += bytes;
                        } else {
                            indb[clock] = bytes;
                        }
                        inbytes+=bytes;
                    }
                } 
                rclient.zscan(outkey,0,(err,data)=>{
                    if (data && data.length==2) {
                        let array = data[1];
                        for (let i=0;i<array.length;i++) {
                            let clock = Number(array[i]);
                            let bytes = Number(array[i+1]);
                            i++;
                            if (clock<from) {
                                continue;
                            }
                            if (outdb[clock]) {
                                outdb[clock] += bytes;
                            } else {
                                outdb[clock] = bytes;
                            }<commit_msg>fixed zscan not returning all stats<commit_after>/*    Copyright 2016 Rottiesoft LLC 
 *
 *    This program is free software: you can redistribute it and/or  modify
 *    it under the terms of the GNU Affero General Public License, version 3,
 *    as published by the Free Software Foundation.
 *
 *    This program is distributed in the hope that it will be useful,
 *    but WITHOUT ANY WARRANTY; without even the implied warranty of
 *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *    GNU Affero General Public License for more details.
 *
 *    You should have received a copy of the GNU Affero General Public License
 *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */
'use strict';
var log;
var ip = require('ip');
var os = require('os');
var network = require('network');
var stats = require('stats-lite');

var redis = require("redis");
var rclient = redis.createClient();

var SysManager = require('./SysManager.js');
var sysManager = new SysManager('info');
var DNSManager = require('./DNSManager.js');
var dnsManager = new DNSManager('info');
var bone = require("../lib/Bone.js");

rclient.on("error", function (err) {
    console.log("Redis(alarm) Error " + err);
});

var async = require('async');
var instance = null;

var maxflow = 10000;

class FlowGraph {
    constructor(name,flowarray) {
         if (flowarray) {
             this.flowarray = flowarray;
         } else {
             this.flowarray = [];
         }
         this.name = name;
    }


    flowarraySorted(recent) {
        if (recent == true) {
             this.flowarray.sort(function (a, b) {
                 return Number(b[1]) - Number(a[1]);
             })
             return this.flowarray;
        } else {
             this.flowarray.sort(function (a, b) {
                 return Number(a[1]) - Number(b[1]);
             })
             return this.flowarray;
        }
    }


    addFlow(flow) {
         if (flow.flows == null) {
             let flowStart = Math.ceil(Number(flow.__ts) - Number(flow.du));
             let flowEnd = Math.ceil(Number(flow.__ts));
             if (flow.__ts==null) {
                 flowStart = Math.ceil(Number(flow._ts) - Number(flow.du));
                 flowEnd = Math.ceil(Number(flow._ts));
             }
             let ob = Number(flow.ob);
             let rb = Number(flow.rb);
                
             this.addRawFlow(flowStart,flowEnd,ob,rb,flow.ct);
         } else {
             //console.log("$$$ before ",flow.flows);
             for (let i in flow.flows) {
                 let f = flow.flows[i];
                 this.addRawFlow(f[0],f[1],f[2],f[3],1);
             }
             //console.log("$$$ after",this.flowarray);
         }
    }

    addRawFlow(flowStart, flowEnd, ob,rb,ct) {
         let insertindex = 0;

         for (let i in this.flowarray) {
             let e = this.flowarray[i];
             if (flowStart < e[0]) {
                 break;
             }
             if (flowStart<e[1]) {
                 flowStart = e[0];
                 break;
             }
             insertindex = Number(i)+Number(1);
         }

         let removed = Number(0);
         for (let i = insertindex; i <this.flowarray.length;i++) {
             let e = this.flowarray[Number(i)];
             if (e[1]<flowEnd) {
                 ob += e[2];
                 rb += e[3];
                 ct += e[4];
                 removed++;
                 continue;
             } else if (e[1]>=flowEnd) {
                 ob += e[2];
                 rb += e[3];
                 ct += e[4];
                 flowEnd = e[1];
                 removed++;
                 break;
             }
         }

         this.flowarray.splice(insertindex,removed, [flowStart,flowEnd, ob,rb,ct]);
    //     console.log("insertindex",insertindex,"removed",removed,this.flowarray,"<=end");

    }

}

module.exports = class FlowManager {
    constructor(loglevel) {
        if (instance == null) {
            let cache = {};
            instance = this;
            log = require("./logger.js")("FlowManager", loglevel);
        }
        return instance;
    }

    // stats are 'hour', 'day'
    // stats:hour:ip_address score=bytes key=_ts-_ts%3600
    // ip = 0.0.0.0 is system
    recordStats(ip,type,ts,inBytes,outBytes,callback) {
        let period = 3600;
        if (type === "day") {
            period = 60*60*24;
        } else if (type ==="month") {
            period = 60*60*24*30;
        }
        let inkey = "stats:"+type+":in:"+ip;
        let outkey = "stats:"+type+":out:"+ip;
        let subkey = ts-ts%period;
        rclient.zincrby(inkey,Number(inBytes),subkey,(err,data)=>{
            rclient.zincrby(outkey,Number(outBytes),subkey,(err,data)=>{
                if (callback) {
                    callback(err);
                } 
            }); 
        }); 
    }

    getStats(iplist,type,from,to,callback) {
        let outdb = {};
        let indb = {};
        let inbytes = 0;
        let outbytes = 0;
        let lotsofkeys = 24*60;
        async.eachLimit(iplist, 1, (ip, cb) => {
            let inkey = "stats:"+type+":in:"+ip;
            let outkey = "stats:"+type+":out:"+ip;
            rclient.zscan(inkey,0,'count',lotsofkeys,(err,data)=>{
                if (data && data.length==2) {
                    let array = data[1];
                    for (let i=0;i<array.length;i++) {
                        let clock = Number(array[i]);
                        let bytes = Number(array[i+1]);
                        i++;
                        if (clock<from) {
                            continue;
                        }
                        if (to!=-1 &&  clock>to) {
                            continue;
                        }
                        
                        if (indb[clock]) {
                            indb[clock] += bytes;
                        } else {
                            indb[clock] = bytes;
                        }
                        inbytes+=bytes;
                    }
                } 
                rclient.zscan(outkey,0,'count',lotsofkeys,(err,data)=>{
                    if (data && data.length==2) {
                        let array = data[1];
                        for (let i=0;i<array.length;i++) {
                            let clock = Number(array[i]);
                            let bytes = Number(array[i+1]);
                            i++;
                            if (clock<from) {
                                continue;
                            }
                            if (outdb[clock]) {
                                outdb[clock] += bytes;
                            } else {
                                outdb[clock] = bytes;
                            }
{'fraction_non_alphanumeric': 0.13214917452830188, 'fraction_numerical': 0.011202830188679245, 'mean_word_length': 1.2974940738232306, 'pattern_counts': {'":': 8, '<': 23, '<?xml version=': 0, '>': 19, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 2, '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': '2314858', 'n_tokens_mistral': 3934, 'n_tokens_neox': 3767, 'n_words': 852}