refactoring(reporter: influx): Use an asynchronous API for InfluxDB.

 Some crashes (timeouts) occurred while overloading the system with the native API for InfluxDB.
 We now use an asynchronous API.

Author: mcolmant

README.md

@@ -58,7 +58,7 @@ We all stand on the shoulders of giants and get by with a little help from our f
 * [Sigar](https://support.hyperic.com/display/SIGAR/Home) (version 1.6.5 under [Apache 2 license](http://www.apache.org/licenses/LICENSE-2.0)), for providing a portable interface for gathering system information.
 * [spray-json](http://spray.io/) (version 1.3.2 under [Apache 2 license](http://www.apache.org/licenses/LICENSE-2.0)), for (de)serializing JSON.
 * [docker-java](https://github.com/docker-java/docker-java) (version 2.1.4 under [Apache 2 license](http://www.apache.org/licenses/LICENSE-2.0)), for using the JAVA Docker API.
-* [influxdb-java](https://github.com/influxdata/influxdb-java) (version 2.1 under [MIT license](https://github.com/influxdata/influxdb-java/blob/master/LICENSE)), for using the JAVA InfluxDB API.
+* [scala-influxdb-client](https://github.com/paulgoldbaum/scala-influxdb-client) (version 0.4.5 under [MIT license](https://github.com/paulgoldbaum/scala-influxdb-client/blob/master/LICENSE)), for using an asynchronous scala API for InfluxDB.
 
 # License
 This software is licensed under the *GNU Affero General Public License*, quoted below.

powerapi-cli/src/main/scala/org/powerapi/app/PowerAPI.scala

@@ -24,6 +24,7 @@ package org.powerapi.app
 
 import java.lang.management.ManagementFactory
 
+import scala.concurrent.Await
 import scala.concurrent.duration.DurationInt
 import scala.sys
 import scala.sys.process.stringSeqToProcess
@@ -98,7 +99,7 @@ object PowerAPI extends App {
         |                            --frequency $MILLISECONDS
         |                            --self (0, 1) --pids [pid, ...] (0, *) --apps [app, ...] (0, *) --containers [id, ...] (0, *) | all (0, 1)
         |                            --agg max|min|geomean|logsum|mean|median|stdev|sum|variance
-        |                            --console (0, 1) --file $FILEPATH (0, *) --chart (0, 1) --influx $HOST $USER $PWD $DB $MEASUREMENT (0, *)
+        |                            --console (0, 1) --file $FILEPATH (0, *) --chart (0, 1) --influx $HOST $PORT $USER $PWD $DB $MEASUREMENT (0, *)
         |                  duration [s]
         |
         |example: ./powerapi modules procfs-cpu-simple monitor --frequency 1000 --apps firefox,chrome --agg max --console \
@@ -175,10 +176,12 @@ object PowerAPI extends App {
       cliMonitorsSubcommand(options, currentMonitor + ('displays ->
         (currentMonitor.getOrElse('displays, Set[Any]()).asInstanceOf[Set[Any]] + new JFreeChartDisplay)
       ), tail)
-    case "--influx" :: host :: user :: pwd :: db :: measurement :: tail =>
-      cliMonitorsSubcommand(options, currentMonitor + ('displays ->
-        (currentMonitor.getOrElse('displays, Set[Any]()).asInstanceOf[Set[Any]] + new InfluxDisplay(host, user, pwd, db, measurement))
-      ), tail)
+    case "--influx" :: host :: port :: user :: pwd :: db :: measurement :: tail =>
+      cliMonitorsSubcommand(options, currentMonitor + ('displays -> {
+        val influxDisplay = new InfluxDisplay(host, port.toInt, user, pwd, db, measurement)
+        Await.result(influxDisplay.database.create(), 30.seconds)
+        currentMonitor.getOrElse('displays, Set[Any]()).asInstanceOf[Set[Any]] + influxDisplay
+      }), tail)
     case option :: tail =>
       println(s"unknown monitor option $option")
       sys.exit(1)

powerapi-core/build.sbt

@@ -24,7 +24,7 @@ libraryDependencies ++= Seq(
   "net.java.dev.jna" % "jna" % "4.2.1",
   "io.spray" %% "spray-json" % "1.3.2",
   "com.github.docker-java" % "docker-java" % "2.1.4",
-  "org.influxdb" % "influxdb-java" % "2.1"
+  "com.paulgoldbaum" %% "scala-influxdb-client" % "0.4.5"
 )
 
 // Tests

powerapi-core/src/main/scala/org/powerapi/reporter/InfluxDisplay.scala

@@ -23,31 +23,29 @@
 package org.powerapi.reporter
 
 import java.util.UUID
-import java.util.concurrent.TimeUnit
 
-import collection.JavaConversions._
+import com.paulgoldbaum.influxdbclient.Parameter.Precision
+import com.paulgoldbaum.influxdbclient.{Point, InfluxDB}
 
-
-import org.influxdb.InfluxDBFactory
-import org.influxdb.dto.Point
 import org.powerapi.PowerDisplay
 import org.powerapi.core.power.Power
 import org.powerapi.core.target.Target
 
 /**
   * Write power information inside an InfluxDB database.
   */
-class InfluxDisplay(host: String, user: String, pwd: String, dbName: String, measurement: String) extends PowerDisplay {
+class InfluxDisplay(host: String, port: Int, user: String, pwd: String, dbName: String, measurement: String) extends PowerDisplay {
 
-  val influxdb = InfluxDBFactory.connect(host, user, pwd)
+  val influxdb = InfluxDB.connect(host, port, user, pwd)
+  val database = influxdb.selectDatabase(dbName)
 
-  def display(muid: UUID, timestamp: Long, targets: Set[Target], devices: Set[String], power: Power) {
-    val point = Point.measurement(measurement)
-      .time(timestamp, TimeUnit.MILLISECONDS)
-      .field("power", power.toMilliWatts)
-      .tag(Map("muid" -> s"$muid", "targets" -> s"${targets.mkString(",")}", "devices" -> s"${devices.mkString(",")}"))
-      .build()
+  def display(muid: UUID, timestamp: Long, targets: Set[Target], devices: Set[String], power: Power): Unit = {
+    val point = Point(measurement, timestamp)
+      .addField("power", power.toMilliWatts)
+      .addTag("muid", s"$muid")
+      .addTag("targets", s"${targets.mkString(",")}")
+      .addTag("devices", s"${devices.mkString(",")}")
 
-    influxdb.write(dbName, "default", point)
+    database.write(point, precision = Precision.MILLISECONDS)
   }
 }

powerapi-core/src/test/scala/org/powerapi/reporter/InfluxDisplaySuite.scala

@@ -23,17 +23,16 @@
 package org.powerapi.reporter
 
 import java.util.UUID
-import java.util.concurrent.TimeUnit
 
 import scala.concurrent.duration.DurationInt
-import collection.JavaConversions._
 
 import akka.util.Timeout
 
-import org.influxdb.dto.Query
+import org.joda.time.format.ISODateTimeFormat
 import org.powerapi.UnitTest
 import org.powerapi.core.power._
 import org.powerapi.core.target.{Application, Process, Target}
+import org.scalatest.time.{Seconds, Span}
 
 class InfluxDisplaySuite extends UnitTest {
 
@@ -50,13 +49,30 @@ class InfluxDisplaySuite extends UnitTest {
     val devices = Set[String]("cpu", "gpu", "ssd")
     val power = 10.W
 
-    val influxDisplay = new InfluxDisplay("http://localhost:8086", "powerapi", "powerapi", "test", "event.powerapi")
-    influxDisplay.influxdb.createDatabase("test")
-    influxDisplay.display(muid, timestamp, targets, devices, power)
-    val query = new Query("SELECT * FROM \"event.powerapi\"", "test")
-    val result = influxDisplay.influxdb.query(query, TimeUnit.MILLISECONDS)
-    result.getResults.head.getSeries.head.getValues.head should contain theSameElementsAs Seq(timestamp.toDouble, s"$muid", devices.mkString(","), targets.mkString(","), power.toMilliWatts)
-    influxDisplay.influxdb.deleteDatabase("test")
+    val influxDisplay = new InfluxDisplay("localhost", 8086, "powerapi", "powerapi", "test", "event.powerapi")
+
+    whenReady(influxDisplay.database.create(), timeout(Span(30, Seconds))) {
+      _ =>
+        influxDisplay.display(muid, timestamp, targets, devices, power)
+
+        awaitCond({
+          whenReady(influxDisplay.database.query("SELECT * FROM \"event.powerapi\"")) {
+            result =>
+              result.series.size == 1 &&
+              result.series.head.records.size == 1 &&
+              ISODateTimeFormat.dateTimeParser().parseDateTime(result.series.head.records.head("time").toString).getMillis == timestamp &&
+              result.series.head.records.head("devices") == devices.mkString(",") &&
+              result.series.head.records.head("muid") == s"$muid" &&
+              result.series.head.records.head("power") == power.toMilliWatts &&
+              result.series.head.records.head("targets") == targets.mkString(",")
+          }
+        }, 30.seconds, 1.seconds)
+
+        whenReady(influxDisplay.database.drop(), timeout(Span(30, Seconds))) {
+          _ =>
+            assert(true)
+        }
+    }
   }
 }
 

powerapi-daemon/src/main/scala/org/powerapi/daemon/PowerAPId.scala

@@ -22,6 +22,8 @@
  */
 package org.powerapi.daemon
 
+import scala.concurrent.Await
+import scala.concurrent.duration.DurationInt
 import scala.sys.process.stringSeqToProcess
 
 import org.apache.commons.daemon.{Daemon, DaemonContext}
@@ -131,9 +133,10 @@ class PowerAPId extends Daemon {
             val fileDisplay = new FileDisplay(file.split("=>")(1))
             monitor.to(fileDisplay)
           case influx: String if output.startsWith("influx") =>
-            // influx=>http://locahost:8086,powerapi,powerapi,test,event.powerapi
+            // influx=>locahost,8086,powerapi,powerapi,test,event.powerapi
             val parameters = influx.split("=>")(1).split(",")
-            val influxDisplay = new InfluxDisplay(parameters(0), parameters(1), parameters(2), parameters(3), parameters(4))
+            val influxDisplay = new InfluxDisplay(parameters(0), parameters(1).toInt, parameters(2), parameters(3), parameters(4), parameters(5))
+            Await.result(influxDisplay.database.create(), 30.seconds)
             monitor.to(influxDisplay)
           case "chart" =>
             val chartDisplay = new JFreeChartDisplay()