LogDog: Update to archival V2.

appengine/logdog/coordinator/endpoints/services/archiveStream_test.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 services
 
 import (
         "errors"
         "testing"
 
         "github.com/luci/gae/filter/featureBreaker"
         "github.com/luci/gae/impl/memory"
         ds "github.com/luci/gae/service/datastore"
         "github.com/luci/luci-go/appengine/logdog/coordinator"
         ct "github.com/luci/luci-go/appengine/logdog/coordinator/coordinatorTest"
         "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/proto/logdog/svcconfig"
         "github.com/luci/luci-go/server/auth"
         "github.com/luci/luci-go/server/auth/authtest"
         "golang.org/x/net/context"
 
         . "github.com/luci/luci-go/common/testing/assertions"
         . "github.com/smartystreets/goconvey/convey"
 )
 
 func TestArchiveStream(t *testing.T) {
         t.Parallel()
 
         Convey(`With a testing configuration`, t, func() {
-                c := memory.Use(context.Background())
+                c, tc := testclock.UseTime(context.Background(), testclock.TestTimeUTC)
+                c = memory.Use(c)
                 be := Server{}
 
-                c = ct.UseConfig(c, &svcconfig.Coordinator{
+                now := ds.RoundTime(tc.Now().UTC())
+                ccfg := svcconfig.Coordinator{
                         ServiceAuthGroup: "test-services",
-                })
+                        ArchiveRetries:   3,
+                }
+                c = ct.UseConfig(c, &ccfg)
+
                 fs := authtest.FakeState{}
                 c = auth.WithState(c, &fs)
 
-                // Register a testing log stream (not archived).
+                // Register a testing log stream with an archive tasked.
                 ls := ct.TestLogStream(c, ct.TestLogStreamDescriptor(c, "foo"))
-                if err := ls.Put(ds.Get(c)); err != nil {
+                ls.State = coordinator.LSArchiveTasked
+                if err := ds.Get(c).Put(ls); err != nil {
                         panic(err)
                 }
 
                 req := &logdog.ArchiveStreamRequest{
                         Path:          string(ls.Path()),
-                        Complete:      true,
                         TerminalIndex: 13,
+                        LogEntryCount: 14,
                         StreamUrl:     "gs://fake.stream",
                         StreamSize:    10,
                         IndexUrl:      "gs://fake.index",
                         IndexSize:     20,
                         DataUrl:       "gs://fake.data",
                         DataSize:      30,
                 }
 
                 Convey(`Returns Forbidden error if not a service.`, func() {
                         _, err := be.ArchiveStream(c, req)
                         So(err, ShouldBeRPCPermissionDenied)
                 })
 
                 Convey(`When logged in as a service`, func() {
                         fs.IdentityGroups = []string{"test-services"}
 
                         Convey(`Will mark the stream as archived.`, func() {
                                 _, err := be.ArchiveStream(c, req)
                                 So(err, ShouldBeNil)
 
                                 So(ds.Get(c).Get(ls), ShouldBeNil)
+                                So(ls.Terminated(), ShouldBeTrue)
                                 So(ls.Archived(), ShouldBeTrue)
-                                So(ls.ArchiveWhole, ShouldBeTrue)
+                                So(ls.ArchiveComplete(), ShouldBeTrue)
+
+                                So(ls.State, ShouldEqual, coordinator.LSArchived)
+                                So(ls.TerminatedTime, ShouldResemble, now)
+                                So(ls.ArchivedTime, ShouldResemble, now)
                                 So(ls.TerminalIndex, ShouldEqual, 13)
+                                So(ls.ArchiveLogEntryCount, ShouldEqual, 14)
                                 So(ls.ArchiveStreamURL, ShouldEqual, "gs://fake.stream")
                                 So(ls.ArchiveStreamSize, ShouldEqual, 10)
                                 So(ls.ArchiveIndexURL, ShouldEqual, "gs://fake.index")
+                                So(ls.ArchiveIndexSize, ShouldEqual, 20)
+                                So(ls.ArchiveDataURL, ShouldEqual, "gs://fake.data")
+                                So(ls.ArchiveDataSize, ShouldEqual, 30)
+                        })
+
+                        Convey(`Will mark the stream as partially archived if not complete.`, func() {
+                                req.LogEntryCount = 13
+
+                                _, err := be.ArchiveStream(c, req)
+                                So(err, ShouldBeNil)
+
+                                So(ds.Get(c).Get(ls), ShouldBeNil)
+                                So(ls.Terminated(), ShouldBeTrue)
+                                So(ls.Archived(), ShouldBeTrue)
+                                So(ls.ArchiveComplete(), ShouldBeFalse)
+
+                                So(ls.State, ShouldEqual, coordinator.LSArchived)
+                                So(ls.TerminatedTime, ShouldResemble, now)
+                                So(ls.ArchivedTime, ShouldResemble, now)
+                                So(ls.TerminalIndex, ShouldEqual, 13)
+                                So(ls.ArchiveLogEntryCount, ShouldEqual, 13)
+                                So(ls.ArchiveStreamURL, ShouldEqual, "gs://fake.stream")
+                                So(ls.ArchiveStreamSize, ShouldEqual, 10)
+                                So(ls.ArchiveIndexURL, ShouldEqual, "gs://fake.index")
                                 So(ls.ArchiveIndexSize, ShouldEqual, 20)
                                 So(ls.ArchiveDataURL, ShouldEqual, "gs://fake.data")
                                 So(ls.ArchiveDataSize, ShouldEqual, 30)
                         })
 
                         Convey(`Will refuse to process an invalid stream path.`, func() {
                                 req.Path = "!!!invalid!!!"
                                 _, err := be.ArchiveStream(c, req)
                                 So(err, ShouldBeRPCInvalidArgument, "invalid log stream path")
                         })
 
                         Convey(`If index URL is missing, will refuse to mark the stream archived.`, func() {
                                 req.IndexUrl = ""
 
                                 _, err := be.ArchiveStream(c, req)
                                 So(err, ShouldBeRPCInvalidArgument)
                         })
 
                         Convey(`If stream URL is missing, will refuse to mark the stream archived.`, func() {
                                 req.StreamUrl = ""
 
                                 _, err := be.ArchiveStream(c, req)
                                 So(err, ShouldBeRPCInvalidArgument)
                         })
 
                         Convey(`If stream is already archived, will not update and return success.`, func() {
                                 ls.State = coordinator.LSArchived
                                 ls.TerminalIndex = 1337
+                                ls.ArchiveLogEntryCount = 42
+                                ls.ArchivedTime = now
+                                ls.TerminatedTime = now
+                                So(ds.Get(c).Put(ls), ShouldBeNil)
+                                So(ls.Terminated(), ShouldBeTrue)
                                 So(ls.Archived(), ShouldBeTrue)
-                                So(ls.Put(ds.Get(c)), ShouldBeNil)
 
                                 _, err := be.ArchiveStream(c, req)
                                 So(err, ShouldBeNil)
 
                                 ls.TerminalIndex = -1 // To make sure it reloaded.
                                 So(ds.Get(c).Get(ls), ShouldBeNil)
+                                So(ls.Terminated(), ShouldBeTrue)
                                 So(ls.Archived(), ShouldBeTrue)
+
+                                So(ls.State, ShouldEqual, coordinator.LSArchived)
                                 So(ls.TerminalIndex, ShouldEqual, 1337)
+                                So(ls.ArchiveLogEntryCount, ShouldEqual, 42)
+                        })
+
+                        Convey(`If the archive has failed`, func() {
+                                req.Error = true
+
+                                Convey(`If the stream is below error threshold, will increment error count and return Aborted.`, func() {
+                                        ls.ArchiveErrors = 0
+                                        So(ds.Get(c).Put(ls), ShouldBeNil)
+
+                                        _, err := be.ArchiveStream(c, req)
+                                        So(err, ShouldBeRPCAborted)
+
+                                        So(ds.Get(c).Get(ls), ShouldBeNil)
+                                        So(ls.Archived(), ShouldBeFalse)
+
+                                        So(ls.State, ShouldEqual, coordinator.LSArchiveTasked)
+                                        So(ls.ArchiveErrors, ShouldEqual, 1)
+                                })
+
+                                Convey(`If the stream is above error threshold, will succeed with empty archival.`, func() {
+                                        ls.ArchiveErrors = int(ccfg.ArchiveRetries)
+                                        So(ds.Get(c).Put(ls), ShouldBeNil)
+
+                                        _, err := be.ArchiveStream(c, req)
+                                        So(err, ShouldBeNil)
+                                        So(ds.Get(c).Get(ls), ShouldBeNil)
+                                        So(ls.Archived(), ShouldBeTrue)
+
+                                        So(ls.State, ShouldEqual, coordinator.LSArchived)
+                                        So(ls.ArchiveErrors, ShouldEqual, ccfg.ArchiveRetries+1)
+                                        So(ls.TerminalIndex, ShouldEqual, -1)
+                                        So(ls.ArchiveLogEntryCount, ShouldEqual, 0)
+                                })
                         })
 
                         Convey(`When datastore Get fails, returns internal error.`, func() {
                                 c, fb := featureBreaker.FilterRDS(c, nil)
                                 fb.BreakFeatures(errors.New("test error"), "GetMulti")
 
                                 _, err := be.ArchiveStream(c, req)
                                 So(err, ShouldBeRPCInternal)
                         })
 
                         Convey(`When datastore Put fails, returns internal error.`, func() {
                                 c, fb := featureBreaker.FilterRDS(c, nil)
                                 fb.BreakFeatures(errors.New("test error"), "PutMulti")
 
                                 _, err := be.ArchiveStream(c, req)
                                 So(err, ShouldBeRPCInternal)
                         })
                 })
         })
 }