<commit_before>/*    Copyright 2016 Firewalla 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';

let log = require('./logger.js')(__filename);

let redis = require('redis');
let rclient = redis.createClient();

let Promise = require('bluebird');
Promise.promisifyAll(redis.RedisClient.prototype);
Promise.promisifyAll(redis.Multi.prototype);

let async = require('asyncawait/async');
let await = require('asyncawait/await');

let async2 = require('async');

let util = require('util');

let instance = null;

const MAX_FLOW_PER_AGGR = 2000
const MAX_FLOW_PER_SUM = 30000
const MAX_FLOW_PER_HOUR = 7000

function toInt(n){ return Math.floor(Number(n)); };

class FlowAggrTool {
  constructor() {
    if(!instance) {
      instance = this;
    }
    return instance;
  }

  toFloorInt(n) {
    return toInt(n);
  }

  getFlowKey(mac, trafficDirection, interval, ts) {
    return util.format("aggrflow:%s:%s:%s:%s", mac, trafficDirection, interval, ts);
  }

  getSumFlowKey(mac, trafficDirection, begin, end) {
    if(mac) {
      return util.format("sumflow:%s:%s:%s:%s", mac, trafficDirection, begin, end);
    } else {
      return util.format("syssumflow:%s:%s:%s", trafficDirection, begin, end);
    }
  }

  // aggrflow:<device_mac>:download:10m:<ts>
  flowExists(mac, trafficDirection, interval, ts) {
    let key = this.getFlowKey(mac, trafficDirection, interval, ts);
    return rclient.keysAsync(key)
      .then((results) => {
        return results.length === 1;
      })
  }

  // key: aggrflow:<device_mac>:<direction>:<interval>:<ts>
  // content: destination ip address
  // score: traffic size
  addFlow(mac, trafficDirection, interval, ts, destIP, traffic) {
    let key = this.getFlowKey(mac, trafficDirection, interval, ts);
    return rclient.zaddAsync(key, traffic, destIP);
  }

  addAppActivityFlows(mac, interval, ts, traffics, expire) {
    return this.addXActivityFlows(mac, "app", interval, ts, traffics, expire)
  }

  addCategoryActivityFlows(mac, interval, ts, traffics, expire) {
    return this.addXActivityFlows(mac, "category", interval, ts, traffics, expire)
  }

  addXActivityFlows(mac, x, interval, ts, traffics, expire) {
    expire = expire || 24 * 3600; // by default keep 24 hours

    let key = this.getFlowKey(mac, x, interval, ts);
    let args = [key];
    for(let t in traffics) {
      let duration = (traffics[t] && traffics[t]['duration']) || 0;
      args.push(duration)

      let payload = {}
      payload.device = mac
      payload[x] = t
      args.push(JSON.stringify(payload))
    }

    args.push(0);
    args.push("_"); // placeholder to keep key exists

    return rclient.zaddAsync(args)
      .then(() => {
        return rclient.expireAsync(key, expire)
      });
  }

  // this is to make sure flow data is not flooded enough to consume all memory
  trimFlow(mac, trafficDirection, interval, ts) {
    return async(() => {
      const key = this.getFlowKey(mac, trafficDirection, interval, ts);

      let count = await (rclient.zremrangebyscoreAsync(key, 0, -1 * MAX_FLOW_PER_AGGR)) // only keep the MAX_FLOW_PER_SUM highest flows
      if(count > 0) {
        log.warn(`${count} flows are removed from ${key} for self protection`)
      }
    })()
  }
  
  addFlows(mac, trafficDirection, interval, ts, traffics, expire) {
    expire = expire || 24 * 3600; // by default keep 24 hours

    const length = Object.keys(traffics).length // number of dest ips in this aggr flow
    const key = this.getFlowKey(mac, trafficDirection, interval, ts);

    let args = [key];

    if(length > MAX_FLOW_PER_AGGR) { // self protection      
      args.push(length)
      args.push(JSON.stringify({
        device: mac,
        destIP: "0.0.0.0"       // special ip address to indicate some flows were skipped due to overflow protection
      }))
    }

    for(let destIP in traffics) {
      let traffic = (traffics[destIP] && traffics[destIP][trafficDirection]) || 0;
      args.push(traffic)
      args.push(JSON.stringify({
        device: mac,
        destIP: destIP
      }))
    }

    args.push(0);
    args.push("_"); // placeholder to keep key exists
<commit_msg>bug bug<commit_after>/*    Copyright 2016 Firewalla 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';

let log = require('./logger.js')(__filename);

let redis = require('redis');
let rclient = redis.createClient();

let Promise = require('bluebird');
Promise.promisifyAll(redis.RedisClient.prototype);
Promise.promisifyAll(redis.Multi.prototype);

let async = require('asyncawait/async');
let await = require('asyncawait/await');

let async2 = require('async');

let util = require('util');

let instance = null;

const MAX_FLOW_PER_AGGR = 2000
const MAX_FLOW_PER_SUM = 30000
const MAX_FLOW_PER_HOUR = 7000

function toInt(n){ return Math.floor(Number(n)); };

class FlowAggrTool {
  constructor() {
    if(!instance) {
      instance = this;
    }
    return instance;
  }

  toFloorInt(n) {
    return toInt(n);
  }

  getFlowKey(mac, trafficDirection, interval, ts) {
    return util.format("aggrflow:%s:%s:%s:%s", mac, trafficDirection, interval, ts);
  }

  getSumFlowKey(mac, trafficDirection, begin, end) {
    if(mac) {
      return util.format("sumflow:%s:%s:%s:%s", mac, trafficDirection, begin, end);
    } else {
      return util.format("syssumflow:%s:%s:%s", trafficDirection, begin, end);
    }
  }

  // aggrflow:<device_mac>:download:10m:<ts>
  flowExists(mac, trafficDirection, interval, ts) {
    let key = this.getFlowKey(mac, trafficDirection, interval, ts);
    return rclient.keysAsync(key)
      .then((results) => {
        return results.length === 1;
      })
  }

  // key: aggrflow:<device_mac>:<direction>:<interval>:<ts>
  // content: destination ip address
  // score: traffic size
  addFlow(mac, trafficDirection, interval, ts, destIP, traffic) {
    let key = this.getFlowKey(mac, trafficDirection, interval, ts);
    return rclient.zaddAsync(key, traffic, destIP);
  }

  addAppActivityFlows(mac, interval, ts, traffics, expire) {
    return this.addXActivityFlows(mac, "app", interval, ts, traffics, expire)
  }

  addCategoryActivityFlows(mac, interval, ts, traffics, expire) {
    return this.addXActivityFlows(mac, "category", interval, ts, traffics, expire)
  }

  addXActivityFlows(mac, x, interval, ts, traffics, expire) {
    expire = expire || 24 * 3600; // by default keep 24 hours

    let key = this.getFlowKey(mac, x, interval, ts);
    let args = [key];
    for(let t in traffics) {
      let duration = (traffics[t] && traffics[t]['duration']) || 0;
      args.push(duration)

      let payload = {}
      payload.device = mac
      payload[x] = t
      args.push(JSON.stringify(payload))
    }

    args.push(0);
    args.push("_"); // placeholder to keep key exists

    return rclient.zaddAsync(args)
      .then(() => {
        return rclient.expireAsync(key, expire)
      });
  }

  // this is to make sure flow data is not flooded enough to consume all memory
  trimFlow(mac, trafficDirection, interval, ts) {
    return async(() => {
      const key = this.getFlowKey(mac, trafficDirection, interval, ts);

      let count = await (rclient.zremrangebyrankAsync(key, 0, -1 * MAX_FLOW_PER_AGGR)) // only keep the MAX_FLOW_PER_SUM highest flows
      if(count > 0) {
        log.warn(`${count} flows are removed from ${key} for self protection`)
      }
    })()
  }
  
  addFlows(mac, trafficDirection, interval, ts, traffics, expire) {
    expire = expire || 24 * 3600; // by default keep 24 hours

    const length = Object.keys(traffics).length // number of dest ips in this aggr flow
    const key = this.getFlowKey(mac, trafficDirection, interval, ts);

    let args = [key];

    if(length > MAX_FLOW_PER_AGGR) { // self protection
      args.push(length)
      args.push(JSON.stringify({
        device: mac,
        destIP: "0.0.0.0"       // special ip address to indicate some flows were skipped due to overflow protection
      }))
    }

    for(let destIP in traffics) {
      let traffic = (traffics[destIP] && traffics[destIP][trafficDirection]) || 0;
      args.push(traffic)
      args.push(JSON.stringify({
        device: mac,
        destIP: destIP
      }))
    }

    args.push(0);
    args.push("_"); // placeholder to keep key exists

{'fraction_non_alphanumeric': 0.12540974939198477, 'fraction_numerical': 0.010362694300518135, 'mean_word_length': 3.2488769092542675, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 27, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1264337', 'n_tokens_mistral': 3071, 'n_tokens_neox': 2936, 'n_words': 859}