LogDog: Update to archival V2.

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"
+        "time"
 
         "github.com/golang/protobuf/proto"
         "github.com/luci/luci-go/common/api/logdog_coordinator/services/v1"
         "github.com/luci/luci-go/common/errors"
         "github.com/luci/luci-go/common/gcloud/gs"
         "github.com/luci/luci-go/common/logdog/types"
         log "github.com/luci/luci-go/common/logging"
         "github.com/luci/luci-go/common/proto/logdog/logpb"
         "github.com/luci/luci-go/server/logdog/archive"
         "github.com/luci/luci-go/server/logdog/storage"
@@ skipping 26 matching lines @@
         // ByteRange is the maximum number of stream data bytes in between index
         // entries. See archive.Manifest for more information.
         ByteRange int
 }
 
 // storageBufferSize is the size, in bytes, of the LogEntry buffer that is used
 // to during archival. This should be greater than the maximum LogEntry size.
 const storageBufferSize = types.MaxLogEntryDataSize * 64
 
 // ArchiveTask processes and executes a single log stream archive task.
-func (a *Archivist) ArchiveTask(c context.Context, desc []byte) error {
+//
+// It returns true on success (delete the task) and false on failure (don't
+// delete the task). The return value of true should only be used if the task
+// is truely complete, as deleting a named task in the task queue will cause it

[Vadim Sh., 2016/04/07 01:21:32]

typo: truly

also you don't use named task (anymore?), so the comment about 10 days is no
longer relevant

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

Done.

+// to not be reschedulable for a long period of time (10 days).
+//
+// If the supplied Context is Done, operation may terminate before completion,
+// returning the Context's error.
+func (a *Archivist) ArchiveTask(c context.Context, desc []byte, age time.Duration) bool {
         var task logdog.ArchiveTask
         if err := proto.Unmarshal(desc, &task); err != nil {
                 log.WithError(err).Errorf(c, "Failed to decode archive task.")
-                return err
+                return false
         }
-        return a.Archive(c, &task)
+        if err := a.Archive(c, &task, age); err != nil {
+                log.WithError(err).Errorf(c, "Archival task failed.")
+                return false
+        }
+        return true
 }
 
-// Archive archives a single log stream. If unsuccessful, an error is returned.
+// Archive processes and executes a single log stream archive task.
 //
-// This error may be wrapped in errors.Transient if it is believed to have been
-// caused by a transient failure.
+// It returns an error if the archival isn't complete and confirmed by the
+// Coordinator.
 //
 // If the supplied Context is Done, operation may terminate before completion,
 // returning the Context's error.
-func (a *Archivist) Archive(c context.Context, t *logdog.ArchiveTask) error {
+func (a *Archivist) Archive(c context.Context, t *logdog.ArchiveTask, age time.Duration) error {
+        complete := (age <= t.CompletePeriod.Duration())
+
         // Load the log stream's current state. If it is already archived, we will
         // return an immediate success.
         ls, err := a.Service.LoadStream(c, &logdog.LoadStreamRequest{
                 Path: t.Path,
                 Desc: true,
         })
         switch {
         case err != nil:
-                log.WithError(err).Errorf(c, "Failed to load log stream.")
-                return err
+                return fmt.Errorf("failed to load log stream: %v", err)
         case ls.State == nil:
-                return errors.New("missing state")
+                return errors.New("log stream did not include state")
         case ls.State.ProtoVersion != logpb.Version:
                 log.Fields{
                         "protoVersion":    ls.State.ProtoVersion,
                         "expectedVersion": logpb.Version,
                 }.Errorf(c, "Unsupported log stream protobuf version.")
-                return errors.New("unsupported protobuf version")
+                return errors.New("unsupported log stream protobuf version")
         case ls.Desc == nil:
-                return errors.New("missing descriptor")
+                return errors.New("log stream did not include a descriptor")
 
         case ls.State.Purged:
                 log.Warningf(c, "Log stream is purged.")
                 return nil
         case ls.State.Archived:
                 log.Infof(c, "Log stream is already archived.")
                 return nil
+
+        case complete && ls.State.TerminalIndex < 0:
+                return errors.WrapTransient(errors.New("cannot archive complete stream with no terminal index"))
         }
 
         // Deserialize and validate the descriptor protobuf.
         var desc logpb.LogStreamDescriptor
         if err := proto.Unmarshal(ls.Desc, &desc); err != nil {
                 log.Fields{
                         log.ErrorKey:   err,
                         "protoVersion": ls.State.ProtoVersion,
                 }.Errorf(c, "Failed to unmarshal descriptor protobuf.")
                 return err
         }
 
         task := &archiveTask{
                 Archivist:   a,
                 ArchiveTask: t,
+                complete:    complete,
                 ls:          ls,
                 desc:        &desc,
         }
-        if err := task.archive(c); err != nil {
-                log.WithError(err).Errorf(c, "Failed to perform archival operation.")
-                return err
+        aerr := task.archive(c)
+        if aerr != nil {
+                // If this is a transient error, fail immediately with it. Use
+                // isTransientError because err can be an errors.MultiError.
+                //
+                // Returning a transient error here will cause task queue to reschedule.
+                if isTransientError(aerr) {
+                        log.WithError(aerr).Errorf(c, "TRANSIENT error during archival operation.")
+                        return aerr
+                }
+
+                // This is a non-transient error, so we will report it as an error to the
+                // service. It may accept the archival (success), in which case we'll delete
+                // our task, or respond with an error code (Aborted) indicating that this
+                // archival run should be retried and that we should not delete the task.
+                log.WithError(aerr).Errorf(c, "Archival failed with non-transient error.")
+                task.ar.Error = true
         }
+
         log.Fields{
                 "streamURL":     task.ar.StreamUrl,
                 "indexURL":      task.ar.IndexUrl,
                 "dataURL":       task.ar.DataUrl,
                 "terminalIndex": task.ar.TerminalIndex,
-                "complete":      task.ar.Complete,
-        }.Debugf(c, "Finished archive construction.")
+                "logEntryCount": task.ar.LogEntryCount,
+                "hadError":      task.ar.Error,
+                "complete":      task.ar.Complete(),
+        }.Debugf(c, "Finished archive construction; reporting archive state.")
 
         if _, err := a.Service.ArchiveStream(c, &task.ar); err != nil {
-                log.WithError(err).Errorf(c, "Failed to mark log stream as archived.")
-                return err
+                // All service errors will be reported and returned as transient. We will
+                // NOT delete this archival task unless the service has explicitly
+                // instructed us to via success code.
+                log.WithError(err).Errorf(c, "Failed to report archive state.")
+                return errors.WrapTransient(errors.New("failed to report archive state"))
         }
         return nil
 }
 
 // archiveTask is the set of parameters for a single archival.
 type archiveTask struct {
         *Archivist
         *logdog.ArchiveTask
 
+        // complete, if true, will cause incomplete streams to return transient
+        // errors.
+        complete bool
         // 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 41 matching lines @@
         defer deleteOnFail(indexO)
         defer deleteOnFail(dataO)
 
         // Close our GS object managers on exit. If any of them fail to close, marh
         // the archival as a failure.
         closeOM := func(o *gsObject) {
                 if o == nil {
                         return
                 }
                 if ierr := o.Close(); ierr != nil {
-                        err = ierr
+                        err = joinArchiveErrors(err, ierr)
                 }
         }
         defer closeOM(streamO)
         defer closeOM(indexO)
         defer closeOM(dataO)
 
         // Read our log entries from intermediate storage.
         ss := storageSource{
                 Context:       c,
                 st:            t.Storage,
                 path:          types.StreamPath(t.Path),
-                contiguous:    t.Complete,
+                contiguous:    t.complete,
                 terminalIndex: types.MessageIndex(t.ls.State.TerminalIndex),
                 lastIndex:     -1,
         }
 
         m := archive.Manifest{
                 Desc:             t.desc,
                 Source:           &ss,
                 LogWriter:        streamO,
                 IndexWriter:      indexO,
                 DataWriter:       dataO,
                 StreamIndexRange: t.StreamIndexRange,
                 PrefixIndexRange: t.PrefixIndexRange,
                 ByteRange:        t.ByteRange,
 
                 Logger: log.Get(c),
         }
-        err = archive.Archive(m)
-        if err != nil {
+        if err = archive.Archive(m); err != nil {
                 log.WithError(err).Errorf(c, "Failed to archive log stream.")
                 return
         }
 
-        t.ar.TerminalIndex = int64(ss.lastIndex)
         if tidx := t.ls.State.TerminalIndex; tidx != t.ar.TerminalIndex {
-                // Fail, if we were requested to archive only the complete log.
-                if t.Complete {
-                        log.Fields{
-                                "terminalIndex": tidx,
-                                "lastIndex":     t.ar.TerminalIndex,
-                        }.Errorf(c, "Log stream archival stopped prior to terminal index.")
-                        return errors.New("stream finished short of terminal index")
-                }
+                // Fail if we were requested to archive only the complete log. We consider
+                // this a transient error with the expectation that the missing entries will
+                // show up in future retries.
+                switch {
+                case t.complete && ss.hasMissingEntries:
+                        log.Errorf(c, "Log stream has missing entries, but completeness is required.")
+                        return errors.WrapTransient(errors.New("stream has missing entries"))
 
-                if t.ar.TerminalIndex < 0 {
+                case ss.logEntryCount == 0:
                         // If our last log index was <0, then no logs were archived.
                         log.Warningf(c, "No log entries were archived.")
-                } else {
+
+                default:
                         // Update our terminal index.
                         log.Fields{
-                                "from": tidx,
-                                "to":   t.ar.TerminalIndex,
-                        }.Infof(c, "Updated log stream terminal index.")
+                                "terminalIndex":     ss.lastIndex,
+                                "logEntryCount":     ss.logEntryCount,
+                                "hasMissingEntries": ss.hasMissingEntries,
+                        }.Debugf(c, "Finished archiving log stream.")
                 }
         }
 
         // Update our state with archival results.
-        t.ar.Path = t.Path
+        t.ar.TerminalIndex = int64(ss.lastIndex)
+        t.ar.LogEntryCount = ss.logEntryCount
         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{
                 gs:     t.GSClient,
                 bucket: p.Bucket(),
                 path:   p.Filename(),
         }
@@ skipping 49 matching lines @@
 type storageSource struct {
         context.Context
 
         st            storage.Storage    // the storage instance to read from
         path          types.StreamPath   // the path of the log stream
         contiguous    bool               // if true, enforce contiguous entries
         terminalIndex types.MessageIndex // if >= 0, discard logs beyond this
 
         buf               []*logpb.LogEntry
         lastIndex         types.MessageIndex
+        logEntryCount     int64
         hasMissingEntries bool // true if some log entries were missing.
 }
 
 func (s *storageSource) bufferEntries(start types.MessageIndex) error {
         bytes := 0
 
         req := storage.GetRequest{
                 Path:  s.path,
                 Index: start,
         }
@@ skipping 13 matching lines @@
                 return bytes < storageBufferSize
         })
 }
 
 func (s *storageSource) NextLogEntry() (*logpb.LogEntry, error) {
         if len(s.buf) == 0 {
                 s.buf = s.buf[:0]
                 if err := s.bufferEntries(s.lastIndex + 1); err != nil {
                         if err == storage.ErrDoesNotExist {
                                 log.Warningf(s, "Archive target stream does not exist in intermediate storage.")
+                                if s.terminalIndex >= 0 {
+                                        s.hasMissingEntries = true
+                                }
                                 return nil, archive.ErrEndOfStream
                         }
 
                         log.WithError(err).Errorf(s, "Failed to retrieve log stream from storage.")
                         return nil, err
                 }
         }
 
+        // If we have no more buffered entries, we have exhausted our log stream.
         if len(s.buf) == 0 {
-                log.Fields{
-                        "lastIndex": s.lastIndex,
-                }.Debugf(s, "Encountered end of stream.")
+                // If we have a terminal index, but we didn't actually emit that index,
+                // mark that we have missing entries.
+                if s.terminalIndex >= 0 && s.lastIndex != s.terminalIndex {
+                        log.Fields{
+                                "terminalIndex": s.terminalIndex,
+                                "lastIndex":     s.lastIndex,
+                        }.Warningf(s, "Log stream stopped before terminal index.")
+                        s.hasMissingEntries = true
+                } else {
+                        log.Fields{
+                                "lastIndex": s.lastIndex,
+                        }.Debugf(s, "Encountered end of stream.")
+                }
+
                 return nil, archive.ErrEndOfStream
         }
 
+        // Pop the next log entry and advance the stream.
         var le *logpb.LogEntry
         le, s.buf = s.buf[0], s.buf[1:]
 
         // If we're enforcing a contiguous log stream, error if this LogEntry is not
         // contiguous.
         sidx := types.MessageIndex(le.StreamIndex)
         nidx := (s.lastIndex + 1)
         if sidx != nidx {
                 s.hasMissingEntries = true
-
-                if s.contiguous {
-                        log.Fields{
-                                "index":     sidx,
-                                "nextIndex": nidx,
-                        }.Errorf(s, "Non-contiguous log stream while enforcing.")
-                        return nil, errors.New("non-contiguous log stream")
-                }
+        }
+        if s.contiguous && s.hasMissingEntries {
+                log.Fields{
+                        "index":     sidx,
+                        "nextIndex": nidx,
+                }.Warningf(s, "Non-contiguous log stream while enforcing.")
+                return nil, archive.ErrEndOfStream
         }
 
         // If we're enforcing a maximum terminal index, return end of stream if this
         // LogEntry exceeds that index.
         if s.terminalIndex >= 0 && sidx > s.terminalIndex {
                 log.Fields{
                         "index":         sidx,
                         "terminalIndex": s.terminalIndex,
                 }.Warningf(s, "Discarding log entries beyond expected terminal index.")
                 return nil, archive.ErrEndOfStream
         }
 
         s.lastIndex = sidx
+        s.logEntryCount++
         return le, nil
 }
+
+func joinArchiveErrors(err error, add error) error {
+        switch {
+        case err == nil:
+                return add
+        case add == nil:
+                return err
+
+        default:
+                merr, ok := err.(errors.MultiError)
+                if !ok {
+                        merr = errors.MultiError{err, nil}[:1]
+                }
+                merr = append(merr, add)
+                return merr
+        }
+}
+
+func isTransientError(err error) bool {
+        if errors.IsTransient(err) {
+                return true
+        }
+
+        if merr, ok := err.(errors.MultiError); ok {
+                for _, ierr := range merr {
+                        if isTransientError(ierr) {
+                                return true
+                        }
+                }
+        }
+        return false
+}