@@ -62,6 +62,22 @@ case class TimeInStates(times: Map[Long, Long]) {
6262 */
6363case class GlobalCpuTimes (idleTime : Long , activeTime : Long )
6464
65+ /**
66+ * Wrapper class for disk info.
67+ */
68+ case class Disk (name : String , major : Int , minor : Int , bytesRead : Long = 0 , bytesWritten : Long = 0 ) {
69+ def - (other : Disk ): Disk = {
70+ if (name == other.name && major == other.major && minor == other.minor) {
71+ val diffBytesRead = bytesRead - other.bytesRead
72+ val diffBytesWrite = bytesWritten - other.bytesWritten
73+
74+ Disk (name, major, minor, if (diffBytesRead > 0 ) diffBytesRead else 0 , if (diffBytesWrite > 0 ) diffBytesWrite else 0 )
75+ }
76+
77+ else this
78+ }
79+ }
80+
6581/**
6682 * Base trait use for implementing os specific methods.
6783 *
@@ -111,6 +127,49 @@ trait OSHelper {
111127 case _ => 0L
112128 }
113129 }
130+
131+ /**
132+ * Create a cgroup attached to a given subsystem.
133+ */
134+ def createCGroup (subsystem : String , name : String ): Unit
135+
136+ /**
137+ * Test if a cgroup exists.
138+ */
139+ def existsCGroup (subsystem : String , name : String ): Boolean
140+
141+ /**
142+ * Attach pids (`toAttach` string) to an existing cgroup.
143+ */
144+ def attachToCGroup (subsystem : String , name : String , toAttach : String ): Unit
145+
146+ /**
147+ * Delete a cgroup attached to each given subsystem.
148+ */
149+ def deleteCGroup (subsystem : String , name : String ): Unit
150+
151+ /**
152+ * Get information for selected disks.
153+ */
154+ def getDiskInfo (names : Seq [String ]): Seq [Disk ]
155+
156+ /**
157+ * Get the global read/written bytes information for selected disks.
158+ */
159+ def getGlobalDiskBytes (disks : Seq [Disk ]): Seq [Disk ]
160+
161+ /**
162+ * Get the target read/written bytes information for a given target on selected disks.
163+ */
164+ def getTargetDiskBytes (disks : Seq [Disk ], target : Target ): Seq [Disk ]
165+
166+ /**
167+ * Get all directories recursively inside a given path.
168+ */
169+ def getAllDirectories (dir : File ): Seq [File ] = {
170+ val current = dir.listFiles.filter(_.isDirectory)
171+ current ++ current.flatMap(getAllDirectories)
172+ }
114173}
115174
116175/**
@@ -166,6 +225,24 @@ class LinuxHelper extends Configuration(None) with OSHelper {
166225 case ConfigValue (p) => p
167226 case _ => " /sys/devices/system/cpu/cpu%?index/cpufreq/stats/time_in_state"
168227 }
228+ /**
229+ * Cgroup base directory.
230+ */
231+ lazy val cgroupSysFSPath = load {
232+ _.getString(" powerapi.sysfs.cgroup-sysfs-path"
233+ } match {
234+ case ConfigValue (p) => p
235+ case _ => " /sys/fs/cgroup"
236+ }
237+ /**
238+ * Disk stats filepath.
239+ */
240+ lazy val diskStatPath = load {
241+ _.getString(" powerapi.procfs.disk-stats-path"
242+ } match {
243+ case ConfigValue (p) => p
244+ case _ => " /proc/diskstats"
245+ }
169246 /**
170247 * CPU's topology.
171248 */
@@ -297,6 +374,91 @@ class LinuxHelper extends Configuration(None) with OSHelper {
297374
298375 TimeInStates (result.toMap[Long , Long ])
299376 }
377+
378+ def createCGroup (subsystem : String , name : String ): Unit = {
379+ Seq (" cgcreate" " -g" s " $subsystem:/ $name" ).!
380+ }
381+
382+ def existsCGroup (subsystem : String , name : String ): Boolean = {
383+ new File (s " $cgroupSysFSPath/ $subsystem/ $name" ).exists()
384+ }
385+
386+ def attachToCGroup (subsystem : String , name : String , toAttach : String ): Unit = {
387+ Seq (" cgclassify" " -g" s " $subsystem:/ $name" , s " $toAttach" ).!
388+ }
389+
390+ def deleteCGroup (subsystem : String , name : String ): Unit = {
391+ Seq (" cgdelete" s " $subsystem:/ $name" ).!
392+ }
393+
394+ def getDiskInfo (names : Seq [String ]): Seq [Disk ] = {
395+ val Regex = s " \\ s+([0-9]+) \\ s+([0-9]+) \\ s+( ${names.mkString(" |" ) \\ s+.* " .r
396+
397+ using(diskStatPath)(source => {
398+ source.getLines().collect {
399+ case line => line match {
400+ case Regex (major, minor, name) => Some (Disk (name, major.toInt, minor.toInt))
401+ case _ => None
402+ }
403+ }.filter(_.isDefined).map(_.get).toList
404+ })
405+ }
406+
407+ def getGlobalDiskBytes (disks : Seq [Disk ]): Seq [Disk ] = {
408+ val directory = new File (s " $cgroupSysFSPath/blkio " )
409+ val directories = Seq (directory) ++ getAllDirectories(directory)
410+ val bytesRead = collection.mutable.Map [String , Long ]()
411+ val bytesWritten = collection.mutable.Map [String , Long ]()
412+
413+ for (dir <- directories) {
414+ using(s " ${dir.getAbsolutePath}/blkio.throttle.io_service_bytes " )(source => {
415+ val lines = source.getLines().toList
416+
417+ for (disk <- disks) {
418+ val ReadRegex = s " ( ${disk.major}):( ${disk.minor}) \\ s+Read \\ s+([0-9]+) " .r
419+ val WriteRegex = s " ( ${disk.major}):( ${disk.minor}) \\ s+Write \\ s+([0-9]+) " .r
420+
421+ for (line <- lines) {
422+ line match {
423+ case ReadRegex (_, _, bytes) => bytesRead += (disk.name -> (bytesRead.getOrElse(disk.name, 0l ) + bytes.toLong))
424+ case WriteRegex (_, _, bytes) => bytesWritten += (disk.name -> (bytesWritten.getOrElse(disk.name, 0l ) + bytes.toLong))
425+ case _ =>
426+ }
427+ }
428+ }
429+ })
430+ }
431+
432+ disks.map(disk => Disk (disk.name, disk.major, disk.minor, bytesRead.getOrElse(disk.name, 0l ), bytesWritten.getOrElse(disk.name, 0l )))
433+ }
434+
435+ def getTargetDiskBytes (disks : Seq [Disk ], target : Target ): Seq [Disk ] = {
436+ val pids = getProcesses(target).map(_.pid)
437+ val directories = pids.map(pid => new File (s " $cgroupSysFSPath/blkio/powerapi/ $pid" ))
438+ val bytesRead = collection.mutable.Map [String , Long ]()
439+ val bytesWritten = collection.mutable.Map [String , Long ]()
440+
441+ for (dir <- directories) {
442+ using(s " ${dir.getAbsolutePath}/blkio.throttle.io_service_bytes " )(source => {
443+ val lines = source.getLines().toList
444+
445+ for (disk <- disks) {
446+ val ReadRegex = s " ( ${disk.major}):( ${disk.minor}) \\ s+Read \\ s+([0-9]+) " .r
447+ val WriteRegex = s " ( ${disk.major}):( ${disk.minor}) \\ s+Write \\ s+([0-9]+) " .r
448+
449+ for (line <- lines) {
450+ line match {
451+ case ReadRegex (_, _, bytes) => bytesRead += (disk.name -> (bytesRead.getOrElse(disk.name, 0l ) + bytes.toLong))
452+ case WriteRegex (_, _, bytes) => bytesWritten += (disk.name -> (bytesWritten.getOrElse(disk.name, 0l ) + bytes.toLong))
453+ case _ =>
454+ }
455+ }
456+ }
457+ })
458+ }
459+
460+ disks.map(disk => Disk (disk.name, disk.major, disk.minor, bytesRead.getOrElse(disk.name, 0l ), bytesWritten.getOrElse(disk.name, 0l )))
461+ }
300462}
301463
302464trait SigarHelperConfiguration extends Configuration {
@@ -356,4 +518,18 @@ class SigarHelper(sigar: SigarProxy) extends OSHelper {
356518 }
357519
358520 def getTimeInStates : TimeInStates = throw new SigarException (" sigar cannot be able to get how many time CPU spent under each frequency"
521+
522+ def createCGroup (subsystem : String , name : String ): Unit = {}
523+
524+ def existsCGroup (subsystem : String , name : String ): Boolean = false
525+
526+ def attachToCGroup (subsystem : String , name : String , toAttach : String ): Unit = {}
527+
528+ def deleteCGroup (subsystem : String , name : String ): Unit = {}
529+
530+ def getDiskInfo (names : Seq [String ]): Seq [Disk ] = Seq ()
531+
532+ def getGlobalDiskBytes (disks : Seq [Disk ]): Seq [Disk ] = Seq ()
533+
534+ def getTargetDiskBytes (disks : Seq [Disk ], target : Target ): Seq [Disk ] = Seq ()
359535}
0 commit comments