lib/crunchstat/crunchstat_test.go

   1 // Copyright (C) The Arvados Authors. All rights reserved.
   2 //
   3 // SPDX-License-Identifier: AGPL-3.0
   4
   5 package crunchstat
   6
   7 import (
   8         "bytes"
   9         "errors"
  10         "fmt"
  11         "os"
  12         "path"
  13         "regexp"
  14         "strconv"
  15         "testing"
  16         "time"
  17
  18         "github.com/sirupsen/logrus"
  19         . "gopkg.in/check.v1"
  20 )
  21
  22 const logMsgPrefix = `(?m)(.*\n)*.* msg="`
  23 const GiB = int64(1024 * 1024 * 1024)
  24
  25 type fakeStat struct {
  26         cgroupRoot string
  27         statName   string
  28         unit       string
  29         value      int64
  30 }
  31
  32 var fakeRSS = fakeStat{
  33         cgroupRoot: "testdata/fakestat",
  34         statName:   "mem rss",
  35         unit:       "bytes",
  36         // Note this is the value of total_rss, not rss, because that's what should
  37         // always be reported for thresholds and maxima.
  38         value: 750 * 1024 * 1024,
  39 }
  40
  41 func Test(t *testing.T) {
  42         TestingT(t)
  43 }
  44
  45 var _ = Suite(&suite{
  46         logger: logrus.New(),
  47 })
  48
  49 type suite struct {
  50         cgroupRoot string
  51         logbuf     bytes.Buffer
  52         logger     *logrus.Logger
  53 }
  54
  55 func (s *suite) SetUpSuite(c *C) {
  56         s.logger.Out = &s.logbuf
  57 }
  58
  59 func (s *suite) SetUpTest(c *C) {
  60         s.cgroupRoot = ""
  61         s.logbuf.Reset()
  62 }
  63
  64 func (s *suite) tempCgroup(c *C, sourceDir string) error {
  65         tempDir := c.MkDir()
  66         dirents, err := os.ReadDir(sourceDir)
  67         if err != nil {
  68                 return err
  69         }
  70         for _, dirent := range dirents {
  71                 srcData, err := os.ReadFile(path.Join(sourceDir, dirent.Name()))
  72                 if err != nil {
  73                         return err
  74                 }
  75                 destPath := path.Join(tempDir, dirent.Name())
  76                 err = os.WriteFile(destPath, srcData, 0o600)
  77                 if err != nil {
  78                         return err
  79                 }
  80         }
  81         s.cgroupRoot = tempDir
  82         return nil
  83 }
  84
  85 func (s *suite) addPidToCgroup(pid int) error {
  86         if s.cgroupRoot == "" {
  87                 return errors.New("cgroup has not been set up for this test")
  88         }
  89         procsPath := path.Join(s.cgroupRoot, "cgroup.procs")
  90         procsFile, err := os.OpenFile(procsPath, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o600)
  91         if err != nil {
  92                 return err
  93         }
  94         pidLine := strconv.Itoa(pid) + "\n"
  95         _, err = procsFile.Write([]byte(pidLine))
  96         if err != nil {
  97                 procsFile.Close()
  98                 return err
  99         }
 100         return procsFile.Close()
 101 }
 102
 103 func (s *suite) TestReadAllOrWarnFail(c *C) {
 104         rep := Reporter{Logger: s.logger}
 105
 106         // The special file /proc/self/mem can be opened for
 107         // reading, but reading from byte 0 returns an error.
 108         f, err := os.Open("/proc/self/mem")
 109         c.Assert(err, IsNil)
 110         defer f.Close()
 111         _, err = rep.readAllOrWarn(f)
 112         c.Check(err, NotNil)
 113         c.Check(s.logbuf.String(), Matches, ".* msg=\"warning: read /proc/self/mem: .*\n")
 114 }
 115
 116 func (s *suite) TestReadAllOrWarnSuccess(c *C) {
 117         rep := Reporter{Logger: s.logger}
 118
 119         f, err := os.Open("./crunchstat_test.go")
 120         c.Assert(err, IsNil)
 121         defer f.Close()
 122         data, err := rep.readAllOrWarn(f)
 123         c.Check(err, IsNil)
 124         c.Check(string(data), Matches, "(?ms).*\npackage crunchstat\n.*")
 125         c.Check(s.logbuf.String(), Equals, "")
 126 }
 127
 128 func (s *suite) TestReportPIDs(c *C) {
 129         r := Reporter{
 130                 Logger:     s.logger,
 131                 CgroupRoot: "/sys/fs/cgroup",
 132                 PollPeriod: time.Second,
 133         }
 134         r.Start()
 135         r.ReportPID("init", 1)
 136         r.ReportPID("test_process", os.Getpid())
 137         r.ReportPID("nonexistent", 12345) // should be silently ignored/omitted
 138         for deadline := time.Now().Add(10 * time.Second); ; time.Sleep(time.Millisecond) {
 139                 if time.Now().After(deadline) {
 140                         c.Error("timed out")
 141                         break
 142                 }
 143                 if m := regexp.MustCompile(`(?ms).*procmem \d+ init (\d+) test_process.*`).FindSubmatch(s.logbuf.Bytes()); len(m) > 0 {
 144                         size, err := strconv.ParseInt(string(m[1]), 10, 64)
 145                         c.Check(err, IsNil)
 146                         // Expect >1 MiB and <100 MiB -- otherwise we
 147                         // are probably misinterpreting /proc/N/stat
 148                         // or multiplying by the wrong page size.
 149                         c.Check(size > 1000000, Equals, true)
 150                         c.Check(size < 100000000, Equals, true)
 151                         break
 152                 }
 153         }
 154         c.Logf("%s", s.logbuf.String())
 155 }
 156
 157 func (s *suite) testRSSThresholds(c *C, rssPercentages []int64, alertCount int) {
 158         c.Assert(alertCount <= len(rssPercentages), Equals, true)
 159         rep := Reporter{
 160                 CgroupRoot: fakeRSS.cgroupRoot,
 161                 Logger:     s.logger,
 162                 MemThresholds: map[string][]Threshold{
 163                         "rss": NewThresholdsFromPercentages(GiB, rssPercentages),
 164                 },
 165                 PollPeriod:      time.Second * 10,
 166                 ThresholdLogger: s.logger,
 167         }
 168         rep.Start()
 169         rep.Stop()
 170         logs := s.logbuf.String()
 171         c.Logf("%s", logs)
 172
 173         for index, expectPercentage := range rssPercentages[:alertCount] {
 174                 var logCheck Checker
 175                 if index < alertCount {
 176                         logCheck = Matches
 177                 } else {
 178                         logCheck = Not(Matches)
 179                 }
 180                 pattern := fmt.Sprintf(`%sContainer using over %d%% of memory \(rss %d/%d bytes\)"`,
 181                         logMsgPrefix, expectPercentage, fakeRSS.value, GiB)
 182                 c.Check(logs, logCheck, pattern)
 183         }
 184 }
 185
 186 func (s *suite) TestZeroRSSThresholds(c *C) {
 187         s.testRSSThresholds(c, []int64{}, 0)
 188 }
 189
 190 func (s *suite) TestOneRSSThresholdPassed(c *C) {
 191         s.testRSSThresholds(c, []int64{55}, 1)
 192 }
 193
 194 func (s *suite) TestOneRSSThresholdNotPassed(c *C) {
 195         s.testRSSThresholds(c, []int64{85}, 0)
 196 }
 197
 198 func (s *suite) TestMultipleRSSThresholdsNonePassed(c *C) {
 199         s.testRSSThresholds(c, []int64{95, 97, 99}, 0)
 200 }
 201
 202 func (s *suite) TestMultipleRSSThresholdsSomePassed(c *C) {
 203         s.testRSSThresholds(c, []int64{60, 70, 80, 90}, 2)
 204 }
 205
 206 func (s *suite) TestMultipleRSSThresholdsAllPassed(c *C) {
 207         s.testRSSThresholds(c, []int64{1, 2, 3}, 3)
 208 }
 209
 210 func (s *suite) TestLogMaxima(c *C) {
 211         err := s.tempCgroup(c, fakeRSS.cgroupRoot)
 212         c.Assert(err, IsNil)
 213         rep := Reporter{
 214                 CgroupRoot: s.cgroupRoot,
 215                 Logger:     s.logger,
 216                 PollPeriod: time.Second * 10,
 217                 TempDir:    s.cgroupRoot,
 218         }
 219         rep.Start()
 220         rep.Stop()
 221         rep.LogMaxima(s.logger, map[string]int64{"rss": GiB})
 222         logs := s.logbuf.String()
 223         c.Logf("%s", logs)
 224
 225         expectRSS := fmt.Sprintf(`Maximum container memory rss usage was %d%%, %d/%d bytes`,
 226                 100*fakeRSS.value/GiB, fakeRSS.value, GiB)
 227         for _, expected := range []string{
 228                 `Maximum disk usage was \d+%, \d+/\d+ bytes`,
 229                 `Maximum container memory cache usage was 73400320 bytes`,
 230                 `Maximum container memory swap usage was 320 bytes`,
 231                 `Maximum container memory pgmajfault usage was 20 faults`,
 232                 expectRSS,
 233         } {
 234                 pattern := logMsgPrefix + expected + `"`
 235                 c.Check(logs, Matches, pattern)
 236         }
 237 }
 238
 239 func (s *suite) TestLogProcessMemMax(c *C) {
 240         err := s.tempCgroup(c, fakeRSS.cgroupRoot)
 241         c.Assert(err, IsNil)
 242         pid := os.Getpid()
 243         err = s.addPidToCgroup(pid)
 244         c.Assert(err, IsNil)
 245
 246         rep := Reporter{
 247                 CgroupRoot: s.cgroupRoot,
 248                 Logger:     s.logger,
 249                 PollPeriod: time.Second * 10,
 250                 TempDir:    s.cgroupRoot,
 251         }
 252         rep.ReportPID("test-run", pid)
 253         rep.Start()
 254         rep.Stop()
 255         rep.LogProcessMemMax(s.logger)
 256         logs := s.logbuf.String()
 257         c.Logf("%s", logs)
 258
 259         pattern := logMsgPrefix + `Maximum test-run memory rss usage was \d+ bytes"`
 260         c.Check(logs, Matches, pattern)
 261 }