lib/crunchrun/logging_test.go

   1 // Copyright (C) The Arvados Authors. All rights reserved.
   2 //
   3 // SPDX-License-Identifier: AGPL-3.0
   4
   5 package crunchrun
   6
   7 import (
   8         "bytes"
   9         "fmt"
  10         "io"
  11         "strings"
  12         "testing"
  13         "time"
  14
  15         "git.arvados.org/arvados.git/sdk/go/arvados"
  16         "git.arvados.org/arvados.git/sdk/go/arvadosclient"
  17         . "gopkg.in/check.v1"
  18         check "gopkg.in/check.v1"
  19 )
  20
  21 type LoggingTestSuite struct {
  22         client *arvados.Client
  23 }
  24
  25 type TestTimestamper struct {
  26         count int
  27 }
  28
  29 func (stamper *TestTimestamper) Timestamp(t time.Time) string {
  30         stamper.count++
  31         t, err := time.ParseInLocation(time.RFC3339Nano, fmt.Sprintf("2015-12-29T15:51:45.%09dZ", stamper.count), t.Location())
  32         if err != nil {
  33                 panic(err)
  34         }
  35         return RFC3339Timestamp(t)
  36 }
  37
  38 // Gocheck boilerplate
  39 var _ = Suite(&LoggingTestSuite{})
  40
  41 func (s *LoggingTestSuite) SetUpTest(c *C) {
  42         s.client = arvados.NewClientFromEnv()
  43         crunchLogUpdatePeriod = time.Hour * 24 * 365
  44         crunchLogUpdateSize = 1 << 50
  45 }
  46
  47 func (s *LoggingTestSuite) TestWriteLogs(c *C) {
  48         api := &ArvTestClient{}
  49         kc := &KeepTestClient{}
  50         defer kc.Close()
  51         cr, err := NewContainerRunner(s.client, api, kc, "zzzzz-zzzzzzzzzzzzzzz")
  52         c.Assert(err, IsNil)
  53         cr.CrunchLog.Timestamper = (&TestTimestamper{}).Timestamp
  54
  55         cr.CrunchLog.Print("Hello world!")
  56         cr.CrunchLog.Print("Goodbye")
  57         cr.CrunchLog.Close()
  58
  59         c.Check(api.Calls, Equals, 1)
  60
  61         mt, err := cr.LogCollection.MarshalManifest(".")
  62         c.Check(err, IsNil)
  63         c.Check(mt, Equals, ". 74561df9ae65ee9f35d5661d42454264+83 0:83:crunch-run.txt\n")
  64
  65         logtext := "2015-12-29T15:51:45.000000001Z Hello world!\n" +
  66                 "2015-12-29T15:51:45.000000002Z Goodbye\n"
  67
  68         c.Check(api.Content[0]["log"].(arvadosclient.Dict)["event_type"], Equals, "crunch-run")
  69         c.Check(api.Content[0]["log"].(arvadosclient.Dict)["properties"].(map[string]string)["text"], Equals, logtext)
  70         c.Check(string(kc.Content), Equals, logtext)
  71 }
  72
  73 func (s *LoggingTestSuite) TestWriteLogsLarge(c *C) {
  74         if testing.Short() {
  75                 return
  76         }
  77         api := &ArvTestClient{}
  78         kc := &KeepTestClient{}
  79         defer kc.Close()
  80         cr, err := NewContainerRunner(s.client, api, kc, "zzzzz-zzzzzzzzzzzzzzz")
  81         c.Assert(err, IsNil)
  82         cr.CrunchLog.Timestamper = (&TestTimestamper{}).Timestamp
  83         cr.CrunchLog.Immediate = nil
  84
  85         for i := 0; i < 2000000; i++ {
  86                 cr.CrunchLog.Printf("Hello %d", i)
  87         }
  88         cr.CrunchLog.Print("Goodbye")
  89         cr.CrunchLog.Close()
  90
  91         c.Check(api.Calls > 0, Equals, true)
  92         c.Check(api.Calls < 2000000, Equals, true)
  93
  94         mt, err := cr.LogCollection.MarshalManifest(".")
  95         c.Check(err, IsNil)
  96         c.Check(mt, Equals, ". 9c2c05d1fae6aaa8af85113ba725716d+67108864 80b821383a07266c2a66a4566835e26e+21780065 0:88888929:crunch-run.txt\n")
  97 }
  98
  99 func (s *LoggingTestSuite) TestWriteMultipleLogs(c *C) {
 100         api := &ArvTestClient{}
 101         kc := &KeepTestClient{}
 102         defer kc.Close()
 103         cr, err := NewContainerRunner(s.client, api, kc, "zzzzz-zzzzzzzzzzzzzzz")
 104         c.Assert(err, IsNil)
 105         ts := &TestTimestamper{}
 106         cr.CrunchLog.Timestamper = ts.Timestamp
 107         w, err := cr.NewLogWriter("stdout")
 108         c.Assert(err, IsNil)
 109         stdout := NewThrottledLogger(w)
 110         stdout.Timestamper = ts.Timestamp
 111
 112         cr.CrunchLog.Print("Hello world!")
 113         stdout.Print("Doing stuff")
 114         cr.CrunchLog.Print("Goodbye")
 115         stdout.Print("Blurb")
 116         cr.CrunchLog.Close()
 117         stdout.Close()
 118
 119         logText := make(map[string]string)
 120         for _, content := range api.Content {
 121                 log := content["log"].(arvadosclient.Dict)
 122                 logText[log["event_type"].(string)] += log["properties"].(map[string]string)["text"]
 123         }
 124
 125         c.Check(logText["crunch-run"], Equals, `2015-12-29T15:51:45.000000001Z Hello world!
 126 2015-12-29T15:51:45.000000003Z Goodbye
 127 `)
 128         c.Check(logText["stdout"], Equals, `2015-12-29T15:51:45.000000002Z Doing stuff
 129 2015-12-29T15:51:45.000000004Z Blurb
 130 `)
 131
 132         mt, err := cr.LogCollection.MarshalManifest(".")
 133         c.Check(err, IsNil)
 134         c.Check(mt, Equals, ". 48f9023dc683a850b1c9b482b14c4b97+163 0:83:crunch-run.txt 83:80:stdout.txt\n")
 135 }
 136
 137 func (s *LoggingTestSuite) TestLogUpdate(c *C) {
 138         for _, trial := range []struct {
 139                 maxBytes    int64
 140                 maxDuration time.Duration
 141         }{
 142                 {1000, 10 * time.Second},
 143                 {1000000, time.Millisecond},
 144         } {
 145                 c.Logf("max %d bytes, %s", trial.maxBytes, trial.maxDuration)
 146                 crunchLogUpdateSize = trial.maxBytes
 147                 crunchLogUpdatePeriod = trial.maxDuration
 148
 149                 api := &ArvTestClient{}
 150                 kc := &KeepTestClient{}
 151                 defer kc.Close()
 152                 cr, err := NewContainerRunner(s.client, api, kc, "zzzzz-zzzzzzzzzzzzzzz")
 153                 c.Assert(err, IsNil)
 154                 ts := &TestTimestamper{}
 155                 cr.CrunchLog.Timestamper = ts.Timestamp
 156                 w, err := cr.NewLogWriter("stdout")
 157                 c.Assert(err, IsNil)
 158                 stdout := NewThrottledLogger(w)
 159                 stdout.Timestamper = ts.Timestamp
 160
 161                 c.Check(cr.logUUID, Equals, "")
 162                 cr.CrunchLog.Printf("Hello %1000s", "space")
 163                 for i, t := 0, time.NewTicker(time.Millisecond); i < 5000 && cr.logUUID == ""; i++ {
 164                         <-t.C
 165                 }
 166                 c.Check(cr.logUUID, Not(Equals), "")
 167                 cr.CrunchLog.Print("Goodbye")
 168                 fmt.Fprint(stdout, "Goodbye\n")
 169                 cr.CrunchLog.Close()
 170                 stdout.Close()
 171                 w.Close()
 172
 173                 mt, err := cr.LogCollection.MarshalManifest(".")
 174                 c.Check(err, IsNil)
 175                 // Block packing depends on whether there's an update
 176                 // between the two Goodbyes -- either way the first
 177                 // block will be 4dc76.
 178                 c.Check(mt, Matches, `. 4dc76e0a212bfa30c39d76d8c16da0c0\+1038 (afc503bc1b9a828b4bb543cb629e936c\+78|90699dc22545cd74a0664303f70bc05a\+39 276b49339fd5203d15a93ff3de11bfb9\+39) 0:1077:crunch-run.txt 1077:39:stdout.txt\n`)
 179         }
 180 }
 181
 182 func (s *LoggingTestSuite) TestWriteLogsWithRateLimitThrottleBytes(c *C) {
 183         s.testWriteLogsWithRateLimit(c, "crunchLogThrottleBytes", 50, 65536, "Exceeded rate 50 bytes per 60 seconds")
 184 }
 185
 186 func (s *LoggingTestSuite) TestWriteLogsWithRateLimitThrottleLines(c *C) {
 187         s.testWriteLogsWithRateLimit(c, "crunchLogThrottleLines", 1, 1024, "Exceeded rate 1 lines per 60 seconds")
 188 }
 189
 190 func (s *LoggingTestSuite) TestWriteLogsWithRateLimitThrottleBytesPerEvent(c *C) {
 191         s.testWriteLogsWithRateLimit(c, "crunchLimitLogBytesPerJob", 50, 67108864, "Exceeded log limit 50 bytes (crunch_limit_log_bytes_per_job)")
 192 }
 193
 194 func (s *LoggingTestSuite) TestWriteLogsWithZeroBytesPerJob(c *C) {
 195         s.testWriteLogsWithRateLimit(c, "crunchLimitLogBytesPerJob", 0, 67108864, "Exceeded log limit 0 bytes (crunch_limit_log_bytes_per_job)")
 196 }
 197
 198 func (s *LoggingTestSuite) testWriteLogsWithRateLimit(c *C, throttleParam string, throttleValue int, throttleDefault int, expected string) {
 199         discoveryMap[throttleParam] = float64(throttleValue)
 200         defer func() {
 201                 discoveryMap[throttleParam] = float64(throttleDefault)
 202         }()
 203
 204         api := &ArvTestClient{}
 205         kc := &KeepTestClient{}
 206         defer kc.Close()
 207         cr, err := NewContainerRunner(s.client, api, kc, "zzzzz-zzzzzzzzzzzzzzz")
 208         c.Assert(err, IsNil)
 209         cr.CrunchLog.Timestamper = (&TestTimestamper{}).Timestamp
 210
 211         cr.CrunchLog.Print("Hello world!")
 212         cr.CrunchLog.Print("Goodbye")
 213         cr.CrunchLog.Close()
 214
 215         c.Check(api.Calls, Equals, 1)
 216
 217         mt, err := cr.LogCollection.MarshalManifest(".")
 218         c.Check(err, IsNil)
 219         c.Check(mt, Equals, ". 74561df9ae65ee9f35d5661d42454264+83 0:83:crunch-run.txt\n")
 220
 221         logtext := "2015-12-29T15:51:45.000000001Z Hello world!\n" +
 222                 "2015-12-29T15:51:45.000000002Z Goodbye\n"
 223
 224         c.Check(api.Content[0]["log"].(arvadosclient.Dict)["event_type"], Equals, "crunch-run")
 225         stderrLog := api.Content[0]["log"].(arvadosclient.Dict)["properties"].(map[string]string)["text"]
 226         c.Check(true, Equals, strings.Contains(stderrLog, expected))
 227         c.Check(string(kc.Content), Equals, logtext)
 228 }
 229
 230 type filterSuite struct{}
 231
 232 var _ = Suite(&filterSuite{})
 233
 234 func (*filterSuite) TestFilterKeepstoreErrorsOnly(c *check.C) {
 235         var buf bytes.Buffer
 236         f := filterKeepstoreErrorsOnly{WriteCloser: nopCloser{&buf}}
 237         for _, s := range []string{
 238                 "not j",
 239                 "son\n" + `{"msg":"foo"}` + "\n{}\n" + `{"msg":"request"}` + "\n" + `{"msg":1234}` + "\n\n",
 240                 "\n[\n",
 241                 `{"msg":"response","respStatusCode":404,"foo": "bar"}` + "\n",
 242                 `{"msg":"response","respStatusCode":206}` + "\n",
 243         } {
 244                 f.Write([]byte(s))
 245         }
 246         c.Check(buf.String(), check.Equals, `not json
 247 {"msg":"foo"}
 248 {}
 249 {"msg":1234}
 250 [
 251 {"msg":"response","respStatusCode":404,"foo": "bar"}
 252 `)
 253 }
 254
 255 type nopCloser struct {
 256         io.Writer
 257 }
 258
 259 func (nopCloser) Close() error { return nil }