<commit_before>var util = require('util'),
    dgram = require('dgram'),
    queue = require('../lib/priority_queue')

function StatsDedupe(startupTime, config, emitter) {
	var self = this
	self.config = config.statsdedupe || {};
	self.statsd_config = config;
	self.metrics = {};
	self.queue = queue.PriorityQueue({low: true});
	self.hosts = config.statsdedupe.hosts || [{
		address: config.address || '127.0.0.1',
		port: config.port || 8125
	}];
	self.default_retention = self.config.default_retention || '60000';
	
	emitter.on('flush', function(time, metrics) { self.flush(time, metrics); });
	emitter.on('packet', function(packet, rinfo) { self.decode(packet, rinfo); });
}

StatsDedupe.prototype.send = function(packet, host, port) {
  var self = this
  self.sock.send(packet, 0, packet.length, port, host, function(err,bytes) {
    if (err) {
      console.log(err)
    }
  })
}

StatsDedupe.prototype.format = function (key, value, suffix) {
  return new Buffer("'" + key + "':" + value + "|" + suffix)
}

StatsDedupe.prototype.flush = function(time, metrics) {
	var self = this
	hosts = self.hosts;
	var stats, packet
	
	self.sock = dgram.createSocket("udp4")
	self.evict();
	for (var i = 0; i < hosts.length; i++) {
		for(key in self.metrics) {
			var set_size = Object.keys(self.metrics[key]).length;
			packet = self.format(key, set_size, "g")
			self.send(packet, hosts[i].host, hosts[i].port)
		}
	}
}

StatsDedupe.prototype.flush_one = function(key) {
	var self = this
	hosts = self.hosts;
	var stats, packet
	self.sock = dgram.createSocket("udp4")
	self.evict();
	for (var i = 0; i < hosts.length; i++) {
		var set_size = Object.keys(self.metrics[key]).length;
		packet = self.format(key, set_size, "g")
		self.send(packet, hosts[i].host, hosts[i].port)
	}
}

StatsDedupe.prototype.evict = function() {
	var self = this;
	var date = new Date();
	var now = date.getTime();
	while(!self.queue.empty() && self.queue.top().priority < now) {
		var obj = self.queue.pop();
		if(obj.priority >= self.metrics[obj.key][obj.value]) {
			delete self.metrics[obj.key][obj.value];
		}
	}
}

