refactoring(docker): use the cgroup statistics to retrieve the cpu usage of a docker container

The LinuxHelper used the docker-java api to retrieve the targets which belong to a container. Each target was monitored to obtain the power consumption of a container.
We are now using the cgroup capabilities to obtain directly the stats at the container level and therefore reducing the runtime overhead.

Author: mcolmant

README.md

@@ -57,7 +57,6 @@ We all stand on the shoulders of giants and get by with a little help from our f
 * [Saddle](http://saddle.github.io/) (version 1.3.4 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-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.
 * [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

powerapi-core/build.sbt

@@ -23,7 +23,6 @@ libraryDependencies ++= Seq(
   "org.hyperic" % "sigar" % "1.6.5.132",
   "net.java.dev.jna" % "jna" % "4.2.1",
   "io.spray" %% "spray-json" % "1.3.2",
-  "com.github.docker-java" % "docker-java" % "2.1.4",
   "com.paulgoldbaum" %% "scala-influxdb-client" % "0.4.5"
 )
 

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

@@ -25,9 +25,9 @@ package org.powerapi.core
 import java.io.{File, IOException}
 
 import scala.collection.JavaConversions._
+import scala.io.Source
 import scala.sys.process.stringSeqToProcess
 
-import com.github.dockerjava.core.DockerClientBuilder
 import com.typesafe.config.Config
 import org.apache.logging.log4j.LogManager
 import org.hyperic.sigar.ptql.ProcessFinder
@@ -106,6 +106,13 @@ trait OSHelper {
     */
   def getProcessCpuTime(process: Process): Long
 
+  /**
+    * Get the docker container execution time on the cpu.
+    *
+    * @param container targeted docker container
+    */
+  def getDockerContainerCpuTime(container: Container): Long
+
   /**
     * Get the global execution times for the cpu.
     */
@@ -122,8 +129,9 @@ trait OSHelper {
   def getTargetCpuTime(target: Target): Long = {
     target match {
       case process: Process => getProcessCpuTime(process)
-      case wrapper: Target if wrapper.isInstanceOf[Application] || wrapper.isInstanceOf[Container] =>
-        getProcesses(wrapper).toSeq.map(process => getProcessCpuTime(process)).sum
+      case container: Container => getDockerContainerCpuTime(container)
+      case application: Application =>
+        getProcesses(application).toSeq.map(process => getProcessCpuTime(process)).sum
       case _ => 0L
     }
   }
@@ -253,14 +261,31 @@ class LinuxHelper extends Configuration(None) with OSHelper {
     case ConfigValue(values) => values
     case _ => Map()
   }
-  lazy val docker = DockerClientBuilder.getInstance("unix:///var/run/docker.sock").build()
+  /**
+    * Mount path.
+    */
+  lazy val mountsPath = load {
+    _.getString("powerapi.procfs.mounts-path")
+  } match {
+    case ConfigValue(p) => p
+    case _ => "/proc/mounts"
+  }
+
   private val log = LogManager.getLogger
   private val PSFormat = """^\s*(\d+)\s*""".r
   private val GlobalStatFormat = """cpu\s+([\d\s]+)""".r
   private val TimeInStateFormat = """(\d+)\s+(\d+)""".r
-  private val procUserTimeIndex = 13
-  private val procSysTimeIndex = 14
-  private val procGlobalIdleTime = 3
+  private val MountFormat = """^.+\s+(.+)\s+(.+)\s+(.+)\s+.+\s+.+$""".r
+
+  /**
+    * Returns the mount path of a given cgroup if it exists.
+    */
+  def cgroupMntPoint(name: String): Option[String] = {
+    Source.fromFile(mountsPath).getLines().collectFirst {
+      case MountFormat(mnt, typ, tokens) if typ == "cgroup" && tokens.contains(name) =>
+        mnt
+    }
+  }
 
   def getCPUFrequencies: Set[Long] = {
     (for (index <- topology.values.flatten) yield {
@@ -281,10 +306,6 @@ class LinuxHelper extends Configuration(None) with OSHelper {
       Seq("ps", "-C", app.name, "-o", "pid", "--no-headers").lineStream_!.map {
         case PSFormat(pid) => Process(pid.toInt)
       }.toSet
-    case cont: Container =>
-      docker.topContainerCmd(cont.id).withPsArgs("-Aopid").exec.getProcesses.flatten.map(
-        process => Process(process.toInt)
-      ).toSet
     case process: Process =>
       Set(process)
     case _ =>
@@ -310,7 +331,7 @@ class LinuxHelper extends Configuration(None) with OSHelper {
 
         val statLine = source.getLines.toIndexedSeq(0).split("\\s")
         // User time + System time
-        statLine(procUserTimeIndex).toLong + statLine(procSysTimeIndex).toLong
+        statLine(13).toLong + statLine(14).toLong
       })
     }
     catch {
@@ -320,6 +341,20 @@ class LinuxHelper extends Configuration(None) with OSHelper {
     }
   }
 
+  /**
+    * Use the cpuacct cgroup to retrieve the overall cpu statistics for a docker container.
+    */
+  def getDockerContainerCpuTime(container: Container): Long = {
+    // TODO: Could be also replaced by a direct call to the Docker API with stream = 0. The main benefit would be to handle a distant docker server.
+    cgroupMntPoint("cpuacct") match {
+      case Some(path) if new File(s"$path/docker/${container.id}").isDirectory =>
+        Source.fromFile(s"$path/docker/${container.id}/cpuacct.stat").getLines().map(_.split("\\s")(1).toLong).sum
+      case _ =>
+        log.warn("i/o exception, cpuacct cgroup not mounted for the container {}", s"${container.id}")
+        0l
+    }
+  }
+
   def getGlobalCpuTimes: GlobalCpuTimes = {
     try {
       using(globalStatPath)(source => {
@@ -332,7 +367,7 @@ class LinuxHelper extends Configuration(None) with OSHelper {
             * @see http://lxr.free-electrons.com/source/kernel/sched/cputime.c#L165
             */
           case GlobalStatFormat(times) =>
-            val idleTime = times.split("\\s")(procGlobalIdleTime).toLong
+            val idleTime = times.split("\\s")(3).toLong
             val activeTime = times.split("\\s").slice(0, 8).map(_.toLong).sum - idleTime
 
             GlobalCpuTimes(idleTime, activeTime)
@@ -503,6 +538,8 @@ class SigarHelper(sigar: SigarProxy) extends OSHelper {
     }
   }
 
+  def getDockerContainerCpuTime(container: Container): Long = throw new SigarException("not yet implemented with sigar")
+
   def getGlobalCpuTimes: GlobalCpuTimes = {
     try {
       val idleTime = sigar.getCpu.getIdle

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

@@ -56,9 +56,10 @@ case class Application(name: String) extends Target {
   * Monitoring targets for a specific container.
   *
   * @param id id of the container.
+  * @param name name of the container.
   * @author <a href="mailto:l.huertas.pro@gmail.com">Loïc Huertas</a>
   */
-case class Container(id: String) extends Target {
+case class Container(id: String, name: String = "unknown") extends Target {
   override def toString: String = id
 }
 

powerapi-core/src/test/resources/configuration-suite.conf

@@ -116,52 +116,3 @@ rapl.supported-architectures = [
   {id = 63, model = "Haswell-EP"}
   {id = 61, model = "Broadwell"}
 ]
-
-
-powerapi.disk.formulae = [
-  { name = "sda",
-    models = {
-      read = [
-        { condition = "<= 78966784", coeffs = [ 0.00, 1.01e-07 ] }
-        { condition = "> 78966784", coeffs = [ 7.62, 1.72e-10 ] }
-      ],
-      write = [
-        { condition = "<= 66674688", coeffs = [ 0.00, 1.13e-07 ] }
-        { condition = "> 66674688", coeffs = [ 8.33, 1.79e-09 ] }
-      ]
-    }
-  }
-
-  { name = "sdb",
-    models = {
-      read = [
-        { condition = "<= 10", coeffs = [ 0.00, 0.003 ] }
-        { condition = "> 10", coeffs = [ 2, 0.15 ] }
-      ],
-      write = [
-        { condition = "<= 5", coeffs = [ 0.00, 0.001 ] }
-        { condition = "> 5", coeffs = [ 3, 0.25 ] }
-      ]
-    }
-  }
-]
-powerapi.disk.interval = 1s
-
-disk-test {
-  powerapi.disk.formulae = [
-    { name = "sdb",
-      models = {
-        read = [
-          { condition = "<= 100", coeffs = [ 0.00, 0.0013 ] }
-          { condition = "> 100", coeffs = [ 2.13, 0.28 ] }
-        ],
-        write = [
-          { condition = "<= 1000", coeffs = [ 0.00, 0.0016 ] }
-          { condition = "> 1000", coeffs = [ 3.15, 0.40 ] }
-        ]
-      }
-    }
-  ]
-
-  powerapi.disk.interval = 250ms
-}

powerapi-core/src/test/resources/proc/mounts

@@ -0,0 +1,40 @@
+rootfs / rootfs rw 0 0
+sysfs /sys sysfs rw,nosuid,nodev,noexec,relatime 0 0
+proc /proc proc rw,nosuid,nodev,noexec,relatime 0 0
+udev /dev devtmpfs rw,relatime,size=8196944k,nr_inodes=2049236,mode=755 0 0
+devpts /dev/pts devpts rw,nosuid,noexec,relatime,gid=5,mode=620,ptmxmode=000 0 0
+tmpfs /run tmpfs rw,nosuid,noexec,relatime,size=1641624k,mode=755 0 0
+/dev/disk/by-uuid/8141fc2e-ec4b-4475-a334-91678f8f1fe9 / ext4 rw,relatime,errors=remount-ro,data=ordered 0 0
+none /sys/fs/cgroup tmpfs rw,relatime,size=4k,mode=755 0 0
+none /sys/fs/fuse/connections fusectl rw,relatime 0 0
+none /sys/kernel/debug debugfs rw,relatime 0 0
+none /sys/kernel/security securityfs rw,relatime 0 0
+none /run/lock tmpfs rw,nosuid,nodev,noexec,relatime,size=5120k 0 0
+none /run/shm tmpfs rw,nosuid,nodev,relatime 0 0
+cgroup /sys/fs/cgroup/cpuset cgroup rw,relatime,cpuset 0 0
+none /run/user tmpfs rw,nosuid,nodev,noexec,relatime,size=102400k,mode=755 0 0
+none /sys/fs/pstore pstore rw,relatime 0 0
+cgroup /sys/fs/cgroup/cpu cgroup rw,relatime,cpu 0 0
+cgroup /sys/fs/cgroup/cpuacct cgroup rw,relatime,cpuacct 0 0
+cgroup /sys/fs/cgroup/memory cgroup rw,relatime,memory 0 0
+cgroup /sys/fs/cgroup/devices cgroup rw,relatime,devices 0 0
+cgroup /sys/fs/cgroup/freezer cgroup rw,relatime,freezer 0 0
+cgroup /sys/fs/cgroup/blkio cgroup rw,relatime,blkio 0 0
+cgroup /sys/fs/cgroup/perf_event cgroup rw,relatime,perf_event 0 0
+cgroup /sys/fs/cgroup/hugetlb cgroup rw,relatime,hugetlb 0 0
+systemd /sys/fs/cgroup/systemd cgroup rw,nosuid,nodev,noexec,relatime,name=systemd 0 0
+/dev/disk/by-uuid/8141fc2e-ec4b-4475-a334-91678f8f1fe9 /var/lib/docker/aufs ext4 rw,relatime,errors=remount-ro,data=ordered 0 0
+gvfsd-fuse /run/user/1000/gvfs fuse.gvfsd-fuse rw,nosuid,nodev,relatime,user_id=1000,group_id=1000 0 0
+none /var/lib/docker/aufs/mnt/797e80d132c3c82ab9891574a358d4614097d1315945441a0bc04a713c490421 aufs rw,relatime,si=6b8a6d0dfd2c1c66,dio 0 0
+shm /var/lib/docker/containers/797e80d132c3c82ab9891574a358d4614097d1315945441a0bc04a713c490421/shm tmpfs rw,nosuid,nodev,noexec,relatime,size=65536k 0 0
+mqueue /var/lib/docker/containers/797e80d132c3c82ab9891574a358d4614097d1315945441a0bc04a713c490421/mqueue mqueue rw,nosuid,nodev,noexec,relatime 0 0
+proc /run/docker/netns/628d2274152b proc rw,nosuid,nodev,noexec,relatime 0 0
+none /var/lib/docker/aufs/mnt/b20b6f81720ba6686b36d8a43727025cd239bfda45203a7710851fea853bda4c aufs rw,relatime,si=6b8a6d0d66fa0c66,dio 0 0
+shm /var/lib/docker/containers/b20b6f81720ba6686b36d8a43727025cd239bfda45203a7710851fea853bda4c/shm tmpfs rw,nosuid,nodev,noexec,relatime,size=65536k 0 0
+mqueue /var/lib/docker/containers/b20b6f81720ba6686b36d8a43727025cd239bfda45203a7710851fea853bda4c/mqueue mqueue rw,nosuid,nodev,noexec,relatime 0 0
+proc /run/docker/netns/2bff8e7c00b5 proc rw,nosuid,nodev,noexec,relatime 0 0
+proc /run/docker/netns/default proc rw,nosuid,nodev,noexec,relatime 0 0
+none /var/lib/docker/aufs/mnt/98f216c23b2f3f643577b18660825144a69fa5af141cc05e76e8c1c2b4f21658 aufs rw,relatime,si=6b8a6d0ac968fc66,dio 0 0
+shm /var/lib/docker/containers/98f216c23b2f3f643577b18660825144a69fa5af141cc05e76e8c1c2b4f21658/shm tmpfs rw,nosuid,nodev,noexec,relatime,size=65536k 0 0
+mqueue /var/lib/docker/containers/98f216c23b2f3f643577b18660825144a69fa5af141cc05e76e8c1c2b4f21658/mqueue mqueue rw,nosuid,nodev,noexec,relatime 0 0
+proc /run/docker/netns/a8ec38ebc3f0 proc rw,nosuid,nodev,noexec,relatime 0 0

powerapi-core/src/test/resources/sys/fs/cgroup/cpuacct/docker/abcd/cpuacct.stat

@@ -0,0 +1,2 @@
+user 2502902
+system 277405

powerapi-core/src/test/scala/org/powerapi/core/OSHelperSuite.scala

@@ -62,16 +62,17 @@ class OSHelperSuite extends UnitTest {
       def getProcessCpuTime(process: Process): Long = process match {
         case Process(1) => 10
         case Process(2) => 11
-        case Process(3) => 20
-        case Process(4) => 21
         case Process(10) => 30
       }
 
+      def getDockerContainerCpuTime(container: Container): Long = container match {
+        case Container("abcd", "n") => 20 + 21
+      }
+
       def getGlobalCpuTimes: GlobalCpuTimes = ???
 
       def getProcesses(target: Target): Set[Process] = target match {
         case Application("firefox") => Set(1, 2)
-        case Container("docker") => Set(3, 4)
         case Process(10) => Set(10)
       }
 
@@ -81,11 +82,10 @@ class OSHelperSuite extends UnitTest {
     }
 
     helper.getProcesses(Application("firefox")) should equal(Set(Process(1), Process(2)))
-    helper.getProcesses(Container("docker")) should equal(Set(Process(3), Process(4)))
     helper.getProcesses(Process(10)) should equal(Set(Process(10)))
     helper.getTargetCpuTime(Process(10)) should equal(30)
     helper.getTargetCpuTime(Application("firefox")) should equal(10 + 11)
-    helper.getTargetCpuTime(Container("docker")) should equal(20 + 21)
+    helper.getTargetCpuTime(Container("abcd", "n")) should equal(20 + 21)
     helper.getTargetCpuTime(All) should equal(0)
     helper.getAllDirectories(new File(s"${basepath}/sys/fs/cgroup/blkio")) should contain theSameElementsAs Seq(
       new File(s"${basepath}/sys/fs/cgroup/blkio/powerapi"),
@@ -106,6 +106,7 @@ class OSHelperSuite extends UnitTest {
     linuxHelper.cgroupSysFSPath should equal("p6")
     linuxHelper.diskStatPath should equal("p7")
     linuxHelper.topology should equal(Map(0 -> Set(0, 4), 1 -> Set(1, 5), 2 -> Set(2, 6), 3 -> Set(3, 7)))
+    linuxHelper.mountsPath should equal("p1/mounts")
   }
 
   it should "return the list of available frequencies" in {
@@ -134,6 +135,24 @@ class OSHelperSuite extends UnitTest {
     helper.getProcessCpuTime(10) should equal(0)
   }
 
+  it should "return the cgroup mount point if it exists" in {
+    val helper = new LinuxHelper {
+      override lazy val mountsPath = s"${basepath}proc/mounts"
+    }
+
+    helper.cgroupMntPoint("cpuacct") should equal (Some("/sys/fs/cgroup/cpuacct"))
+    helper.cgroupMntPoint("test") should equal (None)
+  }
+
+  it should "return the cpu time of a given docker container" in {
+    val helper = new LinuxHelper {
+      override def cgroupMntPoint(name: String): Option[String] = Some(s"${basepath}sys/fs/cgroup/cpuacct")
+    }
+
+    helper.getDockerContainerCpuTime(Container("abcd", "n")) should equal(2502902 + 277405)
+    helper.getDockerContainerCpuTime(Container("test", "n2")) should equal(0)
+  }
+
   it should "return the global cpu time" in {
     val helper = new LinuxHelper {
       override lazy val globalStatPath = s"${basepath}proc/stat"