Add Butler global tags.

logdog/client/butler/butler.go

 // Copyright 2015 The LUCI Authors. All rights reserved.
 // Use of this source code is governed under the Apache License, Version 2.0
 // that can be found in the LICENSE file.
 
 package butler
 
 import (
         "errors"
         "fmt"
         "io"
         "sort"
         "sync"
         "time"
 
         "github.com/luci/luci-go/common/clock"
         "github.com/luci/luci-go/common/config"
         "github.com/luci/luci-go/common/data/stringset"
         "github.com/luci/luci-go/common/iotools"
         log "github.com/luci/luci-go/common/logging"
         "github.com/luci/luci-go/common/proto/google"
         "github.com/luci/luci-go/common/runtime/paniccatcher"
         "github.com/luci/luci-go/common/sync/parallel"
         "github.com/luci/luci-go/logdog/client/butler/bundler"
         "github.com/luci/luci-go/logdog/client/butler/output"
         "github.com/luci/luci-go/logdog/client/butler/streamserver"
         "github.com/luci/luci-go/logdog/client/butlerlib/streamproto"
         "github.com/luci/luci-go/logdog/common/types"
+
         "golang.org/x/net/context"
 )
 
 const (
         // DefaultMaxBufferAge is the default amount of time that a log entry may
         // be buffered before being dispatched.
         DefaultMaxBufferAge = time.Duration(5 * time.Second)
 
         // DefaultOutputWorkers is the default number of output workers to use.
         DefaultOutputWorkers = 16
@@ skipping 11 matching lines @@
         Output output.Output
         // OutputWorkers is the number of simultaneous goroutines that will be used
         // to output Butler log data. If zero, DefaultOutputWorkers will be used.
         OutputWorkers int
 
         // Project is the project that the log stream will be bound to.
         Project config.ProjectName
         // Prefix is the log stream common prefix value.
         Prefix types.StreamName
 
+        // GlobalTags are a set of global log stream tags to apply to individual
+        // streams on registration. Individual stream tags will override tags with
+        // the same key.
+        GlobalTags streamproto.TagMap
+
         // BufferLogs, if true, instructs the butler to buffer collected log data
         // before sending it to Output.
         BufferLogs bool
         // If buffering logs, this is the maximum amount of time that a log will
         // be buffered before being marked for dispatch. If this is zero,
         // DefaultMaxBufferAge will be used.
         MaxBufferAge time.Duration
 
         // TeeStdout, if not nil, is the Writer that will be used for streams
         // requesting STDOUT tee.
@@ skipping 322 matching lines @@
 // is responsible for closing it.
 func (b *Butler) AddStream(rc io.ReadCloser, p *streamproto.Properties) error {
         p = p.Clone()
         if p.Timestamp == nil || p.Timestamp.Time().IsZero() {
                 p.Timestamp = google.NewTimestamp(clock.Now(b.ctx))
         }
         if err := p.Validate(); err != nil {
                 return err
         }
 
+        // Build per-stream tag map.
+        if l := len(b.c.GlobalTags); l > 0 {
+                for k, v := range b.c.GlobalTags {
+                        // Add only global flags that aren't already present (overridden) in
+                        // stream tags.
+                        if _, ok := p.Tags[k]; !ok {
+                                p.Tags[k] = v
+                        }
+                }
+        }
+
         if p.Timeout > 0 {
                 if rts, ok := rc.(iotools.ReadTimeoutSetter); ok {
                         if err := rts.SetReadTimeout(p.Timeout); err != nil {
                                 log.Fields{
                                         log.ErrorKey: err,
                                         "timeout":    p.Timeout,
                                 }.Warningf(b.ctx, "Failed to set stream timeout.")
                         }
                 } else {
                         log.Fields{
@@ skipping 230 matching lines @@
         // shutdown prematurely, so this should be reasonably quick.
         b.Activate()
 }
 
 // Returns the configured Butler error.
 func (b *Butler) getRunErr() error {
         b.shutdownMu.Lock()
         defer b.shutdownMu.Unlock()
         return b.runErr
 }