LogDog: Add Storage-layer data caching.

logdog/common/storage/bigtable/bigtable.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 (
         "fmt"
         "time"
 
@@ skipping 47 matching lines @@
         // If keysOnly is true, then the callback will return nil row data.
         getLogData(c context.Context, rk *rowKey, limit int, keysOnly bool, cb btGetCallback) error
 
         // setMaxLogAge updates the maximum log age policy for the log family.
         setMaxLogAge(context.Context, time.Duration) error
 }
 
 // btTableProd is an implementation of the btTable interface that uses a real
 // production BigTable connection.
 type btTableProd struct {
-        *btStorage
+        base *btStorage
 }
 
 func (t *btTableProd) putLogData(c context.Context, rk *rowKey, data []byte) error {
         m := bigtable.NewMutation()
         m.Set(logColumnFamily, logColumn, bigtable.ServerTime, data)
         cm := bigtable.NewCondMutation(bigtable.RowKeyFilter(rk.encode()), nil, m)
 
         rowExists := false
-        if err := t.logTable.Apply(c, rk.encode(), cm, bigtable.GetCondMutationResult(&rowExists)); err != nil {
+        if err := t.base.logTable.Apply(c, rk.encode(), cm, bigtable.GetCondMutationResult(&rowExists)); err != nil {
                 return grpcutil.WrapIfTransient(err)
         }
         if rowExists {
                 return storage.ErrExists
         }
         return nil
 }
 
 func (t *btTableProd) getLogData(c context.Context, rk *rowKey, limit int, keysOnly bool, cb btGetCallback) error {
         // Construct read options based on Get request.
         ropts := []bigtable.ReadOption{
                 bigtable.RowFilter(bigtable.FamilyFilter(logColumnFamily)),
                 bigtable.RowFilter(bigtable.ColumnFilter(logColumn)),
                 nil,
         }[:2]
         if keysOnly {
                 ropts = append(ropts, bigtable.RowFilter(bigtable.StripValueFilter()))
         }
         if limit > 0 {
                 ropts = append(ropts, bigtable.LimitRows(int64(limit)))
         }
 
         // This will limit the range to the immediate row key ("ASDF~INDEX") to
         // immediately after the row key ("ASDF~~"). See rowKey for more information.
         rng := bigtable.NewRange(rk.encode(), rk.pathPrefixUpperBound())
 
         var innerErr error
-        err := t.logTable.ReadRows(c, rng, func(row bigtable.Row) bool {
+        err := t.base.logTable.ReadRows(c, rng, func(row bigtable.Row) bool {
                 data, err := getLogRowData(row)
                 if err != nil {
                         innerErr = storage.ErrBadData
                         return false
                 }
 
                 drk, err := decodeRowKey(row.Key())
                 if err != nil {
                         innerErr = err
                         return false
@@ skipping 12 matching lines @@
                 return innerErr
         }
         return nil
 }
 
 func (t *btTableProd) setMaxLogAge(c context.Context, d time.Duration) error {
         var logGCPolicy bigtable.GCPolicy
         if d > 0 {
                 logGCPolicy = bigtable.MaxAgePolicy(d)
         }
-        if err := t.adminClient.SetGCPolicy(c, t.LogTable, logColumnFamily, logGCPolicy); err != nil {
+        if err := t.base.adminClient.SetGCPolicy(c, t.base.LogTable, logColumnFamily, logGCPolicy); err != nil {
                 return grpcutil.WrapIfTransient(err)
         }
         return nil
 }
 
 // getLogRowData loads the []byte contents of the supplied log row.
 //
 // If the row doesn't exist, storage.ErrDoesNotExist will be returned.
 func getLogRowData(row bigtable.Row) (data []byte, err error) {
         items, ok := row[logColumnFamily]
@@ skipping 25 matching lines @@
 
         // Get the specific ReadItem for our column
         colName := fmt.Sprintf("%s:%s", family, column)
         for _, item := range items {
                 if item.Column == colName {
                         return &item
                 }
         }
         return nil
 }