LogDog: Fix archival Get/Tail implementations.

logdog/server/archivist/storageSource.go

 // Copyright 2016 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 archivist
 
 import (
         "io"
 
-        "github.com/golang/protobuf/proto"
         "github.com/luci/luci-go/common/config"
         log "github.com/luci/luci-go/common/logging"
         "github.com/luci/luci-go/logdog/api/logpb"
         "github.com/luci/luci-go/logdog/common/storage"
         "github.com/luci/luci-go/logdog/common/types"
         "golang.org/x/net/context"
 )
 
 // storageSource is a renderer.Source that pulls log entries from intermediate
 // storage via its storage.Storage instance.
@@ skipping 11 matching lines @@
 }
 
 func (s *storageSource) bufferEntries(start types.MessageIndex) error {
         bytes := 0
 
         req := storage.GetRequest{
                 Project: s.project,
                 Path:    s.path,
                 Index:   start,
         }
-        return s.st.Get(req, func(idx types.MessageIndex, d []byte) bool {
-                le := logpb.LogEntry{}
-                if err := proto.Unmarshal(d, &le); err != nil {
-                        log.Fields{
-                                log.ErrorKey:  err,
-                                "streamIndex": idx,
-                        }.Errorf(s, "Failed to unmarshal LogEntry.")
+        return s.st.Get(req, func(e *storage.Entry) bool {
+                le, err := e.GetLogEntry()
+                if err != nil {
+                        log.WithError(err).Errorf(s, "Failed to unmarshal LogEntry.")
                         return false
                 }
-                s.buf = append(s.buf, &le)
+                s.buf = append(s.buf, le)
 
                 // Stop loading if we've reached or exceeded our buffer size.
-                bytes += len(d)
+                bytes += len(e.D)
                 return bytes < storageBufferSize
         })
 }
 
 func (s *storageSource) NextLogEntry() (*logpb.LogEntry, error) {
         if len(s.buf) == 0 {
                 s.buf = s.buf[:0]
                 if err := s.bufferEntries(s.lastIndex + 1); err != nil {
                         if err == storage.ErrDoesNotExist {
                                 log.Warningf(s, "Archive target stream does not exist in intermediate storage.")
@@ skipping 35 matching lines @@
                         "index":         sidx,
                         "terminalIndex": s.terminalIndex,
                 }.Warningf(s, "Discarding log entries beyond expected terminal index.")
                 return nil, io.EOF
         }
 
         s.lastIndex = sidx
         s.logEntryCount++
         return le, nil
 }