aggregate.js

function insertSomeData(callback) {
    db.event.insert({
        date: 1451606400000, // 2016-01-01T00:00:00Z
        action: 'A',
        src_ip: '10.1.1.1',
        src_port: 11111,
        dst_ip: '10.1.1.2',
        dst_port: 80
    });
    
    db.event.insert({
        date: 1451606460000, // 2016-01-01T00:01:00Z
        action: 'R',
        src_ip: '10.1.1.1',
        src_port: 22222,
        dst_ip: '10.1.1.2',
        dst_port: 80
    });
    
    db.event.insert({
        date: 1451606520000, // 2016-01-01T00:02:00Z
        action: 'A',
        src_ip: '10.1.1.1',
        src_port: 33333,
        dst_ip: '10.1.1.2',
        dst_port: 22
    });
    
    db.event.insert({
        date: 1451606580000, // 2016-01-01T00:03:00Z
        action: 'R',
        src_ip: '10.1.1.1',
        src_port: 33333,
        dst_ip: '10.1.1.2',
        dst_port: 22
    });
    
    db.event.insert({
        date: 1451606640000, // 2016-01-01T00:04:00Z
        action: 'A',
        src_ip: '10.1.1.3',
        src_port: 33333,
        dst_ip: '10.1.1.4',
        dst_port: 443
    });
    
    if (callback && typeof callback === 'function') {
        callback();    
    }
}

function getTopAcceptedWebDestIps (timeMillisStart, limit) {
    limit = limit || 10;
    
    return db.event.aggregate([
        { $match: { date: { $gte: timeMillisStart }, action: 'A', dst_port: { $in: [ 80, 443 ] } } },
        { $group: { _id: { src_ip: '$src_ip', dst_ip: '$dst_ip' }, hits: { $sum: 1 } } },
        { $sort:  { hits: -1, dst_ip: -1, src_ip: -1 } },
        { $limit: limit }
    ]);
}

function printData () {
    var cursor = getTopAcceptedWebDestIps(lastHour);
    cursor.forEach(function(doc) {
        print('A: ' + doc['_id']['src_ip'] + ' -> ' + doc['_id']['dst_ip'] + '(:80|:443) = ' + doc['hits']);
    });
}

var minuteInMillis = 60000;
var fiveMinutes    =  5 * minuteInMillis;
var tenMinutes     = 10 * minuteInMillis;
var hour           = 60 * minuteInMillis;

var now             = new Date('2016-01-01T00:05:00Z').getTime();
var lastMinute      = now - minuteInMillis;
var lastFiveMinutes = now - fiveMinutes;
var lastTenMinutes  = now - tenMinutes;
var lastHour        = now - hour;

db.event.createIndex({
    action: -1,
    dst_port: -1,
    date: 1
});

insertSomeData(printData);