LogDog: CLI can use stream URLs, fix auth.

logdog/client/cli/subcommandCat.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 cli
 
 import (
-        "errors"
         "io"
         "os"
         "strconv"
         "strings"
         "time"
 
+        "github.com/luci/luci-go/common/errors"
         "github.com/luci/luci-go/common/flag/flagenum"
         log "github.com/luci/luci-go/common/logging"
         "github.com/luci/luci-go/common/proto/google"
         "github.com/luci/luci-go/common/proto/milo"
         "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/renderer"
         "github.com/luci/luci-go/logdog/common/types"
-        "github.com/luci/luci-go/luci_config/common/cfgtypes"
 
         "github.com/golang/protobuf/proto"
         "github.com/maruel/subcommands"
         "golang.org/x/net/context"
 )
 
 var errDatagramNotSupported = errors.New("datagram not supported")
 
 type timestampsFlag string
 
@@ skipping 52 matching lines @@
 }
 
 func (cmd *catCommandRun) Run(scApp subcommands.Application, args []string, _ subcommands.Env) int {
         a := scApp.(*application)
 
         if len(args) == 0 {
                 log.Errorf(a, "At least one log path must be supplied.")
                 return 1
         }
 
-        // Validate and construct our cat paths.
-        paths := make([]*streamPath, len(args))
+        // Validate and construct our cat addresses.
+        addrs := make([]*types.StreamAddr, len(args))
         for i, arg := range args {
+                // If the address parses as a URL, use it directly.
+                var err error
+                if addrs[i], err = types.ParseURL(arg); err == nil {
+                        continue
+                }
+
                 // User-friendly: trim any leading or trailing slashes from the path.
                 project, path, _, err := a.splitPath(arg)
                 if err != nil {
                         log.WithError(err).Errorf(a, "Invalid path specifier.")
                         return 1
                 }
 
-                sp := streamPath{project, types.StreamPath(path)}
-                if err := sp.path.Validate(); err != nil {
+                addr := types.StreamAddr{Project: project, Path: types.StreamPath(path)}
+                if err := addr.Path.Validate(); err != nil {
                         log.Fields{
                                 log.ErrorKey: err,
                                 "index":      i,
-                                "project":    sp.project,
-                                "path":       sp.path,
+                                "project":    addr.Project,
+                                "path":       addr.Path,
                         }.Errorf(a, "Invalid command-line stream path.")
                         return 1
                 }
 
-                paths[i] = &sp
+                if addr.Host, err = a.resolveHost(""); err != nil {
+                        err = errors.Annotate(err).Reason("failed to resolve host: %(host)q").
+                                D("host", addr.Host).Err()
+                        errors.Log(a, err)
+                        return 1
+                }
+
+                addrs[i] = &addr
         }
         if cmd.buffer <= 0 {
                 log.Fields{
                         "value": cmd.buffer,
                 }.Errorf(a, "Buffer size must be >0.")
         }
 
+        coords := make(map[string]*coordinator.Client, len(addrs))
+        for _, addr := range addrs {
+                if _, ok := coords[addr.Host]; ok {
+                        continue
+                }
+
+                var err error
+                if coords[addr.Host], err = a.coordinatorClient(addr.Host); err != nil {
+                        err = errors.Annotate(err).Reason("failed to create Coordinator client for %(host)q").
+                                D("host", addr.Host).Err()
+
+                        errors.Log(a, err)
+                        return 1
+                }
+        }
+
         tctx, _ := a.timeoutCtx(a)
-        for i, sp := range paths {
-                if err := cmd.catPath(tctx, a.coord, sp); err != nil {
+        for i, addr := range addrs {
+                if err := cmd.catPath(tctx, coords[addr.Host], addr); err != nil {
                         log.Fields{
                                 log.ErrorKey: err,
-                                "project":    sp.project,
-                                "path":       sp.path,
+                                "project":    addr.Project,
+                                "path":       addr.Path,
                                 "index":      i,
                         }.Errorf(a, "Failed to fetch log stream.")
 
                         if err == context.DeadlineExceeded {
                                 return 2
                         }
                         return 1
                 }
         }
 
         return 0
 }
 
-// streamPath is a single path to fetch.
-type streamPath struct {
-        project cfgtypes.ProjectName
-        path    types.StreamPath
-}
-
-func (cmd *catCommandRun) catPath(c context.Context, coord *coordinator.Client, sp *streamPath) error {
+func (cmd *catCommandRun) catPath(c context.Context, coord *coordinator.Client, addr *types.StreamAddr) error {
         // Pull stream information.
         src := coordinatorSource{
-                stream: coord.Stream(sp.project, sp.path),
+                stream: coord.Stream(addr.Project, addr.Path),
         }
         src.tidx = -1 // Must be set to probe for state.
 
         f := fetcher.New(c, fetcher.Options{
                 Source:      &src,
                 Index:       types.MessageIndex(cmd.index),
                 Count:       cmd.count,
                 BufferCount: cmd.fetchSize,
                 BufferBytes: int64(cmd.fetchBytes),
         })
@@ skipping 67 matching lines @@
                 }
 
                 if err := proto.MarshalText(w, pb); err != nil {
                         log.WithError(err).Errorf(c, "Failed to marshal datagram as text.")
                         return false
                 }
 
                 return true
         }
 }