Merge pull request #74 from Spirals-Team/feature/docker

feature(docker): implement energy monitoring of a container

Author: rouvoy

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

@@ -23,19 +23,20 @@
 package org.powerapi.app
 
 import java.lang.management.ManagementFactory
-import org.powerapi.core.target.{Application, All, Process, Target}
+import org.powerapi.core.target._
 import org.powerapi.module.rapl.RAPLModule
 import org.powerapi.reporter.{FileDisplay, JFreeChartDisplay, ConsoleDisplay}
-import org.powerapi.{PowerMonitoring, PowerMeter}
+import org.powerapi.{PowerDisplay, PowerMonitoring, PowerMeter}
 import org.powerapi.core.power._
 import org.powerapi.module.cpu.dvfs.CpuDvfsModule
 import org.powerapi.module.cpu.simple.{SigarCpuSimpleModule, ProcFSCpuSimpleModule}
-import org.powerapi.module.libpfm.{LibpfmModule, LibpfmHelper, LibpfmCoreProcessModule, LibpfmCoreModule, LibpfmProcessModule}
+import org.powerapi.module.libpfm._
 import org.powerapi.module.extPMeter.powerspy.PowerSpyModule
 import org.powerapi.module.extPMeter.g5k.G5kOmegaWattModule
 import scala.concurrent.duration.DurationInt
 import scala.sys
 import scala.sys.process.stringSeqToProcess
