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 curlify = require('request-as-curl');

var mongo = require('mongodb'),
    MongoClient = mongo.MongoClient,
    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) {
  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);
  console.log(curlify(req, req.body));
  res.send('https://grafana.com/plugins/grafana-simple-json-datasource\n');
  res.end();
});

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

app.all('/query', function (req, res) {
    console.log(curlify(req, req.body));
    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;
    });
    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

	var collections = []
    _.each(names, function(name) {
        _.each(["gauges"], function(prefix){
            collections.push({coll : name, prefix: prefix, name: name});
        })
    });
    
	console.log(collections);
	var results = collections.map(
        function(coll) {
            return  new Promise((resolve, reject) => {
                
                db.collection(coll.coll).find({ time: { $gte: from_str, $lte: to_str } }, { type: 1, time: 1, gauge: 1, $slice: maxDataPoints }).sort({ time: 1 }).toArray(function (err, docs) {
                    if (err) {console.log(err);reject(err);}
                        
                    console.time('loop');
                    var name = coll.name;        
                    var result = {};
                    result[coll.name] = new Array();
    		        console.log("query success " + name + " found " + docs.length);
    		        var prefix = coll.prefix;
    		
                    for (var x = 0; x < docs.length; x++) {
                        var doc = docs[x];
                        var value = doc.gauge;
                        (result[name]).push([value, 1000 * doc.time]);
                    }
    
                    var data = {target: name,datapoints: result[name]};

                    console.timeEnd('loop');
    		        resolve(data);
                    //mongo_query_result.push(data); 
                })  //find...
                
	       }); //promise
	}); //map

	Promise.all(results).then(mongo_query_result=> { 
		setCORSHeaders(res);
		var rtr = [];
		for(var i = 0; i < mongo_query_result.length; i++){
			rtr = rtr.concat(mongo_query_result[i]);
		}
		rtr = rtr.filter(function(v){ return v.datapoints.length > 0;});
		res.json(rtr);
		res.end();

	}, reason => {
	  console.log(reason)
	  res.status(500).send(reason);
	});
	    });
});

app.all('/last', function (req, res) {
    console.log(curlify(req, req.body));
    var mongo_query_result = [];
    var names = _.map(req.body.targets, function (t) {
        return t.target;
    });
    
    var maxDataPoints = req.body.maxDataPoints;

    mongo.connect("mongodb://" + options.host + "/" + options.name, function (err, db) {
        if (err) {console.log(err);}
    
        var collections = []
        _.each(names, function(name) {
            _.each(["gauges"], function(prefix){
                collections.push({coll : name, prefix: prefix, name: name});
            })
        });
        
        //console.log(collections);
        var results = collections.map(
            function(coll) {
                return  new Promise((resolve, reject) => {
                    
                    db.collection(coll.coll).find({}, { time: 1, gauge: 1 }).limit(maxDataPoints).sort({ time: -1 }).toArray(function (err, docs) {
                        if (err) {console.log(err);reject(err);}
                            
                        console.time('loop');
                        var name = coll.name;        
                        var result = {};
                        result[coll.name] = new Array();
                        console.log("query success " + name + " found " + docs.length);
                        var prefix = coll.prefix;
                
                        for (var x = 0; x < docs.length; x++) {
                            var doc = docs[x];
                            var value = doc.gauge;
                            (result[name]).push([value, 1000 * doc.time]);
                        }
        
                        var data = {target: name,datapoints: result[name]};
    
                        console.timeEnd('loop');
                        resolve(data);
                        //mongo_query_result.push(data); 
                    })  //find...
                    
               }); //promise
        }); //map
    
        Promise.all(results).then(mongo_query_result=> { 
            setCORSHeaders(res);
            var rtr = [];
            for(var i = 0; i < mongo_query_result.length; i++){
                rtr = rtr.concat(mongo_query_result[i]);
            }
            rtr = rtr.filter(function(v){ return v.datapoints.length > 0;});
            res.json(rtr);
            res.end();
    
        }, reason => {console.log(reason); res.status(500).send(reason);});
    });
});

app.listen(8000);

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