LogDog: Add signed GS URL fetching.

logdog/appengine/coordinator/endpoints/logs/get_test.go

 // Copyright 2015 The LUCI Authors. All rights reserved.
 // Use of this source code is governed under the Apache License, Version 2.0
 // that can be found in the LICENSE file.
 
 package logs
 
 import (
         "bytes"
         "errors"
         "fmt"
         "math"
         "testing"
         "time"
 
+        "github.com/luci/luci-go/common/clock"
         "github.com/luci/luci-go/common/config"
         "github.com/luci/luci-go/common/data/recordio"
         "github.com/luci/luci-go/common/iotools"
+        "github.com/luci/luci-go/common/proto/google"
         "github.com/luci/luci-go/logdog/api/endpoints/coordinator/logs/v1"
         "github.com/luci/luci-go/logdog/api/logpb"
         ct "github.com/luci/luci-go/logdog/appengine/coordinator/coordinatorTest"
         "github.com/luci/luci-go/logdog/common/archive"
         "github.com/luci/luci-go/logdog/common/renderer"
         "github.com/luci/luci-go/logdog/common/storage"
-        "github.com/luci/luci-go/logdog/common/storage/bigtable"
         "github.com/luci/luci-go/logdog/common/types"
 
         "github.com/luci/gae/filter/featureBreaker"
 
         "github.com/golang/protobuf/proto"
         "golang.org/x/net/context"
 
         . "github.com/luci/luci-go/common/testing/assertions"
         . "github.com/smartystreets/goconvey/convey"
 )