+import scala.util.matching.Regex
 
 /**
  * PowerAPI CLI.
@@ -47,8 +48,9 @@ object PowerAPI extends App {
   val modulesR = """(procfs-cpu-simple|sigar-cpu-simple|cpu-dvfs|libpfm|libpfm-process|libpfm-core|libpfm-core-process|powerspy|g5k-omegawatt|rapl)(,(procfs-cpu-simple|sigar-cpu-simple|cpu-dvfs|libpfm|libpfm-process|libpfm-core|libpfm-core-process|powerspy|g5k-omegawatt|rapl))*""".r
   val aggR = """max|min|geomean|logsum|mean|median|stdev|sum|variance""".r
   val durationR = """\d+""".r
-  val pidR = """(\d+)""".r
-  val appR = """(.+)""".r
+  val pidsR = """(\d+)(,(\d+))*""".r
+  val appsR = """([^,]+)(,([^,]+))*""".r
+  val containersR = """([^,]+)(,([^,]+))*""".r
 
   @volatile var powerMeters = Seq[PowerMeter]()
   @volatile var monitors = Seq[PowerMonitoring]()
@@ -60,13 +62,8 @@ object PowerAPI extends App {
     powerMeters = Seq()
   }
 
-  def validateModules(str: String) = str match {
-    case modulesR(_*) => true
-    case _ => false
-  }
-
-  def validateAgg(str: String): Boolean = str match {
-    case aggR(_*) => true
+  def validate(regex: Regex, str: String) = str match {
+    case regex(_*) => true
     case _ => false
   }
 
@@ -84,42 +81,25 @@ object PowerAPI extends App {
     }
   }
 
-  def validateDuration(str: String): Boolean = str match {
-    case durationR(_*) => true
-    case _ => false
-  }
-
-  implicit def targetsStrToTargets(str: String): Seq[Target] = {
-    val strTargets = if(str.split(",").contains("all")) {
-      "all"
-    }
-    else str
-
-    (for(target <- strTargets.split(",")) yield {
-      target match {
-        case "" => Process(ManagementFactory.getRuntimeMXBean.getName.split("@")(0).toInt)
-        case "all" => All
-        case pidR(pid) => Process(pid.toInt)
-        case appR(app) => Application(app)
-      }
-    }).toSeq
-  }
-
   def printHelp(): Unit = {
     val str =
       """
         |PowerAPI, Spirals Team
         |
         |Build a software-defined power meter. Do not forget to configure correctly the modules.
-        |You can use different settings per software-defined power meter for some modules by using the optional prefix option.
+        |Different settings can be used per software-defined power meter by using the prefix option.
         |Please, refer to the documentation inside the GitHub wiki for further details.
         |
-        |usage: ./powerapi modules [procfs-cpu-simple|sigar-cpu-simple|cpu-dvfs|libpfm|libpfm-process|libpfm-core|libpfm-core-proces|powerspy|g5k-omegawatt|rapl,...] *--prefix [name]* \
-        |                          monitor --frequency [ms] --targets [pid, ..., app, ...|all] --agg [max|min|geomean|logsum|mean|median|stdev|sum|variance] --[console,file [filepath],chart] \
+        |usage: ./powerapi modules procfs-cpu-simple|sigar-cpu-simple|cpu-dvfs|libpfm|libpfm-process|libpfm-core|libpfm-core-process|powerspy|g5k-omegawatt|rapl (1, *) *--prefix [name]*
+        |                          monitor (1, *)
+        |                            --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)
         |                  duration [s]
         |
-        |example: ./powerapi modules procfs-cpu-simple monitor --frequency 1000 --targets firefox,chrome --agg max --console \
-        |                    modules powerspy --prefix powermeter2 monitor --frequency 1000 --targets all --agg max --console \
+        |example: ./powerapi modules procfs-cpu-simple monitor --frequency 1000 --apps firefox,chrome --agg max --console \
+        |                    modules powerspy --prefix powermeter2 monitor --frequency 1000 --all --agg max --console \
         |                    duration 30
       """.stripMargin
 
@@ -128,30 +108,34 @@ object PowerAPI extends App {
 
   def cli(options: List[Map[Symbol, Any]], duration: String, args: List[String]): (List[Map[Symbol, Any]], String) = args match {
     case Nil => (options, duration)
-    case "modules" :: value :: "--prefix" :: prefix :: "monitor" :: tail if validateModules(value) => {
+    case "modules" :: value :: "--prefix" :: prefix :: "monitor" :: tail if validate(modulesR, value) => {
       val (remainingArgs, monitors) = cliMonitorsSubcommand(List(), Map(), tail.map(_.toString))
       cli(options :+ Map('modules -> value, 'prefix -> Some(prefix), 'monitors -> monitors), duration, remainingArgs)
     }
-    case "modules" :: value :: "monitor" :: tail if validateModules(value) => {
+    case "modules" :: value :: "monitor" :: tail if validate(modulesR, value) => {
       val (remainingArgs, monitors) = cliMonitorsSubcommand(List(), Map(), tail.map(_.toString))
       cli(options :+ Map('modules -> value, 'prefix -> None, 'monitors -> monitors), duration, remainingArgs)
     }
-    case "duration" :: value :: tail if validateDuration(value) => cli(options, value, tail)
+    case "duration" :: value :: tail if validate(durationR, value) => cli(options, value, tail)
     case option :: tail => println(s"unknown cli option $option"); sys.exit(1)
   }
 
   def cliMonitorsSubcommand(options: List[Map[Symbol, Any]], currentMonitor: Map[Symbol, Any], args: List[String]): (List[String], List[Map[Symbol, Any]]) = args match {
     case Nil => (List(), options :+ currentMonitor)
-    case "modules" :: value :: "--prefix" :: prefix :: "monitor" :: tail if validateModules(value) => (List("modules", value, "--prefix", prefix, "monitor") ++ tail, options :+ currentMonitor)
-    case "modules" :: value :: "monitor" :: tail if validateModules(value) => (List("modules", value, "monitor") ++ tail, options :+ currentMonitor)
-    case "duration" :: value :: tail if validateDuration(value) => (List("duration", value) ++ tail, options :+ currentMonitor)
+    case "modules" :: value :: "--prefix" :: prefix :: "monitor" :: tail if validate(modulesR, value) => (List("modules", value, "--prefix", prefix, "monitor") ++ tail, options :+ currentMonitor)
+    case "modules" :: value :: "monitor" :: tail if validate(modulesR, value) => (List("modules", value, "monitor") ++ tail, options :+ currentMonitor)
+    case "duration" :: value :: tail if validate(durationR, value) => (List("duration", value) ++ tail, options :+ currentMonitor)
     case "monitor" :: tail => cliMonitorsSubcommand(options :+ currentMonitor, Map(), tail)
-    case "--frequency" :: value :: tail if validateDuration(value) => cliMonitorsSubcommand(options, currentMonitor ++ Map('frequency -> value), tail)
-    case "--targets" :: value :: tail => cliMonitorsSubcommand(options, currentMonitor ++ Map('targets -> value), tail)
-    case "--agg" :: value :: tail if validateAgg(value) => cliMonitorsSubcommand(options, currentMonitor ++ Map('agg -> value), tail)
-    case "--console" :: tail => cliMonitorsSubcommand(options, currentMonitor ++ Map('console -> "true"), tail)
-    case "--file" :: value :: tail => cliMonitorsSubcommand(options, currentMonitor ++ Map('file -> value), tail)
-    case "--chart" :: tail => cliMonitorsSubcommand(options, currentMonitor ++ Map('chart -> "true"), tail)
+    case "--frequency" :: value :: tail if validate(durationR, value) => cliMonitorsSubcommand(options, currentMonitor ++ Map('frequency -> value), tail)
+    case "--self" :: tail => cliMonitorsSubcommand(options, currentMonitor + ('targets -> (currentMonitor.getOrElse('targets, Set[Any]()).asInstanceOf[Set[Any]] + Process(ManagementFactory.getRuntimeMXBean.getName.split("@")(0).toInt))), tail)
+    case "--pids" :: value :: tail if validate(pidsR, value) => cliMonitorsSubcommand(options, currentMonitor + ('targets -> (currentMonitor.getOrElse('targets, Set[Any]()).asInstanceOf[Set[Any]] ++ value.split(",").map(pid => Process(pid.toInt)))), tail)
+    case "--apps" :: value :: tail if validate(appsR, value) => cliMonitorsSubcommand(options, currentMonitor + ('targets -> (currentMonitor.getOrElse('targets, Set[Any]()).asInstanceOf[Set[Any]] ++ value.split(",").map(app => Application(app)))), tail)
+    case "--containers" :: value :: tail if validate(containersR, value) => cliMonitorsSubcommand(options, currentMonitor + ('targets -> (currentMonitor.getOrElse('targets, Set[Any]()).asInstanceOf[Set[Any]] ++ value.split(",").map(container => Container(container)))), tail)
+    case "--all" :: tail => cliMonitorsSubcommand(options, currentMonitor + ('targets -> (currentMonitor.getOrElse('targets, Set[Any]()).asInstanceOf[Set[Any]] + All)), tail)
+    case "--agg" :: value :: tail if validate(aggR, value) => cliMonitorsSubcommand(options, currentMonitor ++ Map('agg -> value), tail)
+    case "--console" :: tail => cliMonitorsSubcommand(options, currentMonitor + ('displays -> (currentMonitor.getOrElse('displays, Set[Any]()).asInstanceOf[Set[Any]] + new ConsoleDisplay)), tail)
+    case "--file" :: value :: tail => cliMonitorsSubcommand(options, currentMonitor + ('displays -> (currentMonitor.getOrElse('displays, Set[Any]()).asInstanceOf[Set[Any]] + new FileDisplay(value))), tail)
+    case "--chart" :: tail => cliMonitorsSubcommand(options, currentMonitor + ('displays -> (currentMonitor.getOrElse('displays, Set[Any]()).asInstanceOf[Set[Any]] + new JFreeChartDisplay)), tail)
     case option :: tail => println(s"unknown monitor option $option"); sys.exit(1)
   }
 
@@ -162,6 +146,8 @@ object PowerAPI extends App {
 
   else {
     if(System.getProperty("os.name").toLowerCase.indexOf("nix") >= 0 || System.getProperty("os.name").toLowerCase.indexOf("nux") >= 0) Seq("bash", "scripts/system.bash").!
+    System.setProperty("java.library.path", "lib")
+
     val (configuration, duration) = cli(List(), "3600", args.toList)
 
     var libpfmHelper: Option[LibpfmHelper] = None
@@ -192,28 +178,18 @@ object PowerAPI extends App {
 
       for(monitorConf <- powerMeterConf('monitors).asInstanceOf[List[Map[Symbol, Any]]]) {
         val frequency = monitorConf.getOrElse('frequency, "1000").toString.toInt.milliseconds
-        val targets: Seq[Target] = monitorConf.getOrElse('targets, "").toString.toLowerCase
+        val targets = {
+          val uniqueTargets = monitorConf.getOrElse('targets, Set(Process(ManagementFactory.getRuntimeMXBean.getName.split("@")(0).toInt))).asInstanceOf[Set[Target]].toSeq
+          if(uniqueTargets.contains(All)) Seq(All) else uniqueTargets
+        }
         val agg: Seq[Power] => Power = aggStrToAggFunction(monitorConf.getOrElse('agg, "max").toString.toLowerCase)
-        val console = monitorConf.getOrElse('console, "").toString
-        val file = monitorConf.getOrElse('file, "").toString
-        val chart = monitorConf.getOrElse('chart, "").toString
+        val displays = monitorConf.getOrElse('displays, Set(new ConsoleDisplay)).asInstanceOf[Set[PowerDisplay]]
 
         val monitor = powerMeter.monitor(frequency)(targets: _*)(agg)
         monitors :+= monitor
 
-        if(console != "") {
-          val consoleDisplay = new ConsoleDisplay()
-          monitor.to(consoleDisplay)
-        }
-
-        if(file != "") {
-          val fileDisplay = new FileDisplay(file)
-          monitor.to(fileDisplay)
-        }
-
-        if(chart != "") {
-          val chartDisplay = new JFreeChartDisplay()
-          monitor.to(chartDisplay)
+        for(display <- displays) {
+          monitor.to(display)
         }
       }
     }
@@ -231,3 +207,4 @@ object PowerAPI extends App {
   shutdownHookThread.remove()
   sys.exit(0)
 }
+

powerapi-core/build.sbt

@@ -3,7 +3,8 @@ name := "powerapi-core"
 organization := "org.powerapi"
 
 resolvers ++= Seq(
-  "JBoss Thirdparty Uploads" at "https://repository.jboss.org/nexus/content/repositories/thirdparty-uploads/"
+  "JBoss Thirdparty Uploads" at "https://repository.jboss.org/nexus/content/repositories/thirdparty-uploads/",
+  "softprops-maven" at "http://dl.bintray.com/content/softprops/maven"
 )
 
 // App
@@ -22,7 +23,8 @@ libraryDependencies ++= Seq(
   "io.spray" %% "spray-routing" % "1.3.3",
   "io.spray" %% "spray-json" % "1.3.2",
   "com.github.nscala-time" %% "nscala-time" % "2.0.0",
-  "net.java.dev.jna" % "jna" % "4.1.0"
+  "net.java.dev.jna" % "jna" % "4.1.0",
+  "com.github.docker-java" % "docker-java" % "2.1.1"
 )
 
 // Tests

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

@@ -29,10 +29,11 @@ import org.apache.logging.log4j.LogManager
 import org.hyperic.sigar.{Sigar, SigarException, SigarProxyCache}
 import org.hyperic.sigar.ptql.ProcessFinder
 import org.powerapi.core.FileHelper.using
-import org.powerapi.core.target.{All, Application, Process, Target, TargetUsageRatio}
+import org.powerapi.core.target.{All, Application, Container, Process, Target, TargetUsageRatio}
 import org.powerapi.module.{Cache, CacheKey}
 import scala.collection.JavaConversions._
 import scala.sys.process.stringSeqToProcess
+import com.github.dockerjava.core.DockerClientBuilder
 
 /**
  * This is not a monitoring target. It's an internal wrapper for the Thread IDentifier.
@@ -78,6 +79,13 @@ trait OSHelper {
    * @param application: targeted application.
    */
   def getProcesses(application: Application): Set[Process]
