LogDog: Update to archival V2.

server/internal/logdog/archivist/archivist_test.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 (
-        "errors"
         "fmt"
         "strings"
         "sync"
         "testing"
         "time"
 
         "github.com/golang/protobuf/proto"
         "github.com/luci/luci-go/common/api/logdog_coordinator/services/v1"
         "github.com/luci/luci-go/common/clock/testclock"
+        "github.com/luci/luci-go/common/errors"
         "github.com/luci/luci-go/common/gcloud/gs"
         "github.com/luci/luci-go/common/logdog/types"
         "github.com/luci/luci-go/common/proto/google"
         "github.com/luci/luci-go/common/proto/logdog/logpb"
         "github.com/luci/luci-go/server/logdog/storage"
         "github.com/luci/luci-go/server/logdog/storage/memory"
         "golang.org/x/net/context"
         "google.golang.org/grpc"
 
         . "github.com/luci/luci-go/common/testing/assertions"
@@ skipping 190 matching lines @@
                                 Path:          string(desc.Path()),
                                 ProtoVersion:  logpb.Version,
                                 TerminalIndex: -1,
                                 Archived:      false,
                                 Purged:        false,
                         },
                         Desc: descBytes,
                 }
 
                 var archiveRequest *logdog.ArchiveStreamRequest
+                var archiveStreamErr error
                 sc := testServicesClient{
                         lsCallback: func(req *logdog.LoadStreamRequest) (*logdog.LoadStreamResponse, error) {
                                 return &stream, nil
                         },
                         asCallback: func(req *logdog.ArchiveStreamRequest) error {
                                 archiveRequest = req
-                                return nil
+                                return archiveStreamErr
                         },
                 }
 
                 ar := Archivist{
                         Service:  &sc,
                         Storage:  &st,
                         GSClient: &gsc,
                         GSBase:   gs.Path("gs://archive-test/path/to/archive/"), // Extra slashes to test concatenation.
                 }
 
+                expired := 10 * time.Minute
                 task := logdog.ArchiveTask{
-                        Path:     stream.State.Path,
-                        Complete: true,
+                        Path:           stream.State.Path,
+                        CompletePeriod: google.NewDuration(expired),
                 }
+                expired++ // This represents a time PAST CompletePeriod.
 
                 gsURL := func(p string) string {
                         return fmt.Sprintf("gs://archive-test/path/to/archive/%s/%s", desc.Path(), p)
                 }
 
                 // hasStreams can be called to check that the retained archiveRequest had
                 // data sizes for the named archive stream types.
                 //
                 // After checking, the values are set to zero. This allows us to use
                 // ShouldEqual without hardcoding specific archival sizes into the results.
                 hasStreams := func(log, index, data bool) bool {
                         So(archiveRequest, ShouldNotBeNil)
                         if (log && archiveRequest.StreamSize <= 0) ||
                                 (index && archiveRequest.IndexSize <= 0) ||
                                 (data && archiveRequest.DataSize <= 0) {
                                 return false
                         }
 
                         archiveRequest.StreamSize = 0
                         archiveRequest.IndexSize = 0
                         archiveRequest.DataSize = 0
                         return true
                 }
 
                 Convey(`Will fail to archive if the specified stream state could not be loaded.`, func() {
                         sc.lsCallback = func(*logdog.LoadStreamRequest) (*logdog.LoadStreamResponse, error) {
                                 return nil, errors.New("does not exist")
                         }
-                        So(ar.Archive(c, &task), ShouldErrLike, "does not exist")
+                        So(ar.Archive(c, &task, 0), ShouldErrLike, "does not exist")
                 })
 
                 Convey(`Will refrain from archiving if the stream is already archived.`, func() {
                         stream.State.Archived = true
-                        So(ar.Archive(c, &task), ShouldBeNil)
+                        So(ar.Archive(c, &task, 0), ShouldBeNil)
                         So(archiveRequest, ShouldBeNil)
                 })
 
                 Convey(`Will refrain from archiving if the stream is purged.`, func() {
                         stream.State.Purged = true
-                        So(ar.Archive(c, &task), ShouldBeNil)
+                        So(ar.Archive(c, &task, 0), ShouldBeNil)
                         So(archiveRequest, ShouldBeNil)
                 })
 
                 // Weird case: the log has been marked for archival, has not been
