feature(rest): implement the REST reporter

Author: huertas

README.md

@@ -55,6 +55,11 @@ We all stand on the shoulders of giants and get by with a little help from our f
 * [Scala IO](http://jesseeichar.github.io/scala-io-doc/0.4.3/index.html#!/overview) (version 0.4.3 under [3-clause BSD license](http://www.scala-lang.org/license.html)), for an extensions of IO.
 * [Saddle](http://saddle.github.io/) (version 1.3.3 under [Apache 2 license](http://www.apache.org/licenses/LICENSE-2.0)), for data manipulation.
 * [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-can](http://spray.io/) (version 1.3.3 under [Apache 2 license](http://www.apache.org/licenses/LICENSE-2.0)), for providing a low-level, low-overhead, high-performance HTTP server and client.
+* [spray-client](http://spray.io/) (version 1.3.3 under [Apache 2 license](http://www.apache.org/licenses/LICENSE-2.0)), for providing high-level HTTP client.
+* [spray-routing](http://spray.io/) (version 1.3.3 under [Apache 2 license](http://www.apache.org/licenses/LICENSE-2.0)), for providing a high-level, very flexible routing DSL for elegantly defining RESTful web services.
+* [spray-json](http://spray.io/) (version 1.2.2 under [Apache 2 license](http://www.apache.org/licenses/LICENSE-2.0)), for (de)serializing JSON.
+* [nscala-time](https://github.com/nscala-time/nscala-time) (version 2.0.0 under [Apache 2 license](http://www.apache.org/licenses/LICENSE-2.0)), for using a scala wrapper for Joda Time (quality replacement for the Java date and time classes).
 
 # License
 This software is licensed under the *GNU Affero General Public License*, quoted below.

powerapi-core/build.sbt

@@ -16,14 +16,20 @@ libraryDependencies ++= Seq(
   "com.github.scala-incubator.io" %% "scala-io-file" % "0.4.3",
   "org.jfree" % "jfreechart" % "1.0.19",
   "org.scala-saddle" %% "saddle-core" % "1.3.3",
-  "org.hyperic" % "sigar" % "1.6.5.132"
+  "org.hyperic" % "sigar" % "1.6.5.132",
+  "io.spray" %% "spray-can" % "1.3.3",
+  "io.spray" %% "spray-client" % "1.3.3",
+  "io.spray" %% "spray-routing" % "1.3.3",
+  "io.spray" %% "spray-json" % "1.3.2",
+  "com.github.nscala-time" %% "nscala-time" % "2.0.0"
 )
 
 // Tests
 libraryDependencies ++= Seq(
   "com.typesafe.akka" %% "akka-testkit" % "2.3.11" % "test",
   "org.scalatest" %% "scalatest" % "2.2.5" % "test",
-  "org.scalamock" %% "scalamock-scalatest-support" % "3.2.2" % "test"
+  "org.scalamock" %% "scalamock-scalatest-support" % "3.2.2" % "test",
+  "io.spray" %% "spray-testkit" % "1.3.3" % "test"
 )
 
 startYear := Some(2014)

powerapi-core/src/main/scala/org/powerapi/PowerMeter.scala

@@ -31,7 +31,8 @@ import akka.util.Timeout
 import java.util.concurrent.TimeUnit
 import org.powerapi.PowerMeterMessages.StopAll
 import org.powerapi.core.{ActorComponent, Clocks, ConfigValue, Configuration, APIComponent, Monitor, Monitors, MessageBus}
-import org.powerapi.core.MonitorChannel.MonitorStart
+import org.powerapi.core.MonitorChannel.{ GetMonitoredProcesses, MonitorStart }
+import org.powerapi.core.MonitorChannel.stopMonitor
 import org.powerapi.core.target.Target
 import org.powerapi.core.power._
 import scala.concurrent.{ Await, Future }
@@ -80,6 +81,12 @@ class PowerMeterActor(eventBus: MessageBus, modules: Seq[PowerModule], timeout:
         case _ => log.error("The monitor supervisor is not created")
       }
     }
+    case GetMonitoredProcesses => {
+      monitorSupervisor match {
+        case Some(actorRef) => sender ! Await.result(actorRef.ask(GetMonitoredProcesses)(timeout), timeout.duration)
+        case _ => log.error("The monitor supervisor is not created")
+      }
+    }
     case StopAll => stopAll()
   } orElse default
 
@@ -122,6 +129,17 @@ class PowerMeter(system: ActorSystem, modules: Seq[PowerModule]) extends PowerMe
     Await.result(powerMeterActor.ask(MonitorStart("", monitor.muid, frequency, targets.toList))(timeout), timeout.duration)
     monitor
   }
+  
+  /**
+   * Allows to get the processes which are monitored by a power meter.
+   */
+  def getMonitoredProcesses: Iterable[Set[Target]] = {
+    Await.result(
+      Await.result(
+        powerMeterActor.ask(GetMonitoredProcesses)(timeout), timeout.duration
+      ).asInstanceOf[Future[Iterable[Set[Target]]]], timeout.duration
+    ).asInstanceOf[Iterable[Set[Target]]]
+  }
 
   /**
    * Blocks and actively waits for a specific duration before returning.
@@ -136,6 +154,13 @@ class PowerMeter(system: ActorSystem, modules: Seq[PowerModule]) extends PowerMe
 
     this
   }
+  
+  /**
+   * Cancel a subscription and stops the associated monitoring.
+   */
+  def stopMonitor(muid: UUID): Unit = {
+    org.powerapi.core.MonitorChannel.stopMonitor(muid)(eventBus)
+  }
 
   /**
    * Shuts down the current instance of power meter.
@@ -243,3 +268,4 @@ trait PowerDisplay {
    */
   def display(muid: UUID, timestamp: Long, targets: Set[Target], devices: Set[String], power: Power)
 }
+

powerapi-core/src/main/scala/org/powerapi/core/MonitorActors.scala

@@ -22,21 +22,37 @@
  */
 package org.powerapi.core
 
+import akka.util.Timeout
 import akka.actor.SupervisorStrategy.{Directive, Resume}
 import akka.actor.{Actor, ActorRef, ActorSystem, PoisonPill, Props}
 import akka.event.LoggingReceive
+import akka.pattern.ask
 import java.util.UUID
+import java.util.concurrent.TimeUnit
 import org.powerapi.core.ClockChannel.ClockTick
 import org.powerapi.core.power._
 import org.powerapi.core.target.Target
 import org.powerapi.{PowerDisplay, PowerMonitoring}
 import org.powerapi.core.ClockChannel.{startClock, stopClock, subscribeClockTick, unsubscribeClockTick}
-import org.powerapi.core.MonitorChannel.{MonitorAggFunction, MonitorStart, MonitorStarted, MonitorStop, MonitorStopAll, formatMonitorChildName, subscribeMonitorsChannel}
+import org.powerapi.core.MonitorChannel.{GetMonitoredProcesses, MonitorAggFunction, MonitorStart, MonitorStarted, MonitorStop, MonitorStopAll, formatMonitorChildName, subscribeMonitorsChannel}
 import org.powerapi.core.MonitorChannel.{publishMonitorTick, setAggFunction, stopMonitor}
 import org.powerapi.module.PowerChannel.{AggregatePowerReport, RawPowerReport, render, subscribeRawPowerReport, unsubscribeRawPowerReport, subscribeAggPowerReport}
 import org.powerapi.module.SensorChannel.{monitorAllStopped, monitorStopped}
 import org.powerapi.reporter.ReporterComponent
-import scala.concurrent.duration.FiniteDuration
+import scala.concurrent.duration.{ DurationLong, FiniteDuration }
+import scala.concurrent.{ Await, Future }
+import scala.concurrent.ExecutionContext.Implicits.global
+import scala.util.{ Failure, Success }
+
+/**
+ * Main Configuration
+ */
+trait MonitorConfiguration extends Configuration {
+  lazy val timeout: Timeout = load { _.getDuration("powerapi.actors.timeout", TimeUnit.MILLISECONDS) } match {
+    case ConfigValue(value) => Timeout(value.milliseconds)
+    case _ => Timeout(15l.seconds)
+  }
+}
 
 /**
  * One child represents one monitor.
@@ -61,6 +77,7 @@ class MonitorChild(eventBus: MessageBus,
     case tick: ClockTick => produceMessages(tick)
     case MonitorAggFunction(_, id, aggF) if muid == id => setAggregatingFunction(aggR, aggF)
     case powerReport: RawPowerReport => aggregate(aggR, powerReport, aggFunction)
+    case GetMonitoredProcesses => sender ! targets
     case MonitorStop(_, id) if muid == id => stop()
     case _: MonitorStopAll => stop()
   } orElse default
@@ -125,7 +142,7 @@ class MonitorChild(eventBus: MessageBus,
  *
  * @author Maxime Colmant <maxime.colmant@gmail.com>
  */
-class Monitors(eventBus: MessageBus) extends Supervisor {
+class Monitors(eventBus: MessageBus) extends MonitorConfiguration with Supervisor {
 
   override def preStart(): Unit = {
     subscribeMonitorsChannel(eventBus)(self)
@@ -144,6 +161,7 @@ class Monitors(eventBus: MessageBus) extends Supervisor {
     case msg: MonitorAggFunction => setAggregatingFunction(msg)
     case msg: MonitorStop => stop(msg)
     case msg: MonitorStopAll => stopAll(msg)
+    case GetMonitoredProcesses => getMonitoredProcesses
   } orElse default
 
   /**
@@ -189,6 +207,20 @@ class Monitors(eventBus: MessageBus) extends Supervisor {
     monitorAllStopped()(eventBus)
     context.become(receive)
   }
+  
+  /**
+   * Allows to get the processes which are monitored by a power meter.
+   */
+  def getMonitoredProcesses: Unit = {
+    val monitoredProcesses = Future.sequence(for (child <- context.children) yield {
+      child.ask(GetMonitoredProcesses)(timeout).map(Success(_)).recover({
+        case e => Failure(e)
+      })
+    }).map(_.collect {
+      case Success(list) => list.asInstanceOf[List[Target]].toSet
+    })
+    sender ! monitoredProcesses
+  }
 }
 
 /**
@@ -226,3 +258,4 @@ class Monitor(eventBus: MessageBus, system: ActorSystem) extends PowerMonitoring
     reporters = Array()
   }
 }
+

powerapi-core/src/main/scala/org/powerapi/core/MonitorChannel.scala

@@ -84,6 +84,12 @@ object MonitorChannel extends Channel {
    */
   object MonitorStarted
 
+  /**
+   * Dedicated type of message to get the processes which are monitored
+   * by a power meter.
+   */
+  object GetMonitoredProcesses
+
   /**
    * MonitorStop is represented as a dedicated type of message.
    *
@@ -156,3 +162,4 @@ object MonitorChannel extends Channel {
     s"monitor-$muid"
   }
 }
+