LogDog: Add collector service implementation.

common/gcloud/gcps/ackbuffer/ackbuffer.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 ackbuffer implements a Pub/Sub acknowledgement buffer capability.
 // Pub/Sub ACKs will be collected and batched before being sent to Pub/Sub,
 // with specific deadline enforcement.
 package ackbuffer
 
 import (
         "sync"
         "time"
 
         "github.com/luci/luci-go/common/gcloud/gcps"
         log "github.com/luci/luci-go/common/logging"
         "github.com/luci/luci-go/common/meter"
         "github.com/luci/luci-go/common/parallel"
-        "github.com/luci/luci-go/common/retry"
         "golang.org/x/net/context"
 )
 
 const (
         // DefaultMaxBufferTime is the default amount of time that an ACK will remain
         // buffered before being sent.
         //
         // We base this off the default acknowledgement delay.
         DefaultMaxBufferTime = (gcps.DefaultMaxAckDelay / 6)
 
         // DefaultMaxParallelACK is the default maximum number of simultaneous
         // parallel ACK request goroutines.
         DefaultMaxParallelACK = 16
 )
 
 // DiscardCallback is a callback method that will be invoked if ACK IDs must
 // be discarded.
 type DiscardCallback func(ackIDs []string)
 
-// PubSubACK sends ACKs to a Pub/Sub interface.
-//
-// gcps.PubSub naturally implements this interface.
-type PubSubACK interface {
-        // Ack acknowledges one or more Pub/Sub message ACK IDs.
-        Ack(s gcps.Subscription, ackIDs ...string) error
-}
-
 // Config is a set of configuration parameters for an AckBuffer.
 type Config struct {
-        // PubSub is the Pub/Sub instance to ACK with.
-        PubSub PubSubACK
-
-        // Subscription is the name of the Pub/Sub subscription to ACK.
-        Subscription gcps.Subscription
+        // Ack is the Pub/Sub instance to ACK with.
+        Ack Acknowledger
 
         // MaxBufferTime is the maximum amount of time to buffer an ACK before sending it.
         MaxBufferTime time.Duration
 
         // The maximum number of parallel ACK requests that can be simultaneously
         // open. If zero, DefaultMaxParallelACK will be used.
         MaxParallelACK int
 
         // DiscardCallback is invoked when a series of ACK IDs is discarded after
         // repeated failures to ACK. If this is nil, no callback will be invoked.
@@ skipping 19 matching lines @@
 // New instantiates a new AckBuffer. The returned AckBuffer must have its
 // CloseAndFlush method invoked before terminating, else data loss may occur.
 func New(ctx context.Context, c Config) *AckBuffer {
         if c.MaxBufferTime <= 0 {
                 c.MaxBufferTime = DefaultMaxBufferTime
         }
         if c.MaxParallelACK <= 0 {
                 c.MaxParallelACK = DefaultMaxParallelACK
         }
 
-        ctx = log.SetField(ctx, "subscription", c.Subscription)
-
         b := &AckBuffer{
                 cfg:          &c,
                 ctx:          ctx,
                 ackRequestC:  make(chan []string),
                 ackFinishedC: make(chan struct{}),
         }
         b.meter = meter.New(ctx, meter.Config{
-                Count:    gcps.MaxMessageAckPerRequest,
+                Count:    b.cfg.Ack.AckBatchSize(),
                 Delay:    b.cfg.MaxBufferTime,
                 Callback: b.meterCallback,
         })
 
         // Start our ACK loop.
         wg := sync.WaitGroup{}
         go func() {
                 defer close(b.ackFinishedC)
 
                 // Allocate and populate a set of ACK tokens. This will be used as a
@@ skipping 42 matching lines @@
 // This shouldn't block if possible, else the Meter will block. However, if
 // ACK requests build up, this will block until they are finished.
 func (b *AckBuffer) meterCallback(work []interface{}) {
         ackIDs := make([]string, len(work))
         for idx, w := range work {
                 ackIDs[idx] = w.(string)
         }
         b.ackRequestC <- ackIDs
 }
 
-// acknowledge acknowledges a set of IDs. It will retry on transient errors.
+// acknowledge acknowledges a set of IDs.
 //
 // This method will discard the ACKs if they fail.
 func (b *AckBuffer) acknowledge(ackIDs []string) {
-        err := retry.Retry(b.ctx, retry.TransientOnly(retry.Default()), func() error {
-                return b.cfg.PubSub.Ack(b.cfg.Subscription, ackIDs...)
-        }, func(err error, delay time.Duration) {
-                log.Fields{
-                        log.ErrorKey: err,
-                        "delay":      delay,
-                }.Warningf(b.ctx, "Error sending ACK; retrying.")
-        })
-        if err != nil {
+        if err := b.cfg.Ack.Ack(b.ctx, ackIDs...); err != nil {

[dnj (Google), 2016/01/21 04:36:24]

User must supply a retry-enabled Pub/Sub instance now. This is, fortunately,
super-easy.

                 log.Fields{
                         log.ErrorKey: err,
                         "count":      len(ackIDs),
                 }.Errorf(b.ctx, "Failed to ACK.")
                 if b.cfg.DiscardCallback != nil {
                         b.cfg.DiscardCallback(ackIDs)
                 }
         }
 }