LogDog: Fix archival Get/Tail implementations.

logdog/appengine/coordinator/endpoints/logs/get.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 logs
 
 import (
         "time"
 
-        "github.com/golang/protobuf/proto"
         ds "github.com/luci/gae/service/datastore"
         "github.com/luci/luci-go/common/config"
         log "github.com/luci/luci-go/common/logging"
         "github.com/luci/luci-go/common/retry"
         "github.com/luci/luci-go/grpc/grpcutil"
         "github.com/luci/luci-go/logdog/api/endpoints/coordinator/logs/v1"
         "github.com/luci/luci-go/logdog/api/logpb"
         "github.com/luci/luci-go/logdog/appengine/coordinator"
         "github.com/luci/luci-go/logdog/common/storage"
         "github.com/luci/luci-go/logdog/common/storage/archive"
@@ skipping 141 matching lines @@
                 defer func() {
                         if err := gs.Close(); err != nil {
                                 log.WithError(err).Warningf(c, "Failed to close Google Storage client.")
                         }
                 }()
 
                 st, err = archive.New(c, archive.Options{
                         IndexURL:  lst.ArchiveIndexURL,
                         StreamURL: lst.ArchiveStreamURL,
                         Client:    gs,
-                        MaxBytes:  byteLimit,

[dnj, 2016/10/19 23:18:59]

Getting rid of this is no big deal, since it's enforced in this code anyway.
This was a misguided attempt to not have GS fetch too much, but GS reader
fetches in chunks anyway, so we're fine.

                 })
                 if err != nil {
                         log.WithError(err).Errorf(c, "Failed to create Google Storage storage instance.")
                         return nil, err
                 }
         }
         defer st.Close()
 
         project, path := coordinator.Project(c), ls.Path()
 
-        var fetchedLogs [][]byte
+        var fetchedLogs []*logpb.LogEntry
         var err error
         if tail {
                 fetchedLogs, err = getTail(c, st, project, path)
         } else {
                 fetchedLogs, err = getHead(c, req, st, project, path, byteLimit)
         }
         if err != nil {
                 log.WithError(err).Errorf(c, "Failed to fetch log records.")
                 return nil, err
         }
 
-        logEntries := make([]*logpb.LogEntry, len(fetchedLogs))
-        for idx, ld := range fetchedLogs {
-                // Deserialize the log entry, then convert it to output value.
-                le := logpb.LogEntry{}
-                if err := proto.Unmarshal(ld, &le); err != nil {
-                        log.Fields{
-                                log.ErrorKey: err,
-                                "index":      idx,
-                        }.Errorf(c, "Failed to generate response log entry.")
-                        return nil, err
-                }
-                logEntries[idx] = &le
-        }
-        return logEntries, nil
+        return fetchedLogs, nil
 }
 
 func getHead(c context.Context, req *logdog.GetRequest, st storage.Storage, project config.ProjectName,
-        path types.StreamPath, byteLimit int) ([][]byte, error) {
+        path types.StreamPath, byteLimit int) ([]*logpb.LogEntry, error) {
         log.Fields{
                 "project":       project,
                 "path":          path,
                 "index":         req.Index,
                 "count":         req.LogCount,
                 "bytes":         req.ByteCount,
                 "noncontiguous": req.NonContiguous,
         }.Debugf(c, "Issuing Get request.")
 
         // Allocate result logs array.
         logCount := int(req.LogCount)
         asz := getInitialArraySize
         if logCount > 0 && logCount < asz {
                 asz = logCount
         }
-        logs := make([][]byte, 0, asz)
+        logs := make([]*logpb.LogEntry, 0, asz)
 
         sreq := storage.GetRequest{
                 Project: project,
                 Path:    path,
                 Index:   types.MessageIndex(req.Index),
                 Limit:   logCount,
         }
 
         count := 0
+        var ierr error
         err := retry.Retry(c, retry.TransientOnly(retry.Default), func() error {
                 // Issue the Get request. This may return a transient error, in which case
                 // we will retry.
-                return st.Get(sreq, func(idx types.MessageIndex, ld []byte) bool {
-                        if count > 0 && byteLimit-len(ld) < 0 {
+                return st.Get(sreq, func(e *storage.Entry) bool {
+                        var le *logpb.LogEntry
+                        if le, ierr = e.GetLogEntry(); ierr != nil {
+                                return false
+                        }
+
+                        if count > 0 && byteLimit-len(e.D) < 0 {
                                 // Not the first log, and we've exceeded our byte limit.
                                 return false
                         }
-                        byteLimit -= len(ld)
+                        byteLimit -= len(e.D)

[nodir, 2016/10/26 20:25:33]

you are mutating byteLimit in a retry loop. I think you have to copy byteLimit
at each attempt beginning

 
-                        if !(req.NonContiguous || idx == sreq.Index) {
+                        sidx, _ := e.GetStreamIndex() // GetLogEntry succeeded, so this must.
+                        if !(req.NonContiguous || sidx == sreq.Index) {
                                 return false
                         }
-                        logs = append(logs, ld)
-                        sreq.Index = idx + 1
+                        logs = append(logs, le)
+                        sreq.Index = sidx + 1
                         count++

[nodir, 2016/10/26 20:25:33]

same here

                         return !(logCount > 0 && count >= logCount)
                 })
         }, func(err error, delay time.Duration) {
                 log.Fields{
                         log.ErrorKey:   err,
                         "delay":        delay,
                         "initialIndex": req.Index,
                         "nextIndex":    sreq.Index,
                         "count":        len(logs),
                 }.Warningf(c, "Transient error while loading logs; retrying.")
         })
         switch err {
         case nil:
+                if ierr != nil {
+                        log.WithError(ierr).Errorf(c, "Bad log entry data.")
+                        return nil, ierr
+                }
                 return logs, nil
 
         case storage.ErrDoesNotExist:
                 return nil, nil
 
         default:
                 log.Fields{
                         log.ErrorKey:   err,
                         "initialIndex": req.Index,
                         "nextIndex":    sreq.Index,
                         "count":        len(logs),
                 }.Errorf(c, "Failed to execute range request.")
                 return nil, err
         }
 }
 
-func getTail(c context.Context, st storage.Storage, project config.ProjectName, path types.StreamPath) ([][]byte, error) {
+func getTail(c context.Context, st storage.Storage, project config.ProjectName, path types.StreamPath) (
+        []*logpb.LogEntry, error) {
         log.Fields{
                 "project": project,
                 "path":    path,
         }.Debugf(c, "Issuing Tail request.")
 
-        var data []byte
+        var e *storage.Entry
         err := retry.Retry(c, retry.TransientOnly(retry.Default), func() (err error) {
-                data, _, err = st.Tail(project, path)
+                e, err = st.Tail(project, path)
                 return
         }, func(err error, delay time.Duration) {
                 log.Fields{
                         log.ErrorKey: err,
                         "delay":      delay,
                 }.Warningf(c, "Transient error while fetching tail log; retrying.")
         })
         switch err {
         case nil:
-                return [][]byte{data}, nil
+                le, err := e.GetLogEntry()
+                if err != nil {
+                        log.WithError(err).Errorf(c, "Failed to load tail entry data.")
+                        return nil, err
+                }
+                return []*logpb.LogEntry{le}, nil
 
         case storage.ErrDoesNotExist:
                 return nil, nil
 
         default:
                 log.WithError(err).Errorf(c, "Failed to fetch tail log.")
                 return nil, err
         }
 }