LogDog: Update to archival V2.

appengine/logdog/coordinator/endpoints/logs/get_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 logs
 
 import (
         "bytes"
         "errors"
         "fmt"
         "io"
         "testing"
         "time"
 
         "github.com/golang/protobuf/proto"
         "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/logs/v1"
         "github.com/luci/luci-go/common/clock/testclock"
         "github.com/luci/luci-go/common/gcloud/gs"
         "github.com/luci/luci-go/common/iotools"
         "github.com/luci/luci-go/common/logdog/types"
         "github.com/luci/luci-go/common/proto/logdog/logpb"
-        "github.com/luci/luci-go/common/proto/logdog/svcconfig"
         "github.com/luci/luci-go/common/recordio"
         "github.com/luci/luci-go/server/auth"
         "github.com/luci/luci-go/server/auth/authtest"
         "github.com/luci/luci-go/server/logdog/archive"
         "github.com/luci/luci-go/server/logdog/storage"
         memoryStorage "github.com/luci/luci-go/server/logdog/storage/memory"
         "golang.org/x/net/context"
 
         . "github.com/luci/luci-go/common/testing/assertions"
         . "github.com/smartystreets/goconvey/convey"
 )
 
 type staticArchiveSource []*logpb.LogEntry
 
 func (s *staticArchiveSource) NextLogEntry() (le *logpb.LogEntry, err error) {
         if len(*s) == 0 {
                 err = archive.ErrEndOfStream
         } else {
                 le, *s = (*s)[0], (*s)[1:]
         }
         return
 }
 
-type testGSClient map[string][]byte
+type testGSClient map[gs.Path][]byte
 
