LogDog: Add collector service implementation.

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"
-        "strings"
 
         "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")
 
 // 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.
 type btGetCallback func(*rowKey, []byte) error
 
 // btTable is a general interface for BigTable operations intended to enable
 // unit tests to stub out BigTable without adding runtime inefficiency.
-//
-// If any of these methods fails with a transient error, it will be wrapped
-// as an errors.Transient error.
 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
 
         // getLogData retrieves rows belonging to the supplied stream record, starting
         // with the first index owned by that record. The supplied callback is invoked
         // once per retrieved row.
         //
         // rk is the starting row key.
         //
         // If the supplied limit is nonzero, no more than limit rows will be
         // retrieved.
         //
         // 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
 }
 
-// btTransientSubstrings is the set of known error substrings returned by
-// BigTable that indicate failures that aren't related to the specific data
-// content.
-var btTransientSubstrings = []string{
-        "Internal error encountered",
-        "interactive login is required",
-}
-
 // btTableProd is an implementation of the btTable interface that uses a real
 // production BigTable connection.
 type btTableProd struct {
         *bigtable.Table
 }
 
 func (t *btTableProd) putLogData(c context.Context, rk *rowKey, data []byte) error {
         m := bigtable.NewMutation()
         m.Set("log", "data", bigtable.ServerTime, data)
         cm := bigtable.NewCondMutation(bigtable.RowKeyFilter(rk.encode()), nil, m)
@@ skipping 70 matching lines @@
         if isTransient(err) {
                 err = errors.WrapTransient(err)
         }
         return err
 }
 
 // isTransient tests if a BigTable SDK error is transient.
 //
 // Since the BigTable API doesn't give us this information, we will identify
 // transient errors by parsing their error string :(
-//
-// TODO(dnj): File issue to add error qualifier functions to BigTable API.
 func isTransient(err error) bool {
-        if err == nil {
-                return false
-        }
-
-        msg := err.Error()
-        for _, s := range btTransientSubstrings {
-                if strings.Contains(msg, s) {
-                        return true
-                }
-        }
-        return false
+        return grpcutil.IsTransient(err)
 }
 
 // getLogData loads the "data" column from the "log" column family and returns
 // its []byte contents.
 //
 // If the row doesn't exist, storage.ErrDoesNotExist will be returned.
 func getLogData(row bigtable.Row) ([]byte, error) {
         ri := getReadItem(row, "log", "data")
         if ri == nil {
                 return nil, storage.ErrDoesNotExist
@@ skipping 11 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
 }