lib/crunchstat/crunchstat.go

   1 // Copyright (C) The Arvados Authors. All rights reserved.
   2 //
   3 // SPDX-License-Identifier: AGPL-3.0
   4
   5 // Package crunchstat reports resource usage (CPU, memory, disk,
   6 // network) for a cgroup.
   7 package crunchstat
   8
   9 import (
  10         "bufio"
  11         "bytes"
  12         "errors"
  13         "fmt"
  14         "io"
  15         "io/fs"
  16         "io/ioutil"
  17         "os"
  18         "path/filepath"
  19         "regexp"
  20         "sort"
  21         "strconv"
  22         "strings"
  23         "sync"
  24         "syscall"
  25         "time"
  26 )
  27
  28 // crunchstat collects all memory statistics, but only reports these.
  29 var memoryStats = [...]string{"cache", "swap", "pgmajfault", "rss"}
  30
  31 type logPrinter interface {
  32         Printf(fmt string, args ...interface{})
  33 }
  34
  35 // A Reporter gathers statistics for a cgroup and writes them to a
  36 // log.Logger.
  37 type Reporter struct {
  38         // Func that returns the pid of a process inside the desired
  39         // cgroup. Reporter will call Pid periodically until it
  40         // returns a positive number, then start reporting stats for
  41         // the cgroup that process belongs to.
  42         //
  43         // Pid is used when cgroups v2 is available. For cgroups v1,
  44         // see below.
  45         Pid func() int
  46
  47         // Interval between samples. Must be positive.
  48         PollPeriod time.Duration
  49
  50         // Temporary directory, will be monitored for available, used
  51         // & total space.
  52         TempDir string
  53
  54         // Where to write statistics. Must not be nil.
  55         Logger logPrinter
  56
  57         // When stats cross thresholds configured in the fields below,
  58         // they are reported to this logger.
  59         ThresholdLogger logPrinter
  60
  61         // MemThresholds maps memory stat names to slices of thresholds.
  62         // When the corresponding stat exceeds a threshold, that will be logged.
  63         MemThresholds map[string][]Threshold
  64
  65         // Filesystem to read /proc entries and cgroup stats from.
  66         // Non-nil for testing, nil for real root filesystem.
  67         FS fs.FS
  68
  69         // Enable debug messages.
  70         Debug bool
  71
  72         // available cgroup hierarchies
  73         statFiles struct {
  74                 cpuMax            string // v2
  75                 cpusetCpus        string // v1,v2 (via /proc/$PID/cpuset)
  76                 cpuacctStat       string // v1 (via /proc/$PID/cgroup => cpuacct)
  77                 cpuStat           string // v2
  78                 ioServiceBytes    string // v1 (via /proc/$PID/cgroup => blkio)
  79                 ioStat            string // v2
  80                 memoryStat        string // v1 and v2 (but v2 is missing some entries)
  81                 memoryCurrent     string // v2
  82                 memorySwapCurrent string // v2
  83                 netDev            string // /proc/$PID/net/dev
  84         }
  85
  86         kernelPageSize      int64
  87         lastNetSample       map[string]ioSample
  88         lastDiskIOSample    map[string]ioSample
  89         lastCPUSample       cpuSample
  90         lastDiskSpaceSample diskSpaceSample
  91         lastMemSample       memSample
  92         maxDiskSpaceSample  diskSpaceSample
  93         maxMemSample        map[memoryKey]int64
  94
  95         // process returned by Pid(), whose cgroup stats we are
  96         // reporting
  97         pid int
  98
  99         // individual processes whose memory size we are reporting
 100         reportPIDs   map[string]int
 101         reportPIDsMu sync.Mutex
 102
 103         done    chan struct{} // closed when we should stop reporting
 104         ready   chan struct{} // have pid and stat files
 105         flushed chan struct{} // closed when we have made our last report
 106 }
 107
 108 type Threshold struct {
 109         percentage int64
 110         threshold  int64
 111         total      int64
 112 }
 113
 114 func NewThresholdFromPercentage(total int64, percentage int64) Threshold {
 115         return Threshold{
 116                 percentage: percentage,
 117                 threshold:  total * percentage / 100,
 118                 total:      total,
 119         }
 120 }
 121
 122 func NewThresholdsFromPercentages(total int64, percentages []int64) (thresholds []Threshold) {
 123         for _, percentage := range percentages {
 124                 thresholds = append(thresholds, NewThresholdFromPercentage(total, percentage))
 125         }
 126         return
 127 }
 128
 129 // memoryKey is a key into Reporter.maxMemSample.
 130 // Initialize it with just statName to get the host/cgroup maximum.
 131 // Initialize it with all fields to get that process' maximum.
 132 type memoryKey struct {
 133         processID   int
 134         processName string
 135         statName    string
 136 }
 137
 138 // Start starts monitoring in a new goroutine, and returns
 139 // immediately.
 140 //
 141 // The monitoring goroutine waits for a non-empty CIDFile to appear
 142 // (unless CID is non-empty). Then it waits for the accounting files
 143 // to appear for the monitored container. Then it collects and reports
 144 // statistics until Stop is called.
 145 //
 146 // Callers should not call Start more than once.
 147 //
 148 // Callers should not modify public data fields after calling Start.
 149 func (r *Reporter) Start() {
 150         r.done = make(chan struct{})
 151         r.ready = make(chan struct{})
 152         r.flushed = make(chan struct{})
 153         if r.FS == nil {
 154                 r.FS = os.DirFS("/")
 155         }
 156         go r.run()
 157 }
 158
 159 // ReportPID starts reporting stats for a specified process.
 160 func (r *Reporter) ReportPID(name string, pid int) {
 161         r.reportPIDsMu.Lock()
 162         defer r.reportPIDsMu.Unlock()
 163         if r.reportPIDs == nil {
 164                 r.reportPIDs = map[string]int{name: pid}
 165         } else {
 166                 r.reportPIDs[name] = pid
 167         }
 168 }
 169
 170 // Stop reporting. Do not call more than once, or before calling
 171 // Start.
 172 //
 173 // Nothing will be logged after Stop returns unless you call a Log* method.
 174 func (r *Reporter) Stop() {
 175         close(r.done)
 176         <-r.flushed
 177 }
 178
 179 var v1keys = map[string]bool{
 180         "blkio":   true,
 181         "cpuacct": true,
 182         "cpuset":  true,
 183         "memory":  true,
 184 }
 185
 186 // Find cgroup hierarchies in /proc/mounts, e.g.,
 187 //
 188 //      {
 189 //              "blkio": "/sys/fs/cgroup/blkio",
 190 //              "unified": "/sys/fs/cgroup/unified",
 191 //      }
 192 func (r *Reporter) cgroupMounts() map[string]string {
 193         procmounts, err := fs.ReadFile(r.FS, "proc/mounts")
 194         if err != nil {
 195                 r.Logger.Printf("error reading /proc/mounts: %s", err)
 196                 return nil
 197         }
 198         mounts := map[string]string{}
 199         for _, line := range bytes.Split(procmounts, []byte{'\n'}) {
 200                 fields := bytes.SplitN(line, []byte{' '}, 6)
 201                 if len(fields) != 6 {
 202                         continue
 203                 }
 204                 switch string(fields[2]) {
 205                 case "cgroup2":
 206                         // cgroup /sys/fs/cgroup/unified cgroup2 rw,nosuid,nodev,noexec,relatime 0 0
 207                         mounts["unified"] = string(fields[1])
 208                 case "cgroup":
 209                         // cgroup /sys/fs/cgroup/blkio cgroup rw,nosuid,nodev,noexec,relatime,blkio 0 0
 210                         options := bytes.Split(fields[3], []byte{','})
 211                         for _, option := range options {
 212                                 option := string(option)
 213                                 if v1keys[option] {
 214                                         mounts[option] = string(fields[1])
 215                                         break
 216                                 }
 217                         }
 218                 }
 219         }
 220         return mounts
 221 }
 222
 223 // generate map of cgroup controller => path for r.pid.
 224 //
 225 // the "unified" controller represents cgroups v2.
 226 func (r *Reporter) cgroupPaths(mounts map[string]string) map[string]string {
 227         if len(mounts) == 0 {
 228                 return nil
 229         }
 230         procdir := fmt.Sprintf("proc/%d", r.pid)
 231         buf, err := fs.ReadFile(r.FS, procdir+"/cgroup")
 232         if err != nil {
 233                 r.Logger.Printf("error reading cgroup file: %s", err)
 234                 return nil
 235         }
 236         paths := map[string]string{}
 237         for _, line := range bytes.Split(buf, []byte{'\n'}) {
 238                 // The entry for cgroup v2 is always in the format
 239                 // "0::$PATH" --
 240                 // https://docs.kernel.org/admin-guide/cgroup-v2.html
 241                 if bytes.HasPrefix(line, []byte("0::/")) && mounts["unified"] != "" {
 242                         paths["unified"] = mounts["unified"] + string(line[3:])
 243                         continue
 244                 }
 245                 // cgroups v1 entries look like
 246                 // "6:cpu,cpuacct:/user.slice"
 247                 fields := bytes.SplitN(line, []byte{':'}, 3)
 248                 if len(fields) != 3 {
 249                         continue
 250                 }
 251                 for _, key := range bytes.Split(fields[1], []byte{','}) {
 252                         key := string(key)
 253                         if mounts[key] != "" {
 254                                 paths[key] = mounts[key] + string(fields[2])
 255                         }
 256                 }
 257         }
 258         // In unified mode, /proc/$PID/cgroup doesn't have a cpuset
 259         // entry, but we still need it -- there's no cpuset.cpus file
 260         // in the cgroup2 subtree indicated by the 0::$PATH entry. We
 261         // have to get the right path from /proc/$PID/cpuset.
 262         if _, found := paths["cpuset"]; !found && mounts["unified"] != "" {
 263                 buf, _ := fs.ReadFile(r.FS, procdir+"/cpuset")
 264                 cpusetPath := string(bytes.TrimRight(buf, "\n"))
 265                 paths["cpuset"] = mounts["unified"] + cpusetPath
 266         }
 267         return paths
 268 }
 269
 270 func (r *Reporter) findStatFiles() {
 271         mounts := r.cgroupMounts()
 272         paths := r.cgroupPaths(mounts)
 273         done := map[*string]bool{}
 274         for _, try := range []struct {
 275                 statFile *string
 276                 pathkey  string
 277                 file     string
 278         }{
 279                 {&r.statFiles.cpuMax, "unified", "cpu.max"},
 280                 {&r.statFiles.cpusetCpus, "cpuset", "cpuset.cpus.effective"},
 281                 {&r.statFiles.cpusetCpus, "cpuset", "cpuset.cpus"},
 282                 {&r.statFiles.cpuacctStat, "cpuacct", "cpuacct.stat"},
 283                 {&r.statFiles.cpuStat, "unified", "cpu.stat"},
 284                 // blkio.throttle.io_service_bytes must precede
 285                 // blkio.io_service_bytes -- on ubuntu1804, the latter
 286                 // is present but reports 0
 287                 {&r.statFiles.ioServiceBytes, "blkio", "blkio.throttle.io_service_bytes"},
 288                 {&r.statFiles.ioServiceBytes, "blkio", "blkio.io_service_bytes"},
 289                 {&r.statFiles.ioStat, "unified", "io.stat"},
 290                 {&r.statFiles.memoryStat, "unified", "memory.stat"},
 291                 {&r.statFiles.memoryStat, "memory", "memory.stat"},
 292                 {&r.statFiles.memoryCurrent, "unified", "memory.current"},
 293                 {&r.statFiles.memorySwapCurrent, "unified", "memory.swap.current"},
 294         } {
 295                 startpath, ok := paths[try.pathkey]
 296                 if !ok || done[try.statFile] {
 297                         continue
 298                 }
 299                 // /proc/$PID/cgroup says cgroup path is
 300                 // /exa/mple/exa/mple, however, sometimes the file we
 301                 // need is not under that path, it's only available in
 302                 // a parent cgroup's dir.  So we start at
 303                 // /sys/fs/cgroup/unified/exa/mple/exa/mple/ and walk
 304                 // up to /sys/fs/cgroup/unified/ until we find the
 305                 // desired file.
 306                 //
 307                 // This might mean our reported stats include more
 308                 // cgroups in the cgroup tree, but it's the best we
 309                 // can do.
 310                 for path := startpath; path != "" && path != "/" && (path == startpath || strings.HasPrefix(path, mounts[try.pathkey])); path, _ = filepath.Split(strings.TrimRight(path, "/")) {
 311                         target := strings.TrimLeft(filepath.Join(path, try.file), "/")
 312                         buf, err := fs.ReadFile(r.FS, target)
 313                         if err != nil || len(buf) == 0 || bytes.Equal(buf, []byte{'\n'}) {
 314                                 if r.Debug {
 315                                         if os.IsNotExist(err) {
 316                                                 // don't stutter
 317                                                 err = os.ErrNotExist
 318                                         }
 319                                         r.Logger.Printf("skip /%s: %s", target, err)
 320                                 }
 321                                 continue
 322                         }
 323                         *try.statFile = target
 324                         done[try.statFile] = true
 325                         r.Logger.Printf("notice: reading stats from /%s", target)
 326                         break
 327                 }
 328         }
 329
 330         netdev := fmt.Sprintf("proc/%d/net/dev", r.pid)
 331         if buf, err := fs.ReadFile(r.FS, netdev); err == nil && len(buf) > 0 {
 332                 r.statFiles.netDev = netdev
 333                 r.Logger.Printf("using /%s", netdev)
 334         }
 335 }
 336
 337 func (r *Reporter) reportMemoryMax(logger logPrinter, source, statName string, value, limit int64) {
 338         var units string
 339         switch statName {
 340         case "pgmajfault":
 341                 units = "faults"
 342         default:
 343                 units = "bytes"
 344         }
 345         if limit > 0 {
 346                 percentage := 100 * value / limit
 347                 logger.Printf("Maximum %s memory %s usage was %d%%, %d/%d %s",
 348                         source, statName, percentage, value, limit, units)
 349         } else {
 350                 logger.Printf("Maximum %s memory %s usage was %d %s",
 351                         source, statName, value, units)
 352         }
 353 }
 354
 355 func (r *Reporter) LogMaxima(logger logPrinter, memLimits map[string]int64) {
 356         if r.lastCPUSample.hasData {
 357                 logger.Printf("Total CPU usage was %f user and %f sys on %.2f CPUs",
 358                         r.lastCPUSample.user, r.lastCPUSample.sys, r.lastCPUSample.cpus)
 359         }
 360         for disk, sample := range r.lastDiskIOSample {
 361                 logger.Printf("Total disk I/O on %s was %d bytes written and %d bytes read",
 362                         disk, sample.txBytes, sample.rxBytes)
 363         }
 364         if r.maxDiskSpaceSample.total > 0 {
 365                 percentage := 100 * r.maxDiskSpaceSample.used / r.maxDiskSpaceSample.total
 366                 logger.Printf("Maximum disk usage was %d%%, %d/%d bytes",
 367                         percentage, r.maxDiskSpaceSample.used, r.maxDiskSpaceSample.total)
 368         }
 369         for _, statName := range memoryStats {
 370                 value, ok := r.maxMemSample[memoryKey{statName: "total_" + statName}]
 371                 if !ok {
 372                         value, ok = r.maxMemSample[memoryKey{statName: statName}]
 373                 }
 374                 if ok {
 375                         r.reportMemoryMax(logger, "container", statName, value, memLimits[statName])
 376                 }
 377         }
 378         for ifname, sample := range r.lastNetSample {
 379                 logger.Printf("Total network I/O on %s was %d bytes written and %d bytes read",
 380                         ifname, sample.txBytes, sample.rxBytes)
 381         }
 382 }
 383
 384 func (r *Reporter) LogProcessMemMax(logger logPrinter) {
 385         for memKey, value := range r.maxMemSample {
 386                 if memKey.processName == "" {
 387                         continue
 388                 }
 389                 r.reportMemoryMax(logger, memKey.processName, memKey.statName, value, 0)
 390         }
 391 }
 392
 393 func (r *Reporter) readAllOrWarn(in io.Reader) ([]byte, error) {
 394         content, err := ioutil.ReadAll(in)
 395         if err != nil {
 396                 r.Logger.Printf("warning: %v", err)
 397         }
 398         return content, err
 399 }
 400
 401 type ioSample struct {
 402         sampleTime time.Time
 403         txBytes    int64
 404         rxBytes    int64
 405 }
 406
 407 func (r *Reporter) doBlkIOStats() {
 408         var sampleTime = time.Now()
 409         newSamples := make(map[string]ioSample)
 410
 411         if r.statFiles.ioStat != "" {
 412                 statfile, err := fs.ReadFile(r.FS, r.statFiles.ioStat)
 413                 if err != nil {
 414                         return
 415                 }
 416                 for _, line := range bytes.Split(statfile, []byte{'\n'}) {
 417                         // 254:16 rbytes=72163328 wbytes=117370880 rios=3811 wios=3906 dbytes=0 dios=0
 418                         words := bytes.Split(line, []byte{' '})
 419                         if len(words) < 2 {
 420                                 continue
 421                         }
 422                         thisSample := ioSample{sampleTime, -1, -1}
 423                         for _, kv := range words[1:] {
 424                                 if bytes.HasPrefix(kv, []byte("rbytes=")) {
 425                                         fmt.Sscanf(string(kv[7:]), "%d", &thisSample.rxBytes)
 426                                 } else if bytes.HasPrefix(kv, []byte("wbytes=")) {
 427                                         fmt.Sscanf(string(kv[7:]), "%d", &thisSample.txBytes)
 428                                 }
 429                         }
 430                         if thisSample.rxBytes >= 0 && thisSample.txBytes >= 0 {
 431                                 newSamples[string(words[0])] = thisSample
 432                         }
 433                 }
 434         } else if r.statFiles.ioServiceBytes != "" {
 435                 statfile, err := fs.ReadFile(r.FS, r.statFiles.ioServiceBytes)
 436                 if err != nil {
 437                         return
 438                 }
 439                 for _, line := range bytes.Split(statfile, []byte{'\n'}) {
 440                         var device, op string
 441                         var val int64
 442                         if _, err := fmt.Sscanf(string(line), "%s %s %d", &device, &op, &val); err != nil {
 443                                 continue
 444                         }
 445                         var thisSample ioSample
 446                         var ok bool
 447                         if thisSample, ok = newSamples[device]; !ok {
 448                                 thisSample = ioSample{sampleTime, -1, -1}
 449                         }
 450                         switch op {
 451                         case "Read":
 452                                 thisSample.rxBytes = val
 453                         case "Write":
 454                                 thisSample.txBytes = val
 455                         }
 456                         newSamples[device] = thisSample
 457                 }
 458         }
 459
 460         for dev, sample := range newSamples {
 461                 if sample.txBytes < 0 || sample.rxBytes < 0 {
 462                         continue
 463                 }
 464                 delta := ""
 465                 if prev, ok := r.lastDiskIOSample[dev]; ok {
 466                         delta = fmt.Sprintf(" -- interval %.4f seconds %d write %d read",
 467                                 sample.sampleTime.Sub(prev.sampleTime).Seconds(),
 468                                 sample.txBytes-prev.txBytes,
 469                                 sample.rxBytes-prev.rxBytes)
 470                 }
 471                 r.Logger.Printf("blkio:%s %d write %d read%s\n", dev, sample.txBytes, sample.rxBytes, delta)
 472                 r.lastDiskIOSample[dev] = sample
 473         }
 474 }
 475
 476 type memSample struct {
 477         sampleTime time.Time
 478         memStat    map[string]int64
 479 }
 480
 481 func (r *Reporter) getMemSample() {
 482         thisSample := memSample{time.Now(), make(map[string]int64)}
 483
 484         // memory.stat contains "pgmajfault" in cgroups v1 and v2. It
 485         // also contains "rss", "swap", and "cache" in cgroups v1.
 486         c, err := r.FS.Open(r.statFiles.memoryStat)
 487         if err != nil {
 488                 return
 489         }
 490         defer c.Close()
 491         b := bufio.NewScanner(c)
 492         for b.Scan() {
 493                 var stat string
 494                 var val int64
 495                 if _, err := fmt.Sscanf(string(b.Text()), "%s %d", &stat, &val); err != nil {
 496                         continue
 497                 }
 498                 thisSample.memStat[stat] = val
 499         }
 500
 501         // In cgroups v2, we need to read "memory.current" and
 502         // "memory.swap.current" as well.
 503         for stat, fnm := range map[string]string{
 504                 // memory.current includes cache. We don't get
 505                 // separate rss/cache values, so we call
 506                 // memory usage "rss" for compatibility, and
 507                 // omit "cache".
 508                 "rss":  r.statFiles.memoryCurrent,
 509                 "swap": r.statFiles.memorySwapCurrent,
 510         } {
 511                 if fnm == "" {
 512                         continue
 513                 }
 514                 buf, err := fs.ReadFile(r.FS, fnm)
 515                 if err != nil {
 516                         continue
 517                 }
 518                 var val int64
 519                 _, err = fmt.Sscanf(string(buf), "%d", &val)
 520                 if err != nil {
 521                         continue
 522                 }
 523                 thisSample.memStat[stat] = val
 524         }
 525         for stat, val := range thisSample.memStat {
 526                 maxKey := memoryKey{statName: stat}
 527                 if val > r.maxMemSample[maxKey] {
 528                         r.maxMemSample[maxKey] = val
 529                 }
 530         }
 531         r.lastMemSample = thisSample
 532
 533         if r.ThresholdLogger != nil {
 534                 for statName, thresholds := range r.MemThresholds {
 535                         statValue, ok := thisSample.memStat["total_"+statName]
 536                         if !ok {
 537                                 statValue, ok = thisSample.memStat[statName]
 538                                 if !ok {
 539                                         continue
 540                                 }
 541                         }
 542                         var index int
 543                         var statThreshold Threshold
 544                         for index, statThreshold = range thresholds {
 545                                 if statValue < statThreshold.threshold {
 546                                         break
 547                                 } else if statThreshold.percentage > 0 {
 548                                         r.ThresholdLogger.Printf("Container using over %d%% of memory (%s %d/%d bytes)",
 549                                                 statThreshold.percentage, statName, statValue, statThreshold.total)
 550                                 } else {
 551                                         r.ThresholdLogger.Printf("Container using over %d of memory (%s %s bytes)",
 552                                                 statThreshold.threshold, statName, statValue)
 553                                 }
 554                         }
 555                         r.MemThresholds[statName] = thresholds[index:]
 556                 }
 557         }
 558 }
 559
 560 func (r *Reporter) reportMemSample() {
 561         var outstat bytes.Buffer
 562         for _, key := range memoryStats {
 563                 // Use "total_X" stats (entire hierarchy) if enabled,
 564                 // otherwise just the single cgroup -- see
 565                 // https://www.kernel.org/doc/Documentation/cgroup-v1/memory.txt
 566                 if val, ok := r.lastMemSample.memStat["total_"+key]; ok {
 567                         fmt.Fprintf(&outstat, " %d %s", val, key)
 568                 } else if val, ok := r.lastMemSample.memStat[key]; ok {
 569                         fmt.Fprintf(&outstat, " %d %s", val, key)
 570                 }
 571         }
 572         r.Logger.Printf("mem%s\n", outstat.String())
 573 }
 574
 575 func (r *Reporter) doProcmemStats() {
 576         if r.kernelPageSize == 0 {
 577                 // assign "don't try again" value in case we give up
 578                 // and return without assigning the real value
 579                 r.kernelPageSize = -1
 580                 buf, err := fs.ReadFile(r.FS, "proc/self/smaps")
 581                 if err != nil {
 582                         r.Logger.Printf("error reading /proc/self/smaps: %s", err)
 583                         return
 584                 }
 585                 m := regexp.MustCompile(`\nKernelPageSize:\s*(\d+) kB\n`).FindSubmatch(buf)
 586                 if len(m) != 2 {
 587                         r.Logger.Printf("error parsing /proc/self/smaps: KernelPageSize not found")
 588                         return
 589                 }
 590                 size, err := strconv.ParseInt(string(m[1]), 10, 64)
 591                 if err != nil {
 592                         r.Logger.Printf("error parsing /proc/self/smaps: KernelPageSize %q: %s", m[1], err)
 593                         return
 594                 }
 595                 r.kernelPageSize = size * 1024
 596         } else if r.kernelPageSize < 0 {
 597                 // already failed to determine page size, don't keep
 598                 // trying/logging
 599                 return
 600         }
 601
 602         r.reportPIDsMu.Lock()
 603         defer r.reportPIDsMu.Unlock()
 604         procnames := make([]string, 0, len(r.reportPIDs))
 605         for name := range r.reportPIDs {
 606                 procnames = append(procnames, name)
 607         }
 608         sort.Strings(procnames)
 609         procmem := ""
 610         for _, procname := range procnames {
 611                 pid := r.reportPIDs[procname]
 612                 buf, err := fs.ReadFile(r.FS, fmt.Sprintf("proc/%d/stat", pid))
 613                 if err != nil {
 614                         continue
 615                 }
 616                 // If the executable name contains a ')' char,
 617                 // /proc/$pid/stat will look like '1234 (exec name)) S
 618                 // 123 ...' -- the last ')' is the end of the 2nd
 619                 // field.
 620                 paren := bytes.LastIndexByte(buf, ')')
 621                 if paren < 0 {
 622                         continue
 623                 }
 624                 fields := bytes.SplitN(buf[paren:], []byte{' '}, 24)
 625                 if len(fields) < 24 {
 626                         continue
 627                 }
 628                 // rss is the 24th field in .../stat, and fields[0]
 629                 // here is the last char ')' of the 2nd field, so
 630                 // rss is fields[22]
 631                 rss, err := strconv.ParseInt(string(fields[22]), 10, 64)
 632                 if err != nil {
 633                         continue
 634                 }
 635                 value := rss * r.kernelPageSize
 636                 procmem += fmt.Sprintf(" %d %s", value, procname)
 637                 maxKey := memoryKey{pid, procname, "rss"}
 638                 if value > r.maxMemSample[maxKey] {
 639                         r.maxMemSample[maxKey] = value
 640                 }
 641         }
 642         if procmem != "" {
 643                 r.Logger.Printf("procmem%s\n", procmem)
 644         }
 645 }
 646
 647 func (r *Reporter) doNetworkStats() {
 648         if r.statFiles.netDev == "" {
 649                 return
 650         }
 651         sampleTime := time.Now()
 652         stats, err := r.FS.Open(r.statFiles.netDev)
 653         if err != nil {
 654                 return
 655         }
 656         defer stats.Close()
 657         scanner := bufio.NewScanner(stats)
 658         for scanner.Scan() {
 659                 var ifName string
 660                 var rx, tx int64
 661                 words := strings.Fields(scanner.Text())
 662                 if len(words) != 17 {
 663                         // Skip lines with wrong format
 664                         continue
 665                 }
 666                 ifName = strings.TrimRight(words[0], ":")
 667                 if ifName == "lo" || ifName == "" {
 668                         // Skip loopback interface and lines with wrong format
 669                         continue
 670                 }
 671                 if tx, err = strconv.ParseInt(words[9], 10, 64); err != nil {
 672                         continue
 673                 }
 674                 if rx, err = strconv.ParseInt(words[1], 10, 64); err != nil {
 675                         continue
 676                 }
 677                 nextSample := ioSample{}
 678                 nextSample.sampleTime = sampleTime
 679                 nextSample.txBytes = tx
 680                 nextSample.rxBytes = rx
 681                 var delta string
 682                 if prev, ok := r.lastNetSample[ifName]; ok {
 683                         interval := nextSample.sampleTime.Sub(prev.sampleTime).Seconds()
 684                         delta = fmt.Sprintf(" -- interval %.4f seconds %d tx %d rx",
 685                                 interval,
 686                                 tx-prev.txBytes,
 687                                 rx-prev.rxBytes)
 688                 }
 689                 r.Logger.Printf("net:%s %d tx %d rx%s\n", ifName, tx, rx, delta)
 690                 r.lastNetSample[ifName] = nextSample
 691         }
 692 }
 693
 694 type diskSpaceSample struct {
 695         hasData    bool
 696         sampleTime time.Time
 697         total      uint64
 698         used       uint64
 699         available  uint64
 700 }
 701
 702 func (r *Reporter) doDiskSpaceStats() {
 703         s := syscall.Statfs_t{}
 704         err := syscall.Statfs(r.TempDir, &s)
 705         if err != nil {
 706                 return
 707         }
 708         bs := uint64(s.Bsize)
 709         nextSample := diskSpaceSample{
 710                 hasData:    true,
 711                 sampleTime: time.Now(),
 712                 total:      s.Blocks * bs,
 713                 used:       (s.Blocks - s.Bfree) * bs,
 714                 available:  s.Bavail * bs,
 715         }
 716         if nextSample.used > r.maxDiskSpaceSample.used {
 717                 r.maxDiskSpaceSample = nextSample
 718         }
 719
 720         var delta string
 721         if r.lastDiskSpaceSample.hasData {
 722                 prev := r.lastDiskSpaceSample
 723                 interval := nextSample.sampleTime.Sub(prev.sampleTime).Seconds()
 724                 delta = fmt.Sprintf(" -- interval %.4f seconds %d used",
 725                         interval,
 726                         int64(nextSample.used-prev.used))
 727         }
 728         r.Logger.Printf("statfs %d available %d used %d total%s\n",
 729                 nextSample.available, nextSample.used, nextSample.total, delta)
 730         r.lastDiskSpaceSample = nextSample
 731 }
 732
 733 type cpuSample struct {
 734         hasData    bool // to distinguish the zero value from real data
 735         sampleTime time.Time
 736         user       float64
 737         sys        float64
 738         cpus       float64
 739 }
 740
 741 // Return the number of virtual CPUs available in the container. This
 742 // can be based on a scheduling ratio (which is not necessarily a
 743 // whole number) or a restricted set of accessible CPUs.
 744 //
 745 // Return the number of host processors based on /proc/cpuinfo if
 746 // cgroupfs doesn't reveal anything.
 747 //
 748 // Return 0 if even that doesn't work.
 749 func (r *Reporter) getCPUCount() float64 {
 750         if buf, err := fs.ReadFile(r.FS, r.statFiles.cpuMax); err == nil {
 751                 // cpu.max looks like "150000 100000" if CPU usage is
 752                 // restricted to 150% (docker run --cpus=1.5), or "max
 753                 // 100000\n" if not.
 754                 var max, period int64
 755                 if _, err := fmt.Sscanf(string(buf), "%d %d", &max, &period); err == nil {
 756                         return float64(max) / float64(period)
 757                 }
 758         }
 759         if buf, err := fs.ReadFile(r.FS, r.statFiles.cpusetCpus); err == nil {
 760                 // cpuset.cpus looks like "0,4-7\n" if only CPUs
 761                 // 0,4,5,6,7 are available.
 762                 cpus := 0
 763                 for _, v := range bytes.Split(buf, []byte{','}) {
 764                         var min, max int
 765                         n, _ := fmt.Sscanf(string(v), "%d-%d", &min, &max)
 766                         if n == 2 {
 767                                 cpus += (max - min) + 1
 768                         } else {
 769                                 cpus++
 770                         }
 771                 }
 772                 return float64(cpus)
 773         }
 774         if buf, err := fs.ReadFile(r.FS, "proc/cpuinfo"); err == nil {
 775                 // cpuinfo has a line like "processor\t: 0\n" for each
 776                 // CPU.
 777                 cpus := 0
 778                 for _, line := range bytes.Split(buf, []byte{'\n'}) {
 779                         if bytes.HasPrefix(line, []byte("processor\t:")) {
 780                                 cpus++
 781                         }
 782                 }
 783                 return float64(cpus)
 784         }
 785         return 0
 786 }
 787
 788 func (r *Reporter) doCPUStats() {
 789         var nextSample cpuSample
 790         if r.statFiles.cpuStat != "" {
 791                 // v2
 792                 f, err := r.FS.Open(r.statFiles.cpuStat)
 793                 if err != nil {
 794                         return
 795                 }
 796                 defer f.Close()
 797                 nextSample = cpuSample{
 798                         hasData:    true,
 799                         sampleTime: time.Now(),
 800                         cpus:       r.getCPUCount(),
 801                 }
 802                 for {
 803                         var stat string
 804                         var val int64
 805                         n, err := fmt.Fscanf(f, "%s %d\n", &stat, &val)
 806                         if err != nil || n != 2 {
 807                                 break
 808                         }
 809                         if stat == "user_usec" {
 810                                 nextSample.user = float64(val) / 1000000
 811                         } else if stat == "system_usec" {
 812                                 nextSample.sys = float64(val) / 1000000
 813                         }
 814                 }
 815         } else if r.statFiles.cpuacctStat != "" {
 816                 // v1
 817                 b, err := fs.ReadFile(r.FS, r.statFiles.cpuacctStat)
 818                 if err != nil {
 819                         return
 820                 }
 821
 822                 var userTicks, sysTicks int64
 823                 fmt.Sscanf(string(b), "user %d\nsystem %d", &userTicks, &sysTicks)
 824                 userHz := float64(100)
 825                 nextSample = cpuSample{
 826                         hasData:    true,
 827                         sampleTime: time.Now(),
 828                         user:       float64(userTicks) / userHz,
 829                         sys:        float64(sysTicks) / userHz,
 830                         cpus:       r.getCPUCount(),
 831                 }
 832         }
 833
 834         delta := ""
 835         if r.lastCPUSample.hasData {
 836                 delta = fmt.Sprintf(" -- interval %.4f seconds %.4f user %.4f sys",
 837                         nextSample.sampleTime.Sub(r.lastCPUSample.sampleTime).Seconds(),
 838                         nextSample.user-r.lastCPUSample.user,
 839                         nextSample.sys-r.lastCPUSample.sys)
 840         }
 841         r.Logger.Printf("cpu %.4f user %.4f sys %.2f cpus%s\n",
 842                 nextSample.user, nextSample.sys, nextSample.cpus, delta)
 843         r.lastCPUSample = nextSample
 844 }
 845
 846 func (r *Reporter) doAllStats() {
 847         r.reportMemSample()
 848         r.doProcmemStats()
 849         r.doCPUStats()
 850         r.doBlkIOStats()
 851         r.doNetworkStats()
 852         r.doDiskSpaceStats()
 853 }
 854
 855 // Report stats periodically until we learn (via r.done) that someone
 856 // called Stop.
 857 func (r *Reporter) run() {
 858         defer close(r.flushed)
 859
 860         r.maxMemSample = make(map[memoryKey]int64)
 861
 862         if !r.waitForPid() {
 863                 return
 864         }
 865         r.findStatFiles()
 866         close(r.ready)
 867
 868         r.lastNetSample = make(map[string]ioSample)
 869         r.lastDiskIOSample = make(map[string]ioSample)
 870
 871         if len(r.TempDir) == 0 {
 872                 // Temporary dir not provided, try to get it from the environment.
 873                 r.TempDir = os.Getenv("TMPDIR")
 874         }
 875         if len(r.TempDir) > 0 {
 876                 r.Logger.Printf("notice: monitoring temp dir %s\n", r.TempDir)
 877         }
 878
 879         r.getMemSample()
 880         r.doAllStats()
 881
 882         if r.PollPeriod < 1 {
 883                 r.PollPeriod = time.Second * 10
 884         }
 885
 886         memTicker := time.NewTicker(time.Second)
 887         mainTicker := time.NewTicker(r.PollPeriod)
 888         for {
 889                 select {
 890                 case <-r.done:
 891                         return
 892                 case <-memTicker.C:
 893                         r.getMemSample()
 894                 case <-mainTicker.C:
 895                         r.doAllStats()
 896                 }
 897         }
 898 }
 899
 900 // Wait for Pid() to return a real pid.  Return true if this succeeds
 901 // before Stop is called.
 902 func (r *Reporter) waitForPid() bool {
 903         ticker := time.NewTicker(100 * time.Millisecond)
 904         defer ticker.Stop()
 905         warningTimer := time.After(r.PollPeriod)
 906         for {
 907                 r.pid = r.Pid()
 908                 if r.pid > 0 {
 909                         break
 910                 }
 911                 select {
 912                 case <-ticker.C:
 913                 case <-warningTimer:
 914                         r.Logger.Printf("warning: Pid() did not return a process ID after %v (config error?) -- still waiting...", r.PollPeriod)
 915                 case <-r.done:
 916                         r.Logger.Printf("warning: Pid() never returned a process ID")
 917                         return false
 918                 }
 919         }
 920         return true
 921 }
 922
 923 func (r *Reporter) dumpSourceFiles(destdir string) error {
 924         select {
 925         case <-r.done:
 926                 return errors.New("reporter was never ready")
 927         case <-r.ready:
 928         }
 929         todo := []string{
 930                 fmt.Sprintf("proc/%d/cgroup", r.pid),
 931                 fmt.Sprintf("proc/%d/cpuset", r.pid),
 932                 "proc/cpuinfo",
 933                 "proc/mounts",
 934                 "proc/self/smaps",
 935                 r.statFiles.cpuMax,
 936                 r.statFiles.cpusetCpus,
 937                 r.statFiles.cpuacctStat,
 938                 r.statFiles.cpuStat,
 939                 r.statFiles.ioServiceBytes,
 940                 r.statFiles.ioStat,
 941                 r.statFiles.memoryStat,
 942                 r.statFiles.memoryCurrent,
 943                 r.statFiles.memorySwapCurrent,
 944                 r.statFiles.netDev,
 945         }
 946         for _, path := range todo {
 947                 if path == "" {
 948                         continue
 949                 }
 950                 err := r.createParentsAndCopyFile(destdir, path)
 951                 if err != nil {
 952                         return err
 953                 }
 954         }
 955         r.reportPIDsMu.Lock()
 956         r.reportPIDsMu.Unlock()
 957         for _, pid := range r.reportPIDs {
 958                 path := fmt.Sprintf("proc/%d/stat", pid)
 959                 err := r.createParentsAndCopyFile(destdir, path)
 960                 if err != nil {
 961                         return err
 962                 }
 963         }
 964         if proc, err := os.FindProcess(r.pid); err != nil || proc.Signal(syscall.Signal(0)) != nil {
 965                 return fmt.Errorf("process %d no longer exists, snapshot is probably broken", r.pid)
 966         }
 967         return nil
 968 }
 969
 970 func (r *Reporter) createParentsAndCopyFile(destdir, path string) error {
 971         buf, err := fs.ReadFile(r.FS, path)
 972         if os.IsNotExist(err) {
 973                 return nil
 974         } else if err != nil {
 975                 return err
 976         }
 977         if parent, _ := filepath.Split(path); parent != "" {
 978                 err = os.MkdirAll(destdir+"/"+parent, 0777)
 979                 if err != nil {
 980                         return fmt.Errorf("mkdir %s: %s", destdir+"/"+parent, err)
 981                 }
 982         }
 983         destfile := destdir + "/" + path
 984         r.Logger.Printf("copy %s to %s -- size %d", path, destfile, len(buf))
 985         return os.WriteFile(destfile, buf, 0777)
 986 }