blob: 3f82fe1200e9ffcf2dbe73d90f4d30919a15614f [file] [log] [blame]
// Copyright 2016 The LUCI Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// Package main implements a simple CLI tool to load and interact with Google
// Storage archived data.
package main
import (
"bytes"
"context"
"flag"
"io"
"io/ioutil"
"math"
"os"
"strings"
"go.chromium.org/luci/auth"
"go.chromium.org/luci/auth/client/authcli"
"go.chromium.org/luci/common/cli"
"go.chromium.org/luci/common/data/rand/mathrand"
"go.chromium.org/luci/common/data/recordio"
"go.chromium.org/luci/common/errors"
"go.chromium.org/luci/common/gcloud/gs"
log "go.chromium.org/luci/common/logging"
"go.chromium.org/luci/common/logging/gologger"
"go.chromium.org/luci/logdog/api/logpb"
"go.chromium.org/luci/logdog/common/storage"
"go.chromium.org/luci/logdog/common/storage/archive"
"go.chromium.org/luci/logdog/common/types"
"github.com/golang/protobuf/proto"
"github.com/maruel/subcommands"
"go.chromium.org/luci/hardcoded/chromeinfra"
)
////////////////////////////////////////////////////////////////////////////////
// main
////////////////////////////////////////////////////////////////////////////////
type application struct {
cli.Application
authOpts auth.Options
}
func getApplication(base subcommands.Application) (*application, context.Context) {
app := base.(*application)
return app, app.Context(context.Background())
}
func (a *application) getGSClient(c context.Context) (gs.Client, error) {
authenticator := auth.NewAuthenticator(c, auth.OptionalLogin, a.authOpts)
transport, err := authenticator.Transport()
if err != nil {
return nil, errors.Annotate(err, "failed to get auth transport").Err()
}
return gs.NewProdClient(c, transport)
}
func mainImpl(c context.Context, defaultAuthOpts auth.Options, args []string) int {
c = gologger.StdConfig.Use(c)
logConfig := log.Config{
Level: log.Warning,
}
defaultAuthOpts.Scopes = append([]string{auth.OAuthScopeEmail}, gs.ReadOnlyScopes...)
var authFlags authcli.Flags
a := application{
Application: cli.Application{
Name: "Archive Storage Utility",
Title: "Archive Storage Utility",
Context: func(c context.Context) context.Context {
// Install configured logger.
c = logConfig.Set(gologger.StdConfig.Use(c))
return c
},
Commands: []*subcommands.Command{
subcommands.CmdHelp,
&subcommandDumpIndex,
&subcommandDumpStream,
&subcommandGet,
&subcommandTail,
authcli.SubcommandLogin(defaultAuthOpts, "auth-login", false),
authcli.SubcommandLogout(defaultAuthOpts, "auth-logout", false),
authcli.SubcommandInfo(defaultAuthOpts, "auth-info", false),
},
},
}
fs := flag.NewFlagSet("flags", flag.ExitOnError)
logConfig.AddFlags(fs)
authFlags.Register(fs, defaultAuthOpts)
fs.Parse(args)
// Process authentication options.
var err error
a.authOpts, err = authFlags.Options()
if err != nil {
log.WithError(err).Errorf(c, "Failed to create auth options.")
return 1
}
// Execute our subcommand.
return subcommands.Run(&a, fs.Args())
}
func main() {
mathrand.SeedRandomly()
os.Exit(mainImpl(context.Background(), chromeinfra.DefaultAuthOptions(), os.Args[1:]))
}
func renderErr(c context.Context, err error) {
log.Errorf(c, "Error encountered during operation: %s\n%s", err,
strings.Join(errors.RenderStack(err), "\n"))
}
////////////////////////////////////////////////////////////////////////////////
// Subcommand: dump-index
////////////////////////////////////////////////////////////////////////////////
type cmdRunDumpIndex struct {
subcommands.CommandRunBase
path string
}
var subcommandDumpIndex = subcommands.Command{
UsageLine: "dump-index",
ShortDesc: "Dumps the contents of an index protobuf.",
CommandRun: func() subcommands.CommandRun {
var cmd cmdRunDumpIndex
cmd.Flags.StringVar(&cmd.path, "path", "", "Google Storage path to the index protobuf.")
return &cmd
},
}
func (cmd *cmdRunDumpIndex) Run(baseApp subcommands.Application, args []string, _ subcommands.Env) int {
app, c := getApplication(baseApp)
if cmd.path == "" {
log.Errorf(c, "Missing required argument (-path).")
return 1
}
path := gs.Path(cmd.path)
client, err := app.getGSClient(c)
if err != nil {
log.WithError(err).Errorf(c, "Failed to create GS client.")
return 1
}
defer client.Close()
reader, err := client.NewReader(path, 0, -1)
if err != nil {
log.WithError(err).Errorf(c, "Failed to create GS reader.")
return 1
}
defer reader.Close()
data, err := ioutil.ReadAll(reader)
if err != nil {
log.WithError(err).Errorf(c, "Failed to read index data from GS.")
return 1
}
log.Debugf(c, "Loaded %d byte(s).", len(data))
var index logpb.LogIndex
if err := unmarshalAndDump(c, os.Stdout, data, &index); err != nil {
log.WithError(err).Errorf(c, "Failed to dump index protobuf.")
return 1
}
return 0
}
func unmarshalAndDump(c context.Context, out io.Writer, data []byte, msg proto.Message) error {
if data != nil {
if err := proto.Unmarshal(data, msg); err != nil {
log.WithError(err).Errorf(c, "Failed to unmarshal protobuf.")
return err
}
}
if err := proto.MarshalText(out, msg); err != nil {
log.WithError(err).Errorf(c, "Failed to dump protobuf to output.")
return err
}
return nil
}
////////////////////////////////////////////////////////////////////////////////
// Subcommand: dump-stream
////////////////////////////////////////////////////////////////////////////////
type cmdRunDumpStream struct {
subcommands.CommandRunBase
path string
}
var subcommandDumpStream = subcommands.Command{
UsageLine: "dump-stream",
ShortDesc: "Dumps the contents of a log stream protobuf.",
CommandRun: func() subcommands.CommandRun {
var cmd cmdRunDumpStream
cmd.Flags.StringVar(&cmd.path, "path", "", "Google Storage path to the stream protobuf.")
return &cmd
},
}
func (cmd *cmdRunDumpStream) Run(baseApp subcommands.Application, args []string, _ subcommands.Env) int {
app, c := getApplication(baseApp)
if cmd.path == "" {
log.Errorf(c, "Missing required argument (-path).")
return 1
}
path := gs.Path(cmd.path)
client, err := app.getGSClient(c)
if err != nil {
log.WithError(err).Errorf(c, "Failed to create GS client.")
return 1
}
defer client.Close()
reader, err := client.NewReader(path, 0, -1)
if err != nil {
log.WithError(err).Errorf(c, "Failed to create GS reader.")
return 1
}
defer reader.Close()
descFrame := true
err = dumpRecordIO(c, reader, func(c context.Context, d []byte) error {
if descFrame {
descFrame = false
var desc logpb.LogStreamDescriptor
if err := unmarshalAndDump(c, os.Stdout, d, &desc); err != nil {
return errors.Annotate(err, "failed to dump log descriptor").Err()
}
return nil
}
var entry logpb.LogEntry
if err := unmarshalAndDump(c, os.Stdout, d, &entry); err != nil {
return errors.Annotate(err, "failed to dump log entry").Err()
}
return nil
})
if err != nil {
renderErr(c, err)
return 1
}
return 0
}
////////////////////////////////////////////////////////////////////////////////
// Subcommand: get
////////////////////////////////////////////////////////////////////////////////
type cmdRunGet struct {
subcommands.CommandRunBase
indexPath string
streamPath string
index int
limit int
}
var subcommandGet = subcommands.Command{
UsageLine: "get",
ShortDesc: "Performs a Storage Get operation.",
CommandRun: func() subcommands.CommandRun {
var cmd cmdRunGet
cmd.Flags.StringVar(&cmd.indexPath, "index-path", "", "Google Storage path to the index protobuf.")
cmd.Flags.StringVar(&cmd.streamPath, "stream-path", "", "Google Storage path to the stream protobuf.")
cmd.Flags.IntVar(&cmd.index, "index", 0, "The index to fetch.")
cmd.Flags.IntVar(&cmd.limit, "limit", 0, "The log entry limit.")
return &cmd
},
}
func (cmd *cmdRunGet) Run(baseApp subcommands.Application, args []string, _ subcommands.Env) int {
app, c := getApplication(baseApp)
switch {
case cmd.indexPath == "":
log.Errorf(c, "Missing required argument (-index-path).")
return 1
case cmd.streamPath == "":
log.Errorf(c, "Missing required argument (-stream-path).")
return 1
}
client, err := app.getGSClient(c)
if err != nil {
log.WithError(err).Errorf(c, "Failed to create GS client.")
return 1
}
defer client.Close()
stClient, err := archive.New(archive.Options{
Index: gs.Path(cmd.indexPath),
Stream: gs.Path(cmd.streamPath),
Client: client,
})
if err != nil {
log.WithError(err).Errorf(c, "Failed to create storage client.")
return 1
}
defer stClient.Close()
var innerErr error
err = stClient.Get(c, storage.GetRequest{
Index: types.MessageIndex(cmd.index),
Limit: cmd.limit,
}, func(e *storage.Entry) bool {
le, err := e.GetLogEntry()
if err != nil {
log.WithError(err).Errorf(c, "Failed to unmarshal log entry.")
return false
}
log.Fields{
"index": le.StreamIndex,
}.Infof(c, "Fetched log entry.")
if innerErr = unmarshalAndDump(c, os.Stdout, nil, le); innerErr != nil {
return false
}
return true
})
switch {
case innerErr != nil:
log.WithError(innerErr).Errorf(c, "Failed to process fetched log entries.")
return 1
case err != nil:
log.WithError(err).Errorf(c, "Failed to Get log entries.")
return 1
default:
return 0
}
}
////////////////////////////////////////////////////////////////////////////////
// Subcommand: tail
////////////////////////////////////////////////////////////////////////////////
type cmdRunTail struct {
subcommands.CommandRunBase
indexPath string
streamPath string
}
var subcommandTail = subcommands.Command{
UsageLine: "tail",
ShortDesc: "Performs a Storage Tail operation.",
CommandRun: func() subcommands.CommandRun {
var cmd cmdRunTail
cmd.Flags.StringVar(&cmd.indexPath, "index-path", "", "Google Storage path to the index protobuf.")
cmd.Flags.StringVar(&cmd.streamPath, "stream-path", "", "Google Storage path to the stream protobuf.")
return &cmd
},
}
func (cmd *cmdRunTail) Run(baseApp subcommands.Application, args []string, _ subcommands.Env) int {
app, c := getApplication(baseApp)
switch {
case cmd.indexPath == "":
log.Errorf(c, "Missing required argument (-index-path).")
return 1
case cmd.streamPath == "":
log.Errorf(c, "Missing required argument (-stream-path).")
return 1
}
client, err := app.getGSClient(c)
if err != nil {
log.WithError(err).Errorf(c, "Failed to create GS client.")
return 1
}
defer client.Close()
stClient, err := archive.New(archive.Options{
Index: gs.Path(cmd.indexPath),
Stream: gs.Path(cmd.streamPath),
Client: client,
})
if err != nil {
log.WithError(err).Errorf(c, "Failed to create storage client.")
return 1
}
defer stClient.Close()
e, err := stClient.Tail(c, "", "")
if err != nil {
log.WithError(err).Errorf(c, "Failed to Tail log entries.")
return 1
}
if e == nil {
log.Infof(c, "No log data to tail.")
return 0
}
le, err := e.GetLogEntry()
if err != nil {
log.WithError(err).Errorf(c, "Failed to unmarshal log entry.")
return 1
}
log.Fields{
"index": le.StreamIndex,
"size": len(e.D),
}.Debugf(c, "Dumping tail entry.")
if err := unmarshalAndDump(c, os.Stdout, nil, le); err != nil {
log.WithError(err).Errorf(c, "Failed to dump tail entry.")
return 1
}
return 0
}
func dumpRecordIO(c context.Context, r io.Reader, cb func(context.Context, []byte) error) error {
var (
frameReader = recordio.NewReader(r, math.MaxInt64)
frameIndex = 0
buf bytes.Buffer
eof bool
)
for !eof {
frameSize, r, err := frameReader.ReadFrame()
switch err {
case nil:
break
case io.EOF:
log.Debugf(c, "Encountered EOF.")
eof = true
break
default:
return errors.Annotate(err, "Encountered error reading log stream.").
InternalReason("frameIndex(%d)", frameIndex).Err()
}
if frameSize > 0 {
buf.Reset()
buf.Grow(int(frameSize))
if _, err := buf.ReadFrom(r); err != nil {
return errors.Annotate(err, "Failed to buffer frame.").
InternalReason("frameIndex(%d)", frameIndex).Err()
}
if err := cb(c, buf.Bytes()); err != nil {
return err
}
}
log.Fields{
"index": frameIndex,
"size": buf.Len(),
}.Debugf(c, "Read frame.")
frameIndex++
}
return nil
}