//
// This package is useful for maintaining a log object that is updated
// over time. Every time the object is updated, it will be written to
-// the log. Writes will be throttled to no more than one every
-// WriteFrequencySeconds
+// the log. Writes will be throttled to no more frequent than
+// WriteInterval.
//
// This package is safe for concurrent use as long as:
// The maps passed to a LogMutator are not accessed outside of the
import (
"git.curoverse.com/arvados.git/sdk/go/arvadosclient"
"log"
- "sync"
"time"
)
type LoggerParams struct {
- Client arvadosclient.ArvadosClient // The client we use to write log entries
- EventType string // The event type to assign to the log entry.
- MinimumWriteInterval time.Duration // Wait at least this long between log writes
+ Client arvadosclient.ArvadosClient // The client we use to write log entries
+ EventType string // The event type to assign to the log entry.
+ WriteInterval time.Duration // Wait at least this long between log writes
}
// A LogMutator is a function which modifies the log entry.
entry map[string]interface{} // Convenience shortcut into data
properties map[string]interface{} // Convenience shortcut into data
- lock sync.Locker // Synchronizes access to this struct
params LoggerParams // Parameters we were given
- // Variables used to determine when and if we write to the log.
- nextWriteAllowed time.Time // The next time we can write, respecting MinimumWriteInterval
- modified bool // Has this data been modified since the last write?
- writeScheduled bool // Is a write been scheduled for the future?
+ // Variables to coordinate updating and writing.
+ modified bool // Has this data been modified since the last write?
+ workToDo chan LogMutator // Work to do in the worker thread.
+ writeTicker *time.Ticker // On each tick we write the log data to arvados, if it has been modified.
writeHooks []LogMutator // Mutators we call before each write.
}
}
l := &Logger{data: make(map[string]interface{}),
- lock: &sync.Mutex{},
params: params}
l.entry = make(map[string]interface{})
l.data["log"] = l.entry
l.properties = make(map[string]interface{})
l.entry["properties"] = l.properties
- return l
-}
-// Updates the log data and then writes it to the api server. If the
-// log has been recently written then the write will be postponed to
-// respect MinimumWriteInterval and this function will return before
-// the write occurs.
-func (l *Logger) Update(mutator LogMutator) {
- l.lock.Lock()
+ l.workToDo = make(chan LogMutator, 10)
+ l.writeTicker = time.NewTicker(params.WriteInterval)
- mutator(l.properties, l.entry)
- l.modified = true // We assume the mutator modified the log, even though we don't know for sure.
+ // Start the worker goroutine.
+ go l.work()
- l.considerWriting()
-
- l.lock.Unlock()
+ return l
}
-// Similar to Update(), but forces a write without respecting the
-// MinimumWriteInterval. This is useful if you know that you're about
-// to quit (e.g. if you discovered a fatal error, or you're finished),
-// since go will not wait for timers (including the pending write
-// timer) to go off before exiting.
-func (l *Logger) ForceUpdate(mutator LogMutator) {
- l.lock.Lock()
+// Exported functions will be called from other goroutines, therefore
+// all they are allowed to do is enqueue work to be done in the worker
+// goroutine.
- mutator(l.properties, l.entry)
- l.modified = true // We assume the mutator modified the log, even though we don't know for sure.
-
- l.write()
- l.lock.Unlock()
-}
-
-// Adds a hook which will be called every time this logger writes an entry.
-func (l *Logger) AddWriteHook(hook LogMutator) {
- l.lock.Lock()
- l.writeHooks = append(l.writeHooks, hook)
- // TODO(misha): Consider setting modified and attempting a write.
- l.lock.Unlock()
+// Enqueues an update. This will happen in another goroutine after
+// this method returns.
+func (l *Logger) Update(mutator LogMutator) {
+ l.workToDo <- mutator
}
-// This function is called on a timer when we have something to write,
-// but need to schedule the write for the future to respect
-// MinimumWriteInterval.
-func (l *Logger) acquireLockConsiderWriting() {
- l.lock.Lock()
+// Similar to Update(), but writes the log entry as soon as possible
+// (ignoring MinimumWriteInterval) and blocks until the entry has been
+// written. This is useful if you know that you're about to quit
+// (e.g. if you discovered a fatal error, or you're finished), since
+// go will not wait for timers (including the pending write timer) to
+// go off before exiting.
+func (l *Logger) FinalUpdate(mutator LogMutator) {
+ // Block on this channel until everything finishes
+ done := make(chan bool)
+
+ // TODO(misha): Consider not accepting any future updates somehow,
+ // since they won't get written if they come in after this.
+
+ // Stop the periodic write ticker. We'll perform the final write
+ // before returning from this function.
+ l.workToDo <- func(p map[string]interface{}, e map[string]interface{}) {
+ l.writeTicker.Stop()
+ }
- // We are the scheduled write, so there are no longer future writes
- // scheduled.
- l.writeScheduled = false
+ // Apply the final update
+ l.workToDo <- mutator
- l.considerWriting()
+ // Perform the write and signal that we can return.
+ l.workToDo <- func(p map[string]interface{}, e map[string]interface{}) {
+ // TODO(misha): Add a boolean arg to write() to indicate that it's
+ // final so that we can set the appropriate event type.
+ l.write()
+ done <- true
+ }
- l.lock.Unlock()
+ // Wait until we've performed the write.
+ <-done
}
-// The above methods each acquire the lock and release it.
-// =======================================================
-// The below methods all assume we're holding a lock.
-
-// Check whether we have anything to write. If we do, then either
-// write it now or later, based on what we're allowed.
-func (l *Logger) considerWriting() {
- if !l.modified {
- // Nothing to write
- } else if l.writeAllowedNow() {
- l.write()
- } else if l.writeScheduled {
- // A future write is already scheduled, we don't need to do anything.
- } else {
- writeAfter := l.nextWriteAllowed.Sub(time.Now())
- time.AfterFunc(writeAfter, l.acquireLockConsiderWriting)
- l.writeScheduled = true
+// Adds a hook which will be called every time this logger writes an entry.
+func (l *Logger) AddWriteHook(hook LogMutator) {
+ // We do the work in a LogMutator so that it happens in the worker
+ // goroutine.
+ l.workToDo <- func(p map[string]interface{}, e map[string]interface{}) {
+ l.writeHooks = append(l.writeHooks, hook)
}
}
-// Whether writing now would respect MinimumWriteInterval
-func (l *Logger) writeAllowedNow() bool {
- return l.nextWriteAllowed.Before(time.Now())
+// The worker loop
+func (l *Logger) work() {
+ for {
+ select {
+ case <-l.writeTicker.C:
+ if l.modified {
+ l.write()
+ l.modified = false
+ }
+ case mutator := <-l.workToDo:
+ mutator(l.properties, l.entry)
+ l.modified = true
+ }
+ }
}
// Actually writes the log entry.
}
// Update the event type in case it was modified or is missing.
+ // TODO(misha): Fix this to write different event types.
l.entry["event_type"] = l.params.EventType
// Write the log entry.
// This is a network write and will take a while, which is bad
- // because we're holding a lock and all other goroutines will back
- // up behind it.
+ // because we're blocking all the other work on this goroutine.
//
// TODO(misha): Consider rewriting this so that we can encode l.data
- // into a string, release the lock, write the string, and then
- // acquire the lock again to note that we succeeded in writing. This
- // will be tricky and will require support in the client.
+ // into a string, and then perform the actual write in another
+ // routine. This will be tricky and will require support in the
+ // client.
err := l.params.Client.Create("logs", l.data, nil)
if err != nil {
log.Printf("Attempted to log: %v", l.data)
log.Fatalf("Received error writing log: %v", err)
}
-
- // Update stats.
- l.nextWriteAllowed = time.Now().Add(l.params.MinimumWriteInterval)
- l.modified = false
}
KeepServers []ServerAddress `json:"items"`
}
-// Methods to implement util.SdkListResponse Interface
-func (k KeepServiceList) NumItemsAvailable() (numAvailable int, err error) {
- return k.ItemsAvailable, nil
-}
-
-func (k KeepServiceList) NumItemsContained() (numContained int, err error) {
- return len(k.KeepServers), nil
-}
-
var (
// Don't access the token directly, use getDataManagerToken() to
// make sure it's been read.
keepServer.Port)
if arvLogger != nil {
+ now := time.Now()
arvLogger.Update(func(p map[string]interface{}, e map[string]interface{}) {
keepInfo := p["keep_info"].(map[string]interface{})
serverInfo := make(map[string]interface{})
- serverInfo["time_status_request_sent"] = time.Now()
+ serverInfo["time_status_request_sent"] = now
keepInfo[keepServer.String()] = serverInfo
})
}
if arvLogger != nil {
+ now := time.Now()
arvLogger.Update(func(p map[string]interface{}, e map[string]interface{}) {
keepInfo := p["keep_info"].(map[string]interface{})
serverInfo := keepInfo[keepServer.String()].(map[string]interface{})
- serverInfo["time_status_response_processed"] = time.Now()
+ serverInfo["time_status_response_processed"] = now
serverInfo["status"] = keepStatus
})
}
log.Println("About to fetch keep server contents from " + url)
if arvLogger != nil {
+ now := time.Now()
arvLogger.Update(func(p map[string]interface{}, e map[string]interface{}) {
keepInfo := p["keep_info"].(map[string]interface{})
serverInfo := keepInfo[keepServer.String()].(map[string]interface{})
- serverInfo["time_index_request_sent"] = time.Now()
+ serverInfo["time_index_request_sent"] = now
})
}
}
if arvLogger != nil {
+ now := time.Now()
arvLogger.Update(func(p map[string]interface{}, e map[string]interface{}) {
keepInfo := p["keep_info"].(map[string]interface{})
serverInfo := keepInfo[keepServer.String()].(map[string]interface{})
-
- serverInfo["time_index_response_received"] = time.Now()
+ serverInfo["time_index_response_received"] = now
})
}
numSizeDisagreements)
if arvLogger != nil {
+ now := time.Now()
arvLogger.Update(func(p map[string]interface{}, e map[string]interface{}) {
keepInfo := p["keep_info"].(map[string]interface{})
serverInfo := keepInfo[keepServer.String()].(map[string]interface{})
- serverInfo["time_processing_finished"] = time.Now()
+ serverInfo["time_processing_finished"] = now
serverInfo["lines_received"] = numLines
serverInfo["duplicates_seen"] = numDuplicates
serverInfo["size_disagreements_seen"] = numSizeDisagreements
projects / arvados.git / commitdiff