Butler: fix global tags applied to empty map.

logdog/client/butler/bundler/bundler.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 bundler
 
 import (
         "container/heap"
         "fmt"
         "sync"
@@ skipping 34 matching lines @@
         // bundled data. Other factors can cause the bundle to be sent before this,
         // but it is an upper bound.
         MaxBufferDelay time.Duration
 }
 
 type bundlerStream interface {
         isDrained() bool
         name() string
         expireTime() (time.Time, bool)
         nextBundleEntry(*builder, bool) bool
+        streamDesc() *logpb.LogStreamDescriptor
 }
 
 // Bundler is the main Bundler instance. It exposes goroutine-safe endpoints for
 // stream registration and bundle consumption.
 type Bundler struct {
         c *Config
 
         // finishedC is closed when makeBundles goroutine has terminated.
         finishedC chan struct{}
         bundleC   chan *logpb.ButlerLogBundle
@@ skipping 69 matching lines @@
                 return nil, fmt.Errorf("a Stream is already registered for %q", p.Name)
         }
 
         // Create a new stream. This will kick off its processing goroutine, which
         // will not stop until it is closed.
         s := newStream(c)
         b.registerStreamLocked(s)
         return s, nil
 }
 
+// GetStreamDescs returns the set of registered stream names mapped to their
+// descriptors.
+//
+// This is intended for testing purposes. DO NOT modify the resulting
+// descriptors.
+func (b *Bundler) GetStreamDescs() map[string]*logpb.LogStreamDescriptor {
+        b.streamsLock.Lock()
+        defer b.streamsLock.Unlock()
+
+        if len(b.streams) == 0 {
+                return nil
+        }
+
+        streams := make(map[string]*logpb.LogStreamDescriptor, len(b.streams))
+        for k, s := range b.streams {
+                streams[k] = s.streamDesc()
+        }
+        return streams
+}
+
 // CloseAndFlush closes the Bundler, alerting it that no more streams will be
 // added and that existing data may be aggressively output.
 //
 // CloseAndFlush will block until all buffered data has been consumed.
 func (b *Bundler) CloseAndFlush() {
         // Mark that we're flushing. This will cause us to perform more aggressive
         // bundling in Next().
         b.startFlushing()
         <-b.finishedC
 }
@@ skipping 339 matching lines @@
 
 func (s *streamState) Push(x interface{}) {
         s.streams = append(s.streams, x.(bundlerStream))
 }
 
 func (s *streamState) Pop() interface{} {
         last := s.streams[len(s.streams)-1]
         s.streams = s.streams[:len(s.streams)-1]
         return last
 }