Merge pull request #533 from isocolsky/ref/stats
Improve stats performance
This commit is contained in:
commit
21afe4d9b7
267
lib/stats.js
267
lib/stats.js
|
@ -14,14 +14,17 @@ var moment = require('moment');
|
||||||
var config = require('../config');
|
var config = require('../config');
|
||||||
var storage = require('./storage');
|
var storage = require('./storage');
|
||||||
|
|
||||||
|
|
||||||
|
var INITIAL_DATE = '2015-01-01';
|
||||||
|
|
||||||
function Stats(opts) {
|
function Stats(opts) {
|
||||||
opts = opts || {};
|
opts = opts || {};
|
||||||
|
|
||||||
this.network = opts.network || 'livenet';
|
this.network = opts.network || 'livenet';
|
||||||
this.from = moment(opts.from || '2015-01-01');
|
this.from = moment(opts.from || INITIAL_DATE);
|
||||||
this.to = moment(opts.to);
|
this.to = moment(opts.to);
|
||||||
this.fromTs = Math.floor(this.from.startOf('day').valueOf() / 1000);
|
this.fromTs = this.from.startOf('day').valueOf();
|
||||||
this.toTs = Math.floor(this.to.endOf('day').valueOf() / 1000);
|
this.toTs = this.to.endOf('day').valueOf();
|
||||||
};
|
};
|
||||||
|
|
||||||
Stats.prototype.run = function(cb) {
|
Stats.prototype.run = function(cb) {
|
||||||
|
@ -62,91 +65,239 @@ Stats.prototype._getStats = function(cb) {
|
||||||
});
|
});
|
||||||
};
|
};
|
||||||
|
|
||||||
Stats.prototype._countBy = function(data, key) {
|
|
||||||
return _.map(_.groupBy(data, key), function(v, k) {
|
|
||||||
var item = {};
|
|
||||||
item[key] = k;
|
|
||||||
item['count'] = v.length;
|
|
||||||
return item;
|
|
||||||
});
|
|
||||||
};
|
|
||||||
|
|
||||||
Stats.prototype._sumBy = function(data, key, attr) {
|
|
||||||
return _.map(_.groupBy(data, key), function(v, k) {
|
|
||||||
var item = {};
|
|
||||||
item[key] = k;
|
|
||||||
item[attr] = _.reduce(v, function(memo, x) {
|
|
||||||
return memo + x[attr];
|
|
||||||
}, 0);
|
|
||||||
return item;
|
|
||||||
});
|
|
||||||
};
|
|
||||||
|
|
||||||
Stats.prototype._getNewWallets = function(cb) {
|
Stats.prototype._getNewWallets = function(cb) {
|
||||||
var self = this;
|
var self = this;
|
||||||
|
|
||||||
self.db.collection(storage.collections.WALLETS)
|
function getLastDate(cb) {
|
||||||
.find({
|
self.db.collection('stats_wallets')
|
||||||
network: self.network,
|
.find({})
|
||||||
|
.sort({
|
||||||
|
'_id.day': -1
|
||||||
|
})
|
||||||
|
.limit(1)
|
||||||
|
.toArray(function(err, lastRecord) {
|
||||||
|
if (_.isEmpty(lastRecord)) return cb(null, moment(INITIAL_DATE));
|
||||||
|
return cb(null, moment(lastRecord[0]._id.day));
|
||||||
|
});
|
||||||
|
};
|
||||||
|
|
||||||
|
function updateStats(from, cb) {
|
||||||
|
var to = moment().subtract(1, 'day').endOf('day');
|
||||||
|
var map = function() {
|
||||||
|
var day = new Date(this.createdOn * 1000);
|
||||||
|
day.setHours(0);
|
||||||
|
day.setMinutes(0);
|
||||||
|
day.setSeconds(0);
|
||||||
|
var key = {
|
||||||
|
day: +day,
|
||||||
|
network: this.network,
|
||||||
|
};
|
||||||
|
var value = {
|
||||||
|
count: 1
|
||||||
|
};
|
||||||
|
emit(key, value);
|
||||||
|
};
|
||||||
|
var reduce = function(k, v) {
|
||||||
|
var count = 0;
|
||||||
|
for (var i = 0; i < v.length; i++) {
|
||||||
|
count += v[i].count;
|
||||||
|
}
|
||||||
|
return {
|
||||||
|
count: count,
|
||||||
|
};
|
||||||
|
};
|
||||||
|
var opts = {
|
||||||
|
query: {
|
||||||
createdOn: {
|
createdOn: {
|
||||||
|
$gt: from.unix(),
|
||||||
|
$lte: to.unix(),
|
||||||
|
},
|
||||||
|
},
|
||||||
|
out: {
|
||||||
|
merge: 'stats_wallets',
|
||||||
|
}
|
||||||
|
};
|
||||||
|
self.db.collection(storage.collections.WALLETS)
|
||||||
|
.mapReduce(map, reduce, opts, function(err, collection, stats) {
|
||||||
|
return cb(err);
|
||||||
|
});
|
||||||
|
};
|
||||||
|
|
||||||
|
function queryStats(cb) {
|
||||||
|
self.db.collection('stats_wallets')
|
||||||
|
.find({
|
||||||
|
'_id.network': self.network,
|
||||||
|
'_id.day': {
|
||||||
$gte: self.fromTs,
|
$gte: self.fromTs,
|
||||||
$lte: self.toTs,
|
$lte: self.toTs,
|
||||||
},
|
},
|
||||||
})
|
})
|
||||||
.toArray(function(err, wallets) {
|
.sort({
|
||||||
|
'_id.day': 1
|
||||||
|
})
|
||||||
|
.toArray(function(err, results) {
|
||||||
if (err) return cb(err);
|
if (err) return cb(err);
|
||||||
|
var stats = {};
|
||||||
var data = _.map(wallets, function(wallet) {
|
stats.byDay = _.map(results, function(record) {
|
||||||
|
var day = moment(record._id.day).format('YYYYMMDD');
|
||||||
return {
|
return {
|
||||||
day: moment(wallet.createdOn * 1000).format('YYYYMMDD'),
|
day: day,
|
||||||
type: (wallet.m == 1 && wallet.n == 1) ? 'personal' : 'shared',
|
count: record.value.count,
|
||||||
config: wallet.m + '-of-' + wallet.n,
|
|
||||||
};
|
};
|
||||||
});
|
});
|
||||||
|
|
||||||
var stats = {
|
|
||||||
byDay: self._countBy(data, 'day'),
|
|
||||||
byConfig: self._countBy(data, 'config'),
|
|
||||||
};
|
|
||||||
|
|
||||||
stats.byTypeThenDay = _.groupBy(data, 'type');
|
|
||||||
_.each(stats.byTypeThenDay, function(v, k) {
|
|
||||||
stats.byTypeThenDay[k] = self._countBy(v, 'day');
|
|
||||||
});
|
|
||||||
|
|
||||||
return cb(null, stats);
|
return cb(null, stats);
|
||||||
});
|
});
|
||||||
};
|
};
|
||||||
|
|
||||||
|
async.series([
|
||||||
|
|
||||||
|
function(next) {
|
||||||
|
getLastDate(function(err, lastDate) {
|
||||||
|
if (err) return next(err);
|
||||||
|
|
||||||
|
lastDate = lastDate.startOf('day');
|
||||||
|
var yesterday = moment().subtract(1, 'day').startOf('day');
|
||||||
|
if (lastDate.isBefore(yesterday)) {
|
||||||
|
// Needs update
|
||||||
|
return updateStats(lastDate, next);
|
||||||
|
}
|
||||||
|
next();
|
||||||
|
});
|
||||||
|
},
|
||||||
|
function(next) {
|
||||||
|
queryStats(next);
|
||||||
|
},
|
||||||
|
],
|
||||||
|
function(err, res) {
|
||||||
|
if (err) {
|
||||||
|
log.error(err);
|
||||||
|
}
|
||||||
|
return cb(err, res[1]);
|
||||||
|
});
|
||||||
|
};
|
||||||
|
|
||||||
Stats.prototype._getTxProposals = function(cb) {
|
Stats.prototype._getTxProposals = function(cb) {
|
||||||
var self = this;
|
var self = this;
|
||||||
|
|
||||||
self.db.collection(storage.collections.TXS)
|
function getLastDate(cb) {
|
||||||
.find({
|
self.db.collection('stats_txps')
|
||||||
network: self.network,
|
.find({})
|
||||||
|
.sort({
|
||||||
|
'_id.day': -1
|
||||||
|
})
|
||||||
|
.limit(1)
|
||||||
|
.toArray(function(err, lastRecord) {
|
||||||
|
if (_.isEmpty(lastRecord)) return cb(null, moment(INITIAL_DATE));
|
||||||
|
return cb(null, moment(lastRecord[0]._id.day));
|
||||||
|
});
|
||||||
|
};
|
||||||
|
|
||||||
|
function updateStats(from, cb) {
|
||||||
|
var to = moment().subtract(1, 'day').endOf('day');
|
||||||
|
var map = function() {
|
||||||
|
var day = new Date(this.broadcastedOn * 1000);
|
||||||
|
day.setHours(0);
|
||||||
|
day.setMinutes(0);
|
||||||
|
day.setSeconds(0);
|
||||||
|
var key = {
|
||||||
|
day: +day,
|
||||||
|
network: this.network,
|
||||||
|
};
|
||||||
|
var value = {
|
||||||
|
count: 1,
|
||||||
|
amount: this.amount
|
||||||
|
};
|
||||||
|
emit(key, value);
|
||||||
|
};
|
||||||
|
var reduce = function(k, v) {
|
||||||
|
var count = 0,
|
||||||
|
amount = 0;
|
||||||
|
for (var i = 0; i < v.length; i++) {
|
||||||
|
count += v[i].count;
|
||||||
|
amount += v[i].amount;
|
||||||
|
}
|
||||||
|
return {
|
||||||
|
count: count,
|
||||||
|
amount: amount,
|
||||||
|
};
|
||||||
|
};
|
||||||
|
var opts = {
|
||||||
|
query: {
|
||||||
status: 'broadcasted',
|
status: 'broadcasted',
|
||||||
createdOn: {
|
broadcastedOn: {
|
||||||
|
$gt: from.unix(),
|
||||||
|
$lte: to.unix(),
|
||||||
|
},
|
||||||
|
},
|
||||||
|
out: {
|
||||||
|
merge: 'stats_txps',
|
||||||
|
}
|
||||||
|
};
|
||||||
|
self.db.collection(storage.collections.TXS)
|
||||||
|
.mapReduce(map, reduce, opts, function(err, collection, stats) {
|
||||||
|
return cb(err);
|
||||||
|
});
|
||||||
|
};
|
||||||
|
|
||||||
|
function queryStats(cb) {
|
||||||
|
self.db.collection('stats_txps')
|
||||||
|
.find({
|
||||||
|
'_id.network': self.network,
|
||||||
|
'_id.day': {
|
||||||
$gte: self.fromTs,
|
$gte: self.fromTs,
|
||||||
$lte: self.toTs,
|
$lte: self.toTs,
|
||||||
},
|
},
|
||||||
})
|
})
|
||||||
.toArray(function(err, txps) {
|
.sort({
|
||||||
|
'_id.day': 1
|
||||||
|
})
|
||||||
|
.toArray(function(err, results) {
|
||||||
if (err) return cb(err);
|
if (err) return cb(err);
|
||||||
|
|
||||||
var data = _.map(txps, function(txp) {
|
|
||||||
return {
|
|
||||||
day: moment(txp.createdOn * 1000).format('YYYYMMDD'),
|
|
||||||
amount: txp.amount,
|
|
||||||
};
|
|
||||||
});
|
|
||||||
|
|
||||||
var stats = {
|
var stats = {
|
||||||
nbByDay: self._countBy(data, 'day'),
|
nbByDay: [],
|
||||||
amountByDay: self._sumBy(data, 'day', 'amount'),
|
amountByDay: []
|
||||||
};
|
};
|
||||||
|
_.each(results, function(record) {
|
||||||
|
var day = moment(record._id.day).format('YYYYMMDD');
|
||||||
|
stats.nbByDay.push({
|
||||||
|
day: day,
|
||||||
|
count: record.value.count,
|
||||||
|
});
|
||||||
|
stats.amountByDay.push({
|
||||||
|
day: day,
|
||||||
|
amount: record.value.amount,
|
||||||
|
});
|
||||||
|
});
|
||||||
return cb(null, stats);
|
return cb(null, stats);
|
||||||
});
|
});
|
||||||
};
|
};
|
||||||
|
|
||||||
|
async.series([
|
||||||
|
|
||||||
|
function(next) {
|
||||||
|
getLastDate(function(err, lastDate) {
|
||||||
|
if (err) return next(err);
|
||||||
|
|
||||||
|
lastDate = lastDate.startOf('day');
|
||||||
|
var yesterday = moment().subtract(1, 'day').startOf('day');
|
||||||
|
if (lastDate.isBefore(yesterday)) {
|
||||||
|
// Needs update
|
||||||
|
return updateStats(lastDate, next);
|
||||||
|
}
|
||||||
|
next();
|
||||||
|
});
|
||||||
|
},
|
||||||
|
function(next) {
|
||||||
|
queryStats(next);
|
||||||
|
},
|
||||||
|
],
|
||||||
|
function(err, res) {
|
||||||
|
if (err) {
|
||||||
|
log.error(err);
|
||||||
|
}
|
||||||
|
return cb(err, res[1]);
|
||||||
|
});
|
||||||
|
};
|
||||||
|
|
||||||
module.exports = Stats;
|
module.exports = Stats;
|
||||||
|
|
Loading…
Reference in New Issue