LogDog: Add project support to Storage.

server/logdog/storage/bigtable/storage.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 (
         "bytes"
         "errors"
         "fmt"
 
+        "github.com/luci/luci-go/common/config"
         "github.com/luci/luci-go/common/logdog/types"
         log "github.com/luci/luci-go/common/logging"
         "github.com/luci/luci-go/common/recordio"
         "github.com/luci/luci-go/server/logdog/storage"
         "golang.org/x/net/context"
         "google.golang.org/cloud"
         "google.golang.org/cloud/bigtable"
 )
 
 var (
@@ skipping 26 matching lines @@
 )
 
 var (
         // errStop is an internal sentinel error used to indicate "stop iteration"
         // to btTable.getLogData iterator.
         errStop = errors.New("bigtable: stop iteration")
 
         // errRepeatRequest is a transition sentinel that instructs us to repeat
         // our query with KeysOnly set to true.
         //
-        // This can be deleted once BigTable rows without a count have been aged
-        // off.
+        // TODO(dnj): This can be deleted once BigTable rows without a count have been
+        // aged off.
         errRepeatRequest = errors.New("bigtable: repeat request")
 )
 
 // Options is a set of configuration options for BigTable storage.
 type Options struct {
         // Project is the name of the project to connect to.
         Project string
         // Zone is the name of the zone to connect to.
         Zone string
         // Cluster is the name of the cluster to connect to.
@@ skipping 94 matching lines @@
 func (s *btStorage) Put(r storage.PutRequest) error {
         rw := rowWriter{
                 threshold: s.maxRowSize,
         }
 
         for len(r.Values) > 0 {
                 // Add the next entry to the writer.
                 if appended := rw.append(r.Values[0]); !appended {
                         // We have failed to append our maximum BigTable row size. Flush any
                         // currently-buffered row data and try again with an empty buffer.
-                        count, err := rw.flush(s, s.raw, r.Index, r.Path)
+                        count, err := rw.flush(s, s.raw, r.Index, r.Project, r.Path)
                         if err != nil {
                                 return err
                         }
 
                         if count == 0 {
                                 // Nothing was buffered, but we still couldn't append an entry. The
                                 // current entry is too large by itself, so we must fail.
                                 return fmt.Errorf("single row entry exceeds maximum size (%d > %d)", len(r.Values[0]), bigTableRowMaxBytes)
                         }
 
                         r.Index += types.MessageIndex(count)
                         continue
                 }
 
                 // We successfully appended this entry, so advance.
                 r.Values = r.Values[1:]
         }
 
         // Flush any buffered rows.
-        if _, err := rw.flush(s, s.raw, r.Index, r.Path); err != nil {
+        if _, err := rw.flush(s, s.raw, r.Index, r.Project, r.Path); err != nil {
                 return err
         }
         return nil
 }
 
 func (s *btStorage) Get(r storage.GetRequest, cb storage.GetCallback) error {
-        startKey := newRowKey(string(r.Path), int64(r.Index), 0)
+        startKey := newRowKey(string(r.Project), string(r.Path), int64(r.Index), 0)
         ctx := log.SetFields(s, log.Fields{
+                "project":     r.Project,
                 "path":        r.Path,
                 "index":       r.Index,
                 "limit":       r.Limit,
                 "startRowKey": startKey,
                 "keysOnly":    r.KeysOnly,
         })
 
         // If we issue a query and get back a legacy row, it will have no count
         // associated with it. We will fast-exit
 
@@ skipping 17 matching lines @@
 
                 switch {
                 case r.KeysOnly:
                         // Keys only query, so count is the authority.
                         if rk.count == 0 {
                                 // If it's zero, we are dealing with a legacy row, before we started
                                 // keeping count. A keys-only query is insufficient to get the
                                 // inforamtion that we need, so repeat this Get request with KeysOnly
                                 // set to false.
                                 //
-                                // NOTE: This logic can be removed once all uncounted rows are aged off.
+                                // TODO(dnj): This logic can be removed once all uncounted rows are aged
+                                // off.
                                 r.KeysOnly = false
                                 return errRepeatRequest
                         }
                         break
 
                 case rk.count == 0:
                         // This is a non-keys-only query, but we are dealing with a legacy row.
                         // Use the record count as the authority.
                         //
-                        // NOTE: This logic can be removed once all uncounted rows are aged off.
+                        // TODO(dnj): This logic can be removed once all uncounted rows are aged
+                        // off.
                         rk.count = int64(len(records))
 
                 case rk.count == int64(len(records)):
                         // This is the expected case, so we're set.
                         //
-                        // NOTE: Once uncounted rows are aged off, this is the only logic that
-                        // we need, in the form of a sanity assertion.
+                        // TODO(dnj): Once uncounted rows are aged off, this is the only logic
+                        // that we need, in the form of a sanity assertion.
                         break
 
                 default:
                         log.Fields{
                                 "count":       rk.count,
                                 "recordCount": len(records),
                         }.Errorf(ctx, "Record count doesn't match declared count.")
                         return storage.ErrBadData
                 }
 
@@ skipping 51 matching lines @@
 
         case errRepeatRequest:
                 return s.Get(r, cb)
 
         default:
                 log.WithError(err).Errorf(ctx, "Failed to retrieve row range.")
                 return err
         }
 }
 
-func (s *btStorage) Tail(p types.StreamPath) ([]byte, types.MessageIndex, error) {
+func (s *btStorage) Tail(project config.ProjectName, path types.StreamPath) ([]byte, types.MessageIndex, error) {
         ctx := log.SetFields(s, log.Fields{
-                "path": p,
+                "project": project,
+                "path":    path,
         })
 
         // Iterate through all log keys in the stream. Record the latest one.
-        rk := newRowKey(string(p), 0, 0)
+        rk := newRowKey(string(project), string(path), 0, 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,
                         "zone":       s.Zone,
@@ skipping 59 matching lines @@
         }
 
         // If we have exceeded our threshold, report a failure.
         appended = (w.buf.Len() <= w.threshold)
         if appended {
                 w.count++
         }
         return
 }
 
-func (w *rowWriter) flush(ctx context.Context, raw btTable, index types.MessageIndex, path types.StreamPath) (int, error) {
+func (w *rowWriter) flush(ctx context.Context, raw btTable, index types.MessageIndex,
+        project config.ProjectName, path types.StreamPath) (int, error) {
         flushCount := w.count
         if flushCount == 0 {
                 return 0, nil
         }
 
         // Write the current set of buffered rows to the table. Index on the LAST
         // row index.
         lastIndex := int64(index) + int64(flushCount) - 1
-        rk := newRowKey(string(path), lastIndex, int64(w.count))
+        rk := newRowKey(string(project), string(path), lastIndex, int64(w.count))
 
         log.Fields{
                 "rowKey":    rk,
+                "project":   project,
                 "path":      path,
                 "index":     index,
                 "lastIndex": lastIndex,
                 "count":     w.count,
                 "size":      w.buf.Len(),
         }.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
 }