services/crunch-run/logging.go

   1 // Copyright (C) The Arvados Authors. All rights reserved.
   2 //
   3 // SPDX-License-Identifier: AGPL-3.0
   4
   5 package main
   6
   7 import (
   8         "bufio"
   9         "bytes"
  10         "fmt"
  11         "io"
  12         "log"
  13         "regexp"
  14         "strings"
  15         "sync"
  16         "time"
  17
  18         "git.curoverse.com/arvados.git/sdk/go/arvadosclient"
  19 )
  20
  21 // Timestamper is the signature for a function that takes a timestamp and
  22 // return a formated string value.
  23 type Timestamper func(t time.Time) string
  24
  25 // Logging plumbing:
  26 //
  27 // ThrottledLogger.Logger -> ThrottledLogger.Write ->
  28 // ThrottledLogger.buf -> ThrottledLogger.flusher ->
  29 // ArvLogWriter.Write -> CollectionFileWriter.Write | Api.Create
  30 //
  31 // For stdout/stderr ReadWriteLines additionally runs as a goroutine to pull
  32 // data from the stdout/stderr Reader and send to the Logger.
  33
  34 // ThrottledLogger accepts writes, prepends a timestamp to each line of the
  35 // write, and periodically flushes to a downstream writer.  It supports the
  36 // "Logger" and "WriteCloser" interfaces.
  37 type ThrottledLogger struct {
  38         *log.Logger
  39         buf *bytes.Buffer
  40         sync.Mutex
  41         writer   io.WriteCloser
  42         flush    chan struct{}
  43         stopped  chan struct{}
  44         stopping chan struct{}
  45         Timestamper
  46         Immediate    *log.Logger
  47         pendingFlush bool
  48 }
  49
  50 // RFC3339NanoFixed is a fixed-width version of time.RFC3339Nano.
  51 const RFC3339NanoFixed = "2006-01-02T15:04:05.000000000Z07:00"
  52
  53 // RFC3339Timestamp formats t as RFC3339NanoFixed.
  54 func RFC3339Timestamp(t time.Time) string {
  55         return t.Format(RFC3339NanoFixed)
  56 }
  57
  58 // Write prepends a timestamp to each line of the input data and
  59 // appends to the internal buffer. Each line is also logged to
  60 // tl.Immediate, if tl.Immediate is not nil.
  61 func (tl *ThrottledLogger) Write(p []byte) (n int, err error) {
  62         tl.Mutex.Lock()
  63         defer tl.Mutex.Unlock()
  64
  65         if tl.buf == nil {
  66                 tl.buf = &bytes.Buffer{}
  67         }
  68
  69         now := tl.Timestamper(time.Now().UTC())
  70         sc := bufio.NewScanner(bytes.NewBuffer(p))
  71         for err == nil && sc.Scan() {
  72                 out := fmt.Sprintf("%s %s\n", now, sc.Bytes())
  73                 if tl.Immediate != nil {
  74                         tl.Immediate.Print(out[:len(out)-1])
  75                 }
  76                 _, err = io.WriteString(tl.buf, out)
  77         }
  78         if err == nil {
  79                 err = sc.Err()
  80                 if err == nil {
  81                         n = len(p)
  82                 }
  83         }
  84
  85         if int64(tl.buf.Len()) >= crunchLogBytesPerEvent {
  86                 // Non-blocking send.  Try send a flush if it is ready to
  87                 // accept it.  Otherwise do nothing because a flush is already
  88                 // pending.
  89                 select {
  90                 case tl.flush <- struct{}{}:
  91                 default:
  92                 }
  93         }
  94
  95         return
  96 }
  97
  98 // Periodically check the current buffer; if not empty, send it on the
  99 // channel to the goWriter goroutine.
 100 func (tl *ThrottledLogger) flusher() {
 101         ticker := time.NewTicker(time.Duration(crunchLogSecondsBetweenEvents))
 102         defer ticker.Stop()
 103         for stopping := false; !stopping; {
 104                 select {
 105                 case <-tl.stopping:
 106                         // flush tl.buf and exit the loop
 107                         stopping = true
 108                 case <-tl.flush:
 109                 case <-ticker.C:
 110                 }
 111
 112                 var ready *bytes.Buffer
 113
 114                 tl.Mutex.Lock()
 115                 ready, tl.buf = tl.buf, &bytes.Buffer{}
 116                 tl.Mutex.Unlock()
 117
 118                 if ready != nil && ready.Len() > 0 {
 119                         tl.writer.Write(ready.Bytes())
 120                 }
 121         }
 122         close(tl.stopped)
 123 }
 124
 125 // Close the flusher goroutine and wait for it to complete, then close the
 126 // underlying Writer.
 127 func (tl *ThrottledLogger) Close() error {
 128         select {
 129         case <-tl.stopping:
 130                 // already stopped
 131         default:
 132                 close(tl.stopping)
 133         }
 134         <-tl.stopped
 135         return tl.writer.Close()
 136 }
 137
 138 const (
 139         // MaxLogLine is the maximum length of stdout/stderr lines before they are split.
 140         MaxLogLine = 1 << 12
 141 )
 142
 143 // ReadWriteLines reads lines from a reader and writes to a Writer, with long
 144 // line splitting.
 145 func ReadWriteLines(in io.Reader, writer io.Writer, done chan<- bool) {
 146         reader := bufio.NewReaderSize(in, MaxLogLine)
 147         var prefix string
 148         for {
 149                 line, isPrefix, err := reader.ReadLine()
 150                 if err == io.EOF {
 151                         break
 152                 } else if err != nil {
 153                         writer.Write([]byte(fmt.Sprintln("error reading container log:", err)))
 154                 }
 155                 var suffix string
 156                 if isPrefix {
 157                         suffix = "[...]\n"
 158                 }
 159
 160                 if prefix == "" && suffix == "" {
 161                         writer.Write(line)
 162                 } else {
 163                         writer.Write([]byte(fmt.Sprint(prefix, string(line), suffix)))
 164                 }
 165
 166                 // Set up prefix for following line
 167                 if isPrefix {
 168                         prefix = "[...]"
 169                 } else {
 170                         prefix = ""
 171                 }
 172         }
 173         done <- true
 174 }
 175
 176 // NewThrottledLogger creates a new thottled logger that
 177 // (a) prepends timestamps to each line
 178 // (b) batches log messages and only calls the underlying Writer
 179 //  at most once per "crunchLogSecondsBetweenEvents" seconds.
 180 func NewThrottledLogger(writer io.WriteCloser) *ThrottledLogger {
 181         tl := &ThrottledLogger{}
 182         tl.flush = make(chan struct{}, 1)
 183         tl.stopped = make(chan struct{})
 184         tl.stopping = make(chan struct{})
 185         tl.writer = writer
 186         tl.Logger = log.New(tl, "", 0)
 187         tl.Timestamper = RFC3339Timestamp
 188         go tl.flusher()
 189         return tl
 190 }
 191
 192 // Log throttling rate limiting config parameters
 193 var crunchLimitLogBytesPerJob int64 = 67108864
 194 var crunchLogThrottleBytes int64 = 65536
 195 var crunchLogThrottlePeriod time.Duration = time.Second * 60
 196 var crunchLogThrottleLines int64 = 1024
 197 var crunchLogPartialLineThrottlePeriod time.Duration = time.Second * 5
 198 var crunchLogBytesPerEvent int64 = 4096
 199 var crunchLogSecondsBetweenEvents time.Duration = time.Second * 1
 200
 201 // ArvLogWriter is an io.WriteCloser that processes each write by
 202 // writing it through to another io.WriteCloser (typically a
 203 // CollectionFileWriter) and creating an Arvados log entry.
 204 type ArvLogWriter struct {
 205         ArvClient     IArvadosClient
 206         UUID          string
 207         loggingStream string
 208         writeCloser   io.WriteCloser
 209
 210         // for rate limiting
 211         bytesLogged                  int64
 212         logThrottleResetTime         time.Time
 213         logThrottleLinesSoFar        int64
 214         logThrottleBytesSoFar        int64
 215         logThrottleBytesSkipped      int64
 216         logThrottleIsOpen            bool
 217         logThrottlePartialLineNextAt time.Time
 218         logThrottleFirstPartialLine  bool
 219         bufToFlush                   bytes.Buffer
 220         bufFlushedAt                 time.Time
 221         closing                      bool
 222 }
 223
 224 func (arvlog *ArvLogWriter) Write(p []byte) (int, error) {
 225         // Write to the next writer in the chain (a file in Keep)
 226         var err1 error
 227         if arvlog.writeCloser != nil {
 228                 _, err1 = arvlog.writeCloser.Write(p)
 229         }
 230
 231         // write to API after checking rate limit
 232         now := time.Now()
 233
 234         if now.After(arvlog.logThrottleResetTime) {
 235                 // It has been more than throttle_period seconds since the last
 236                 // checkpoint; so reset the throttle
 237                 if arvlog.logThrottleBytesSkipped > 0 {
 238                         arvlog.bufToFlush.WriteString(fmt.Sprintf("%s Skipped %d bytes of log\n", RFC3339Timestamp(now.UTC()), arvlog.logThrottleBytesSkipped))
 239                 }
 240
 241                 arvlog.logThrottleResetTime = now.Add(crunchLogThrottlePeriod)
 242                 arvlog.logThrottleBytesSoFar = 0
 243                 arvlog.logThrottleLinesSoFar = 0
 244                 arvlog.logThrottleBytesSkipped = 0
 245                 arvlog.logThrottleIsOpen = true
 246         }
 247
 248         lines := bytes.Split(p, []byte("\n"))
 249
 250         for _, line := range lines {
 251                 // Short circuit the counting code if we're just going to throw
 252                 // away the data anyway.
 253                 if !arvlog.logThrottleIsOpen {
 254                         arvlog.logThrottleBytesSkipped += int64(len(line))
 255                         continue
 256                 } else if len(line) == 0 {
 257                         continue
 258                 }
 259
 260                 // check rateLimit
 261                 logOpen, msg := arvlog.rateLimit(line, now)
 262                 if logOpen {
 263                         arvlog.bufToFlush.WriteString(string(msg) + "\n")
 264                 }
 265         }
 266
 267         if (int64(arvlog.bufToFlush.Len()) >= crunchLogBytesPerEvent ||
 268                 (now.Sub(arvlog.bufFlushedAt) >= crunchLogSecondsBetweenEvents) ||
 269                 arvlog.closing) && (arvlog.bufToFlush.Len() > 0) {
 270                 // write to API
 271                 lr := arvadosclient.Dict{"log": arvadosclient.Dict{
 272                         "object_uuid": arvlog.UUID,
 273                         "event_type":  arvlog.loggingStream,
 274                         "properties":  map[string]string{"text": arvlog.bufToFlush.String()}}}
 275                 err2 := arvlog.ArvClient.Create("logs", lr, nil)
 276
 277                 arvlog.bufToFlush = bytes.Buffer{}
 278                 arvlog.bufFlushedAt = now
 279
 280                 if err1 != nil || err2 != nil {
 281                         return 0, fmt.Errorf("%s ; %s", err1, err2)
 282                 }
 283         }
 284
 285         return len(p), nil
 286 }
 287
 288 // Close the underlying writer
 289 func (arvlog *ArvLogWriter) Close() (err error) {
 290         arvlog.closing = true
 291         arvlog.Write([]byte{})
 292         if arvlog.writeCloser != nil {
 293                 err = arvlog.writeCloser.Close()
 294                 arvlog.writeCloser = nil
 295         }
 296         return err
 297 }
 298
 299 var lineRegexp = regexp.MustCompile(`^\S+ (.*)`)
 300
 301 // Test for hard cap on total output and for log throttling. Returns whether
 302 // the log line should go to output or not. Returns message if limit exceeded.
 303 func (arvlog *ArvLogWriter) rateLimit(line []byte, now time.Time) (bool, []byte) {
 304         message := ""
 305         lineSize := int64(len(line))
 306
 307         if arvlog.logThrottleIsOpen {
 308                 matches := lineRegexp.FindStringSubmatch(string(line))
 309
 310                 if len(matches) == 2 && strings.HasPrefix(matches[1], "[...]") && strings.HasSuffix(matches[1], "[...]") {
 311                         // This is a partial line.
 312
 313                         if arvlog.logThrottleFirstPartialLine {
 314                                 // Partial should be suppressed.  First time this is happening for this line so provide a message instead.
 315                                 arvlog.logThrottleFirstPartialLine = false
 316                                 arvlog.logThrottlePartialLineNextAt = now.Add(crunchLogPartialLineThrottlePeriod)
 317                                 arvlog.logThrottleBytesSkipped += lineSize
 318                                 return true, []byte(fmt.Sprintf("%s Rate-limiting partial segments of long lines to one every %d seconds.",
 319                                         RFC3339Timestamp(now.UTC()), crunchLogPartialLineThrottlePeriod/time.Second))
 320                         } else if now.After(arvlog.logThrottlePartialLineNextAt) {
 321                                 // The throttle period has passed.  Update timestamp and let it through.
 322                                 arvlog.logThrottlePartialLineNextAt = now.Add(crunchLogPartialLineThrottlePeriod)
 323                         } else {
 324                                 // Suppress line.
 325                                 arvlog.logThrottleBytesSkipped += lineSize
 326                                 return false, line
 327                         }
 328                 } else {
 329                         // Not a partial line so reset.
 330                         arvlog.logThrottlePartialLineNextAt = time.Time{}
 331                         arvlog.logThrottleFirstPartialLine = true
 332                 }
 333
 334                 arvlog.bytesLogged += lineSize
 335                 arvlog.logThrottleBytesSoFar += lineSize
 336                 arvlog.logThrottleLinesSoFar += 1
 337
 338                 if arvlog.bytesLogged > crunchLimitLogBytesPerJob {
 339                         message = fmt.Sprintf("%s Exceeded log limit %d bytes (crunch_limit_log_bytes_per_job). Log will be truncated.",
 340                                 RFC3339Timestamp(now.UTC()), crunchLimitLogBytesPerJob)
 341                         arvlog.logThrottleResetTime = now.Add(time.Duration(365 * 24 * time.Hour))
 342                         arvlog.logThrottleIsOpen = false
 343
 344                 } else if arvlog.logThrottleBytesSoFar > crunchLogThrottleBytes {
 345                         remainingTime := arvlog.logThrottleResetTime.Sub(now)
 346                         message = fmt.Sprintf("%s Exceeded rate %d bytes per %d seconds (crunch_log_throttle_bytes). Logging will be silenced for the next %d seconds.",
 347                                 RFC3339Timestamp(now.UTC()), crunchLogThrottleBytes, crunchLogThrottlePeriod/time.Second, remainingTime/time.Second)
 348                         arvlog.logThrottleIsOpen = false
 349
 350                 } else if arvlog.logThrottleLinesSoFar > crunchLogThrottleLines {
 351                         remainingTime := arvlog.logThrottleResetTime.Sub(now)
 352                         message = fmt.Sprintf("%s Exceeded rate %d lines per %d seconds (crunch_log_throttle_lines), logging will be silenced for the next %d seconds.",
 353                                 RFC3339Timestamp(now.UTC()), crunchLogThrottleLines, crunchLogThrottlePeriod/time.Second, remainingTime/time.Second)
 354                         arvlog.logThrottleIsOpen = false
 355
 356                 }
 357         }
 358
 359         if !arvlog.logThrottleIsOpen {
 360                 // Don't log anything if any limit has been exceeded. Just count lossage.
 361                 arvlog.logThrottleBytesSkipped += lineSize
 362         }
 363
 364         if message != "" {
 365                 // Yes, write to logs, but use our "rate exceeded" message
 366                 // instead of the log message that exceeded the limit.
 367                 message += " A complete log is still being written to Keep, and will be available when the job finishes."
 368                 return true, []byte(message)
 369         } else {
 370                 return arvlog.logThrottleIsOpen, line
 371         }
 372 }
 373
 374 // load the rate limit discovery config parameters
 375 func loadLogThrottleParams(clnt IArvadosClient) {
 376         param, err := clnt.Discovery("crunchLimitLogBytesPerJob")
 377         if err == nil {
 378                 crunchLimitLogBytesPerJob = int64(param.(float64))
 379         }
 380
 381         param, err = clnt.Discovery("crunchLogThrottleBytes")
 382         if err == nil {
 383                 crunchLogThrottleBytes = int64(param.(float64))
 384         }
 385
 386         param, err = clnt.Discovery("crunchLogThrottlePeriod")
 387         if err == nil {
 388                 crunchLogThrottlePeriod = time.Duration(float64(time.Second) * param.(float64))
 389         }
 390
 391         param, err = clnt.Discovery("crunchLogThrottleLines")
 392         if err == nil {
 393                 crunchLogThrottleLines = int64(param.(float64))
 394         }
 395
 396         param, err = clnt.Discovery("crunchLogPartialLineThrottlePeriod")
 397         if err == nil {
 398                 crunchLogPartialLineThrottlePeriod = time.Duration(float64(time.Second) * param.(float64))
 399         }
 400
 401         param, err = clnt.Discovery("crunchLogBytesPerEvent")
 402         if err == nil {
 403                 crunchLogBytesPerEvent = int64(param.(float64))
 404         }
 405
 406         param, err = clnt.Discovery("crunchLogSecondsBetweenEvents")
 407         if err == nil {
 408                 crunchLogSecondsBetweenEvents = time.Duration(float64(time.Second) * param.(float64))
 409         }
 410 }