+  
+  /**
+   * Get the list of processes running on a container.
+   *
+   * @param container: targeted container.
+   */
+  def getProcesses(container: Container): Set[Process] = throw new Exception("The container handling is not available on this operating system.")
 
   /**
    * Get the list of thread behind a Process.
@@ -131,6 +139,16 @@ trait OSHelper {
           }
         }
       )
+      case container: Container => Some(
+        getProcesses(container).foldLeft(0: Long) {
+          (acc, process: Process) => {
+            getProcessCpuTime(process) match {
+              case Some(value) => acc + value
+              case _ => acc
+            }
+          }
+        }
+      )
       case _ => None
     }
   }
@@ -151,6 +169,16 @@ trait OSHelper {
           }
         }
       )
+      case container: Container => TargetUsageRatio(
+        getProcesses(container).foldLeft(0.0: Double) {
+          (acc, process: Process) => {
+            getProcessCpuPercent(muid, process) match {
+              case TargetUsageRatio(value) => acc + value
+              case _ => acc
+            }
+          }
+        }
+      )
       case _ => TargetUsageRatio(0.0)
     }
   }
@@ -168,6 +196,8 @@ class LinuxHelper extends Configuration(None) with OSHelper {
   private val GlobalStatFormat = """cpu\s+([\d\s]+)""".r
   private val TimeInStateFormat = """(\d+)\s+(\d+)""".r
 
+  val docker = DockerClientBuilder.getInstance("unix:///var/run/docker.sock").build()
+
   /**
    * This file allows to get all the cpu frequencies with the help of procfs and cpufreq_utils.
    */
