GangliaLogger.scala

class GangliaLogger(config: ConfigMap) extends Thread {
  val frequency = config.getInt("ganglia_frequency", 15000)
  val gmetricExecutable = config.getString("gmetric_executable", "gmetric")
  val logJvmStats = config.getBool("log_jvm_stats", true)
  val logCounterStats = config.getBool("log_counter_stats", true)
  val logTimingStats = config.getBool("log_timing_stats", true)
  val logGaugeStats = config.getBool("log_gauge_stats", true)
  val quit = new CountDownLatch(1)
  val done = new CountDownLatch(1)
  val runtime = Runtime.getRuntime()
  val log = Logger.get("gmetric")

  override def run() = {
    log.info("Ganglia logger starting, dumping stats every %d ms", frequency)
    while (!quit.await(frequency, TimeUnit.MILLISECONDS)) {
      log.debug("dumping Ganglia stats")
      if (logJvmStats) logStats(Stats.getJvmStats())
      if (logCounterStats) logStats(Stats.getCounterStats())
      if (logTimingStats) logStats(Stats.getTimingStats(true))
      if (logGaugeStats) logStats(Stats.getGaugeStats(true))
    }
    done.countDown()
  }

  def logStats[T](stats: Map[String, T]) {
    for ((key, value) <- stats) {
      logGmetric(key, value.toString)
    }
  }

  def logGmetric(key: String, value: String) {
    try {
      val command ="%s -tuint32 -n%s -v%s".format(gmetricExecutable, key, value)
      log.debug("calling %s", command)
      val proc = runtime.exec(command)
      proc.waitFor()
      log.debug("gmetric exited with code %d", proc.exitValue())
      proc.destroy()
    } catch {
      case ex: IOException => log.warning("exception on gmetric " + ex.toString())
    }
  }
}