LogDog: Add project namespace to Butler/Collector.

client/internal/logdog/butler/bundler/bundler.go

 // Copyright 2015 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 package bundler
 
 import (
         "container/heap"
         "fmt"
         "sync"
         "time"
 
         "github.com/luci/luci-go/client/logdog/butlerlib/streamproto"
         "github.com/luci/luci-go/common/cancelcond"
         "github.com/luci/luci-go/common/clock"
+        "github.com/luci/luci-go/common/config"
         "github.com/luci/luci-go/common/logdog/types"
         "github.com/luci/luci-go/common/proto/google"
         "github.com/luci/luci-go/common/proto/logdog/logpb"
         "golang.org/x/net/context"
 )
 
 // Config is the Bundler configuration.
 type Config struct {
         // Clock is the clock instance that will be used for Bundler and stream
         // timing.
         Clock clock.Clock
 
         // Source is the bundle source string to use. This can be empty if there is no
         // source information to include.
         Source string
 
+        // Project is the project to use.
+        Project config.ProjectName
+        // Prefix is the common prefix for this set of streams.
+        Prefix types.StreamName
+        // Secret is the prefix secret for this set of streams.
+        Secret []byte
+
         // MaxBufferedBytes is the maximum number of bytes to buffer in memory per
         // stream.
         MaxBufferedBytes int64
 
         // MaxBundleSize is the maximum bundle size in bytes that may be generated.
         //
         // If this value is zero, no size constraint will be applied to generated
         // bundles.
         MaxBundleSize int
 
@@ skipping 48 matching lines @@
 }
 
 // Register adds a new stream to the Bundler, returning a reference to the
 // registered stream.
 func (b *Bundler) Register(p streamproto.Properties) (Stream, error) {
         // Our Properties must validate.
         if err := p.Validate(); err != nil {
                 return nil, err
         }
 
+        // Enforce that the log stream descriptor's Prefix is empty.
+        p.Prefix = ""
+
         // Construct a parser for this stream.
         c := streamConfig{
                 name: p.Name,
                 template: logpb.ButlerLogBundle_Entry{
                         Desc: &p.LogStreamDescriptor,
                 },
                 maximumBufferDuration: b.c.MaxBufferDelay,
                 maximumBufferedBytes:  b.c.MaxBufferedBytes,
                 onAppend: func(appended bool) {
                         if appended {
                                 b.signalStreamUpdate()
                         }
                 },
         }
 
         err := error(nil)
         c.parser, err = newParser(&p, &b.prefixCounter)
         if err != nil {
                 return nil, fmt.Errorf("failed to create stream parser: %s", err)
         }
 
-        // Generate a secret for this Stream instance.
-        c.template.Secret, err = types.NewPrefixSecret()
-        if err != nil {
-                return nil, fmt.Errorf("failed to generate stream secret: %s", err)
-        }
-
         b.streamsLock.Lock()
         defer b.streamsLock.Unlock()
 
         // Ensure that this is not a duplicate stream name.
         if s := b.streams[p.Name]; s != nil {
                 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.
@@ skipping 46 matching lines @@
                         b.bundleC <- bb.bundle()
                 }
         }()
 
         for {
                 bb = &builder{
                         size: b.c.MaxBundleSize,
                         template: logpb.ButlerLogBundle{
                                 Source:    b.c.Source,
                                 Timestamp: google.NewTimestamp(b.getClock().Now()),
+                                Project:   string(b.c.Project),
+                                Prefix:    string(b.c.Prefix),
+                                Secret:    b.c.Secret,
                         },
                 }
                 var oldestContentTime time.Time
 
                 for {
                         state := b.getStreamStateLocked()
 
                         // Attempt to create more bundles.
                         sendNow := b.bundleRoundLocked(bb, state)
 
@@ skipping 287 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
 }