LogDog: Add Storage-layer data caching.

logdog/common/storage/bigtable/storage.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 bigtable
 
 import (
         "bytes"
         "errors"
         "fmt"
 
         "github.com/luci/luci-go/common/config"
         "github.com/luci/luci-go/common/data/recordio"
         log "github.com/luci/luci-go/common/logging"
         "github.com/luci/luci-go/logdog/common/storage"
+        "github.com/luci/luci-go/logdog/common/storage/caching"
         "github.com/luci/luci-go/logdog/common/types"
 
         "cloud.google.com/go/bigtable"
         "golang.org/x/net/context"
         "google.golang.org/api/option"
 )
 
 var (
         // StorageScopes is the set of OAuth scopes needed to use the storage
         // functionality.
@@ skipping 34 matching lines @@
         // Project is the name of the project to connect to.
         Project string
         // Instance is the name of the instance to connect to.
         Instance string
         // ClientOptions are additional client options to use when instantiating the
         // client instance.
         ClientOptions []option.ClientOption
 
         // Table is the name of the BigTable table to use for logs.
         LogTable string
+
+        // Cache, if not nil, will be used to cache data.
+        Cache caching.Cache
 }
 
 func (o *Options) client(ctx context.Context) (*bigtable.Client, error) {
         return bigtable.NewClient(ctx, o.Project, o.Instance, o.ClientOptions...)
 }
 
 func (o *Options) adminClient(ctx context.Context) (*bigtable.AdminClient, error) {
         return bigtable.NewAdminClient(ctx, o.Project, o.Instance, o.ClientOptions...)
 }
 
 // btStorage is a storage.Storage implementation that uses Google Cloud BigTable
 // as a backend.
 type btStorage struct {
         *Options
 
         // Context is the bound supplied with New. It is retained (rather than
         // supplied on a per-call basis) because a special Storage Context devoid of
         // gRPC metadata is needed for Storage calls.
         context.Context
 
         client      *bigtable.Client
         logTable    *bigtable.Table
         adminClient *bigtable.AdminClient
 
-        // raw is the underlying btTable instance to use for raw operations.
+        // raw, if not nil, is the raw BigTable interface to use. This is useful for
+        // testing. If nil, this will default to the production isntance.
         raw btTable
+
         // maxRowSize is the maxmium number of bytes that can be stored in a single
         // BigTable row. This is a function of BigTable, and constant in production
         // (bigTableRowMaxBytes), but variable here to allow for testing to control.
         maxRowSize int
 }
 
 // New instantiates a new Storage instance connected to a BigTable instance.
 //
 // The returned Storage instance will close the Client when its Close() method
 // is called.
 func New(ctx context.Context, o Options) (storage.Storage, error) {
         client, err := o.client(ctx)
         if err != nil {
                 return nil, err
         }
 
         admin, err := o.adminClient(ctx)
         if err != nil {
                 return nil, err
         }
 
-        return newBTStorage(ctx, o, client, admin), nil
+        return newBTStorage(ctx, o, client, admin, nil), nil
 }
 
-func newBTStorage(ctx context.Context, o Options, client *bigtable.Client, adminClient *bigtable.AdminClient) *btStorage {
+func newBTStorage(ctx context.Context, o Options, client *bigtable.Client, adminClient *bigtable.AdminClient, raw btTable) *btStorage {
         s := &btStorage{
                 Options: &o,
                 Context: ctx,
 
                 client:      client,
                 adminClient: adminClient,
+                raw:         raw,
                 maxRowSize:  bigTableRowMaxBytes,
         }
         if s.client != nil {
                 s.logTable = s.client.Open(o.LogTable)
         }
-        s.raw = &btTableProd{s}
+        if s.raw == nil {
+                s.raw = &btTableProd{s}
+        }
         return s
 }
 
 func (s *btStorage) Close() {
         if s.client != nil {
                 s.client.Close()
                 s.client = nil
         }
 
         if s.adminClient != nil {
@@ skipping 149 matching lines @@
                 return err
         }
 }
 
 func (s *btStorage) Tail(project config.ProjectName, path types.StreamPath) (*storage.Entry, error) {
         ctx := log.SetFields(s, log.Fields{
                 "project": project,
                 "path":    path,
         })
 
+        // Load the "last tail index" from cache. If we have no cache, start at 0.
+        var startIdx int64
+        if s.Cache != nil {
+                startIdx = getLastTailIndex(s, s.Cache, project, path)
+        }
+
         // Iterate through all log keys in the stream. Record the latest one.
-        rk := newRowKey(string(project), string(path), 0, 0)
+        rk := newRowKey(string(project), string(path), startIdx, 0)
         var latest *rowKey
         err := s.raw.getLogData(ctx, rk, 0, true, func(rk *rowKey, data []byte) error {
                 latest = rk
                 return nil
         })
         if err != nil {
                 log.Fields{
                         log.ErrorKey: err,
                         "project":    s.Project,
                         "instance":   s.Instance,
                         "table":      s.LogTable,
                 }.Errorf(ctx, "Failed to scan for tail.")
         }
 
         if latest == nil {
                 // No rows for the specified stream.
                 return nil, storage.ErrDoesNotExist
         }
 
+        // Update our cache if the tail index has changed.
+        if s.Cache != nil && startIdx != latest.index {
+                putLastTailIndex(s, s.Cache, project, path, latest.index)
+        }
+
         // Fetch the latest row's data.
         var d []byte
         err = s.raw.getLogData(ctx, latest, 1, false, func(rk *rowKey, data []byte) error {
                 records, err := recordio.Split(data)
                 if err != nil || len(records) == 0 {
                         return storage.ErrBadData
                 }
                 d = records[len(records)-1]
                 return errStop
         })
@@ skipping 68 matching lines @@
         }.Debugf(ctx, "Adding entries to BigTable.")
         if err := raw.putLogData(ctx, rk, w.buf.Bytes()); err != nil {
                 return 0, err
         }
 
         // Reset our buffer state.
         w.buf.Reset()
         w.count = 0
         return flushCount, nil
 }