LogDog: Fix archival Get/Tail implementations.

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"
@@ skipping 249 matching lines @@
                         "startIndex": startIndex,
                 }.Debugf(ctx, "Punting row key range [%d - %d]...", startIndex, rk.index)
 
                 for index := startIndex; index <= rk.index; index++ {
                         // If we're not doing keys-only, consume the row.
                         var row []byte
                         if !r.KeysOnly {
                                 row, records = records[0], records[1:]
                         }
 
-                        if !cb(types.MessageIndex(index), row) {
+                        if !cb(storage.MakeEntry(row, types.MessageIndex(index))) {
                                 return errStop
                         }
                         r.Index = types.MessageIndex(index + 1)
 
                         // Artificially apply limit within our row records.
                         if limit > 0 {
                                 limit--
                                 if limit == 0 {
                                         return errStop
                                 }
                         }
                 }
                 return nil
         })
 
         switch err {
         case nil, errStop:
                 return nil
 
         default:
                 log.WithError(err).Errorf(ctx, "Failed to retrieve row range.")
                 return err
         }
 }
 
-func (s *btStorage) Tail(project config.ProjectName, path types.StreamPath) ([]byte, types.MessageIndex, error) {
+func (s *btStorage) Tail(project config.ProjectName, path types.StreamPath) (*storage.Entry, error) {
         ctx := log.SetFields(s, log.Fields{
                 "project": project,
                 "path":    path,
         })
 
         // Iterate through all log keys in the stream. Record the latest one.
         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,
                         "instance":   s.Instance,
                         "table":      s.LogTable,
                 }.Errorf(ctx, "Failed to scan for tail.")
         }
 
         if latest == nil {
                 // No rows for the specified stream.
-                return nil, 0, storage.ErrDoesNotExist
+                return nil, storage.ErrDoesNotExist
         }
 
         // 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
         })
         if err != nil && err != errStop {
                 log.Fields{
                         log.ErrorKey: err,
                         "project":    s.Project,
                         "instance":   s.Instance,
                         "table":      s.LogTable,
                 }.Errorf(ctx, "Failed to retrieve tail row.")
         }
 
-        return d, types.MessageIndex(latest.index), nil
+        return storage.MakeEntry(d, types.MessageIndex(latest.index)), nil
 }
 
 // rowWriter facilitates writing several consecutive data values to a single
 // BigTable row.
 type rowWriter struct {
         // buf is the current set of buffered data.
         buf bytes.Buffer
 
         // count is the number of rows in the writer.
         count int
@@ skipping 48 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
 }