LogDog: BigTable batching schema.

server/logdog/storage/bigtable/bigtable.go

 // Copyright 2015 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 bigtable
 
 import (
         "fmt"
         "time"
 
         "github.com/luci/luci-go/common/errors"
         "github.com/luci/luci-go/common/grpcutil"
-        log "github.com/luci/luci-go/common/logging"
         "github.com/luci/luci-go/server/logdog/storage"
         "golang.org/x/net/context"
         "google.golang.org/cloud/bigtable"
 )
 
-// errStop is an internal sentinel error used to communicate "stop iteration"
-// to btTable.getLogData.
-var errStop = errors.New("stop")
-
 const (
         logColumnFamily = "log"
         logColumn       = "data"
 )
 
+// Limits taken from here:
+// https://cloud.google.com/bigtable/docs/schema-design
+const (
+        // bigTableRowMaxBytes is the maximum number of bytes that a single BigTable
+        // row may hold.
+        bigTableRowMaxBytes = 1024 * 1024 * 10 // 10MB
+)
+
 // btGetCallback is a callback that is invoked for each log data row returned
 // by getLogData.
 //
 // If an error is encountered, no more log data will be fetched. The error will
-// be propagated to the getLogData call unless the returned error is errStop, in
-// which case iteration will stop and getLogData will return nil.
+// be propagated to the getLogData call.
 type btGetCallback func(*rowKey, []byte) error

[nodir, 2016/03/30 18:29:42]

[general feedback] you don't have to prefix types with bt in bigtable package

[dnj, 2016/03/30 18:37:13]

Yep, but in this case, this is a BigTable-specific binding, as opposed to the
rest of the code, which is a higher level abstraction on top of the binding for
testing.

 
 // btTable is a general interface for BigTable operations intended to enable
 // unit tests to stub out BigTable without adding runtime inefficiency.
 type btTable interface {
         // putLogData adds new log data to BigTable.
         //
         // If data already exists for the named row, it will return storage.ErrExists
         // and not add the data.
         putLogData(context.Context, *rowKey, []byte) error
 
@@ skipping 13 matching lines @@
         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
 }
 
 func (t *btTableProd) putLogData(c context.Context, rk *rowKey, data []byte) error {
-        table, err := t.getLogTable()
-        if err != nil {
-                return err
-        }
-
         m := bigtable.NewMutation()
         m.Set(logColumnFamily, logColumn, bigtable.ServerTime, data)
         cm := bigtable.NewCondMutation(bigtable.RowKeyFilter(rk.encode()), nil, m)
 
         rowExists := false
-        if err = table.Apply(c, rk.encode(), cm, bigtable.GetCondMutationResult(&rowExists)); err != nil {
+        if err := t.logTable.Apply(c, rk.encode(), cm, bigtable.GetCondMutationResult(&rowExists)); err != nil {
                 return wrapTransient(err)
         }
         if rowExists {
                 return storage.ErrExists
         }
         return nil
 }
 
 func (t *btTableProd) getLogData(c context.Context, rk *rowKey, limit int, keysOnly bool, cb btGetCallback) error {
-        table, err := t.getLogTable()
-        if err != nil {
-                return err
-        }
         // Construct read options based on Get request.
         ropts := []bigtable.ReadOption{
                 bigtable.RowFilter(bigtable.FamilyFilter(logColumnFamily)),
                 bigtable.RowFilter(bigtable.ColumnFilter(logColumn)),
                 nil,
                 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())
 
         innerErr := error(nil)
-        err = table.ReadRows(c, rng, func(row bigtable.Row) bool {
+        err := t.logTable.ReadRows(c, rng, func(row bigtable.Row) bool {
                 data := []byte(nil)
                 if !keysOnly {
                         err := error(nil)
                         data, err = getLogData(row)
                         if err != nil {
                                 innerErr = storage.ErrBadData
                                 return false
                         }
                 }
 
                 drk, err := decodeRowKey(row.Key())
                 if err != nil {
-                        log.Fields{
-                                log.ErrorKey: err,
-                                "value":      row.Key(),
-                        }.Warningf(c, "Failed to parse row key.")
-                        innerErr = storage.ErrBadData
+                        innerErr = err
                         return false
                 }
 
                 if err := cb(drk, data); err != nil {
-                        if err != errStop {
-                                innerErr = err
-                        }
+                        innerErr = err
                         return false
                 }
                 return true
         }, ropts...)
         if err == nil {
                 err = innerErr
         }
         return wrapTransient(err)
 }
 
 func (t *btTableProd) setMaxLogAge(c context.Context, d time.Duration) error {
-        ac, err := t.getAdminClient()
-        if err != nil {
-                return err
-        }
-
         var logGCPolicy bigtable.GCPolicy
         if d > 0 {
                 logGCPolicy = bigtable.MaxAgePolicy(d)
         }
-        if err := ac.SetGCPolicy(c, t.LogTable, logColumnFamily, logGCPolicy); err != nil {
+        if err := t.adminClient.SetGCPolicy(c, t.LogTable, logColumnFamily, logGCPolicy); err != nil {
                 return wrapTransient(err)
         }
         return nil
 }
 
 // wrapTransient wraps the supplied error in an errors.TransientError if it is
 // transient.
 func wrapTransient(err error) error {
         if isTransient(err) {
                 err = errors.WrapTransient(err)
@@ skipping 31 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
 }