LogDog: Add log rendering view.

appengine/logdog/coordinator/logView/view.go

+// Copyright 2016 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 logView
+
+import (
+        "io"
+        "net/http"
+        "strings"
+        "time"
+
+        "github.com/golang/protobuf/proto"
+        "github.com/julienschmidt/httprouter"
+        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/logdog/fetcher"
+        "github.com/luci/luci-go/common/logdog/renderer"
+        "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/server/logdog/storage"
+        "github.com/luci/luci-go/server/middleware"
+        "golang.org/x/net/context"
+)
+
+const (
+        // DefaultViewTimeout is the default value for the Handler's Timeout.
+        //
+        // It targets the managed VM environment, which allows 24 hours. We stop
+        // 10 minutes short of that to avoid conflict.
+        DefaultViewTimeout = (24 * time.Hour) - (10 * time.Minute)
+
+        // viewLogDelay is the amount of time to wait in between polls when no more
+        // log entries are available during streaming.
+        viewLogDelay = (5 * time.Second)
+)
+
+// Handler offers an HTTP handler that will return the rendered contents of a
+// log stream.
+type Handler struct {
+        coordinator.Service
+
+        // timeout is the maximum amount of time that we are willing to handle a
+        // View request.
+        //
+        // This should be short of the maximum request time allowed by an AppEngine
+        // handler, else the instance may be terminated.
+        timeout time.Duration
+}
+
+// InstallHandlers installs the view handler into the supplied router.
+func (h *Handler) InstallHandlers(r *httprouter.Router, b middleware.Base) {
+        r.GET("/logs/view/*path", b(h.handle))
+}
+
+// handle will render the contents of a log stream to the supplied
+// ResponseWriter.
+//
+// It expects the stream hash/path to be embedded in p as "path".
+func (h *Handler) handle(c context.Context, w http.ResponseWriter, req *http.Request, p httprouter.Params) {
+        if rc := h.handleImpl(c, w, req, p); rc != 0 {
+                w.WriteHeader(rc)
+        }
+}
+
+// handleImpl handles the HTTP request. If it returns a non-zero value, that
+// will be used as the HTTP response code.
+func (h *Handler) handleImpl(c context.Context, w http.ResponseWriter, req *http.Request, p httprouter.Params) int {
+        // Get the requested log path/hash.
+        path := strings.TrimLeft(p.ByName("path"), "/")
+        log.Fields{
+                "path": path,
+        }.Infof(c, "Request to view log stream.")
+
+        ls, err := coordinator.NewLogStream(path)
+        if err != nil {
+                log.WithError(err).Errorf(c, "Invalid log sream path/hash value supplied.")
+                return http.StatusBadRequest
+        }
+
+        // Set a timeout on our Context.
+        timeout := h.timeout
+        if timeout <= 0 {
+                timeout = DefaultViewTimeout
+        }
+        c, cancelFunc := context.WithTimeout(c, timeout)
+        defer cancelFunc()
+
+        // Load the log stream metadata.
+        if err := ds.Get(c).Get(ls); err != nil {
+                log.WithError(err).Errorf(c, "Failed to load LogStream.")
+
+                switch err {
+                case ds.ErrNoSuchEntity:
+                        return http.StatusNotFound
+                default:
+                        return http.StatusInternalServerError
+                }
+        }
+
+        // The user can stream purged logs only if they're an administrator. If not,
+        // pretend that the stream does not exist.
+        if ls.Purged {
+                if err := config.IsAdminUser(c); err != nil {
+                        log.WithError(err).Warningf(c, "Non-admin user requested purged log stream.")
+                        return http.StatusNotFound
+                }
+        }
+
+        // Get our Storage instance.
+        st, err := h.StorageForLogStream(c, ls)
+        if err != nil {
+                log.WithError(err).Errorf(c, "Failed to get Storage instance.")
+                return http.StatusInternalServerError
+        }
+        defer st.Close()
+
+        // Fetch the requested log stream!
+        w.Header().Add("Content-Type", ls.ContentType)
+
+        f := fetcher.New(c, fetcher.Options{
+                Source: &storageLogSource{
+                        ls: ls,
+                        st: st,
+                },
+                Delay: viewLogDelay,
+        })
+        r := renderer.Renderer{
+                Source:    f,
+                Reproduce: true,
+        }
+
+        // Render log data to our HTTP response.
+        count, err := io.Copy(w, &r)
+        if err != nil {
+                log.WithError(err).Errorf(c, "Failed to output log stream.")
+                return http.StatusInternalServerError
+        }
+
+        log.Fields{
+                "bytes": count,
+        }.Infof(c, "Successfully wrote log stream.")
+        return 0
+}
+
+// storageLogSource is a stateful incremental implementation of the
+// fetcher.Source interface which uses direct storage access to pull log entries.
+//
+// The Fetcher will respond to Context cancellation, so we don't have to worry
+// about that here.
+type storageLogSource struct {
+        ls *coordinator.LogStream
+        st storage.Storage
+}
+
+func (f *storageLogSource) LogEntries(c context.Context, req *fetcher.LogRequest) (
+        logs []*logpb.LogEntry, tidx types.MessageIndex, err error) {
+        // If this log stream hasn't been terminated yet, re-load it and check if
+        // there is a terminal index.
+        tidx = types.MessageIndex(f.ls.TerminalIndex)
+        if tidx < 0 {
+                log.Debugf(c, "Log stream is non-terminal, refreshing.")
+                if err := ds.Get(c).Get(f.ls); err != nil {
+                        log.WithError(err).Errorf(c, "Failed to refresh log stream.")
+                }
+
+                tidx = types.MessageIndex(f.ls.TerminalIndex)
+                if tidx >= 0 {
+                        log.Fields{
+                                "terminalIndex": tidx,
+                        }.Debugf(c, "Log stream has been terminated.")
+                }
+        }
+
+        sreq := storage.GetRequest{
+                Path:  f.ls.Path(),
+                Index: types.MessageIndex(req.Index),
+        }
+
+        // If we have a terminal index, set a limit. This is probably unnecessary, but
+        // will handle the error case where logs past the terminal index are written
+        // to intermediate storage before a terminal index is registered.
+        if tidx >= 0 {
+                sreq.Limit = int64(tidx - req.Index + 1)
+                if sreq.Limit <= 0 {
+                        // No more log records in the stream.
+                        return
+                }
+        }
+
+        // Issue the Get request.
+        var logData [][]byte
+        bytes := int64(0)
+        err = f.st.Get(&sreq, func(idx types.MessageIndex, ld []byte) bool {
+                // Enforce contiguous log records.
+                if idx != sreq.Index {
+                        return false
+                }
+                logData = append(logData, ld)
+                sreq.Index++
+                bytes += int64(len(ld))
+
+                // Apply byte/count constraints.
+                return (!(req.Count > 0 && int64(len(logData)) >= req.Count) && !(req.Bytes > 0 && bytes >= req.Bytes))
+        })
+        if err != nil {
+                log.Fields{
+                        log.ErrorKey:   err,
+                        "initialIndex": req.Index,
+                        "nextIndex":    sreq.Index,
+                        "count":        len(logData),
+                }.Errorf(c, "Failed to fetch logs.")
+                return
+        }
+
+        // Convert the logs to protobufs.
+        logs = make([]*logpb.LogEntry, len(logData))
+        for i, ld := range logData {
+                le := logpb.LogEntry{}
+                if err = proto.Unmarshal(ld, &le); err != nil {
+                        log.Fields{
+                                log.ErrorKey: err,
+                                "logIndex":   i,
+                        }.Errorf(c, "Failed to unmarshal LogEntry from log data.")
+                        return
+                }
+                logs[i] = &le
+        }
+        return
+}