mysql-statsd-backend/mysql-backend.js

///////////////////////////////////////////////////////////////////////////////////
//  NodeJS Statsd MySQL Backend 0.1.0-alpha1
// ------------------------------------------------------------------------------
//
// Authors: Nicolas FRADIN, Damien PACAUD
// Date: 31/10/2012
//
///////////////////////////////////////////////////////////////////////////////////

var _mysql = require('mysql'),
    util = require('util'),
    fs = require('fs');

var STATSD_PACKETS_RECEIVED = "statsd.packets_received";
var STATSD_BAD_LINES = "statsd.bad_lines_seen";


/**
 * Backend Constructor
 *
 * Example config :
 *
  mysql: { 
   host: "localhost", 
   port: 3306, 
   user: "root", 
   password: "root", 
   database: "statsd_db",
   tables: ["statsd_users", "statsd_statistics"]
  }
 *
 * @param startupTime
 * @param config
 * @param emmiter
 */
function StatdMySQLBackend(startupTime, config, emitter) {
  var self = this;
  self.config = config.mysql || {};
  self.engines = {
    counters: [],
    gauges: [],
    timers: [],
    sets: []
  };

  // Verifying that the config file contains enough information for this backend to work  
  if(!this.config.host || !this.config.database || !this.config.user) {
    console.log("You need to specify at least host, port, database and user for this mysql backend");
    process.exit(-1);
  }

  // Default port for mysql is 3306, if unset in conf file, we set it here to default
  if(!this.config.port) {
    this.config.port = 3306;
  }

  // Set backend path
  for(var backend_index in config.backends) {
    var currentBackend = config.backends[backend_index];
    if(currentBackend.indexOf('mysql-backend.js') > -1) {
      self.config.backendPath = currentBackend.substring(0, currentBackend.lastIndexOf('/')+1);
    }
  }

  //Default tables
  if(!this.config.tables) {
    this.config.tables = {counters: ["counters_statistics"]};
  }

  // Default engines
  if(!self.config.engines) {
    self.config.engines = {
      counters: ["engines/countersEngine.js"],
      gauges: [],
      timers: [],
      sets: []
    };
  }
  

  // Check if tables exists
  self.checkDatabase();

  // Load backend engines
  self.loadEngines();

  // Attach events
  emitter.on('flush', function(time_stamp, metrics) { self.onFlush(time_stamp, metrics); } );
  emitter.on('status', self.onStatus );
}


/**
 * Load MySQL Backend Query Engines
 *
 */
StatdMySQLBackend.prototype.loadEngines = function() {
	var self = this;

  // Iterate on each engine type defined in configuration
  for(var engineType in self.config.engines) {
    var typeEngines = self.config.engines[engineType];

    // Load engines for current type
    for(var engineIndex in typeEngines) {
      // Get current engine path
      var enginePath = typeEngines[engineIndex];

      // Load current engine
      var currentEngine = require(self.config.backendPath + enginePath).init();
      if(currentEngine === undefined) {
        console.log("Unable to load engine '" + enginePath + "' ! Please check...");
        process.exit(-1);
      }
      // Add engine to MySQL Backend engines
      self.engines.counters.push(currentEngine);
    }
  }

}


/**
 * Open MySQL connection
 *
 * @return boolean Indicates if connection succeed
 */
StatdMySQLBackend.prototype.openMySqlConnection = function() {
  var self = this;
  var canExecuteQuerries = true;

  // Create MySQL connection
  self.sqlConnection = _mysql.createConnection(this.config);
  self.sqlConnection.connect(function(error){
    canExecuteQuerries = false;
  });

  return canExecuteQuerries;
}


/**
 * Close MySQL connection
 *
 */
StatdMySQLBackend.prototype.closeMySqlConnection = function() {
  var self = this;
  self.sqlConnection.end(function(error) {
    if(error){
      console.log("There was an error while trying to close DB connection : " + util.inspect(error));
      //Let's make sure that socket is destroyed
      self.sqlConnection.destroy();
    }
  });

  return;
}



/**
 * Check if required tables are created. If not create them.
 *
 */
