7816: Rename to crunch-run
[arvados.git] / services / crunch-run / logging.go
1 package main
2
3 import (
4         "bufio"
5         "bytes"
6         "fmt"
7         "git.curoverse.com/arvados.git/sdk/go/arvadosclient"
8         "io"
9         "log"
10         "sync"
11         "time"
12 )
13
14 // Timestamper is the signature for a function that takes a timestamp and
15 // return a formated string value.
16 type Timestamper func(t time.Time) string
17
18 // Logging plumbing:
19 //
20 // ThrottledLogger.Logger -> ThrottledLogger.Write ->
21 // ThrottledLogger.buf -> ThrottledLogger.flusher -> goWriter ->
22 // ArvLogWriter.Write -> CollectionFileWriter.Write | Api.Create
23 //
24 // For stdout/stderr CopyReaderToLog additionally runs as a goroutine to pull
25 // data from the stdout/stderr Reader and send to the Logger.
26
27 // ThrottledLogger accepts writes, prepends a timestamp to each line of the
28 // write, and periodically flushes to a downstream writer.  It supports the
29 // "Logger" and "WriteCloser" interfaces.
30 type ThrottledLogger struct {
31         *log.Logger
32         buf *bytes.Buffer
33         sync.Mutex
34         writer      io.WriteCloser
35         stop        bool
36         flusherDone chan bool
37         Timestamper
38 }
39
40 // RFC3339Fixed is a fixed-width version of RFC3339 with microsecond precision,
41 // because the RFC3339Nano format isn't fixed width.
42 const RFC3339Fixed = "2006-01-02T15:04:05.000000Z07:00"
43
44 // RFC3339Timestamp return a RFC3339 formatted timestamp using RFC3339Fixed
45 func RFC3339Timestamp(now time.Time) string {
46         return now.Format(RFC3339Fixed)
47 }
48
49 // Write to the internal buffer.  Prepend a timestamp to each line of the input
50 // data.
51 func (tl *ThrottledLogger) Write(p []byte) (n int, err error) {
52         tl.Mutex.Lock()
53         if tl.buf == nil {
54                 tl.buf = &bytes.Buffer{}
55         }
56         defer tl.Mutex.Unlock()
57
58         now := tl.Timestamper(time.Now().UTC())
59         sc := bufio.NewScanner(bytes.NewBuffer(p))
60         for sc.Scan() {
61                 _, err = fmt.Fprintf(tl.buf, "%s %s\n", now, sc.Text())
62         }
63         return len(p), err
64 }
65
66 // Periodically check the current buffer; if not empty, send it on the
67 // channel to the goWriter goroutine.
68 func (tl *ThrottledLogger) flusher() {
69         bufchan := make(chan *bytes.Buffer)
70         bufterm := make(chan bool)
71
72         // Use a separate goroutine for the actual write so that the writes are
73         // actually initiated closer every 1s instead of every
74         // 1s + (time to it takes to write).
75         go goWriter(tl.writer, bufchan, bufterm)
76         for {
77                 if !tl.stop {
78                         time.Sleep(1 * time.Second)
79                 }
80                 tl.Mutex.Lock()
81                 if tl.buf != nil && tl.buf.Len() > 0 {
82                         oldbuf := tl.buf
83                         tl.buf = nil
84                         tl.Mutex.Unlock()
85                         bufchan <- oldbuf
86                 } else if tl.stop {
87                         tl.Mutex.Unlock()
88                         break
89                 } else {
90                         tl.Mutex.Unlock()
91                 }
92         }
93         close(bufchan)
94         <-bufterm
95         tl.flusherDone <- true
96 }
97
98 // Receive buffers from a channel and send to the underlying Writer
99 func goWriter(writer io.Writer, c <-chan *bytes.Buffer, t chan<- bool) {
100         for b := range c {
101                 writer.Write(b.Bytes())
102         }
103         t <- true
104 }
105
106 // Close the flusher goroutine and wait for it to complete, then close the
107 // underlying Writer.
108 func (tl *ThrottledLogger) Close() error {
109         tl.stop = true
110         <-tl.flusherDone
111         return tl.writer.Close()
112 }
113
114 const (
115         // MaxLogLine is the maximum length of stdout/stderr lines before they are split.
116         MaxLogLine = 1 << 12
117 )
118
119 // CopyReaderToLog reads from a Reader and prints to a Logger, with long line
120 // splitting.
121 func CopyReaderToLog(in io.Reader, logger *log.Logger, done chan<- bool) {
122         reader := bufio.NewReaderSize(in, MaxLogLine)
123         var prefix string
124         for {
125                 line, isPrefix, err := reader.ReadLine()
126                 if err == io.EOF {
127                         break
128                 } else if err != nil {
129                         logger.Print("error reading container log:", err)
130                 }
131                 var suffix string
132                 if isPrefix {
133                         suffix = "[...]"
134                 }
135                 logger.Print(prefix, string(line), suffix)
136                 // Set up prefix for following line
137                 if isPrefix {
138                         prefix = "[...]"
139                 } else {
140                         prefix = ""
141                 }
142         }
143         done <- true
144 }
145
146 // NewThrottledLogger creates a new thottled logger that
147 // (a) prepends timestamps to each line
148 // (b) batches log messages and only calls the underlying Writer at most once
149 // per second.
150 func NewThrottledLogger(writer io.WriteCloser) *ThrottledLogger {
151         alw := &ThrottledLogger{}
152         alw.flusherDone = make(chan bool)
153         alw.writer = writer
154         alw.Logger = log.New(alw, "", 0)
155         alw.Timestamper = RFC3339Timestamp
156         go alw.flusher()
157         return alw
158 }
159
160 // ArvLogWriter implements a writer that writes to each of a WriteCloser
161 // (typically CollectionFileWriter) and creates an API server log entry.
162 type ArvLogWriter struct {
163         ArvClient     IArvadosClient
164         UUID          string
165         loggingStream string
166         writeCloser   io.WriteCloser
167 }
168
169 func (arvlog *ArvLogWriter) Write(p []byte) (n int, err error) {
170         // Write to the next writer in the chain (a file in Keep)
171         var err1 error
172         if arvlog.writeCloser != nil {
173                 _, err1 = arvlog.writeCloser.Write(p)
174         }
175
176         // write to API
177         lr := arvadosclient.Dict{"object_uuid": arvlog.UUID,
178                 "event_type": arvlog.loggingStream,
179                 "properties": map[string]string{"text": string(p)}}
180         err2 := arvlog.ArvClient.Create("logs", lr, nil)
181
182         if err1 != nil || err2 != nil {
183                 return 0, fmt.Errorf("%s ; %s", err1, err2)
184         }
185         return len(p), nil
186 }
187
188 // Close the underlying writer
189 func (arvlog *ArvLogWriter) Close() (err error) {
190         if arvlog.writeCloser != nil {
191                 err = arvlog.writeCloser.Close()
192                 arvlog.writeCloser = nil
193         }
194         return err
195 }