Use native Pub/Sub library primitives.

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"
-        "github.com/luci/luci-go/common/gcloud/pubsub"
+        gcps "github.com/luci/luci-go/common/gcloud/pubsub"
         "github.com/luci/luci-go/common/logdog/butlerproto"
         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"
 )
 
-// Publisher is an interface for something that publishes Pub/Sub messages.
+// Topic is an interface for a Pub/Sub topic.
 //
-// pubsub.Connection implements this interface.
-type Publisher interface {
+// pubsub.TopicHandle 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.Topic, ...*pubsub.Message) ([]string, error)
+        Publish(context.Context, ...*pubsub.Message) ([]string, error)
 }
 
-var _ Publisher = pubsub.Connection(nil)
+var _ Topic = (*pubsub.TopicHandle)(nil)
 
 // Config is a configuration structure for Pub/Sub output.
 type Config struct {
-        // Publisher is the Pub/Sub instance to use.
-        Publisher Publisher
-
-        // Topic is the name of the Cloud Pub/Sub topic to publish to.
-        Topic pubsub.Topic
+        // Topic is the Pub/Sub topic to publish to.
+        Topic Topic
 
         // Compress, if true, enables zlib compression.
         Compress bool
 
         // Track, if true, tracks all log entries that have been successfully
         // submitted.
         Track bool
 }
 
-// Validate validates the Output configuration.
-func (c *Config) Validate() error {
-        if c.Publisher == nil {
-                return errors.New("pubsub: no pub/sub instance configured")
-        }
-        if err := c.Topic.Validate(); err != nil {
-                return fmt.Errorf("pubsub: invalid Topic [%s]: %s", c.Topic, err)
-        }
-        return nil
-}
-
 // buffer
 type buffer struct {
         bytes.Buffer // Output buffer for published message data.
 
         frameWriter recordio.Writer
         protoWriter *butlerproto.Writer
 }
 
 // Butler Output that sends messages into Google Cloud PubSub as compressed
 // protocol buffer blobs.
@@ skipping 18 matching lines @@
 
         if c.Track {
                 o.et = &output.EntryTracker{}
         }
 
         o.Context = log.SetField(ctx, "pubsub", &o)
         return &o
 }
 
 func (o *pubSubOutput) String() string {
-        return fmt.Sprintf("pubsub(%s)", o.Topic)
+        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)
 
         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) > pubsub.MaxPublishSize {
+        if len(message.Data) > gcps.MaxPublishSize {
                 log.Fields{
                         "messageSize":   len(message.Data),
-                        "maxPubSubSize": pubsub.MaxPublishSize,
+                        "maxPubSubSize": gcps.MaxPublishSize,
                 }.Errorf(o, "Constructed message exceeds Pub/Sub maximum size.")
                 return errors.New("pubsub: bundle contents violate Pub/Sub size limit")
         }
         if err := o.publishMessages([]*pubsub.Message{message}); err != nil {
                 st.F.DiscardedMessages++
                 st.F.Errors++
                 return err
         }
 
         if o.et != nil {
                 o.et.Track(bundle)
         }
 
         st.F.SentBytes += len(message.Data)
         st.F.SentMessages++
         return nil
 }
 
 func (*pubSubOutput) MaxSize() int {
-        return pubsub.MaxPublishSize / 2
+        return gcps.MaxPublishSize / 2
 }
 
 func (o *pubSubOutput) Stats() output.Stats {
         o.statsMu.Lock()
         defer o.statsMu.Unlock()
 
         statsCopy := o.stats
         return &statsCopy
 }
 
@@ skipping 34 matching lines @@
         }
 
         return &pubsub.Message{
                 Data: buf.Bytes(),
         }, nil
 }
 
 // publishMessages handles an individual publish request. It will indefinitely
 // retry transient errors until the publish succeeds.
 func (o *pubSubOutput) publishMessages(messages []*pubsub.Message) error {
-        messageIDs, err := o.Publisher.Publish(o, o.Topic, messages...)
+        var messageIDs []string
+        err := retry.Retry(o, retry.TransientOnly(indefiniteRetry), func() (err error) {
+                messageIDs, err = o.Topic.Publish(o, messages...)
+                return
+        }, func(err error, d time.Duration) {
+                log.Fields{
+                        log.ErrorKey: err,
+                        "delay":      d,
+                        "count":      len(messages),
+                }.Warningf(o, "TRANSIENT error publishing messages; retrying...")
+        })
         if err != nil {
-                return err
-        }
-        if err != nil {
-                log.Errorf(log.SetError(o, err), "Failed to send PubSub message.")
+                log.WithError(err).Errorf(o, "Failed to send PubSub message.")
                 return err
         }
 
-        log.Debugf(log.SetField(o, "messageIds", messageIDs), "Published messages.")
+        log.Fields{
+                "messageIds": messageIDs,
+        }.Debugf(o, "Published messages.")
         return nil
 }
 
 func (o *pubSubOutput) mergeStats(s output.Stats) {
         o.statsMu.Lock()
         defer o.statsMu.Unlock()
 
         o.stats.Merge(s)
 }
+
+// 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,
+        }
+}