StatdMySQLBackend.prototype.checkDatabase = function() {
  var self = this;

  var isConnected = self.openMySqlConnection();
  if(!isConnected) {
    console.log("Unable to connect to MySQL database ! Please check...");
    process.exit(-1);
  }

  // Iterate on each stat type (counters, gauges, ...)
  var tables = self.config.tables
  for(var statType in tables) {

    // Get tables for current stat type
    var typeTables = tables[statType];

    // Iterate on each table for current stat type
    for(var table_index in typeTables) {
      var table_name = typeTables[table_index];
      console.log("Check if table exists : '" + table_name + "'");
      self.sqlConnection.query('show tables like "'+table_name+'";', function(err, results, fields) {
        if(err) {
          console.log("Unbale to execute query !");
          process.exit(-1);
        }

        // If table doesn't exists
        if(results.length > 0) {
          console.log("Table '" + table_name + "' was found !");
        } else {
          console.log("Table '" + table_name + "' was not found !");

          // Try to read SQL file for this table
          var sqlFilePath = self.config.backendPath + 'tables/' + table_name + '.sql';
          fs.readFile(sqlFilePath, 'utf8', function (err,data) {
            if (err) {
              console.log("Unable to read file: '" + sqlFilePath + "' ! Exit...");
              process.exit(-1);
            }

            // Split querries
            var querries = data.split("$$");

            // Execute each query
            for(var queryIndex in querries) {
              var query = querries[queryIndex];
              if(query.trim() == "") continue;
              self.sqlConnection.query(query, function(err, results, fields) {
                if(err) {
                  console.log("Unable to execute query: '" + query +"' for table '"+table_name+"' ! Exit...");
                  process.exit(-1);
                } 

              });
            }
            console.log("Table '" + table_name +"' was created with success.");
            
          });
        }
      });
    }

  }

}



/**
 * Method executed when statsd flush received datas
 *
 * @param time_stamp
 * @param metrics
 */
StatdMySQLBackend.prototype.onFlush = function(time_stamp, metrics) {
  var self = this;

  var counters = metrics['counters'];
  var timers = metrics['timers'];  
  var gauges = metrics['gauges'];
  var sets = metrics['sets'];
  var pctThreshold = metrics['pctThreshold'];

  // Handle statsd counters
  self.handleCounters(counters,time_stamp);
  
}



/**
 * Handle and process received counters 
 * 
 * @param _counters received counters
 * @param time_stamp flush time_stamp 
 */
StatdMySQLBackend.prototype.handleCounters = function(_counters, time_stamp) {
  
  var self = this;

  var packets_received = parseInt(_counters[STATSD_PACKETS_RECEIVED]);
  var bad_lines_seen = parseInt(_counters[STATSD_BAD_LINES]);

  if(packets_received > 0) {
    // Get userCounters for this flush
    var userCounters = self.getUserCounters(_counters);
    var querries = [];

    console.log("Preaparing querries...");

    // Open MySQL connection
    var canExecuteQuerries = self.openMySqlConnection();
    if(canExecuteQuerries) {

      //////////////////////////////////////////////////////////////////////
      // Call buildQuerries method on each counterEngine
      for(var countersEngineIndex in self.engines.counters) {
        console.log("countersEngineIndex = " + countersEngineIndex);
        var countersEngine = self.engines.counters[countersEngineIndex];

        // Add current engine querries to querries list
        var engineQuerries = countersEngine.buildQuerries(userCounters, time_stamp);
        querries = querries.concat(engineQuerries);

        // Insert data into database every 100 query
        if(querries.length >= 100) {
          // Execute querries
          self.executeQuerries(querries);
          querries = [];
        }

      }

      if(querries.length > 0) {
        // Execute querries
        self.executeQuerries(querries);
        querries = [];
      }
    }

    // Close MySQL Connection
    self.closeMySqlConnection();
    
  return;

}
  


StatdMySQLBackend.prototype.executeQuerries = function(sqlQuerries) {
  
  var self = this;

  for(var i = 0 ; i < sqlQuerries.length ; i++){
    console.log("Query " + i + " : " + sqlQuerries[i]);
    self.sqlConnection.query(sqlQuerries[i], function(err, rows) {
      if(!err) {
        console.log(" -> Query [SUCCESS]");
      }
      else {
        //TODO : add better error handling code
        console.log(" -> Query [ERROR]"); 
      }
    });  
  }

}



/**
 *
 *
 */
StatdMySQLBackend.prototype.getUserCounters = function(_counters) {
  var userCounters = {};
  for(var counterName in _counters) {
    var counterNameParts = counterName.split('.');
    if(counterNameParts[0] !== "statsd") {
      userCounters[counterName] = _counters[counterName];
    }
  }
  return userCounters;
}



/**
 *
 *
 */
StatdMySQLBackend.prototype.getStatsdCounters = function(_counters) {
  var statsdCounters = {};
  for(var counterName in _counters) {
    var counterNameParts = counterName.split('.');
    if(counterNameParts[0] === "statsd") {
      statsdCounters[counterName] = _counters[counterName];
    }
  }
  return statsdCounters;
}


/**
 *
 * @param error
 * @param backend_name
 * @param stat_name
 * @param stat_value
 */
StatdMySQLBackend.prototype.onStatus = function(error, backend_name, stat_name, stat_value) {

}



exports.init = function(startupTime, config, events) {
  var instance = new StatdMySQLBackend(startupTime, config, events);
  return true;
};