-                // terminated, and is within its completeness delay. This task will not
-                // have been dispatched by our archive cron, but let's assert that it
-                // behaves correctly regardless.
-                Convey(`Will succeed if the log stream had no entries and no terminal index.`, func() {
-                        So(ar.Archive(c, &task), ShouldBeNil)
-
-                        So(hasStreams(true, true, false), ShouldBeTrue)
-                        So(archiveRequest, ShouldResemble, &logdog.ArchiveStreamRequest{
-                                Path:          task.Path,
-                                Complete:      true,
-                                TerminalIndex: -1,
-
-                                StreamUrl: gsURL("logstream.entries"),
-                                IndexUrl:  gsURL("logstream.index"),
-                                DataUrl:   gsURL("data.bin"),
-                        })
+                // terminated, and is within its completeness delay. This task should not
+                // have been dispatched by our expired archive cron, but let's assert that
+                // it behaves correctly regardless.
+                Convey(`Will refuse to archive a complete steram with no terminal index.`, func() {
+                        err := ar.Archive(c, &task, 0)
+                        So(err, ShouldErrLike, "cannot archive complete stream with no terminal index")
+                        So(errors.IsTransient(err), ShouldBeTrue)
                 })
 
                 Convey(`With terminal index "3"`, func() {
                         stream.State.TerminalIndex = 3
 
-                        Convey(`Will fail if the log stream had a terminal index and no entries.`, func() {
-                                So(ar.Archive(c, &task), ShouldErrLike, "stream finished short of terminal index")
+                        Convey(`Will fail transiently if the log stream had no entries.`, func() {
+                                err := ar.Archive(c, &task, 0)
+                                So(err, ShouldErrLike, "stream has missing entries")
+                                So(errors.IsTransient(err), ShouldBeTrue)
                         })
 
                         Convey(`Will fail to archive {0, 1, 2, 4} (incomplete).`, func() {
                                 addTestEntry(0, 1, 2, 4)
-                                So(ar.Archive(c, &task), ShouldErrLike, "non-contiguous log stream")
+                                So(ar.Archive(c, &task, 0), ShouldErrLike, "stream has missing entries")
                         })
 
                         Convey(`Will successfully archive {0, 1, 2, 3, 4}, stopping at the terminal index.`, func() {
                                 addTestEntry(0, 1, 2, 3, 4)
-                                So(ar.Archive(c, &task), ShouldBeNil)
+                                So(ar.Archive(c, &task, 0), ShouldBeNil)
 
                                 So(hasStreams(true, true, true), ShouldBeTrue)
                                 So(archiveRequest, ShouldResemble, &logdog.ArchiveStreamRequest{
                                         Path:          task.Path,
-                                        Complete:      true,
+                                        LogEntryCount: 4,
                                         TerminalIndex: 3,
 
                                         StreamUrl: gsURL("logstream.entries"),
                                         IndexUrl:  gsURL("logstream.index"),
                                         DataUrl:   gsURL("data.bin"),
                                 })
                         })
+
+                        Convey(`When a transient archival error occurs, will return it.`, func() {
+                                gsc.newWriterErr = func(*testGSWriter) error { return errors.WrapTransient(errors.New("test error")) }
+
+                                err := ar.Archive(c, &task, 0)
+                                So(err, ShouldErrLike, "test error")
+                                So(errors.IsTransient(err), ShouldBeTrue)
+                        })
+
+                        Convey(`When a non-transient archival error occurs, will report it`, func() {
+                                gsc.newWriterErr = func(*testGSWriter) error { return errors.New("test error") }
+
+                                Convey(`If remote returns an error, forwards a transient error.`, func() {
+                                        archiveStreamErr = errors.New("test error")
+
+                                        err := ar.Archive(c, &task, 0)
+                                        So(err, ShouldErrLike, "failed to report archive state")
+                                        So(errors.IsTransient(err), ShouldBeTrue)
+
+                                        So(archiveRequest, ShouldResemble, &logdog.ArchiveStreamRequest{
+                                                Path:          task.Path,
+                                                Error:         true,
+                                                TerminalIndex: -1,
+                                        })
+                                })
+
+                                Convey(`If remote returns success, returns nil.`, func() {
+                                        So(ar.Archive(c, &task, 0), ShouldBeNil)
+                                        So(archiveRequest, ShouldResemble, &logdog.ArchiveStreamRequest{
+                                                Path:          task.Path,
+                                                Error:         true,
+                                                TerminalIndex: -1,
+                                        })
+                                })
+                        })
                 })
 
                 Convey(`When not enforcing stream completeness`, func() {
-                        task.Complete = false
 
                         Convey(`With no terminal index`, func() {
                                 Convey(`Will successfully archive if there are no entries.`, func() {
-                                        So(ar.Archive(c, &task), ShouldBeNil)
+                                        So(ar.Archive(c, &task, expired), ShouldBeNil)
 
                                         So(hasStreams(true, true, false), ShouldBeTrue)
                                         So(archiveRequest, ShouldResemble, &logdog.ArchiveStreamRequest{
                                                 Path:          task.Path,
-                                                Complete:      true,
+                                                LogEntryCount: 0,
                                                 TerminalIndex: -1,
 
                                                 StreamUrl: gsURL("logstream.entries"),
                                                 IndexUrl:  gsURL("logstream.index"),
                                                 DataUrl:   gsURL("data.bin"),
                                         })
                                 })
 
                                 Convey(`With {0, 1, 2, 4} (incomplete) will archive the stream and update its terminal index.`, func() {
                                         addTestEntry(0, 1, 2, 4)
-                                        So(ar.Archive(c, &task), ShouldBeNil)
+                                        So(ar.Archive(c, &task, expired), ShouldBeNil)
 
                                         So(hasStreams(true, true, true), ShouldBeTrue)
                                         So(archiveRequest, ShouldResemble, &logdog.ArchiveStreamRequest{
                                                 Path:          task.Path,
-                                                Complete:      false,
+                                                LogEntryCount: 4,
                                                 TerminalIndex: 4,
 
                                                 StreamUrl: gsURL("logstream.entries"),
                                                 IndexUrl:  gsURL("logstream.index"),
                                                 DataUrl:   gsURL("data.bin"),
                                         })
                                 })
                         })
 
                         Convey(`With terminal index 3`, func() {
                                 stream.State.TerminalIndex = 3
 
                                 Convey(`Will successfully archive if there are no entries.`, func() {
-                                        So(ar.Archive(c, &task), ShouldBeNil)
+                                        So(ar.Archive(c, &task, expired), ShouldBeNil)
 
                                         So(hasStreams(true, true, false), ShouldBeTrue)
                                         So(archiveRequest, ShouldResemble, &logdog.ArchiveStreamRequest{
                                                 Path:          task.Path,
-                                                Complete:      true,
+                                                LogEntryCount: 0,
                                                 TerminalIndex: -1,
 
                                                 StreamUrl: gsURL("logstream.entries"),
                                                 IndexUrl:  gsURL("logstream.index"),
                                                 DataUrl:   gsURL("data.bin"),
                                         })
                                 })
 
                                 Convey(`With {0, 1, 2, 4} (incomplete) will archive the stream and update its terminal index to 2.`, func() {
                                         addTestEntry(0, 1, 2, 4)
-                                        So(ar.Archive(c, &task), ShouldBeNil)
+                                        So(ar.Archive(c, &task, expired), ShouldBeNil)
 
                                         So(hasStreams(true, true, true), ShouldBeTrue)
                                         So(archiveRequest, ShouldResemble, &logdog.ArchiveStreamRequest{
                                                 Path:          task.Path,
-                                                Complete:      false,
+                                                LogEntryCount: 3,
                                                 TerminalIndex: 2,
 
                                                 StreamUrl: gsURL("logstream.entries"),
                                                 IndexUrl:  gsURL("logstream.index"),
                                                 DataUrl:   gsURL("data.bin"),
                                         })
                                 })
                         })
                 })
 
                 // Simulate failures during the various stream generation operations.
                 Convey(`Stream generation failures`, func() {
                         stream.State.TerminalIndex = 3
                         addTestEntry(0, 1, 2, 3)
 
                         for _, failName := range []string{"/logstream.entries", "/logstream.index", "/data.bin"} {
                                 for _, testCase := range []struct {
-                                        name  string
-                                        setup func()
+                                        name      string
+                                        transient bool
+                                        setup     func()
                                 }{
-                                        {"delete failure", func() {
+                                        {"delete failure", false, func() {
                                                 gsc.deleteErr = func(b, p string) error {
                                                         if strings.HasSuffix(p, failName) {
                                                                 return errors.New("test error")
                                                         }
                                                         return nil
                                                 }
                                         }},
 
-                                        {"writer create failure", func() {
+                                        {"writer create failure", false, func() {
                                                 gsc.newWriterErr = func(w *testGSWriter) error {
                                                         if strings.HasSuffix(w.path, failName) {
                                                                 return errors.New("test error")
                                                         }
                                                         return nil
                                                 }
                                         }},
 
-                                        {"write failure", func() {
+                                        {"write failure", false, func() {
                                                 gsc.newWriterErr = func(w *testGSWriter) error {
                                                         if strings.HasSuffix(w.path, failName) {
                                                                 w.writeErr = errors.New("test error")
                                                         }
                                                         return nil
                                                 }
                                         }},
 
-                                        {"close failure", func() {
+                                        {"close failure", false, func() {
                                                 gsc.newWriterErr = func(w *testGSWriter) error {
                                                         if strings.HasSuffix(w.path, failName) {
                                                                 w.closeErr = errors.New("test error")
                                                         }
                                                         return nil
                                                 }
                                         }},
 
-                                        {"delete on fail failure (double-failure)", func() {
+                                        {"delete on fail failure (double-failure)", false, func() {
                                                 failed := false
 
                                                 // Simulate a write failure. This is the error that will actually
                                                 // be returned.
                                                 gsc.newWriterErr = func(w *testGSWriter) error {
                                                         if strings.HasSuffix(w.path, failName) {
                                                                 w.writeErr = errors.New("test error")
                                                         }
                                                         return nil
                                                 }
 
                                                 // This will trigger twice per stream, once on create and once on
                                                 // cleanup after the write fails. We want to return an error in
                                                 // the latter case.
                                                 gsc.deleteErr = func(b, p string) error {
                                                         if strings.HasSuffix(p, failName) {
                                                                 if failed {
                                                                         return errors.New("other error")
                                                                 }
 
                                                                 // First delete (on create).
                                                                 failed = true
                                                         }
                                                         return nil
                                                 }
                                         }},
                                 } {
-                                        Convey(fmt.Sprintf(`Can handle %s for %s`, testCase.name, failName), func() {
+                                        Convey(fmt.Sprintf(`Can handle %s for %s, and will not archive.`, testCase.name, failName), func() {
                                                 testCase.setup()
-                                                So(ar.Archive(c, &task), ShouldErrLike, "test error")
+
+                                                err := ar.Archive(c, &task, 0)
+                                                if testCase.transient {
+                                                        // Transient errors are directly returned without consulting the
+                                                        // Coordinator service.
+                                                        So(err, ShouldErrLike, "test error")
+                                                        So(errors.IsTransient(err), ShouldBeTrue)
+                                                        So(archiveRequest, ShouldBeNil)
+                                                } else {
+                                                        So(err, ShouldBeNil)
+                                                        So(archiveRequest, ShouldNotBeNil)
+                                                        So(archiveRequest.Error, ShouldBeTrue)
+                                                }
                                         })
                                 }
                         }
                 })
         })
 }