@@ -242,6 +272,12 @@ class LinuxHelper extends Configuration(None) with OSHelper {
       case PSFormat(pid) => Process(pid.toInt)
     }.toSet
   }
+  
+  override def getProcesses(container: Container): Set[Process] = {
+    docker.topContainerCmd(container.id).withPsArgs("-Aopid").exec.getProcesses.flatten.map(
+      process => Process(process.toInt)
+    ).toSet
+  }
 
   def getThreads(process: Process): Set[Thread] = {
     val pidDirectory = new File(taskPath.replace("%?pid", s"${process.pid}"))

powerapi-core/src/main/scala/org/powerapi/core/target/Target.scala

@@ -39,7 +39,7 @@ trait Target
  * @author <a href="mailto:[email protected]">Maxime Colmant</a>
  */
 case class Process(pid: Int) extends Target {
-  override def toString(): String = s"$pid"
+  override def toString: String = s"$pid"
 }
 
 /**
@@ -51,7 +51,18 @@ case class Process(pid: Int) extends Target {
  * @author <a href="mailto:[email protected]">Maxime Colmant</a>
  */
 case class Application(name: String) extends Target {
-  override def toString(): String = name
+  override def toString: String = name
+}
+
+/**
+ * Monitoring targets for a specific container.
+ *
+ * @param name: id of the container.
+ *
+ * @author <a href="mailto:[email protected]">Loïc Huertas</a>
+ */
+case class Container(id: String) extends Target {
+  override def toString: String = id
 }
 
 /**

powerapi-core/src/main/scala/org/powerapi/module/cpu/simple/CpuSensor.scala

@@ -25,7 +25,7 @@ package org.powerapi.module.cpu.simple
 import org.powerapi.core.{MessageBus, OSHelper}
 import org.powerapi.module.SensorComponent
 import org.powerapi.core.MonitorChannel.MonitorTick
-import org.powerapi.core.target.{All, Application, Process, TargetUsageRatio}
+import org.powerapi.core.target.{All, Application, Container, Process, TargetUsageRatio}
 import org.powerapi.module.cpu.UsageMetricsChannel.publishUsageReport
 import org.powerapi.module.SensorChannel.{MonitorStop, MonitorStopAll}
 import scala.reflect.ClassTag
@@ -44,12 +44,14 @@ class CpuSensor(eventBus: MessageBus, osHelper: OSHelper) extends SensorComponen
   def targetCpuUsageRatio(monitorTick: MonitorTick): TargetUsageRatio = {
     val processClaz = implicitly[ClassTag[Process]].runtimeClass
     val appClaz = implicitly[ClassTag[Application]].runtimeClass
+    val containerClaz = implicitly[ClassTag[Container]].runtimeClass
 
     monitorTick.target match {
-      case target if processClaz.isInstance(target) || appClaz.isInstance(target) => {
+      case target if processClaz.isInstance(target) || appClaz.isInstance(target) || containerClaz.isInstance(target) => {
         osHelper.getTargetCpuPercent(monitorTick.muid, target)
       }
       case All => osHelper.getGlobalCpuPercent(monitorTick.muid)
+      case _ => log.warning("Only Process, Application, Container or All targets can be used with this Sensor"); TargetUsageRatio(0.0)
     }
   }
 

powerapi-core/src/main/scala/org/powerapi/module/extPMeter/ExtPMeterFormula.scala

@@ -26,7 +26,7 @@ import akka.event.LoggingReceive
 import org.powerapi.core.{OSHelper, APIComponent, MessageBus}
 import org.powerapi.core.MonitorChannel.{MonitorTick, subscribeMonitorTick}
 import org.powerapi.core.power._
-import org.powerapi.core.target.{Application, All, Process, TargetUsageRatio}
+import org.powerapi.core.target.{Application, All, Container, Process, TargetUsageRatio}
 import org.powerapi.module.{Cache, CacheKey}
 import org.powerapi.module.PowerChannel.publishRawPowerReport
 import org.powerapi.module.extPMeter.ExtPMeterChannel.{ExtPMeterPower, subscribePMeterPower}
@@ -66,12 +66,13 @@ class ExtPMeterFormula(eventBus: MessageBus, osHelper: OSHelper, idlePower: Powe
 
     val processClaz = implicitly[ClassTag[Process]].runtimeClass
     val appClaz = implicitly[ClassTag[Application]].runtimeClass
+    val containerClaz = implicitly[ClassTag[Container]].runtimeClass
 
     lazy val now = monitorTick.target match {
-      case target if processClaz.isInstance(target) || appClaz.isInstance(target) => {
+      case target if processClaz.isInstance(target) || appClaz.isInstance(target) || containerClaz.isInstance(target) => {
         lazy val targetCpuTime = osHelper.getTargetCpuTime(target) match {
           case Some(time) => time
-          case _ => 0l
+          case _ => log.warning("Only Process, Application, Container or All targets can be used with this Sensor"); 0l
         }
 
         (targetCpuTime, activeCpuTime)