StatsDedupe.prototype.decode = function(msg, rinfo) {
	var self = this;
	var metrics = msg.toString().split("\n");
	var json_metrics;
	var date = new Date();
	var epoch_offset = date.getTime();
	for (var midx in metrics) {
		var bits = metrics[midx].toString().split(':');
		var key = bits.shift()
			.replace(/\s+/g, '_')
			.replace(/\//g, '-')
			.replace(/[^a-zA-Z_\-0-9\.]/g, '');
		for (var i = 0; i < bits.length; i++) {
			var fields = bits[i].split("|")
			if (fields[1] === undefined) {
				continue;
			}
			var suffix = fields[1].trim();
			if (suffix == "s") {
				var value = fields[0] || 0;
				var retention = self.default_retention;
				if(fields[2] !== undefined ) {
					if(fields[2].match(/^#(\d+)/)) {
						var matches = fields[2].match(/^#(\d+)(\w*)/);
						var unit = matches[2];
						retention = Number(matches[1])*self.unit_to_milliseconds(unit);
					}
					else if(fields[3] !== undefined) {
						var matches = fields[3].match(/^#(\d+)(\w*)/);
						var unit = matches[2];
						if(fields[3].match(/^#(\d+)/)) {
							retention = Number(matches[1])*self.unit_to_milliseconds(unit);
						}
					}
				}
				var new_key = false;<commit_msg>Fixed default unit selection<commit_after>var util = require('util'),
    dgram = require('dgram'),
    queue = require('../lib/priority_queue')

function StatsDedupe(startupTime, config, emitter) {
	var self = this
	self.config = config.statsdedupe || {};
	self.statsd_config = config;
	self.metrics = {};
	self.queue = queue.PriorityQueue({low: true});
	self.hosts = config.statsdedupe.hosts || [{
		address: config.address || '127.0.0.1',
		port: config.port || 8125
	}];
	self.default_retention = self.config.default_retention || '60000';
	
	emitter.on('flush', function(time, metrics) { self.flush(time, metrics); });
	emitter.on('packet', function(packet, rinfo) { self.decode(packet, rinfo); });
}

StatsDedupe.prototype.send = function(packet, host, port) {
  var self = this
  self.sock.send(packet, 0, packet.length, port, host, function(err,bytes) {
    if (err) {
      console.log(err)
    }
  })
}

StatsDedupe.prototype.format = function (key, value, suffix) {
  return new Buffer("'" + key + "':" + value + "|" + suffix)
}

StatsDedupe.prototype.flush = function(time, metrics) {
	var self = this
	hosts = self.hosts;
	var stats, packet
	
	self.sock = dgram.createSocket("udp4")
	self.evict();
	for (var i = 0; i < hosts.length; i++) {
		for(key in self.metrics) {
			var set_size = Object.keys(self.metrics[key]).length;
			packet = self.format(key, set_size, "g")
			self.send(packet, hosts[i].host, hosts[i].port)
		}
	}
}

StatsDedupe.prototype.flush_one = function(key) {
	var self = this
	hosts = self.hosts;
	var stats, packet
	self.sock = dgram.createSocket("udp4")
	self.evict();
	for (var i = 0; i < hosts.length; i++) {
		var set_size = Object.keys(self.metrics[key]).length;
		packet = self.format(key, set_size, "g")
		self.send(packet, hosts[i].host, hosts[i].port)
	}
}

StatsDedupe.prototype.evict = function() {
	var self = this;
	var date = new Date();
	var now = date.getTime();
	while(!self.queue.empty() && self.queue.top().priority < now) {
		var obj = self.queue.pop();
		if(obj.priority >= self.metrics[obj.key][obj.value]) {
			delete self.metrics[obj.key][obj.value];
		}
	}
}

StatsDedupe.prototype.decode = function(msg, rinfo) {
	var self = this;
	var metrics = msg.toString().split("\n");
	var json_metrics;
	var date = new Date();
	var epoch_offset = date.getTime();
	for (var midx in metrics) {
		var bits = metrics[midx].toString().split(':');
		var key = bits.shift()
			.replace(/\s+/g, '_')
			.replace(/\//g, '-')
			.replace(/[^a-zA-Z_\-0-9\.]/g, '');
		for (var i = 0; i < bits.length; i++) {
			var fields = bits[i].split("|")
			if (fields[1] === undefined) {
				continue;
			}
			var suffix = fields[1].trim();
			if (suffix == "s") {
				var value = fields[0] || 0;
				var retention = self.default_retention;
				if(fields[2] !== undefined ) {
					if(fields[2].match(/^#(\d+)/)) {
						var matches = fields[2].match(/^#(\d+)(\w*)/);
						var unit = matches[2] || 'ms';
						retention = Number(matches[1])*self.unit_to_milliseconds(unit);
					}
					else if(fields[3] !== undefined) {
						var matches = fields[3].match(/^#(\d+)(\w*)/);
						var unit = matches[2] || 'ms';
						if(fields[3].match(/^#(\d+)/)) {
							retention = Number(matches[1])*self.unit_to_milliseconds(unit);
						}
					}
				}
				var new_key = false;
{'fraction_non_alphanumeric': 0.19545662448543985, 'fraction_numerical': 0.011282207653605733, 'mean_word_length': 3.4747612551159617, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '7521561', 'n_tokens_mistral': 2686, 'n_tokens_neox': 2386, 'n_words': 486}