LogDog: Update to archival V2.

appengine/logdog/coordinator/archivalPublisher.go

+// Copyright 2016 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 coordinator
+
+import (
+        "time"
+
+        "github.com/golang/protobuf/proto"
+        "github.com/luci/luci-go/common/api/logdog_coordinator/services/v1"
+        log "github.com/luci/luci-go/common/logging"
+        "github.com/luci/luci-go/common/retry"
+        "golang.org/x/net/context"
+        gcps "google.golang.org/cloud/pubsub"
+)
+
+// ArchivalPublisher is capable of publishing archival requests.
+type ArchivalPublisher interface {

[dnj, 2016/04/11 17:20:04]

This exists so we can mock it in tests and assert that various archival jobs
were correctly dispatched. It is also sufficiently decoupled from Pub/Sub that
if we switch tasking mechanisms, it will hopefully be far less intrusive.

[iannucci, 2016/04/19 00:55:28]

Since it's just one function, maybe just make this a type and pass it in? E.g.
type ArchivePublish func(...)

[dnj, 2016/04/20 18:15:10]

ATM I'm not binding this to Pub/Sub, so some properties (namely the Pub/Sub
topic, see "pubsubArchivalPublisher") are not suitable as function parameters,
so this makes sense to me.

+        // Publish publishes the supplied ArchiveTask.
+        Publish(context.Context, *logdog.ArchiveTask) error
+}
+
+type pubsubArchivalPublisher struct {
+        // topic is the authenticated Pub/Sub topic handle to publish to.
+        topic *gcps.TopicHandle
+}
+
+func (p *pubsubArchivalPublisher) Publish(c context.Context, t *logdog.ArchiveTask) error {
+        d, err := proto.Marshal(t)
+        if err != nil {
+                log.WithError(err).Errorf(c, "Failed to marshal task.")
+                return err
+        }
+
+        // TODO: Route this through some system (e.g., task queue, tumble) that can
+        // impose a dispatch delay for the settle period.
+        msg := gcps.Message{
+                Data: d,
+        }
+
+        return retry.Retry(c, retry.Default, func() error {
+                _, err := p.topic.Publish(c, &msg)
+                return err
+        }, func(err error, d time.Duration) {
+                log.Fields{
+                        log.ErrorKey: err,
+                        "delay":      d,
+                }.Warningf(c, "Failed to publish task. Retrying...")
+        })
+}