LogDog: Add log rendering view.

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"
 
         "github.com/golang/protobuf/proto"
         ds "github.com/luci/gae/service/datastore"
         "github.com/luci/luci-go/appengine/logdog/coordinator"
         "github.com/luci/luci-go/appengine/logdog/coordinator/config"
         "github.com/luci/luci-go/common/api/logdog_coordinator/logs/v1"
         "github.com/luci/luci-go/common/grpcutil"
         "github.com/luci/luci-go/common/logdog/types"
         log "github.com/luci/luci-go/common/logging"
         "github.com/luci/luci-go/common/proto/logdog/logpb"
         "github.com/luci/luci-go/common/retry"
         "github.com/luci/luci-go/server/logdog/storage"
-        "github.com/luci/luci-go/server/logdog/storage/archive"
         "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
+        getInitialArraySize = int64(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.
         getBytesLimit = 16 * 1024 * 1024
 )
 
 // Get returns state and log data for a single log stream.
 func (s *Server) Get(c context.Context, req *logs.GetRequest) (*logs.GetResponse, error) {
@@ skipping 87 matching lines @@
         }
 
         log.Fields{
                 "logCount": len(resp.Logs),
         }.Debugf(c, "Get request completed successfully.")
         return &resp, nil
 }
 
 func (s *Server) getLogs(c context.Context, req *logs.GetRequest, tail bool, ls *coordinator.LogStream) (
         []*logpb.LogEntry, error) {
-        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 = s.Storage(c)
-                if err != nil {
-                        return nil, err
-                }
-        } else {
-                log.Debugf(c, "Log is archived. Fetching from archive storage.")
-                var err error
-                gs, err := s.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,
-                })
-                if err != nil {
-                        log.WithError(err).Errorf(c, "Failed to create Google Storage storage instance.")
-                        return nil, err
-                }
+        st, err := s.StorageForLogStream(c, ls)
+        if err != nil {
+                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)
         }
         if err != nil {
                 log.WithError(err).Errorf(c, "Failed to fetch log records.")
                 return nil, err
         }
 
@@ skipping 21 matching lines @@
                 "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)
+        logCount := int64(req.LogCount)
         asz := getInitialArraySize
         if logCount > 0 && logCount < asz {
                 asz = logCount
         }
         logs := make([][]byte, 0, asz)
 
         sreq := storage.GetRequest{
                 Path:  p,
                 Index: types.MessageIndex(req.Index),
                 Limit: logCount,
         }
 
-        count := 0
         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 {
+                        if len(logs) > 0 && byteLimit-len(ld) < 0 {
                                 // Not the first log, and we've exceeded our byte limit.
                                 return false
                         }
                         byteLimit -= len(ld)
 
                         if !(req.NonContiguous || idx == sreq.Index) {
                                 return false
                         }
                         logs = append(logs, ld)
                         sreq.Index = idx + 1
-                        count++
-                        return !(logCount > 0 && count >= logCount)
+                        return !(logCount > 0 && int64(len(logs)) >= 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.")
         })
@@ skipping 20 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
 }