<commit_before>/*    Copyright 2021 Firewalla Inc.
 *
 *    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';

const flowTool = require('../net2/FlowTool');
const log = require('../net2/logger.js')(__filename)
const _ = require('lodash');
const Sensor = require('./Sensor.js').Sensor
const fc = require('../net2/config.js');
const featureName = 'compress_flows'
const Promise = require('bluebird');
const zlib = require('zlib');
const extensionManager = require('./ExtensionManager.js')
const { rclient } = require('../util/redis_manager');
const deflateAsync = Promise.promisify(zlib.deflate);
const sem = require('./SensorEventManager.js').getInstance();
const MAX_MEM = 10 * 1000 * 1000

class FlowCompressionSensor extends Sensor {
  constructor() {
    super()
    this.recentlyTickKey = "compressed:flows:lastest:ts"
    this.step = 60 * 60 // one hour
    this.maxInterval = 24 * 60 * 60 // 24 hours
  }

  async run() {
    sem.once('IPTABLES_READY', async () => {
      await this.build()
      setInterval(async () => {
        await this.build()
      }, 30 * 60 * 1000);
    })
  }

  async checkAndCleanMem() {
    let compressedFlowsKeys = await rclient.scanResults(this.getKey("*", "*"), 1000)
    if (compressedFlowsKeys && compressedFlowsKeys.length > 0) {
      compressedFlowsKeys = compressedFlowsKeys.filter(key => key != this.recentlyTickKey).sort((a, b) => {
        const ts1 = a.split(":")[2];
        const ts2 = b.split(":")[2];
        return ts1 > ts2 ? -1 : 2
      })
      let compressedMem = 0
      let delFlag = false
      for (const key of compressedFlowsKeys) {
        if (delFlag) { // delete all earlier keys
          await rclient.delAsync(key);
          continue;
        }
        const mem = Number(await rclient.memoryAsync("usage", key) || 0)
        compressedMem += mem
        if (compressedMem > MAX_MEM) { // accumulate memory size from the latest
          delFlag = true;<commit_msg>fix typo<commit_after>/*    Copyright 2021 Firewalla Inc.
 *
 *    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';

const flowTool = require('../net2/FlowTool');
const log = require('../net2/logger.js')(__filename)
const _ = require('lodash');
const Sensor = require('./Sensor.js').Sensor
const fc = require('../net2/config.js');
const featureName = 'compress_flows'
const Promise = require('bluebird');
const zlib = require('zlib');
const extensionManager = require('./ExtensionManager.js')
const { rclient } = require('../util/redis_manager');
const deflateAsync = Promise.promisify(zlib.deflate);
const sem = require('./SensorEventManager.js').getInstance();
const MAX_MEM = 10 * 1000 * 1000

class FlowCompressionSensor extends Sensor {
  constructor() {
    super()
    this.recentlyTickKey = "compressed:flows:lastest:ts"
    this.step = 60 * 60 // one hour
    this.maxInterval = 24 * 60 * 60 // 24 hours
  }

  async run() {
    sem.once('IPTABLES_READY', async () => {
      await this.build()
      setInterval(async () => {
        await this.build()
      }, 30 * 60 * 1000);
    })
  }

  async checkAndCleanMem() {
    let compressedFlowsKeys = await rclient.scanResults(this.getKey("*", "*"), 1000)
    if (compressedFlowsKeys && compressedFlowsKeys.length > 0) {
      compressedFlowsKeys = compressedFlowsKeys.filter(key => key != this.recentlyTickKey).sort((a, b) => {
        const ts1 = a.split(":")[2];
        const ts2 = b.split(":")[2];
        return ts1 > ts2 ? -1 : 1
      })
      let compressedMem = 0
      let delFlag = false
      for (const key of compressedFlowsKeys) {
        if (delFlag) { // delete all earlier keys
          await rclient.delAsync(key);
          continue;
        }
        const mem = Number(await rclient.memoryAsync("usage", key) || 0)
        compressedMem += mem
        if (compressedMem > MAX_MEM) { // accumulate memory size from the latest
          delFlag = true;
{'fraction_non_alphanumeric': 0.12329579134558388, 'fraction_numerical': 0.020944477376012644, 'mean_word_length': 3.0047468354430378, 'pattern_counts': {'":': 4, '<': 5, '<?xml version=': 0, '>': 19, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '6864658', 'n_tokens_mistral': 1565, 'n_tokens_neox': 1477, 'n_words': 470}