LogDog: Handle archive failures.

server/internal/logdog/archivist/archivist.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 archivist
 
 import (
         "fmt"
 
         "github.com/golang/protobuf/proto"
@@ skipping 98 matching lines @@
         }
 
         task := &archiveTask{
                 Archivist:   a,
                 ArchiveTask: t,
                 ls:          ls,
                 desc:        &desc,
         }
         if err := task.archive(c); err != nil {
                 log.WithError(err).Errorf(c, "Failed to perform archival operation.")
-                return err
+
+                // Fail only if completeness is a requirement. Allow the Coordinator to
+                // handle errors on !Complete streams.
+                if t.Complete {
+                        return err
+                }
+
+                // Report that there was an archival error to the Coordinator's
+                // ArchiveStream endpoint. It may accept the archival or respond with a
+                // FailedPrecondition error code indicating that this archival run should
+                // be considered a failure.
+                task.ar.Error = true

[Vadim Sh., 2016/03/31 22:29:59]

are transient errors possible in "task.archive"? Is coordinator involved in
handling them (better to retry locally a bunch of times before bothering
coordinator).

[dnj, 2016/04/01 22:57:04]

They are retried in task.archive already as part of BigTable and GS shims.

         }
         log.Fields{
                 "streamURL":     task.ar.StreamUrl,
                 "indexURL":      task.ar.IndexUrl,
                 "dataURL":       task.ar.DataUrl,
                 "terminalIndex": task.ar.TerminalIndex,
                 "complete":      task.ar.Complete,
+                "hadError":      task.ar.Error,
         }.Debugf(c, "Finished archive construction.")
 
         if _, err := a.Service.ArchiveStream(c, &task.ar); err != nil {
                 log.WithError(err).Errorf(c, "Failed to mark log stream as archived.")
                 return err
         }
         return nil
 }
 
 // archiveTask is the set of parameters for a single archival.
 type archiveTask struct {
         *Archivist
         *logdog.ArchiveTask
 
         // ls is the log stream state.
         ls *logdog.LoadStreamResponse
         // desc is the unmarshaled log stream descriptor.
         desc *logpb.LogStreamDescriptor
 
         // ar will be populated during archive construction.
         ar logdog.ArchiveStreamRequest
 }
 
 // archiveState performs the archival operation on a stream described by a
 // Coordinator State. Upon success, the State will be updated with the result
 // of the archival operation.
 func (t *archiveTask) archive(c context.Context) (err error) {
+        // Minimal ArchiveRequest parameters.
+        t.ar.Path = t.Path
+        t.ar.TerminalIndex = -1
+
         // Generate our archival object managers.
         bext := t.desc.BinaryFileExt
         if bext == "" {
                 bext = "bin"
         }
 
         path := t.Path
         var streamO, indexO, dataO *gsObject
         streamO, err = t.newGSObject(c, path, "logstream.entries")
         if err != nil {
@@ skipping 100 matching lines @@
                 } else {
                         // Update our terminal index.
                         log.Fields{
                                 "from": tidx,
                                 "to":   t.ar.TerminalIndex,
                         }.Infof(c, "Updated log stream terminal index.")
                 }
         }
 
         // Update our state with archival results.
-        t.ar.Path = t.Path
         t.ar.StreamSize = streamO.Count()
         t.ar.IndexSize = indexO.Count()
         t.ar.DataSize = dataO.Count()
         t.ar.Complete = !ss.hasMissingEntries
         return
 }
 
 func (t *archiveTask) newGSObject(c context.Context, path string, name string) (*gsObject, error) {
         p := t.GSBase.Concat(path, name)
         o := gsObject{
@@ skipping 133 matching lines @@
                 log.Fields{
                         "index":         sidx,
                         "terminalIndex": s.terminalIndex,
                 }.Warningf(s, "Discarding log entries beyond expected terminal index.")
                 return nil, archive.ErrEndOfStream
         }
 
         s.lastIndex = sidx
         return le, nil
 }