1 // Copyright (C) The Arvados Authors. All rights reserved.
3 // SPDX-License-Identifier: AGPL-3.0
5 // Package crunchstat reports resource usage (CPU, memory, disk,
6 // network) for a cgroup.
26 type logPrinter interface {
27 Printf(fmt string, args ...interface{})
30 // A Reporter gathers statistics for a cgroup and writes them to a
32 type Reporter struct {
33 // CID of the container to monitor. If empty, read the CID
34 // from CIDFile (first waiting until a non-empty file appears
35 // at CIDFile). If CIDFile is also empty, report host
39 // Path to a file we can read CID from.
42 // Where cgroup accounting files live on this system, e.g.,
46 // Parent cgroup, e.g., "docker".
49 // Interval between samples. Must be positive.
50 PollPeriod time.Duration
52 // Temporary directory, will be monitored for available, used & total space.
55 // Where to write statistics. Must not be nil.
58 // When stats cross thresholds configured in the fields below,
59 // they are reported to this logger.
60 ThresholdLogger logPrinter
62 // MemThresholds maps memory stat names to slices of thresholds.
63 // When the corresponding stat exceeds a threshold, that will be logged.
64 MemThresholds map[string][]Threshold
67 reportedStatFile map[string]string
68 lastNetSample map[string]ioSample
69 lastDiskIOSample map[string]ioSample
70 lastCPUSample cpuSample
71 lastDiskSpaceSample diskSpaceSample
72 lastMemSample memSample
74 reportPIDs map[string]int
75 reportPIDsMu sync.Mutex
77 done chan struct{} // closed when we should stop reporting
78 flushed chan struct{} // closed when we have made our last report
81 type Threshold struct {
87 func NewThresholdFromPercentage(total int64, percentage int64) Threshold {
89 percentage: percentage,
90 threshold: total * percentage / 100,
95 func NewThresholdsFromPercentages(total int64, percentages []int64) (thresholds []Threshold) {
96 for _, percentage := range percentages {
97 thresholds = append(thresholds, NewThresholdFromPercentage(total, percentage))
102 // Start starts monitoring in a new goroutine, and returns
105 // The monitoring goroutine waits for a non-empty CIDFile to appear
106 // (unless CID is non-empty). Then it waits for the accounting files
107 // to appear for the monitored container. Then it collects and reports
108 // statistics until Stop is called.
110 // Callers should not call Start more than once.
112 // Callers should not modify public data fields after calling Start.
113 func (r *Reporter) Start() {
114 r.done = make(chan struct{})
115 r.flushed = make(chan struct{})
119 // ReportPID starts reporting stats for a specified process.
120 func (r *Reporter) ReportPID(name string, pid int) {
121 r.reportPIDsMu.Lock()
122 defer r.reportPIDsMu.Unlock()
123 if r.reportPIDs == nil {
124 r.reportPIDs = map[string]int{name: pid}
126 r.reportPIDs[name] = pid
130 // Stop reporting. Do not call more than once, or before calling
133 // Nothing will be logged after Stop returns.
134 func (r *Reporter) Stop() {
139 func (r *Reporter) readAllOrWarn(in io.Reader) ([]byte, error) {
140 content, err := ioutil.ReadAll(in)
142 r.Logger.Printf("warning: %v", err)
147 // Open the cgroup stats file in /sys/fs corresponding to the target
148 // cgroup, and return an io.ReadCloser. If no stats file is available,
151 // Log the file that was opened, if it isn't the same file opened on
152 // the last openStatFile for this stat.
154 // Log "not available" if no file is found and either this stat has
155 // been available in the past, or verbose==true.
157 // TODO: Instead of trying all options, choose a process in the
158 // container, and read /proc/PID/cgroup to determine the appropriate
159 // cgroup root for the given statgroup. (This will avoid falling back
160 // to host-level stats during container setup and teardown.)
161 func (r *Reporter) openStatFile(statgroup, stat string, verbose bool) (io.ReadCloser, error) {
164 // Collect container's stats
166 fmt.Sprintf("%s/%s/%s/%s/%s", r.CgroupRoot, statgroup, r.CgroupParent, r.CID, stat),
167 fmt.Sprintf("%s/%s/%s/%s", r.CgroupRoot, r.CgroupParent, r.CID, stat),
170 // Collect this host's stats
172 fmt.Sprintf("%s/%s/%s", r.CgroupRoot, statgroup, stat),
173 fmt.Sprintf("%s/%s", r.CgroupRoot, stat),
179 for _, path = range paths {
180 file, err = os.Open(path)
187 if pathWas := r.reportedStatFile[stat]; pathWas != path {
188 // Log whenever we start using a new/different cgroup
189 // stat file for a given statistic. This typically
190 // happens 1 to 3 times per statistic, depending on
191 // whether we happen to collect stats [a] before any
192 // processes have been created in the container and
193 // [b] after all contained processes have exited.
194 if path == "" && verbose {
195 r.Logger.Printf("notice: stats not available: stat %s, statgroup %s, cid %s, parent %s, root %s\n", stat, statgroup, r.CID, r.CgroupParent, r.CgroupRoot)
196 } else if pathWas != "" {
197 r.Logger.Printf("notice: stats moved from %s to %s\n", r.reportedStatFile[stat], path)
199 r.Logger.Printf("notice: reading stats from %s\n", path)
201 r.reportedStatFile[stat] = path
206 func (r *Reporter) getContainerNetStats() (io.Reader, error) {
207 procsFile, err := r.openStatFile("cpuacct", "cgroup.procs", true)
211 defer procsFile.Close()
212 reader := bufio.NewScanner(procsFile)
214 taskPid := reader.Text()
215 statsFilename := fmt.Sprintf("/proc/%s/net/dev", taskPid)
216 stats, err := ioutil.ReadFile(statsFilename)
218 r.Logger.Printf("notice: %v", err)
221 return strings.NewReader(string(stats)), nil
223 return nil, errors.New("Could not read stats for any proc in container")
226 type ioSample struct {
232 func (r *Reporter) doBlkIOStats() {
233 c, err := r.openStatFile("blkio", "blkio.io_service_bytes", true)
238 b := bufio.NewScanner(c)
239 var sampleTime = time.Now()
240 newSamples := make(map[string]ioSample)
242 var device, op string
244 if _, err := fmt.Sscanf(string(b.Text()), "%s %s %d", &device, &op, &val); err != nil {
247 var thisSample ioSample
249 if thisSample, ok = newSamples[device]; !ok {
250 thisSample = ioSample{sampleTime, -1, -1}
254 thisSample.rxBytes = val
256 thisSample.txBytes = val
258 newSamples[device] = thisSample
260 for dev, sample := range newSamples {
261 if sample.txBytes < 0 || sample.rxBytes < 0 {
265 if prev, ok := r.lastDiskIOSample[dev]; ok {
266 delta = fmt.Sprintf(" -- interval %.4f seconds %d write %d read",
267 sample.sampleTime.Sub(prev.sampleTime).Seconds(),
268 sample.txBytes-prev.txBytes,
269 sample.rxBytes-prev.rxBytes)
271 r.Logger.Printf("blkio:%s %d write %d read%s\n", dev, sample.txBytes, sample.rxBytes, delta)
272 r.lastDiskIOSample[dev] = sample
276 type memSample struct {
278 memStat map[string]int64
281 func (r *Reporter) getMemSample() {
282 c, err := r.openStatFile("memory", "memory.stat", true)
287 b := bufio.NewScanner(c)
288 thisSample := memSample{time.Now(), make(map[string]int64)}
292 if _, err := fmt.Sscanf(string(b.Text()), "%s %d", &stat, &val); err != nil {
295 thisSample.memStat[stat] = val
297 r.lastMemSample = thisSample
299 if r.ThresholdLogger != nil {
300 for statName, thresholds := range r.MemThresholds {
301 statValue, ok := thisSample.memStat["total_"+statName]
303 statValue, ok = thisSample.memStat[statName]
309 var statThreshold Threshold
310 for index, statThreshold = range thresholds {
311 if statValue < statThreshold.threshold {
313 } else if statThreshold.percentage > 0 {
314 r.ThresholdLogger.Printf("Container using over %d%% of memory (%s %d/%d bytes)",
315 statThreshold.percentage, statName, statValue, statThreshold.total)
317 r.ThresholdLogger.Printf("Container using over %d of memory (%s %s bytes)",
318 statThreshold.threshold, statName, statValue)
321 r.MemThresholds[statName] = thresholds[index:]
326 func (r *Reporter) reportMemSample() {
327 var outstat bytes.Buffer
328 wantStats := [...]string{"cache", "swap", "pgmajfault", "rss"}
329 for _, key := range wantStats {
330 // Use "total_X" stats (entire hierarchy) if enabled,
331 // otherwise just the single cgroup -- see
332 // https://www.kernel.org/doc/Documentation/cgroup-v1/memory.txt
333 if val, ok := r.lastMemSample.memStat["total_"+key]; ok {
334 fmt.Fprintf(&outstat, " %d %s", val, key)
335 } else if val, ok := r.lastMemSample.memStat[key]; ok {
336 fmt.Fprintf(&outstat, " %d %s", val, key)
339 r.Logger.Printf("mem%s\n", outstat.String())
342 func (r *Reporter) doProcmemStats() {
343 if r.kernelPageSize == 0 {
344 // assign "don't try again" value in case we give up
345 // and return without assigning the real value
346 r.kernelPageSize = -1
347 buf, err := os.ReadFile("/proc/self/smaps")
349 r.Logger.Printf("error reading /proc/self/smaps: %s", err)
352 m := regexp.MustCompile(`\nKernelPageSize:\s*(\d+) kB\n`).FindSubmatch(buf)
354 r.Logger.Printf("error parsing /proc/self/smaps: KernelPageSize not found")
357 size, err := strconv.ParseInt(string(m[1]), 10, 64)
359 r.Logger.Printf("error parsing /proc/self/smaps: KernelPageSize %q: %s", m[1], err)
362 r.kernelPageSize = size * 1024
363 } else if r.kernelPageSize < 0 {
364 // already failed to determine page size, don't keep
369 r.reportPIDsMu.Lock()
370 defer r.reportPIDsMu.Unlock()
371 procnames := make([]string, 0, len(r.reportPIDs))
372 for name := range r.reportPIDs {
373 procnames = append(procnames, name)
375 sort.Strings(procnames)
377 for _, procname := range procnames {
378 pid := r.reportPIDs[procname]
379 buf, err := os.ReadFile(fmt.Sprintf("/proc/%d/stat", pid))
383 // If the executable name contains a ')' char,
384 // /proc/$pid/stat will look like '1234 (exec name)) S
385 // 123 ...' -- the last ')' is the end of the 2nd
387 paren := bytes.LastIndexByte(buf, ')')
391 fields := bytes.SplitN(buf[paren:], []byte{' '}, 24)
392 if len(fields) < 24 {
395 // rss is the 24th field in .../stat, and fields[0]
396 // here is the last char ')' of the 2nd field, so
398 rss, err := strconv.ParseInt(string(fields[22]), 10, 64)
402 procmem += fmt.Sprintf(" %d %s", rss*r.kernelPageSize, procname)
405 r.Logger.Printf("procmem%s\n", procmem)
409 func (r *Reporter) doNetworkStats() {
410 sampleTime := time.Now()
411 stats, err := r.getContainerNetStats()
416 scanner := bufio.NewScanner(stats)
420 words := strings.Fields(scanner.Text())
421 if len(words) != 17 {
422 // Skip lines with wrong format
425 ifName = strings.TrimRight(words[0], ":")
426 if ifName == "lo" || ifName == "" {
427 // Skip loopback interface and lines with wrong format
430 if tx, err = strconv.ParseInt(words[9], 10, 64); err != nil {
433 if rx, err = strconv.ParseInt(words[1], 10, 64); err != nil {
436 nextSample := ioSample{}
437 nextSample.sampleTime = sampleTime
438 nextSample.txBytes = tx
439 nextSample.rxBytes = rx
441 if prev, ok := r.lastNetSample[ifName]; ok {
442 interval := nextSample.sampleTime.Sub(prev.sampleTime).Seconds()
443 delta = fmt.Sprintf(" -- interval %.4f seconds %d tx %d rx",
448 r.Logger.Printf("net:%s %d tx %d rx%s\n", ifName, tx, rx, delta)
449 r.lastNetSample[ifName] = nextSample
453 type diskSpaceSample struct {
461 func (r *Reporter) doDiskSpaceStats() {
462 s := syscall.Statfs_t{}
463 err := syscall.Statfs(r.TempDir, &s)
467 bs := uint64(s.Bsize)
468 nextSample := diskSpaceSample{
470 sampleTime: time.Now(),
471 total: s.Blocks * bs,
472 used: (s.Blocks - s.Bfree) * bs,
473 available: s.Bavail * bs,
477 if r.lastDiskSpaceSample.hasData {
478 prev := r.lastDiskSpaceSample
479 interval := nextSample.sampleTime.Sub(prev.sampleTime).Seconds()
480 delta = fmt.Sprintf(" -- interval %.4f seconds %d used",
482 int64(nextSample.used-prev.used))
484 r.Logger.Printf("statfs %d available %d used %d total%s\n",
485 nextSample.available, nextSample.used, nextSample.total, delta)
486 r.lastDiskSpaceSample = nextSample
489 type cpuSample struct {
490 hasData bool // to distinguish the zero value from real data
497 // Return the number of CPUs available in the container. Return 0 if
498 // we can't figure out the real number of CPUs.
499 func (r *Reporter) getCPUCount() int64 {
500 cpusetFile, err := r.openStatFile("cpuset", "cpuset.cpus", true)
504 defer cpusetFile.Close()
505 b, err := r.readAllOrWarn(cpusetFile)
509 sp := strings.Split(string(b), ",")
511 for _, v := range sp {
513 n, _ := fmt.Sscanf(v, "%d-%d", &min, &max)
515 cpus += (max - min) + 1
523 func (r *Reporter) doCPUStats() {
524 statFile, err := r.openStatFile("cpuacct", "cpuacct.stat", true)
528 defer statFile.Close()
529 b, err := r.readAllOrWarn(statFile)
534 var userTicks, sysTicks int64
535 fmt.Sscanf(string(b), "user %d\nsystem %d", &userTicks, &sysTicks)
536 userHz := float64(100)
537 nextSample := cpuSample{
539 sampleTime: time.Now(),
540 user: float64(userTicks) / userHz,
541 sys: float64(sysTicks) / userHz,
542 cpus: r.getCPUCount(),
546 if r.lastCPUSample.hasData {
547 delta = fmt.Sprintf(" -- interval %.4f seconds %.4f user %.4f sys",
548 nextSample.sampleTime.Sub(r.lastCPUSample.sampleTime).Seconds(),
549 nextSample.user-r.lastCPUSample.user,
550 nextSample.sys-r.lastCPUSample.sys)
552 r.Logger.Printf("cpu %.4f user %.4f sys %d cpus%s\n",
553 nextSample.user, nextSample.sys, nextSample.cpus, delta)
554 r.lastCPUSample = nextSample
557 func (r *Reporter) doAllStats() {
566 // Report stats periodically until we learn (via r.done) that someone
568 func (r *Reporter) run() {
569 defer close(r.flushed)
571 r.reportedStatFile = make(map[string]string)
573 if !r.waitForCIDFile() || !r.waitForCgroup() {
577 r.lastNetSample = make(map[string]ioSample)
578 r.lastDiskIOSample = make(map[string]ioSample)
580 if len(r.TempDir) == 0 {
581 // Temporary dir not provided, try to get it from the environment.
582 r.TempDir = os.Getenv("TMPDIR")
584 if len(r.TempDir) > 0 {
585 r.Logger.Printf("notice: monitoring temp dir %s\n", r.TempDir)
591 memTicker := time.NewTicker(time.Second)
592 mainTicker := time.NewTicker(r.PollPeriod)
605 // If CID is empty, wait for it to appear in CIDFile. Return true if
606 // we get it before we learn (via r.done) that someone called Stop.
607 func (r *Reporter) waitForCIDFile() bool {
608 if r.CID != "" || r.CIDFile == "" {
612 ticker := time.NewTicker(100 * time.Millisecond)
615 cid, err := ioutil.ReadFile(r.CIDFile)
616 if err == nil && len(cid) > 0 {
623 r.Logger.Printf("warning: CID never appeared in %+q: %v", r.CIDFile, err)
629 // Wait for the cgroup stats files to appear in cgroup_root. Return
630 // true if they appear before r.done indicates someone called Stop. If
631 // they don't appear within one poll interval, log a warning and keep
633 func (r *Reporter) waitForCgroup() bool {
634 ticker := time.NewTicker(100 * time.Millisecond)
636 warningTimer := time.After(r.PollPeriod)
638 c, err := r.openStatFile("cpuacct", "cgroup.procs", false)
646 r.Logger.Printf("warning: cgroup stats files have not appeared after %v (config error?) -- still waiting...", r.PollPeriod)
648 r.Logger.Printf("warning: cgroup stats files never appeared for %v", r.CID)