services/crunchstat/crunchstat.go

   1 package main
   2
   3 import (
   4         "bufio"
   5         "flag"
   6         "errors"
   7         "fmt"
   8         "io"
   9         "io/ioutil"
  10         "log"
  11         "os"
  12         "os/exec"
  13         "os/signal"
  14         "strings"
  15         "syscall"
  16         "time"
  17 )
  18
  19 /*
  20 #include <unistd.h>
  21 #include <sys/types.h>
  22 #include <pwd.h>
  23 #include <stdlib.h>
  24 */
  25 import "C"
  26 // The above block of magic allows us to look up user_hz via _SC_CLK_TCK.
  27
  28 type Cgroup struct {
  29         root   string
  30         parent string
  31         cid    string
  32 }
  33
  34 func CopyPipeToChan(in io.Reader, out chan string, done chan<- bool) {
  35         s := bufio.NewScanner(in)
  36         for s.Scan() {
  37                 out <- s.Text()
  38         }
  39         done <- true
  40 }
  41
  42 func CopyChanToPipe(in <-chan string, out io.Writer) {
  43         for s := range in {
  44                 fmt.Fprintln(out, s)
  45         }
  46 }
  47
  48 func OpenAndReadAll(filename string, log_chan chan<- string) ([]byte, error) {
  49         in, err := os.Open(filename)
  50         if err != nil {
  51                 if log_chan != nil {
  52                         log_chan <- fmt.Sprintf("crunchstat: open %s: %s", filename, err)
  53                 }
  54                 return nil, err
  55         }
  56         defer in.Close()
  57         {
  58                 content, err := ioutil.ReadAll(in)
  59                 if err != nil && log_chan != nil {
  60                         log_chan <- fmt.Sprintf("crunchstat: read %s: %s", filename, err)
  61                 }
  62                 return content, err
  63         }
  64 }
  65
  66 var reportedStatFile map[string]bool
  67 var reportedNoStatFile map[string]bool
  68
  69 func FindStat(stderr chan<- string, cgroup Cgroup, statgroup string, stat string) string {
  70         if reportedStatFile == nil {
  71                 reportedStatFile = make(map[string]bool)
  72                 reportedNoStatFile = make(map[string]bool)
  73         }
  74
  75         var path string
  76         path = fmt.Sprintf("%s/%s/%s/%s/%s", cgroup.root, statgroup, cgroup.parent, cgroup.cid, stat)
  77         if _, err := os.Stat(path); err != nil {
  78                 path = fmt.Sprintf("%s/%s/%s/%s", cgroup.root, cgroup.parent, cgroup.cid, stat)
  79         }
  80         if _, err := os.Stat(path); err != nil {
  81                 path = fmt.Sprintf("%s/%s/%s", cgroup.root, statgroup, stat)
  82         }
  83         if _, err := os.Stat(path); err != nil {
  84                 path = fmt.Sprintf("%s/%s", cgroup.root, stat)
  85         }
  86         if _, err := os.Stat(path); err != nil {
  87                 if _, ok := reportedNoStatFile[path]; !ok {
  88                         stderr <- fmt.Sprintf("crunchstat: did not find stats file (root %s, parent %s, cid %s, statgroup %s, stat %s)", cgroup.root, cgroup.parent, cgroup.cid, statgroup, stat)
  89                         reportedNoStatFile[path] = true
  90                 }
  91                 return ""
  92         }
  93         if _, ok := reportedStatFile[path]; !ok {
  94                 stderr <- fmt.Sprintf("crunchstat: reading stats from %s", path)
  95                 reportedStatFile[path] = true
  96         }
  97         return path
  98 }
  99
 100 func GetContainerNetStats(stderr chan<- string, cgroup Cgroup) (io.Reader, error) {
 101         procsFilename := FindStat(stderr, cgroup, "cpuacct", "cgroup.procs")
 102         procsFile, err := os.Open(procsFilename)
 103         if err != nil {
 104                 stderr <- fmt.Sprintf("crunchstat: open %s: %s", procsFilename, err)
 105                 return nil, err
 106         }
 107         defer procsFile.Close()
 108         reader := bufio.NewScanner(procsFile)
 109         for reader.Scan() {
 110                 taskPid := reader.Text()
 111                 statsFilename := fmt.Sprintf("/proc/%s/net/dev", taskPid)
 112                 stats, err := OpenAndReadAll(statsFilename, stderr)
 113                 if err != nil {
 114                         continue
 115                 }
 116                 return strings.NewReader(string(stats)), nil
 117         }
 118         return nil, errors.New("Could not read stats for any proc in container")
 119 }
 120
 121 type Disk struct {
 122         last_read  int64
 123         next_read  int64
 124         last_write int64
 125         next_write int64
 126 }
 127 var disk map[string]*Disk
 128
 129 func DoBlkIoStats(stderr chan<- string, cgroup Cgroup) {
 130         blkio_io_service_bytes := FindStat(stderr, cgroup, "blkio", "blkio.io_service_bytes")
 131         if blkio_io_service_bytes == "" {
 132                 return
 133         }
 134
 135         if disk == nil {
 136                 disk = make(map[string]*Disk)
 137         }
 138
 139         c, err := os.Open(blkio_io_service_bytes)
 140         if err != nil {
 141                 stderr <- fmt.Sprintf("open %s: %s", blkio_io_service_bytes, err)
 142                 return
 143         }
 144         defer c.Close()
 145         b := bufio.NewScanner(c)
 146         var device, op string
 147         var next int64
 148         for b.Scan() {
 149                 if _, err := fmt.Sscanf(string(b.Text()), "%s %s %d", &device, &op, &next); err != nil {
 150                         continue
 151                 }
 152                 if disk[device] == nil {
 153                         disk[device] = new(Disk)
 154                 }
 155                 if op == "Read" {
 156                         disk[device].last_read = disk[device].next_read
 157                         disk[device].next_read = next
 158                         if disk[device].last_read > 0 && (disk[device].next_read != disk[device].last_read) {
 159                                 stderr <- fmt.Sprintf("crunchstat: blkio.io_service_bytes %s read %v", device, disk[device].next_read-disk[device].last_read)
 160                         }
 161                 }
 162                 if op == "Write" {
 163                         disk[device].last_write = disk[device].next_write
 164                         disk[device].next_write = next
 165                         if disk[device].last_write > 0 && (disk[device].next_write != disk[device].last_write) {
 166                                 stderr <- fmt.Sprintf("crunchstat: blkio.io_service_bytes %s write %v", device, disk[device].next_write-disk[device].last_write)
 167                         }
 168                 }
 169         }
 170 }
 171
 172 func DoMemoryStats(stderr chan<- string, cgroup Cgroup) {
 173         memory_stat := FindStat(stderr, cgroup, "memory", "memory.stat")
 174         if memory_stat == "" {
 175                 return
 176         }
 177         c, err := os.Open(memory_stat)
 178         if err != nil {
 179                 stderr <- fmt.Sprintf("open %s: %s", memory_stat, err)
 180                 return
 181         }
 182         defer c.Close()
 183         b := bufio.NewScanner(c)
 184         var stat string
 185         var val int64
 186         for b.Scan() {
 187                 if _, err := fmt.Sscanf(string(b.Text()), "%s %d", &stat, &val); err != nil {
 188                         continue
 189                 }
 190                 if stat == "rss" {
 191                         stderr <- fmt.Sprintf("crunchstat: mem %d rss", val)
 192                 }
 193         }
 194 }
 195
 196 type NetSample struct {
 197         sampleTime time.Time
 198         txBytes    int64
 199         rxBytes    int64
 200 }
 201
 202 func DoNetworkStats(stderr chan<- string, cgroup Cgroup, lastStat map[string]NetSample) (map[string]NetSample) {
 203         sampleTime := time.Now()
 204         stats, err := GetContainerNetStats(stderr, cgroup)
 205         if err != nil { return lastStat }
 206
 207         if lastStat == nil {
 208                 lastStat = make(map[string]NetSample)
 209         }
 210         scanner := bufio.NewScanner(stats)
 211         Iface: for scanner.Scan() {
 212                 var ifName string
 213                 var rx, tx int64
 214                 words := bufio.NewScanner(strings.NewReader(scanner.Text()))
 215                 words.Split(bufio.ScanWords)
 216                 wordIndex := 0
 217                 for words.Scan() {
 218                         word := words.Text()
 219                         switch wordIndex {
 220                         case 0:
 221                                 ifName = strings.TrimRight(word, ":")
 222                         case 1:
 223                                 if _, err := fmt.Sscanf(word, "%d", &rx); err != nil {
 224                                         continue Iface
 225                                 }
 226                         case 9:
 227                                 if _, err := fmt.Sscanf(word, "%d", &tx); err != nil {
 228                                         continue Iface
 229                                 }
 230                         }
 231                         wordIndex++
 232                 }
 233                 if ifName == "lo" || ifName == "" || wordIndex != 17 {
 234                         // Skip loopback interface and lines with wrong format
 235                         continue
 236                 }
 237                 nextSample := NetSample{}
 238                 nextSample.sampleTime = sampleTime
 239                 nextSample.txBytes = tx
 240                 nextSample.rxBytes = rx
 241                 var delta string
 242                 if lastSample, ok := lastStat[ifName]; ok {
 243                         interval := nextSample.sampleTime.Sub(lastSample.sampleTime).Seconds()
 244                         delta = fmt.Sprintf(" -- interval %.4f seconds %d tx %d rx",
 245                                 interval,
 246                                 tx - lastSample.txBytes,
 247                                 rx - lastSample.rxBytes)
 248                 }
 249                 stderr <- fmt.Sprintf("crunchstat: net:%s %d tx %d rx%s",
 250                         ifName, tx, rx, delta)
 251                 lastStat[ifName] = nextSample
 252         }
 253         return lastStat
 254 }
 255
 256 func PollCgroupStats(cgroup Cgroup, stderr chan string, poll int64, stop_poll_chan <-chan bool) {
 257         var last_user int64 = -1
 258         var last_sys int64 = -1
 259         var last_cpucount int64 = 0
 260
 261         user_hz := float64(C.sysconf(C._SC_CLK_TCK))
 262
 263         var lastNetStat map[string]NetSample = nil
 264
 265         poll_chan := make(chan bool, 1)
 266         go func() {
 267                 // Send periodic poll events.
 268                 poll_chan <- true
 269                 for {
 270                         time.Sleep(time.Duration(poll) * time.Millisecond)
 271                         poll_chan <- true
 272                 }
 273         }()
 274         for {
 275                 bedtime := time.Now()
 276                 select {
 277                 case <-stop_poll_chan:
 278                         return
 279                 case <-poll_chan:
 280                         // Emit stats, then select again.
 281                 }
 282                 morning := time.Now()
 283                 elapsed := morning.Sub(bedtime).Seconds()
 284                 cpuset_cpus := FindStat(stderr, cgroup, "cpuset", "cpuset.cpus")
 285                 if cpuset_cpus != "" {
 286                         b, err := OpenAndReadAll(cpuset_cpus, stderr)
 287                         if err != nil {
 288                                 // cgroup probably gone -- skip other stats too.
 289                                 continue
 290                         }
 291                         sp := strings.Split(string(b), ",")
 292                         cpus := int64(0)
 293                         for _, v := range sp {
 294                                 var min, max int64
 295                                 n, _ := fmt.Sscanf(v, "%d-%d", &min, &max)
 296                                 if n == 2 {
 297                                         cpus += (max - min) + 1
 298                                 } else {
 299                                         cpus += 1
 300                                 }
 301                         }
 302                         last_cpucount = cpus
 303                 }
 304                 cpuacct_stat := FindStat(stderr, cgroup, "cpuacct", "cpuacct.stat")
 305                 if cpuacct_stat != "" {
 306                         b, err := OpenAndReadAll(cpuacct_stat, stderr)
 307                         if err != nil {
 308                                 // Next time around, last_user would
 309                                 // be >1 interval old, so stats will
 310                                 // be incorrect. Start over instead.
 311                                 last_user = -1
 312
 313                                 // cgroup probably gone -- skip other stats too.
 314                                 continue
 315                         }
 316                         var next_user int64
 317                         var next_sys int64
 318                         fmt.Sscanf(string(b), "user %d\nsystem %d", &next_user, &next_sys)
 319
 320                         delta := ""
 321                         if elapsed > 0 && last_user != -1 {
 322                                 delta = fmt.Sprintf(" -- interval %.4f seconds %.4f user %.4f sys",
 323                                         elapsed,
 324                                         float64(next_user - last_user) / user_hz,
 325                                         float64(next_sys - last_sys) / user_hz)
 326                         }
 327                         stderr <- fmt.Sprintf("crunchstat: cpu %.4f user %.4f sys %d cpus%s",
 328                                 float64(next_user) / user_hz,
 329                                 float64(next_sys) / user_hz,
 330                                 last_cpucount,
 331                                 delta)
 332                         last_user = next_user
 333                         last_sys = next_sys
 334                 }
 335
 336                 DoBlkIoStats(stderr, cgroup)
 337                 DoMemoryStats(stderr, cgroup)
 338                 lastNetStat = DoNetworkStats(stderr, cgroup, lastNetStat)
 339         }
 340 }
 341
 342 func run(logger *log.Logger) error {
 343
 344         var (
 345                 cgroup_root    string
 346                 cgroup_parent  string
 347                 cgroup_cidfile string
 348                 wait           int64
 349                 poll           int64
 350         )
 351
 352         flag.StringVar(&cgroup_root, "cgroup-root", "", "Root of cgroup tree")
 353         flag.StringVar(&cgroup_parent, "cgroup-parent", "", "Name of container parent under cgroup")
 354         flag.StringVar(&cgroup_cidfile, "cgroup-cid", "", "Path to container id file")
 355         flag.Int64Var(&wait, "wait", 5, "Maximum time (in seconds) to wait for cid file to show up")
 356         flag.Int64Var(&poll, "poll", 1000, "Polling frequency, in milliseconds")
 357
 358         flag.Parse()
 359
 360         if cgroup_root == "" {
 361                 logger.Fatal("Must provide -cgroup-root")
 362         }
 363
 364         stderr_chan := make(chan string, 1)
 365         defer close(stderr_chan)
 366         finish_chan := make(chan bool)
 367         defer close(finish_chan)
 368
 369         go CopyChanToPipe(stderr_chan, os.Stderr)
 370
 371         var cmd *exec.Cmd
 372
 373         if len(flag.Args()) > 0 {
 374                 // Set up subprocess
 375                 cmd = exec.Command(flag.Args()[0], flag.Args()[1:]...)
 376
 377                 logger.Print("Running ", flag.Args())
 378
 379                 // Child process will use our stdin and stdout pipes
 380                 // (we close our copies below)
 381                 cmd.Stdin = os.Stdin
 382                 cmd.Stdout = os.Stdout
 383
 384                 // Forward SIGINT and SIGTERM to inner process
 385                 term := make(chan os.Signal, 1)
 386                 go func(sig <-chan os.Signal) {
 387                         catch := <-sig
 388                         if cmd.Process != nil {
 389                                 cmd.Process.Signal(catch)
 390                         }
 391                         logger.Print("caught signal: ", catch)
 392                 }(term)
 393                 signal.Notify(term, syscall.SIGTERM)
 394                 signal.Notify(term, syscall.SIGINT)
 395
 396                 // Funnel stderr through our channel
 397                 stderr_pipe, err := cmd.StderrPipe()
 398                 if err != nil {
 399                         logger.Fatal(err)
 400                 }
 401                 go CopyPipeToChan(stderr_pipe, stderr_chan, finish_chan)
 402
 403                 // Run subprocess
 404                 if err := cmd.Start(); err != nil {
 405                         logger.Fatal(err)
 406                 }
 407
 408                 // Close stdin/stdout in this (parent) process
 409                 os.Stdin.Close()
 410                 os.Stdout.Close()
 411         }
 412
 413         // Read the cid file
 414         var container_id string
 415         if cgroup_cidfile != "" {
 416                 // wait up to 'wait' seconds for the cid file to appear
 417                 ok := false
 418                 var i time.Duration
 419                 for i = 0; i < time.Duration(wait)*time.Second; i += (100 * time.Millisecond) {
 420                         cid, err := OpenAndReadAll(cgroup_cidfile, nil)
 421                         if err == nil && len(cid) > 0 {
 422                                 ok = true
 423                                 container_id = string(cid)
 424                                 break
 425                         }
 426                         time.Sleep(100 * time.Millisecond)
 427                 }
 428                 if !ok {
 429                         logger.Printf("Could not read cid file %s", cgroup_cidfile)
 430                 }
 431         }
 432
 433         stop_poll_chan := make(chan bool, 1)
 434         cgroup := Cgroup{cgroup_root, cgroup_parent, container_id}
 435         go PollCgroupStats(cgroup, stderr_chan, poll, stop_poll_chan)
 436
 437         // When the child exits, tell the polling goroutine to stop.
 438         defer func() { stop_poll_chan <- true }()
 439
 440         // Wait for CopyPipeToChan to consume child's stderr pipe
 441         <-finish_chan
 442
 443         return cmd.Wait()
 444 }
 445
 446 func main() {
 447         logger := log.New(os.Stderr, "crunchstat: ", 0)
 448         if err := run(logger); err != nil {
 449                 if exiterr, ok := err.(*exec.ExitError); ok {
 450                         // The program has exited with an exit code != 0
 451
 452                         // This works on both Unix and
 453                         // Windows. Although package syscall is
 454                         // generally platform dependent, WaitStatus is
 455                         // defined for both Unix and Windows and in
 456                         // both cases has an ExitStatus() method with
 457                         // the same signature.
 458                         if status, ok := exiterr.Sys().(syscall.WaitStatus); ok {
 459                                 os.Exit(status.ExitStatus())
 460                         }
 461                 } else {
 462                         logger.Fatalf("cmd.Wait: %v", err)
 463                 }
 464         }
 465 }