LogDog: Add project namespacing to Archivist.

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 (
         "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/config"
         "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"
 
@@ skipping 168 matching lines @@
 func TestHandleArchive(t *testing.T) {
         t.Parallel()
 
         Convey(`A testing archive setup`, t, func() {
                 c, tc := testclock.UseTime(context.Background(), testclock.TestTimeUTC)
 
                 st := memory.Storage{}
                 gsc := testGSClient{}
 
                 // Set up our test log stream.
+                project := "test-project"
                 desc := logpb.LogStreamDescriptor{
                         Prefix:        "testing",
                         Name:          "foo",
                         BinaryFileExt: "bin",
                 }
                 descBytes, err := proto.Marshal(&desc)
                 if err != nil {
                         panic(err)
                 }
 
                 // Utility function to add a log entry for "ls".
-                addTestEntry := func(idxs ...int) {
+                addTestEntry := func(p string, idxs ...int) {
                         for _, v := range idxs {
                                 le := logpb.LogEntry{
                                         PrefixIndex: uint64(v),
                                         StreamIndex: uint64(v),
                                         Content: &logpb.LogEntry_Text{&logpb.Text{
                                                 Lines: []*logpb.Text_Line{
                                                         {
                                                                 Value:     fmt.Sprintf("line #%d", v),
                                                                 Delimiter: "\n",
                                                         },
                                                 },
                                         }},
                                 }
 
                                 d, err := proto.Marshal(&le)
                                 if err != nil {
                                         panic(err)
                                 }
 
                                 err = st.Put(storage.PutRequest{
-                                        Path:   desc.Path(),
-                                        Index:  types.MessageIndex(v),
-                                        Values: [][]byte{d},
+                                        Project: config.ProjectName(p),
+                                        Path:    desc.Path(),
+                                        Index:   types.MessageIndex(v),
+                                        Values:  [][]byte{d},
                                 })
                                 if err != nil {
                                         panic(err)
                                 }
 
                                 // Advance the time for each log entry.
                                 tc.Add(time.Second)
                         }
                 }
 
                 // Set up our testing archival task.
                 expired := 10 * time.Minute
                 archiveTask := logdog.ArchiveTask{
+                        Project:        project,
                         Path:           string(desc.Path()),
                         SettleDelay:    google.NewDuration(10 * time.Second),
                         CompletePeriod: google.NewDuration(expired),
                         Key:            []byte("random archival key"),
                 }
                 expired++ // This represents a time PAST CompletePeriod.
 
                 task := &testTask{
                         task: &archiveTask,
                 }
@@ skipping 27 matching lines @@
                 }
 
                 ar := Archivist{
                         Service:       &sc,
                         Storage:       &st,
                         GSClient:      &gsc,
                         GSBase:        gs.Path("gs://archive-test/path/to/archive/"),         // Extra slashes to test concatenation.
                         GSStagingBase: gs.Path("gs://archive-test-staging/path/to/archive/"), // Extra slashes to test concatenation.
                 }
 
-                gsURL := func(p string) string {
-                        return fmt.Sprintf("gs://archive-test/path/to/archive/%s/%s", desc.Path(), p)
+                gsURL := func(project, name string) string {
+                        return fmt.Sprintf("gs://archive-test/path/to/archive/%s/%s/%s", project, desc.Path(), name)
                 }
 
                 // 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) ||
@@ skipping 79 matching lines @@
                 Convey(`With terminal index "3"`, func() {
                         stream.State.TerminalIndex = 3
 
                         Convey(`Will fail not ACK a log stream with no entries.`, func() {
                                 ack, err := ar.archiveTaskImpl(c, task)
                                 So(err, ShouldEqual, storage.ErrDoesNotExist)
                                 So(ack, ShouldBeFalse)
                         })
 
                         Convey(`Will fail to archive {0, 1, 2, 4} (incomplete).`, func() {
-                                addTestEntry(0, 1, 2, 4)
+                                addTestEntry(project, 0, 1, 2, 4)
 
                                 ack, err := ar.archiveTaskImpl(c, task)
                                 So(err, ShouldErrLike, "missing log entry")
                                 So(ack, ShouldBeFalse)
                         })
 
                         Convey(`Will successfully archive {0, 1, 2, 3, 4}, stopping at the terminal index.`, func() {
-                                addTestEntry(0, 1, 2, 3, 4)
+                                addTestEntry(project, 0, 1, 2, 3, 4)
 
                                 ack, err := ar.archiveTaskImpl(c, task)
                                 So(err, ShouldBeNil)
                                 So(ack, ShouldBeTrue)
 
                                 So(hasStreams(true, true, true), ShouldBeTrue)
                                 So(archiveRequest, ShouldResemble, &logdog.ArchiveStreamRequest{
+                                        Project:       project,
                                         Path:          archiveTask.Path,
                                         LogEntryCount: 4,
                                         TerminalIndex: 3,
 
-                                        StreamUrl: gsURL("logstream.entries"),
-                                        IndexUrl:  gsURL("logstream.index"),
-                                        DataUrl:   gsURL("data.bin"),
+                                        StreamUrl: gsURL(project, "logstream.entries"),
+                                        IndexUrl:  gsURL(project, "logstream.index"),
+                                        DataUrl:   gsURL(project, "data.bin"),
                                 })
                         })
 
                         Convey(`When a transient archival error occurs, will not ACK it.`, func() {
-                                addTestEntry(0, 1, 2, 3, 4)
+                                addTestEntry(project, 0, 1, 2, 3, 4)
                                 gsc.newWriterErr = func(*testGSWriter) error { return errors.WrapTransient(errors.New("test error")) }
 
                                 ack, err := ar.archiveTaskImpl(c, task)
                                 So(err, ShouldErrLike, "test error")
                                 So(ack, ShouldBeFalse)
                         })
 
                         Convey(`When a non-transient archival error occurs`, func() {
-                                addTestEntry(0, 1, 2, 3, 4)
+                                addTestEntry(project, 0, 1, 2, 3, 4)
                                 archiveErr := errors.New("archive failure error")
                                 gsc.newWriterErr = func(*testGSWriter) error { return archiveErr }
 
                                 Convey(`If remote report returns an error, do not ACK.`, func() {
                                         archiveStreamErr = errors.New("test error")
 
                                         ack, err := ar.archiveTaskImpl(c, task)
                                         So(err, ShouldErrLike, "test error")
                                         So(ack, ShouldBeFalse)
 
                                         So(archiveRequest, ShouldResemble, &logdog.ArchiveStreamRequest{
-                                                Path:  archiveTask.Path,
-                                                Error: "archive failure error",
+                                                Project: project,
+                                                Path:    archiveTask.Path,
+                                                Error:   "archive failure error",
                                         })
                                 })
 
                                 Convey(`If remote report returns success, ACK.`, func() {
                                         ack, err := ar.archiveTaskImpl(c, task)
                                         So(err, ShouldBeNil)
                                         So(ack, ShouldBeTrue)
                                         So(archiveRequest, ShouldResemble, &logdog.ArchiveStreamRequest{
-                                                Path:  archiveTask.Path,
-                                                Error: "archive failure error",
+                                                Project: project,
+                                                Path:    archiveTask.Path,
+                                                Error:   "archive failure error",
                                         })
                                 })
 
                                 Convey(`If an empty error string is supplied, the generic error will be filled in.`, func() {
                                         archiveErr = errors.New("")
 
                                         ack, err := ar.archiveTaskImpl(c, task)
                                         So(err, ShouldBeNil)
                                         So(ack, ShouldBeTrue)
                                         So(archiveRequest, ShouldResemble, &logdog.ArchiveStreamRequest{
-                                                Path:  archiveTask.Path,
-                                                Error: "archival error",
+                                                Project: project,
+                                                Path:    archiveTask.Path,
+                                                Error:   "archival error",
                                         })
                                 })
                         })
                 })
 
                 Convey(`When not enforcing stream completeness`, func() {
                         stream.Age = google.NewDuration(expired)
 
                         Convey(`With no terminal index`, func() {
                                 Convey(`Will successfully archive if there are no entries.`, func() {
                                         ack, err := ar.archiveTaskImpl(c, task)
                                         So(err, ShouldBeNil)
                                         So(ack, ShouldBeTrue)
 
                                         So(hasStreams(true, true, false), ShouldBeTrue)
                                         So(archiveRequest, ShouldResemble, &logdog.ArchiveStreamRequest{
+                                                Project:       project,
                                                 Path:          archiveTask.Path,
                                                 LogEntryCount: 0,
                                                 TerminalIndex: -1,
 
-                                                StreamUrl: gsURL("logstream.entries"),
-                                                IndexUrl:  gsURL("logstream.index"),
-                                                DataUrl:   gsURL("data.bin"),
+                                                StreamUrl: gsURL(project, "logstream.entries"),
+                                                IndexUrl:  gsURL(project, "logstream.index"),
+                                                DataUrl:   gsURL(project, "data.bin"),
                                         })
                                 })
 
                                 Convey(`With {0, 1, 2, 4} (incomplete) will archive the stream and update its terminal index.`, func() {
-                                        addTestEntry(0, 1, 2, 4)
+                                        addTestEntry(project, 0, 1, 2, 4)
 
                                         ack, err := ar.archiveTaskImpl(c, task)
                                         So(err, ShouldBeNil)
                                         So(ack, ShouldBeTrue)
 
                                         So(hasStreams(true, true, true), ShouldBeTrue)
                                         So(archiveRequest, ShouldResemble, &logdog.ArchiveStreamRequest{
+                                                Project:       project,
                                                 Path:          archiveTask.Path,
                                                 LogEntryCount: 4,
                                                 TerminalIndex: 4,
 
-                                                StreamUrl: gsURL("logstream.entries"),
-                                                IndexUrl:  gsURL("logstream.index"),
-                                                DataUrl:   gsURL("data.bin"),
+                                                StreamUrl: gsURL(project, "logstream.entries"),
+                                                IndexUrl:  gsURL(project, "logstream.index"),
+                                                DataUrl:   gsURL(project, "data.bin"),
                                         })
                                 })
                         })
 
                         Convey(`With terminal index 3`, func() {
                                 stream.State.TerminalIndex = 3
 
                                 Convey(`Will successfully archive if there are no entries.`, func() {
                                         ack, err := ar.archiveTaskImpl(c, task)
                                         So(err, ShouldBeNil)
                                         So(ack, ShouldBeTrue)
 
                                         So(hasStreams(true, true, false), ShouldBeTrue)
                                         So(archiveRequest, ShouldResemble, &logdog.ArchiveStreamRequest{
+                                                Project:       project,
                                                 Path:          archiveTask.Path,
                                                 LogEntryCount: 0,
                                                 TerminalIndex: -1,
 
-                                                StreamUrl: gsURL("logstream.entries"),
-                                                IndexUrl:  gsURL("logstream.index"),
-                                                DataUrl:   gsURL("data.bin"),
+                                                StreamUrl: gsURL(project, "logstream.entries"),
+                                                IndexUrl:  gsURL(project, "logstream.index"),
+                                                DataUrl:   gsURL(project, "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)
+                                        addTestEntry(project, 0, 1, 2, 4)
 
                                         ack, err := ar.archiveTaskImpl(c, task)
                                         So(err, ShouldBeNil)
                                         So(ack, ShouldBeTrue)
 
                                         So(hasStreams(true, true, true), ShouldBeTrue)
                                         So(archiveRequest, ShouldResemble, &logdog.ArchiveStreamRequest{
+                                                Project:       project,
                                                 Path:          archiveTask.Path,
                                                 LogEntryCount: 3,
                                                 TerminalIndex: 2,
 
-                                                StreamUrl: gsURL("logstream.entries"),
-                                                IndexUrl:  gsURL("logstream.index"),
-                                                DataUrl:   gsURL("data.bin"),
+                                                StreamUrl: gsURL(project, "logstream.entries"),
+                                                IndexUrl:  gsURL(project, "logstream.index"),
+                                                DataUrl:   gsURL(project, "data.bin"),
                                         })
                                 })
                         })
                 })
 
+                Convey(`With an empty project`, func() {
+                        archiveTask.Project = ""
+
+                        Convey(`Will successfully archive {0, 1, 2, 3} with terminal index 3 using "_" for project archive path.`, func() {
+                                stream.State.TerminalIndex = 3
+                                addTestEntry("", 0, 1, 2, 3)
+
+                                ack, err := ar.archiveTaskImpl(c, task)
+                                So(err, ShouldBeNil)
+                                So(ack, ShouldBeTrue)
+
+                                So(hasStreams(true, true, true), ShouldBeTrue)
+                                So(archiveRequest, ShouldResemble, &logdog.ArchiveStreamRequest{
+                                        Project:       "",
+                                        Path:          archiveTask.Path,
+                                        LogEntryCount: 4,
+                                        TerminalIndex: 3,
+
+                                        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)
+                        addTestEntry(project, 0, 1, 2, 3)
 
                         for _, failName := range []string{"/logstream.entries", "/logstream.index", "/data.bin"} {
                                 for _, testCase := range []struct {
                                         name  string
                                         setup func()
                                 }{
                                         {"writer create failure", func() {
                                                 gsc.newWriterErr = func(w *testGSWriter) error {
                                                         if strings.HasSuffix(string(w.path), failName) {
                                                                 return errors.WrapTransient(errors.New("test error"))
@@ skipping 55 matching lines @@
                                                 ack, err := ar.archiveTaskImpl(c, task)
                                                 So(err, ShouldErrLike, "test error")
                                                 So(ack, ShouldBeFalse)
                                                 So(archiveRequest, ShouldBeNil)
                                         })
                                 }
                         }
                 })
         })
 }