LogDog: Implement prefix registration in Butler.

client/internal/logdog/butler/output/pubsub/pubsubOutput.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 pubsub
 
 import (
         "bytes"
         "errors"
         "fmt"
         "sync"
         "time"
 
         "github.com/luci/luci-go/client/internal/logdog/butler/output"
         gcps "github.com/luci/luci-go/common/gcloud/pubsub"
         "github.com/luci/luci-go/common/logdog/butlerproto"
+        "github.com/luci/luci-go/common/logdog/types"
         log "github.com/luci/luci-go/common/logging"
         "github.com/luci/luci-go/common/proto/logdog/logpb"
         "github.com/luci/luci-go/common/recordio"
         "github.com/luci/luci-go/common/retry"
         "golang.org/x/net/context"
         "google.golang.org/cloud/pubsub"
 )
 
 // Topic is an interface for a Pub/Sub topic.
 //
 // pubsub.Topic implements Topic.
 type Topic interface {
         // Name returns the name of the topic.
         Name() string
 
         // Publish mirrors the pubsub.Connection Publish method.
         Publish(context.Context, ...*pubsub.Message) ([]string, error)
 }
 
 var _ Topic = (*pubsub.Topic)(nil)
 
 // Config is a configuration structure for Pub/Sub output.
 type Config struct {
         // Topic is the Pub/Sub topic to publish to.
         Topic Topic
 
+        // Secret, if not nil, is the prefix secret to attach to each outgoing bundle.
+        Secret types.PrefixSecret
+
         // Compress, if true, enables zlib compression.
         Compress bool
 
         // Track, if true, tracks all log entries that have been successfully
         // submitted.
         Track bool
 }
 
 // buffer
 type buffer struct {
@@ skipping 36 matching lines @@
         return fmt.Sprintf("pubsub(%s)", o.Topic.Name())
 }
 
 func (o *pubSubOutput) SendBundle(bundle *logpb.ButlerLogBundle) error {
         st := output.StatsBase{}
         defer o.mergeStats(&st)
 
         b := o.bufferPool.Get().(*buffer)
         defer o.bufferPool.Put(b)
 
+        bundle.Secret = []byte(o.Secret)
         message, err := o.buildMessage(b, bundle)
         if err != nil {
                 log.Fields{
                         log.ErrorKey: err,
                 }.Errorf(o, "Failed to build PubSub Message from bundle.")
                 st.F.DiscardedMessages++
                 st.F.Errors++
                 return err
         }
         if len(message.Data) > gcps.MaxPublishSize {
@@ skipping 106 matching lines @@
 // indefiniteRetry is a retry.Iterator that will indefinitely retry errors with
 // a maximum backoff.
 func indefiniteRetry() retry.Iterator {
         return &retry.ExponentialBackoff{
                 Limited: retry.Limited{
                         Retries: -1,
                 },
                 MaxDelay: 30 * time.Second,
         }
 }