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