@@ skipping 38 matching lines @@
                 trash.ReadFrom(r)
         }
 }
 
 func testGetImpl(t *testing.T, archived bool) {
         Convey(fmt.Sprintf(`With a testing configuration, a Get request (archived=%v)`, archived), t, func() {
                 c, env := ct.Install()
 
                 svr := New()
 
-                // Use an actual BigTable testing instance for our intermediate storage,
-                // instead of the default in-memory instance.
-                is := bigtable.NewMemoryInstance(c, bigtable.Options{
-                        Cache: env.Services.StorageCache(),
-                })
-                env.Services.IS = func() (storage.Storage, error) { return is, nil }
-
                 // di is a datastore bound to the test project namespace.
                 const project = config.ProjectName("proj-foo")
 
                 // Generate our test stream.
                 tls := ct.MakeStream(c, "proj-foo", "testing/+/foo/bar")
 
                 putLogStream := func(c context.Context) {
                         if err := tls.Put(c); err != nil {
                                 panic(err)
                         }
@@ skipping 192 matching lines @@
                                 _, err := svr.Tail(c, &req)
                                 So(err, ShouldBeRPCNotFound)
                         })
                 })
 
                 Convey(`When testing log data is added`, func() {
                         putLogData := func() {
                                 if !archived {
                                         // Add the logs to the in-memory temporary storage.
                                         for _, le := range entries {
-                                                err := is.Put(storage.PutRequest{
+                                                err := env.BigTable.Put(storage.PutRequest{
                                                         Project: project,
                                                         Path:    tls.Path,
                                                         Index:   types.MessageIndex(le.StreamIndex),
                                                         Values:  [][]byte{protobufs[le.StreamIndex]},
                                                 })
                                                 if err != nil {
                                                         panic(fmt.Errorf("failed to Put() LogEntry: %v", err))
                                                 }
                                         }
                                 } else {
@@ skipping 59 matching lines @@
                                         So(err, ShouldBeRPCOK)
                                         So(resp.Logs, ShouldHaveLength, 0)
                                 })
 
                                 Convey(`Will successfully retrieve a stream path.`, func() {
                                         resp, err := svr.Get(c, &req)
                                         So(err, ShouldBeRPCOK)
                                         So(resp, shouldHaveLogs, 0, 1, 2)
 
                                         Convey(`Will successfully retrieve the stream path again (caching).`, func() {
-                                                // Re-put the log data, since the previous request closed the
-                                                // storage instance.
-                                                putLogData()
-
                                                 resp, err := svr.Get(c, &req)
                                                 So(err, ShouldBeRPCOK)
                                                 So(resp, shouldHaveLogs, 0, 1, 2)
                                         })
                                 })
 
                                 Convey(`Will successfully retrieve a stream path offset at 4.`, func() {
                                         req.Index = 4
 
                                         resp, err := svr.Get(c, &req)
@@ skipping 80 matching lines @@
                                         Convey(`Will return Internal if the protobuf descriptor data is corrupt.`, func() {
                                                 tls.Stream.SetDSValidate(false)
                                                 tls.Stream.Descriptor = []byte{0x00} // Invalid protobuf, zero tag.
                                                 putLogStream(c)
 
                                                 _, err := svr.Get(c, &req)
                                                 So(err, ShouldBeRPCInternal)
                                         })
                                 })
 
+                                Convey(`When requesting a signed URL`, func() {
+                                        const duration = 10 * time.Hour
+                                        req.LogCount = -1
+
+                                        sr := logdog.GetRequest_SignURLRequest{
+                                                Lifetime: google.NewDuration(duration),
+                                                Stream:   true,
+                                                Index:    true,
+                                        }
+                                        req.GetSignedUrls = &sr
+
+                                        if archived {
+                                                Convey(`Will successfully retrieve the URL.`, func() {
+                                                        resp, err := svr.Get(c, &req)
+                                                        So(err, ShouldBeNil)
+                                                        So(resp.Logs, ShouldHaveLength, 0)
+
+                                                        So(resp.SignedUrls, ShouldNotBeNil)
+                                                        So(resp.SignedUrls.Stream, ShouldEndWith, "&signed=true")
+                                                        So(resp.SignedUrls.Index, ShouldEndWith, "&signed=true")
+                                                        So(resp.SignedUrls.Expiration.Time(), ShouldResemble, clock.Now(c).Add(duration))
+                                                })
+                                        } else {
+                                                Convey(`Will succeed, but return no URL.`, func() {
+                                                        resp, err := svr.Get(c, &req)
+                                                        So(err, ShouldBeNil)
+                                                        So(resp.Logs, ShouldHaveLength, 0)
+                                                        So(resp.SignedUrls, ShouldBeNil)
+                                                })
+                                        }
+                                })
+
                                 Convey(`Will return Internal if the protobuf log entry data is corrupt.`, func() {
                                         if archived {
                                                 // Corrupt the archive datastream.
                                                 stream := env.GSClient.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 := is.Put(storage.PutRequest{
+                                                err := env.BigTable.Put(storage.PutRequest{
                                                         Project: project,
                                                         Path:    types.StreamPath(req.Path),
                                                         Index:   666,
                                                         Values:  [][]byte{{0x00}}, // Invalid protobuf, zero tag.
                                                 })
                                                 if err != nil {
                                                         panic(err)
                                                 }
                                                 req.Index = 666
                                         }
 
                                         _, err := svr.Get(c, &req)
                                         So(err, ShouldBeRPCInternal)
                                 })
 
                                 Convey(`Will successfully retrieve both logs and stream state.`, func() {
                                         req.State = true
 
                                         resp, err := svr.Get(c, &req)
                                         So(err, ShouldBeRPCOK)
                                         So(resp.State, ShouldResemble, buildLogStreamState(tls.Stream, tls.State))
                                         So(resp, shouldHaveLogs, 0, 1, 2)
                                 })
 
                                 Convey(`Will return Internal if the Storage is not working.`, func() {
                                         if archived {
                                                 env.GSClient["error"] = []byte("test error")
                                         } else {
-                                                is.SetErr(errors.New("not working"))
+                                                env.BigTable.SetErr(errors.New("not working"))
                                         }
 
                                         _, err := svr.Get(c, &req)
                                         So(err, ShouldBeRPCInternal)
                                 })
 
                                 Convey(`Will enforce a maximum count of 2.`, func() {
                                         req.LogCount = 2
                                         resp, err := svr.Get(c, &req)
                                         So(err, ShouldBeRPCOK)
@@ skipping 79 matching lines @@
                                         resp, err := svr.Tail(c, &req)
                                         So(err, ShouldBeRPCOK)
                                         So(resp, shouldHaveLogs, 7)
                                         So(resp.State, ShouldResemble, buildLogStreamState(tls.Stream, tls.State))
 
                                         // For non-archival: 1 miss and 1 put, for the tail row.
                                         // For archival: 1 miss and 1 put, for the index.
                                         So(env.StorageCache.Stats(), ShouldResemble, ct.StorageCacheStats{Puts: 1, Misses: 1})
 
                                         Convey(`Will retrieve the stream path again (caching).`, func() {
-                                                // Re-put the log data, since the previous request closed the
-                                                // storage instance.
-                                                putLogData()
                                                 env.StorageCache.Clear()
 
                                                 resp, err := svr.Tail(c, &req)
                                                 So(err, ShouldBeRPCOK)
                                                 So(resp, shouldHaveLogs, 7)
                                                 So(resp.State, ShouldResemble, buildLogStreamState(tls.Stream, tls.State))
 
                                                 // For non-archival: 1 hit, for the tail row.
                                                 // For archival: 1 hit, for the index.
                                                 So(env.StorageCache.Stats(), ShouldResemble, ct.StorageCacheStats{Hits: 1})
                                         })
                                 })
                         })
                 })
         })
 }
 
 func TestGetIntermediate(t *testing.T) {
         t.Parallel()
 
         testGetImpl(t, false)
 }
 
 func TestGetArchived(t *testing.T) {
         t.Parallel()
 
         testGetImpl(t, true)
 }