Installer/statsd/endpoint/mongodb/index.js

'use strict';

var express = require('express');
var morgan = require('morgan');
var bodyParser = require('body-parser');
var _ = require('lodash');
var app = express();

var mongo = require('mongodb'),
    MongoClient = mongo.MongoClient,
    async = require('async'),
    util = require('util'),
    dbs = {},
    options = {
            debug: false,
            prefix: true,
            size: 100,
            max: 2610,
            name: 'statsd',
            host: 'mongodb',
            port: 27017
    };

app.use(morgan('combined'));
app.use(bodyParser.json());

function setCORSHeaders(res) {
    if (res) {        
        res.setHeader("Access-Control-Allow-Origin", "*");
        res.setHeader("Access-Control-Allow-Methods", "POST");
        res.setHeader("Access-Control-Allow-Headers", "accept, content-type");  
    }
}

app.all('/', function(req, res) {
  setCORSHeaders(res);
  res.send('https://grafana.com/plugins/grafana-simple-json-datasource\n');
  res.end();
});

app.all('/search', function (req, res) {
    var target = req.body.target;
    
    mongo.connect("mongodb://" + options.host + "/" + options.name, function (err, db) {
        if (err) {
            console.log(err);
        }
        if (options.debug) {
            console.log("Connected successfully to server");
        }
        db.listCollections().toArray(function(err, collInfos) {
            var mongo_search_result = [];
            _.each(collInfos, function(collInfo) {
                if (collInfo.name.indexOf(target) && mongo_search_result.indexOf(collInfo.name) === -1) {
                    mongo_search_result.push(collInfo.name);
                }
            });
            
            setCORSHeaders(res);
            res.json(mongo_search_result);
            res.end();
        });
    });
});

// vars global
var global_result;
var global_res;
var global_names;
var checkInterval;

app.all('/query', function (req, res) {
    var mongo_query_result = [];
    var from = new Date(req.body.range.from);
    var to = new Date(req.body.range.to);
    var from_str = Math.floor(from.getTime() / 1000);
    var to_str = Math.floor(to.getTime() / 1000);
    var names = _.map(req.body.targets, function (t) {
        return t.target;
    });
    global_names = names;
    global_res = res;
    var interval = req.body.intervalMs / 1000;
    var maxDataPoints = req.body.maxDataPoints;

    mongo.connect("mongodb://" + options.host + "/" + options.name, function (err, db) {
        if (err) {
            console.log(err);
        }
//        https://docs.mongodb.com/manual/reference/method/db.collection.find/#db.collection.find
        _.each(names, function(name, index) {
            db.collection(""+name).find({ time: { $gte: from_str, $lte: to_str } }, { type: 1, time: 1, count:1, durations: 1, gauge: 1, set: 1, $slice: maxDataPoints }).sort({ time: 1 }).toArray(function (err, docs) {
                if (err) {
                    console.log(err);
                }
                var result = {};
                result[name] = new Array();
                _.each(docs, function (doc) {
                    var value = 0;
                    if (doc.type == 'counters') {
                        value = doc.count;
                    } else if (doc.type == 'timers') {
                        value = doc.durations;
                    } else if (doc.type == 'gauges') {
                        value = doc.gauge;
                    } else if (doc.type == 'sets') {
                        value = doc.set;
                    }
                    (result[name]).push([value, 1000 * doc.time]);
                });
                var data = {
                    target: name,
                    datapoints: result[name]
                };
                mongo_query_result.push(data);
                global_result = mongo_query_result;
            });
            if (index === names.length -1) {
                checkInterval = setInterval(checkResult, 1000);
            }
        });
        
    });
});

function checkResult()
{
    if (global_result.length !== global_names.length) {
        return false;
    }
    
    clearInterval(checkInterval);
    
    if (global_res) {
        setCORSHeaders(global_res);
        global_res.json(global_result);
        global_res.end();
    }
    
    global_res = null;
    
    return true;
}

app.listen(8000);

console.log("Server is listening to port 8000");