Milo flex raw log viewer endpoint

milo/appengine/logs/logs.go

+package main
+
+import (
+        "errors"
+        "fmt"
+        "io"
+        "log"
+        "net/http"
+        "strings"
+        "sync"
+        "time"
+
+        "github.com/luci/luci-go/common/clock"
+        "github.com/luci/luci-go/grpc/prpc"
+        "github.com/luci/luci-go/logdog/api/logpb"
+        "github.com/luci/luci-go/logdog/client/coordinator"
+        "github.com/luci/luci-go/logdog/common/fetcher"
+        "github.com/luci/luci-go/logdog/common/types"
+        "github.com/luci/luci-go/luci_config/common/cfgtypes"
+        "github.com/luci/luci-go/server/auth"
+
+        "golang.org/x/net/context"
+)
+
+var errNoAuth = errors.New("no access")
+
+type streamInfo struct {
+        Project cfgtypes.ProjectName
+        Path    types.StreamPath
+
+        // Client is the HTTP client to use for LogDog communication.
+        Client *coordinator.Client
+}
+
+const noStreamDelay = 5 * time.Second
+
+// coordinatorSource is a fetcher.Source implementation that uses the
+// Coordiantor API.
+type coordinatorSource struct {
+        sync.Mutex
+
+        stream    *coordinator.Stream
+        tidx      types.MessageIndex
+        tailFirst bool
+
+        streamState *coordinator.LogStream
+}
+
+func (s *coordinatorSource) LogEntries(c context.Context, req *fetcher.LogRequest) (
+        []*logpb.LogEntry, types.MessageIndex, error) {
+        s.Lock()
+        defer s.Unlock()
+
+        params := append(make([]coordinator.GetParam, 0, 4),
+                coordinator.LimitBytes(int(req.Bytes)),
+                coordinator.LimitCount(req.Count),
+                coordinator.Index(req.Index),
+        )
+
+        // If we haven't terminated, use this opportunity to fetch/update our stream
+        // state.
+        var streamState coordinator.LogStream
+        reqStream := (s.streamState == nil || s.streamState.State.TerminalIndex < 0)
+        if reqStream {
+                params = append(params, coordinator.WithState(&streamState))
+        }
+
+        for {
+                logs, err := s.stream.Get(c, params...)
+                switch err {
+                case nil:
+                        if reqStream {
+                                s.streamState = &streamState
+                                s.tidx = streamState.State.TerminalIndex
+                        }
+                        return logs, s.tidx, nil
+
+                case coordinator.ErrNoSuchStream:
+                        log.Print("Stream does not exist. Sleeping pending registration.")
+
+                        // Delay, interrupting if our Context is interrupted.
+                        if tr := <-clock.After(c, noStreamDelay); tr.Incomplete() {
+                                return nil, 0, tr.Err
+                        }
+
+                default:
+                        if err != nil {
+                                return nil, 0, err
+                        }
+                }
+        }
+}
+
+func logHandler(c context.Context, w http.ResponseWriter, host, path string) error {
+        // TODO(hinoka): Move this to luci-config.
+        if !(host == "luci-logdog.appspot.com" || host == "luci-logdog-dev.appspot.com") {
+                return fmt.Errorf("unknown host %s", host)
+        }
+        spath := strings.SplitN(path, "/", 2)
+        project := cfgtypes.ProjectName(spath[0])
+        streamPath := types.StreamPath(spath[1])
+
+        t, err := auth.GetRPCTransport(c, auth.AsUser)
+        if err != nil {
+                return err
+        }
+        client := coordinator.NewClient(&prpc.Client{
+                C: &http.Client{
+                        Transport: t,
+                },
+                Host: host,
+        })
+        stream := client.Stream(project, streamPath)
+
+        // Pull stream information.
+        src := coordinatorSource{
+                stream: stream,
+                tidx:   -1, // Must be set to probe for state.
+        }
+        f := fetcher.New(c, fetcher.Options{
+                Source: &src,
+                Index:  types.MessageIndex(0),
+                Count:  0,
+                // Try to buffer as much as possible, with a large window, since this is
+                // basically a cloud-to-cloud connection.
+                BufferCount:    200,
+                BufferBytes:    int64(4 * 1024 * 1024),
+                PrefetchFactor: 10,
+        })
+
+        for {
+                // Read out of the buffer.  This _should_ be bottlenecked on the network
+                // connection between the Flex instance and the client, via Fprintf().
+                entry, err := f.NextLogEntry()
+                switch err {
+                case io.EOF:
+                        break // We're done.
+                case nil:
+                        // Nothing
+                case coordinator.ErrNoAccess:
+                        return errNoAuth // This will force a redirect
+                default:
+                        return err
+                }
+                content := entry.GetText()
+                if content == nil {
+                        break
+                }
+                for _, line := range content.Lines {
+                        fmt.Fprint(w, line.Value)
+                        fmt.Fprint(w, line.Delimiter)
+                }
+        }
+        return nil
+}