sdk/go/logger/logger.go

   1 // Logger periodically writes a log to the Arvados SDK.
   2 //
   3 // This package is useful for maintaining a log object that is updated
   4 // over time. This log object will be periodically written to the log,
   5 // as specified by WriteInterval in the Params.
   6 //
   7 // This package is safe for concurrent use as long as:
   8 // The maps passed to a LogMutator are not accessed outside of the
   9 // LogMutator
  10 //
  11 // Usage:
  12 // arvLogger := logger.NewLogger(params)
  13 // arvLogger.Update(func(properties map[string]interface{},
  14 //      entry map[string]interface{}) {
  15 //   // Modifiy properties and entry however you want
  16 //   // properties is a shortcut for entry["properties"].(map[string]interface{})
  17 //   // properties can take any (valid) values you want to give it,
  18 //   // entry will only take the fields listed at
  19 //   // http://doc.arvados.org/api/schema/Log.html
  20 //   // Valid values for properties are anything that can be json
  21 //   // encoded (i.e. will not error if you call json.Marshal() on it.
  22 // })
  23 package logger
  24
  25 import (
  26         "git.curoverse.com/arvados.git/sdk/go/arvadosclient"
  27         "log"
  28         "time"
  29 )
  30
  31 const (
  32         startSuffix              = "-start"
  33         partialSuffix            = "-partial"
  34         finalSuffix              = "-final"
  35         numberNoMoreWorkMessages = 2 // To return from FinalUpdate() & Work().
  36 )
  37
  38 type LoggerParams struct {
  39         Client          arvadosclient.ArvadosClient // The client we use to write log entries
  40         EventTypePrefix string                      // The prefix we use for the event type in the log entry
  41         WriteInterval   time.Duration               // Wait at least this long between log writes
  42 }
  43
  44 // A LogMutator is a function which modifies the log entry.
  45 // It takes two maps as arguments, properties is the first and entry
  46 // is the second
  47 // properties is a shortcut for entry["properties"].(map[string]interface{})
  48 // properties can take any values you want to give it.
  49 // entry will only take the fields listed at http://doc.arvados.org/api/schema/Log.html
  50 // properties and entry are only safe to access inside the LogMutator,
  51 // they should not be stored anywhere, otherwise you'll risk
  52 // concurrent access.
  53 type LogMutator func(map[string]interface{}, map[string]interface{})
  54
  55 // A Logger is used to build up a log entry over time and write every
  56 // version of it.
  57 type Logger struct {
  58         // The data we write
  59         data       map[string]interface{} // The entire map that we give to the api
  60         entry      map[string]interface{} // Convenience shortcut into data
  61         properties map[string]interface{} // Convenience shortcut into data
  62
  63         params LoggerParams // Parameters we were given
  64
  65         // Variables to coordinate updating and writing.
  66         modified    bool            // Has this data been modified since the last write?
  67         workToDo    chan LogMutator // Work to do in the worker thread.
  68         writeTicker *time.Ticker    // On each tick we write the log data to arvados, if it has been modified.
  69         hasWritten  bool            // Whether we've written at all yet.
  70         noMoreWork  chan bool       // Signals that we're done writing.
  71
  72         writeHooks []LogMutator // Mutators we call before each write.
  73 }
  74
  75 // Create a new logger based on the specified parameters.
  76 func NewLogger(params LoggerParams) *Logger {
  77         // sanity check parameters
  78         if &params.Client == nil {
  79                 log.Fatal("Nil arvados client in LoggerParams passed in to NewLogger()")
  80         }
  81         if params.EventTypePrefix == "" {
  82                 log.Fatal("Empty event type prefix in LoggerParams passed in to NewLogger()")
  83         }
  84
  85         l := &Logger{
  86                 data:        make(map[string]interface{}),
  87                 entry:       make(map[string]interface{}),
  88                 properties:  make(map[string]interface{}),
  89                 params:      params,
  90                 workToDo:    make(chan LogMutator, 10),
  91                 writeTicker: time.NewTicker(params.WriteInterval),
  92                 noMoreWork:  make(chan bool, numberNoMoreWorkMessages)}
  93
  94         l.data["log"] = l.entry
  95         l.entry["properties"] = l.properties
  96
  97         // Start the worker goroutine.
  98         go l.work()
  99
 100         return l
 101 }
 102
 103 // Exported functions will be called from other goroutines, therefore
 104 // all they are allowed to do is enqueue work to be done in the worker
 105 // goroutine.
 106
 107 // Enqueues an update. This will happen in another goroutine after
 108 // this method returns.
 109 func (l *Logger) Update(mutator LogMutator) {
 110         l.workToDo <- mutator
 111 }
 112
 113 // Similar to Update(), but writes the log entry as soon as possible
 114 // (ignoring MinimumWriteInterval) and blocks until the entry has been
 115 // written. This is useful if you know that you're about to quit
 116 // (e.g. if you discovered a fatal error, or you're finished), since
 117 // go will not wait for timers (including the pending write timer) to
 118 // go off before exiting.
 119 func (l *Logger) FinalUpdate(mutator LogMutator) {
 120         // TODO(misha): Consider not accepting any future updates somehow,
 121         // since they won't get written if they come in after this.
 122
 123         // Stop the periodic write ticker. We'll perform the final write
 124         // before returning from this function.
 125         l.workToDo <- func(p map[string]interface{}, e map[string]interface{}) {
 126                 l.writeTicker.Stop()
 127         }
 128
 129         // Apply the final update
 130         l.workToDo <- mutator
 131
 132         // Perform the final write and signal that we can return.
 133         l.workToDo <- func(p map[string]interface{}, e map[string]interface{}) {
 134                 l.write(true)
 135                 for i := 0; i < numberNoMoreWorkMessages; {
 136                         l.noMoreWork <- true
 137                 }
 138         }
 139
 140         // Wait until we've performed the write.
 141         <-l.noMoreWork
 142 }
 143
 144 // Adds a hook which will be called every time this logger writes an entry.
 145 func (l *Logger) AddWriteHook(hook LogMutator) {
 146         // We do the work in a LogMutator so that it happens in the worker
 147         // goroutine.
 148         l.workToDo <- func(p map[string]interface{}, e map[string]interface{}) {
 149                 l.writeHooks = append(l.writeHooks, hook)
 150         }
 151 }
 152
 153 // The worker loop
 154 func (l *Logger) work() {
 155         for {
 156                 select {
 157                 case <-l.writeTicker.C:
 158                         if l.modified {
 159                                 l.write(false)
 160                                 l.modified = false
 161                         }
 162                 case mutator := <-l.workToDo:
 163                         mutator(l.properties, l.entry)
 164                         l.modified = true
 165                 case <-l.noMoreWork:
 166                         return
 167                 }
 168         }
 169 }
 170
 171 // Actually writes the log entry.
 172 func (l *Logger) write(isFinal bool) {
 173
 174         // Run all our hooks
 175         for _, hook := range l.writeHooks {
 176                 hook(l.properties, l.entry)
 177         }
 178
 179         // Update the event type.
 180         if isFinal {
 181                 l.entry["event_type"] = l.params.EventTypePrefix + finalSuffix
 182         } else if l.hasWritten {
 183                 l.entry["event_type"] = l.params.EventTypePrefix + partialSuffix
 184         } else {
 185                 l.entry["event_type"] = l.params.EventTypePrefix + startSuffix
 186         }
 187         l.hasWritten = true
 188
 189         // Write the log entry.
 190         // This is a network write and will take a while, which is bad
 191         // because we're blocking all the other work on this goroutine.
 192         //
 193         // TODO(misha): Consider rewriting this so that we can encode l.data
 194         // into a string, and then perform the actual write in another
 195         // routine. This will be tricky and will require support in the
 196         // client.
 197         err := l.params.Client.Create("logs", l.data, nil)
 198         if err != nil {
 199                 log.Printf("Attempted to log: %v", l.data)
 200                 log.Fatalf("Received error writing log: %v", err)
 201         }
 202 }