LogDog: Update to archival V2.

server/logdog/archive/archive.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 archive constructs a LogDog archive out of log stream components.
 // Records are read from the stream and emitted as an archive.
 package archive
 
 import (
         "io"
 
         "github.com/golang/protobuf/proto"
+        "github.com/luci/luci-go/common/errors"
         "github.com/luci/luci-go/common/logging"
         "github.com/luci/luci-go/common/parallel"
         "github.com/luci/luci-go/common/proto/logdog/logpb"
         "github.com/luci/luci-go/common/recordio"
 )
 
 // Manifest is a set of archival parameters.
 type Manifest struct {
         // Desc is the logpb.LogStreamDescriptor for the stream.
         Desc *logpb.LogStreamDescriptor
@@ skipping 55 matching lines @@
         if m.IndexWriter != nil {
                 idx = &indexBuilder{
                         Manifest: &m,
                         index: logpb.LogIndex{
                                 Desc: m.Desc,
                         },
                         sizeFunc: m.sizeFunc,
                 }
         }
 
-        return parallel.FanOutIn(func(taskC chan<- func() error) {
+        err := parallel.FanOutIn(func(taskC chan<- func() error) {
                 var logC chan *logpb.LogEntry
                 if m.LogWriter != nil {
                         logC = make(chan *logpb.LogEntry)
 
                         taskC <- func() error {
                                 if err := archiveLogs(m.LogWriter, m.Desc, logC, idx); err != nil {
                                         return err
                                 }
 
                                 // If we're building an index, emit it now that the log stream has
@@ skipping 39 matching lines @@
                                                 return nil
                                         }
 
                                         return err
                                 }
 
                                 sendLog(le)
                         }
                 }
         })
+
+        // If any of the returned errors was transient, return a transient error.
+        if errors.Any(err, errors.IsTransient) {

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

This causes the archival process to forward transient errors (e.g., from Google
Storage Writer, etc.)

+                err = errors.WrapTransient(err)
+        }
+        return err
 }
 
 func archiveLogs(w io.Writer, d *logpb.LogStreamDescriptor, logC <-chan *logpb.LogEntry, idx *indexBuilder) error {
         offset := int64(0)
         out := func(pb proto.Message) error {
                 d, err := proto.Marshal(pb)
                 if err != nil {
                         return err
                 }
 
@@ skipping 11 matching lines @@
                 }
 
                 // Add this LogEntry to our index, noting the current offset.
                 if idx != nil {
                         idx.addLogEntry(le, offset)
                 }
                 err = out(le)
         }
         return err
 }