-func (c testGSClient) key(bucket, relpath string) string {
-        return fmt.Sprintf("%s/%s", bucket, relpath)
+func (c testGSClient) put(path gs.Path, d []byte) {
+        c[path] = d
 }
 
-func (c testGSClient) put(bucket, relpath string, d []byte) {
-        c[c.key(bucket, relpath)] = d
-}
-
-func (c testGSClient) get(bucket, relpath string) []byte {
-        return c[c.key(bucket, relpath)]
+func (c testGSClient) get(path gs.Path) []byte {
+        return c[path]
 }
 
 func (c testGSClient) Close() error { return nil }
-func (c testGSClient) NewWriter(string, string) (gs.Writer, error) {
+func (c testGSClient) NewWriter(gs.Path) (gs.Writer, error) {
         return nil, errors.New("not implemented")
 }
-func (c testGSClient) Delete(string, string) error { return errors.New("not implemented") }
+func (c testGSClient) Rename(gs.Path, gs.Path) error { return errors.New("not implemented") }
+func (c testGSClient) Delete(gs.Path) error          { return errors.New("not implemented") }
 
-func (c testGSClient) NewReader(bucket, relpath string, o gs.Options) (io.ReadCloser, error) {
+func (c testGSClient) NewReader(path gs.Path, o gs.Options) (io.ReadCloser, error) {
         if d, ok := c["error"]; ok {
                 return nil, errors.New(string(d))
         }
 
-        d, ok := c[c.key(bucket, relpath)]
+        d, ok := c[path]
         if !ok {
                 return nil, errors.New("does not exist")
         }
 
         to := int(o.To)
         if to == 0 {
                 to = len(d)
         }
         d = d[int(o.From):to]
 
@@ skipping 61 matching lines @@
 func TestGet(t *testing.T) {
         t.Parallel()
 
         Convey(`With a testing configuration, a Get request`, t, func() {
                 c, tc := testclock.UseTime(context.Background(), testclock.TestTimeLocal)
                 c = memory.Use(c)
 
                 fs := authtest.FakeState{}
                 c = auth.WithState(c, &fs)
 
-                c = ct.UseConfig(c, &svcconfig.Coordinator{
-                        AdminAuthGroup: "test-administrators",
-                })
+                ms := memoryStorage.Storage{}
+                gsc := testGSClient{}
+                svcStub := ct.Services{
+                        IS: func() (storage.Storage, error) {
+                                return &ms, nil
+                        },
+                        GS: func() (gs.Client, error) {
+                                return gsc, nil
+                        },
+                }
+                svcStub.InitConfig()
+                svcStub.ServiceConfig.Coordinator.AdminAuthGroup = "test-administrators"
+
+                s := Server{
+                        ServiceBase: coordinator.ServiceBase{&svcStub},
+                }
 
                 // Generate our test stream.
                 desc := ct.TestLogStreamDescriptor(c, "foo/bar")
                 ls := ct.TestLogStream(c, desc)
-                if err := ls.Put(ds.Get(c)); err != nil {
+                if err := ds.Get(c).Put(ls); err != nil {
                         panic(err)
                 }
 
                 tc.Add(time.Second)
                 var entries []*logpb.LogEntry
                 protobufs := map[uint64][]byte{}
                 for _, v := range []int{0, 1, 2, 4, 5, 7} {
                         le := ct.TestLogEntry(c, ls, v)
                         le.GetText().Lines = append(le.GetText().Lines, &logpb.Text_Line{
                                 Value: "another line of text",
@@ skipping 21 matching lines @@
                                 }
                         }
 
                         d, err := proto.Marshal(le)
                         if err != nil {
                                 panic(err)
                         }
                         protobufs[uint64(v)] = d
                 }
 
-                // Define and populate our Storage.
-                s := Server{}
-
-                ms := memoryStorage.Storage{}
-                gsc := testGSClient{}
-                s.StorageFunc = func(context.Context) (storage.Storage, error) {
-                        return &ms, nil
-                }
-                s.GSClientFunc = func(context.Context) (gs.Client, error) {
-                        return gsc, nil
-                }
-
                 Convey(`Testing Get requests (no logs)`, func() {
                         req := logdog.GetRequest{
                                 Path: string(ls.Path()),
                         }
 
                         Convey(`Will fail if the Path is not a stream path or a hash.`, func() {
                                 req.Path = "not/a/full/stream/path"
                                 _, err := s.Get(c, &req)
                                 So(err, ShouldErrLike, "invalid path value")
                         })
@@ skipping 44 matching lines @@
                                                 Desc:             desc,
                                                 Source:           &src,
                                                 LogWriter:        &lbuf,
                                                 IndexWriter:      &ibuf,
                                                 StreamIndexRange: 2,
                                         }
                                         if err := archive.Archive(m); err != nil {
                                                 panic(err)
                                         }
 
-                                        gsc.put("testbucket", "stream", lbuf.Bytes())
-                                        gsc.put("testbucket", "index", ibuf.Bytes())
+                                        now := tc.Now().UTC()
+
+                                        gsc.put("gs://testbucket/stream", lbuf.Bytes())
+                                        gsc.put("gs://testbucket/index", ibuf.Bytes())
                                         ls.State = coordinator.LSArchived
+                                        ls.TerminatedTime = now
+                                        ls.ArchivedTime = now
                                         ls.ArchiveStreamURL = "gs://testbucket/stream"
                                         ls.ArchiveIndexURL = "gs://testbucket/index"
                                 }
-                                if err := ls.Put(ds.Get(c)); err != nil {
+                                if err := ds.Get(c).Put(ls); err != nil {
                                         panic(err)
                                 }
 
                                 Convey(`Testing Get requests`, func() {
                                         req := logdog.GetRequest{
                                                 Path: string(ls.Path()),
                                         }
 
                                         Convey(`When the log stream is purged`, func() {
                                                 ls.Purged = true
-                                                if err := ls.Put(ds.Get(c)); err != nil {
+                                                if err := ds.Get(c).Put(ls); err != nil {
                                                         panic(err)
                                                 }
 
                                                 Convey(`Will return NotFound if the user is not an administrator.`, func() {
                                                         _, err := s.Get(c, &req)
                                                         So(err, ShouldBeRPCNotFound)
                                                 })
 
                                                 Convey(`Will process the request if the user is an administrator.`, func() {
                                                         fs.IdentityGroups = []string{"test-administrators"}
@@ skipping 86 matching lines @@
 
                                         Convey(`With a byte limit of sizeof({0, 1, 2})+1, will return log entries {0, 1, 2}.`, func() {
                                                 req.ByteCount = int32(len(protobufs[0]) + len(protobufs[1]) + len(protobufs[2]) + 1)
 
                                                 resp, err := s.Get(c, &req)
                                                 So(err, ShouldBeRPCOK)
                                                 So(resp, shouldHaveLogs, 0, 1, 2)
                                         })
 
                                         Convey(`Will successfully retrieve a stream path hash.`, func() {
-                                                req.Path = ls.HashID()
+                                                req.Path = ls.HashID
                                                 resp, err := s.Get(c, &req)
                                                 So(err, ShouldBeRPCOK)
                                                 So(resp, shouldHaveLogs, 0, 1, 2)
                                         })
 
                                         Convey(`When requesting state`, func() {
                                                 req.State = true
                                                 req.LogCount = -1
 
                                                 Convey(`Will successfully retrieve stream state.`, func() {
                                                         resp, err := s.Get(c, &req)
                                                         So(err, ShouldBeRPCOK)
                                                         So(resp.State, ShouldResemble, loadLogStreamState(ls))
                                                         So(len(resp.Logs), ShouldEqual, 0)
                                                 })
 
                                                 Convey(`Will return Internal if the protobuf descriptor data is corrupt.`, func() {
-                                                        // We can't use "ls.Put" here because it validates the protobuf!
+                                                        ls.SetDSValidate(false)
                                                         ls.Descriptor = []byte{0x00} // Invalid protobuf, zero tag.
                                                         if err := ds.Get(c).Put(ls); err != nil {
                                                                 panic(err)
                                                         }
 
                                                         _, err := s.Get(c, &req)
                                                         So(err, ShouldBeRPCInternal)
                                                 })
                                         })
 
                                         Convey(`Will return Internal if the protobuf log entry data is corrupt.`, func() {
                                                 if v {
                                                         // Corrupt the archive datastream.
-                                                        stream := gsc.get("testbucket", "stream")
+                                                        stream := gsc.get("gs://testbucket/stream")
                                                         zeroRecords(stream)
                                                 } else {
                                                         // Add corrupted entry to Storage. Create a new entry here, since
                                                         // the storage will reject a duplicate/overwrite.
                                                         err := ms.Put(storage.PutRequest{
                                                                 Path:   types.StreamPath(req.Path),
                                                                 Index:  666,
                                                                 Values: [][]byte{{0x00}}, // Invalid protobuf, zero tag.
                                                         })
                                                         if err != nil {
@@ skipping 106 matching lines @@
                                                 Path: string(ls.Path()),
                                         }
 
                                         Convey(`Will successfully retrieve a stream path.`, func() {
                                                 resp, err := s.Tail(c, &req)
                                                 So(err, ShouldBeRPCOK)
                                                 So(resp, shouldHaveLogs, 7)
                                         })
 
                                         Convey(`Will successfully retrieve a stream path hash and state.`, func() {
-                                                req.Path = ls.HashID()
+                                                req.Path = ls.HashID
                                                 req.State = true
 
                                                 resp, err := s.Tail(c, &req)
                                                 So(err, ShouldBeRPCOK)
                                                 So(resp, shouldHaveLogs, 7)
                                                 So(resp.State, ShouldResemble, loadLogStreamState(ls))
                                         })
                                 })
                         })
                 }
         })
 }