LogDog: Fix archived log stream read errors.

appengine/logdog/coordinator/endpoints/logs/get.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 logs
 
 import (
         "net/url"
         "time"
 
@@ skipping 11 matching lines @@
         "golang.org/x/net/context"
         "google.golang.org/grpc/codes"
 )
 
 const (
         // getInitialArraySize is the initial amount of log slots to allocate for a
         // Get request.
         getInitialArraySize = 256
 
         // getBytesLimit is the maximum amount of data that we are willing to query.
-        // AppEngine limits our response size to 32MB. However, this limit applies
-        // to the raw recovered LogEntry data, so we'll artificially constrain this
-        // to 16MB so the additional JSON overhead doesn't kill it.
+        //
+        // We will limit byte responses to 16MB, based on the following constraints:
+        //      - AppEngine cannot respond with more than 32MB of data. This includes JSON
+        //        overhead, including notation and base64 data expansion.
+        //      - `urlfetch`, which is used for Google Cloud Storage (archival) responses,
+        //        cannot handle responses larger than 32MB.
         getBytesLimit = 16 * 1024 * 1024
 )
 
 // Get returns state and log data for a single log stream.
 func (s *Server) Get(c context.Context, req *logdog.GetRequest) (*logdog.GetResponse, error) {
         return s.getImpl(c, req, false)
 }
 
 // Tail returns the last log entry for a given log stream.
 func (s *Server) Tail(c context.Context, req *logdog.TailRequest) (*logdog.GetResponse, error) {
         r := logdog.GetRequest{
                 Path:  req.Path,
                 State: req.State,
         }
         return s.getImpl(c, &r, true)
 }
 
 // getImpl is common code shared between Get and Tail endpoints.
 func (s *Server) getImpl(c context.Context, req *logdog.GetRequest, tail bool) (*logdog.GetResponse, error) {
         svc := s.GetServices()
+        log.Fields{
+                "path":  req.Path,
+                "index": req.Index,
+                "tail":  tail,
+        }.Debugf(c, "Received get request.")
 
         // Fetch the log stream state for this log stream.
         u, err := url.Parse(req.Path)
         if err != nil {
                 log.Fields{
                         log.ErrorKey: err,
                         "path":       req.Path,
                 }.Errorf(c, "Could not parse path URL.")
                 return nil, grpcutil.Errf(codes.InvalidArgument, "invalid path encoding")
         }
@@ skipping 65 matching lines @@
 
         log.Fields{
                 "logCount": len(resp.Logs),
         }.Debugf(c, "Get request completed successfully.")
         return &resp, nil
 }
 
 func (s *Server) getLogs(c context.Context, svc coordinator.Services, req *logdog.GetRequest, tail bool,
         ls *coordinator.LogStream) (
         []*logpb.LogEntry, error) {
+        byteLimit := int(req.ByteCount)
+        if byteLimit <= 0 || byteLimit > getBytesLimit {
+                byteLimit = getBytesLimit
+        }
+
         var st storage.Storage
         if !ls.Archived() {
                 log.Debugf(c, "Log is not archived. Fetching from intermediate storage.")
 
                 // Logs are not archived. Fetch from intermediate storage.
                 var err error
                 st, err = svc.IntermediateStorage(c)
                 if err != nil {
                         return nil, err
                 }
         } else {
-                log.Debugf(c, "Log is archived. Fetching from archive storage.")
+                log.Fields{
+                        "indexURL":    ls.ArchiveIndexURL,
+                        "streamURL":   ls.ArchiveStreamURL,
+                        "archiveTime": ls.ArchivedTime,
+                }.Debugf(c, "Log is archived. Fetching from archive storage.")
+
                 var err error
                 gs, err := svc.GSClient(c)
                 if err != nil {
                         log.WithError(err).Errorf(c, "Failed to create Google Storage client.")
                         return nil, err
                 }
                 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:  ls.ArchiveIndexURL,
                         StreamURL: ls.ArchiveStreamURL,
                         Client:    gs,
+                        MaxBytes:  byteLimit,
                 })
                 if err != nil {
                         log.WithError(err).Errorf(c, "Failed to create Google Storage storage instance.")
                         return nil, err
                 }
         }
         defer st.Close()
 
         path := ls.Path()
 
         var fetchedLogs [][]byte
         var err error
         if tail {
                 fetchedLogs, err = getTail(c, st, path)
         } else {
-                fetchedLogs, err = getHead(c, req, st, path)
+                fetchedLogs, err = getHead(c, req, st, 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
 }
 
-func getHead(c context.Context, req *logdog.GetRequest, st storage.Storage, p types.StreamPath) ([][]byte, error) {
+func getHead(c context.Context, req *logdog.GetRequest, st storage.Storage, p types.StreamPath, byteLimit int) (
+        [][]byte, error) {
         c = log.SetFields(c, log.Fields{
                 "path":          p,
                 "index":         req.Index,
                 "count":         req.LogCount,
                 "bytes":         req.ByteCount,
                 "noncontiguous": req.NonContiguous,
         })
 
-        byteLimit := int(req.ByteCount)
-        if byteLimit <= 0 || byteLimit > getBytesLimit {
-                byteLimit = getBytesLimit
-        }
-
         // Allocate result logs array.
         logCount := int(req.LogCount)
         asz := getInitialArraySize
         if logCount > 0 && logCount < asz {
                 asz = logCount
         }
         logs := make([][]byte, 0, asz)
 
         sreq := storage.GetRequest{
                 Path:  p,
@@ skipping 52 matching lines @@
                         log.ErrorKey: err,
                         "delay":      delay,
                 }.Warningf(c, "Transient error while fetching tail log; retrying.")
         })
         if err != nil {
                 log.WithError(err).Errorf(c, "Failed to fetch tail log.")
                 return nil, err
         }
         return [][]byte{data}, err
 }