LogDog: Add Storage-layer data caching.

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/gae/filter/featureBreaker"
         "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/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"
 )
 
 func shouldHaveLogs(actual interface{}, expected ...interface{}) string {
         resp := actual.(*logdog.GetResponse)
 
@@ skipping 34 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 188 matching lines @@
                         })
 
                         Convey(`Will fail with NotFound if the log path does not exist (different path).`, func() {
                                 req.Path = "testing/+/does/not/exist"
                                 _, err := svr.Tail(c, &req)
                                 So(err, ShouldBeRPCNotFound)
                         })
                 })
 
                 Convey(`When testing log data is added`, func() {
-                        if !archived {
-                                // Add the logs to the in-memory temporary storage.
-                                for _, le := range entries {
-                                        err := env.IntermediateStorage.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))
+                        putLogData := func() {
+                                if !archived {
+                                        // Add the logs to the in-memory temporary storage.
+                                        for _, le := range entries {
+                                                err := is.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 {
+                                        // Archive this log stream. We will generate one index entry for every
+                                        // 2 log entries.
+                                        src := renderer.StaticSource(entries)
+                                        var lbuf, ibuf bytes.Buffer
+                                        m := archive.Manifest{
+                                                Desc:             tls.Desc,
+                                                Source:           &src,
+                                                LogWriter:        &lbuf,
+                                                IndexWriter:      &ibuf,
+                                                StreamIndexRange: 2,
+                                        }
+                                        if err := archive.Archive(m); err != nil {
+                                                panic(err)
+                                        }
+
+                                        now := env.Clock.Now().UTC()
+
+                                        env.GSClient.Put("gs://testbucket/stream", lbuf.Bytes())
+                                        env.GSClient.Put("gs://testbucket/index", ibuf.Bytes())
+                                        tls.State.TerminatedTime = now
+                                        tls.State.ArchivedTime = now
+                                        tls.State.ArchiveStreamURL = "gs://testbucket/stream"
+                                        tls.State.ArchiveIndexURL = "gs://testbucket/index"
+
+                                        So(tls.State.ArchivalState().Archived(), ShouldBeTrue)
                                 }
-                        } else {
-                                // Archive this log stream. We will generate one index entry for every
-                                // 2 log entries.
-                                src := renderer.StaticSource(entries)
-                                var lbuf, ibuf bytes.Buffer
-                                m := archive.Manifest{
-                                        Desc:             tls.Desc,
-                                        Source:           &src,
-                                        LogWriter:        &lbuf,
-                                        IndexWriter:      &ibuf,
-                                        StreamIndexRange: 2,
-                                }
-                                if err := archive.Archive(m); err != nil {
-                                        panic(err)
-                                }
-
-                                now := env.Clock.Now().UTC()
-
-                                env.GSClient.Put("gs://testbucket/stream", lbuf.Bytes())
-                                env.GSClient.Put("gs://testbucket/index", ibuf.Bytes())
-                                tls.State.TerminatedTime = now
-                                tls.State.ArchivedTime = now
-                                tls.State.ArchiveStreamURL = "gs://testbucket/stream"
-                                tls.State.ArchiveIndexURL = "gs://testbucket/index"
-
-                                So(tls.State.ArchivalState().Archived(), ShouldBeTrue)
                         }
+                        putLogData()
                         putLogStream(c)
 
                         Convey(`Testing Get requests`, func() {
                                 req := logdog.GetRequest{
                                         Project: string(project),
                                         Path:    string(tls.Path),
                                 }
 
                                 Convey(`When the log stream is purged`, func() {
                                         tls.Stream.Purged = true
@@ skipping 19 matching lines @@
 
                                         resp, err := svr.Get(c, &req)
                                         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)
                                         So(err, ShouldBeRPCOK)
                                         So(resp, shouldHaveLogs, 4, 5)
                                 })
 
@@ skipping 84 matching lines @@
                                 })
 
                                 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 := env.IntermediateStorage.Put(storage.PutRequest{
+                                                err := is.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 {
-                                                env.IntermediateStorage.Close()
+                                                is.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 73 matching lines @@
                                         Project: string(project),
                                         Path:    string(tls.Path),
                                         State:   true,
                                 }
 
                                 Convey(`Will successfully retrieve a stream path.`, func() {
                                         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, false)
+        testGetImpl(t, true)
 }