Commit 7ae5bd8f authored by Alan Szlosek's avatar Alan Szlosek

re-work timeseries API to make it easier to avoid calculations on partial time buckets

parent d886de94
/*
1. Our sample metric is called "sample"
2. Every second, calculate the average of the metric values over the past 10 seconds
3. Alert when the average is over 50
4. Populate the metric with random integers every 500ms
*/
var Timeseries = require('./lib/timeseries');
var log = require('./lib/log')('example-timeseries', true);
var moment = require('moment');
......@@ -8,49 +14,53 @@ function getRandomInteger(min, max) {
};
var redisClient = redis.createClient(6379, 'localhost');
var t = new Timeseries(redisClient, 'woof', 60);
/*
Error count
- can push values into lists, where each key is for a second
- when doing stats calculations
- determine start and end timestamp
- pull all values into a single array and run calculations
*/
// Instantiate a timeseries dataset called "sample" that preserves 60 seconds worth of data
var metric = new Timeseries(redisClient, 'sample', 60);
var calc = function() {
var endDateTime = new Date();
var startDateTime = moment(endDateTime).subtract(10, 'seconds').toDate();
t.getCountAverage(
startDateTime,
endDateTime,
function(err, avg) {
if (err) {
log(err);
return;
}
log('Average', avg);
redisClient.quit();
}
);
};
var averageInterval;
var countInterval;
var num = 0;
var stop = 5;
var callback = function() {
var rand = getRandomInteger(1, 100);
t.addCount(new Date(), rand);
var done = function() {
clearInterval(averageInterval);
clearInterval(countInterval);
redisClient.quit();
};
var handle = setInterval(
// Calculate the average of the metric every second, and print the average if it reaches our threshold
var averageInterval = setInterval(
function() {
callback();
// Exclude the current second to avoid partial calculations
var startSeconds = moment().subtract(11, 'seconds').unix();
var endSeconds = startSeconds + 10;
metric.getCountAverage(
startSeconds,
endSeconds,
function(err, avg, secondsSpanned) {
if (err) {
log(err);
done();
return;
}
log('Dataset spans ' + secondsSpanned + ' seconds');
num++;
if (num > stop) {
clearInterval(handle);
calc();
}
if (secondsSpanned == 10 && avg > 50) {
log('Average above 50', avg);
done();
}
}
);
},
1000
);
// Fill the metric with values every 500ms
var countInterval = setInterval(
function() {
var rand = getRandomInteger(1, 100);
metric.addCount(moment().unix(), rand);
},
500
);
......
var log = require('./log')('timeseries', true);
var log = require('./log')('timeseries', false);
/*
NOTES
......@@ -7,6 +7,12 @@ It's possible we can leverage zset and lists in redis for count values.
Perhaps push values onto a list, and increment a score for the second timestamp for each item pushed.
Can we then fetch lexicographically from the zset, get the scores, calculate the list offsets and use lrange?
Might result in our lrange being off by a bit if addCount operations happen between our zset and list fetches.
FEATURES
* All functions should return number of values that were present in Redis
* All functions should return the number of seconds that we have data for. You don't want to alert on a "metric above average" if there's only 1 data point.
*/
......@@ -21,9 +27,8 @@ module.exports = Timeseries;
// Use a zset to sum values into time buckets
Timeseries.prototype.addValue = function(dateTime, type, value) {
Timeseries.prototype.addValue = function(secondsBucket, type, value) {
var self = this;
var secondsBucket = Math.floor(dateTime.getTime() / 1000);
var key = this.key + ':' + type + ':' + secondsBucket;
log('lpush', key, value);
this.redisClient.lpush(
......@@ -35,8 +40,8 @@ Timeseries.prototype.addValue = function(dateTime, type, value) {
);
};
Timeseries.prototype.addCount = function(dateTime, value) {
this.addValue(dateTime, 'count', value);
Timeseries.prototype.addCount = function(secondsBucket, value) {
this.addValue(secondsBucket, 'count', value);
};
......@@ -44,14 +49,15 @@ Timeseries.prototype.addCount = function(dateTime, value) {
Still unsure how to handle multiple types yet: counters, timers, etc
But this method could be used by getAverage() to get the sum and number of items
type: count, timer
*/
Timeseries.prototype.getValues = function(startDateTime, endDateTime, type, callback) {
Timeseries.prototype.getValues = function(startSeconds, endSeconds, type, callback) {
var self = this;
var endSeconds = Math.floor(endDateTime.getTime() / 1000);
var startSeconds = Math.floor(startDateTime.getTime() / 1000);
var earliestSeconds = Number.MAX_SAFE_INTEGER;
var keys = [];
for (var i = startSeconds; i <= endSeconds; i++) {
keys.push( this.key + ':' + type + ':' + i);
keys.push( i );
//keys.push( this.key + ':' + type + ':' + i);
}
var out = [];
......@@ -60,10 +66,11 @@ Timeseries.prototype.getValues = function(startDateTime, endDateTime, type, call
if (keys.length == 0) {
// Done
// Calculate the average now
callback(null, out);
callback(null, out, endSeconds - earliestSeconds);
return;
}
key = keys.shift();
seconds = keys.shift();
key = self.key + ':' + type + ':' + seconds;
// Fetch from redis
log('fetching: ', key);
......@@ -72,6 +79,9 @@ Timeseries.prototype.getValues = function(startDateTime, endDateTime, type, call
callback(err);
return;
}
if (values.length > 0) {
earliestSeconds = Math.min(earliestSeconds, seconds);
}
log(values);
for (var i = 0; i < values.length; i++) {
out.push( parseFloat(values[i]) );
......@@ -85,26 +95,27 @@ Timeseries.prototype.getValues = function(startDateTime, endDateTime, type, call
};
Timeseries.prototype.getSum = function(startDateTime, endDateTime, type, callback) {
Timeseries.prototype.getSum = function(startSeconds, endSeconds, type, callback) {
var self = this;
var endSeconds = Math.floor(endDateTime.getTime() / 1000);
var startSeconds = Math.floor(startDateTime.getTime() / 1000);
var earliestSeconds = Number.MAX_SAFE_INTEGER;
var keys = [];
for (var i = startSeconds; i <= endSeconds; i++) {
keys.push( this.key + ':' + type + ':' + i);
keys.push( i );
}
var sum = 0.00;
var itemCount = 0;
var next = function() {
var seconds;
var key;
if (keys.length == 0) {
// Done
// Calculate the average now
callback(null, sum, itemCount);
callback(null, sum, itemCount, endSeconds - earliestSeconds);
return;
}
key = keys.shift();
seconds = keys.shift();
key = self.key + ':' + type + ':' + seconds;
// Fetch from redis
log('fetching: ', key);
......@@ -114,6 +125,9 @@ Timeseries.prototype.getSum = function(startDateTime, endDateTime, type, callbac
return;
}
log(values);
if (values.length > 0) {
earliestSeconds = Math.min(earliestSeconds, seconds);
}
itemCount += values.length;
for (var i = 0; i < values.length; i++) {
sum += parseFloat(values[i]);
......@@ -126,18 +140,18 @@ Timeseries.prototype.getSum = function(startDateTime, endDateTime, type, callbac
next();
};
Timeseries.prototype.getCountAverage = function(startDateTime, endDateTime, callback) {
Timeseries.prototype.getCountAverage = function(startSeconds, endSeconds, callback) {
// Should we cache this result?
this.getSum(
startDateTime,
endDateTime,
startSeconds,
endSeconds,
'count',
function(err, sum, items) {
function(err, sum, items, secondsSpanned) {
if (err) {
callback(err);
return;
}
callback(null, sum / items);
callback(null, sum / items, secondsSpanned);
}
);
};
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment