LogDog: Implement prefix registration in Butler.

client/cmd/logdog_butler/main.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 main
 
 import (
         "errors"
         "flag"
         "fmt"
         "net/http"
         "os"
         "os/signal"
         "runtime/pprof"
         "sort"
         "strings"
         "time"
 
         "github.com/maruel/subcommands"
         "golang.org/x/net/context"
-        "golang.org/x/oauth2"
 
         "github.com/luci/luci-go/client/authcli"
         "github.com/luci/luci-go/client/internal/logdog/butler"
         "github.com/luci/luci-go/client/internal/logdog/butler/output"
         "github.com/luci/luci-go/common/auth"
         "github.com/luci/luci-go/common/cli"
         "github.com/luci/luci-go/common/clock/clockflag"
+        "github.com/luci/luci-go/common/config"
         "github.com/luci/luci-go/common/flag/multiflag"
-        "github.com/luci/luci-go/common/gcloud/pubsub"
         "github.com/luci/luci-go/common/logdog/types"
         log "github.com/luci/luci-go/common/logging"
         "github.com/luci/luci-go/common/logging/gologger"
         "github.com/luci/luci-go/common/paniccatcher"
 )
 
 const (
         // flagErrorReturnCode is returned when there is an error with the Butler's
         // command-line configuration.
         configErrorReturnCode = 2
 
         // runtimeErrorReturnCode is returned when the execution fails due to a Butler
         // error. This is intended to help differentiate Butler errors from
         // passthrough bootstrapped subprocess errors.
         //
         // This will only be returned for runtime errors. If there is a flag error
         // or a configuration error, standard Butler return codes (likely to overlap
         // with standard process return codes) will be used.
         runtimeErrorReturnCode = 250
 )
 
-// buildScopes consumes a series of independent OAuth2 scope strings and
-// combines them into a single deduplicated list.
-func buildScopes(parts ...[]string) []string {
-        result := []string{}
-        seen := make(map[string]bool)
-        for _, scopes := range parts {
-                for _, s := range scopes {
-                        if _, ok := seen[s]; ok {
-                                continue
-                        }
-                        result = append(result, s)
-                        seen[s] = true
-                }
-        }
-        return result
-}
-
 // application is the Butler application instance and its runtime configuration
 // and state.
 type application struct {
         cli.Application
         context.Context
 
-        butler       butler.Config
-        outputConfig outputConfigFlag
+        project       config.ProjectName
+        prefix        types.StreamName
+        outputWorkers int
+        outputConfig  outputConfigFlag
 
         authFlags authcli.Flags
 
         maxBufferAge clockflag.Duration
         noBufferLogs bool
 
         cpuProfile string
 
         client *http.Client
 
         // ncCtx is a context that will not be cancelled when cancelFunc is called.
         ncCtx      context.Context
         cancelFunc func()
-
-        output output.Output
 }
 
 func (a *application) addFlags(fs *flag.FlagSet) {
         a.outputConfig.Output = os.Stdout
         a.outputConfig.Description = "Select and configure message output adapter."
         a.outputConfig.Options = []multiflag.Option{
                 multiflag.HelpOption(&a.outputConfig.MultiFlag),
         }
 
         // Add registered conditional (build tag) options.
         for _, f := range getOutputFactories() {
                 a.outputConfig.AddFactory(f)
         }
 
         a.maxBufferAge = clockflag.Duration(butler.DefaultMaxBufferAge)
 
-        fs.Var(&a.butler.Project, "project",
+        fs.Var(&a.project, "project",
                 "The log prefix's project name (required).")
-        fs.Var(&a.butler.Prefix, "prefix",
+        fs.Var(&a.prefix, "prefix",
                 "Prefix to apply to all stream names.")
         fs.Var(&a.outputConfig, "output",
                 "The output name and configuration. Specify 'help' for more information.")
         fs.StringVar(&a.cpuProfile,
                 "cpuprofile", "", "If specified, enables CPU profiling and profiles to the specified path.")
-        fs.IntVar(&a.butler.OutputWorkers, "output-workers", butler.DefaultOutputWorkers,
+        fs.IntVar(&a.outputWorkers, "output-workers", butler.DefaultOutputWorkers,
                 "The maximum number of parallel output dispatches.")
         fs.Var(&a.maxBufferAge, "output-max-buffer-age",
                 "Send buffered messages if they've been held for longer than this period.")
         fs.BoolVar(&a.noBufferLogs, "output-no-buffer", false,
                 "If true, dispatch logs immediately. Setting this flag simplifies output at the expense "+
                         "of wire-format efficiency.")
 }
 
 func (a *application) authenticator(ctx context.Context) (*auth.Authenticator, error) {
         opts, err := a.authFlags.Options()
         if err != nil {
                 return nil, err
         }
         return auth.NewAuthenticator(ctx, auth.SilentLogin, opts), nil
 }
 
-func (a *application) authenticatedClient(ctx context.Context) (*http.Client, error) {
-        if a.client == nil {
-                authenticator, err := a.authenticator(ctx)
-                if err != nil {
-                        return nil, err
-                }
-
-                client, err := authenticator.Client()
-                if err != nil {
-                        return nil, err
-                }
-                a.client = client
-        }
-        return a.client, nil
-}
-
-func (a *application) tokenSource(ctx context.Context) (oauth2.TokenSource, error) {
-        authenticator, err := a.authenticator(ctx)
-        if err != nil {
-                return nil, err
-        }
-        return authenticator.TokenSource(), nil
-}
-
 func (a *application) configOutput() (output.Output, error) {
         factory := a.outputConfig.getFactory()
         if factory == nil {
                 return nil, errors.New("main: No output is configured")
         }
 
         output, err := factory.configOutput(a)
         if err != nil {
                 return nil, err
         }
 
         return output, nil
 }
 
-// An execution harness that adds application-level management to a Butler run.
-func (a *application) Main(runFunc func(b *butler.Butler) error) error {
+// runWithButler is an execution harness that adds application-level management
+// to a Butler run.
+func (a *application) runWithButler(out output.Output, runFunc func(b *butler.Butler) error) error {
         // Enable CPU profiling if specified
         if a.cpuProfile != "" {
                 f, err := os.Create(a.cpuProfile)
                 if err != nil {
                         return fmt.Errorf("failed to create CPU profile output: %v", err)
                 }
                 pprof.StartCPUProfile(f)
                 defer pprof.StopCPUProfile()
         }
 
-        // Generate a prefix secret for this Butler session.
-        var err error
-        if a.butler.Secret, err = types.NewPrefixSecret(); err != nil {
-                return fmt.Errorf("failed to generate prefix secret: %s", err)
+        // Instantiate our Butler.
+        butlerOpts := butler.Config{
+                Project:       a.project,
+                Prefix:        a.prefix,
+                MaxBufferAge:  time.Duration(a.maxBufferAge),
+                BufferLogs:    !a.noBufferLogs,
+                Output:        out,
+                OutputWorkers: a.outputWorkers,
+                TeeStdout:     os.Stdout,
+                TeeStderr:     os.Stderr,
         }
-
-        // Instantiate our Butler.
-        a.butler.MaxBufferAge = time.Duration(a.maxBufferAge)
-        a.butler.BufferLogs = !a.noBufferLogs
-        a.butler.Output = a.output
-        a.butler.TeeStdout = os.Stdout
-        a.butler.TeeStderr = os.Stderr
-        if err := a.butler.Validate(); err != nil {
-                return err
-        }
-
-        b, err := butler.New(a, a.butler)
+        b, err := butler.New(a, butlerOpts)
         if err != nil {
                 return err
         }
 
         // Log the Butler's emitted streams.
         defer func() {
-                if r := a.output.Record(); r != nil {
+                if r := out.Record(); r != nil {
                         // Log detail stream record.
                         streams := make([]string, 0, len(r.Streams))
                         for k := range r.Streams {
                                 streams = append(streams, string(k))
                         }
                         sort.Strings(streams)
 
                         for i, stream := range streams {
                                 rec := r.Streams[types.StreamPath(stream)]
 
                                 ranges := make([]string, len(rec.Ranges))
                                 for i, rng := range rec.Ranges {
                                         ranges[i] = rng.String()
                                 }
                                 log.Infof(a, "%d) Stream [%s]: %s", i, stream, strings.Join(ranges, " "))
                         }
                 } else {
                         // No record; display stream overview.
                         s := b.Streams()
                         paths := make([]types.StreamPath, len(s))
                         for i, sn := range s {
-                                paths[i] = a.butler.Prefix.Join(sn)
+                                paths[i] = a.prefix.Join(sn)
                         }
                         log.Fields{
                                 "count":   len(paths),
                                 "streams": paths,
                         }.Infof(a, "Butler emitted %d stream(s).", len(paths))
                 }
         }()
 
         // Execute our Butler run function with the instantiated Butler.
         if err := runFunc(b); err != nil {
                 log.Fields{
                         log.ErrorKey: err,
                 }.Errorf(a, "Butler terminated with error.")
                 a.cancelFunc()
         }
 
         return b.Wait()
 }
 
 func mainImpl(ctx context.Context, argv []string) int {
         authOptions := auth.Options{
-                Scopes: buildScopes(
-                        []string{auth.OAuthScopeEmail},
-                        pubsub.PublisherScopes,
-                ),
+                Scopes: allOutputScopes(),
         }
 
         a := &application{
                 Context: ctx,
                 Application: cli.Application{
                         Name:    "butler",
                         Title:   "Log collection and streaming bootstrap.",
                         Context: func(context.Context) context.Context { return ctx },
                         Commands: []*subcommands.Command{
                                 subcommands.CmdHelp,
@@ skipping 18 matching lines @@
 
         // Parse the top-level flag set.
         if err := flags.Parse(argv); err != nil {
                 log.WithError(err).Errorf(a, "Failed to parse command-line.")
                 return configErrorReturnCode
         }
 
         a.Context = logConfig.Set(a.Context)
 
         // TODO(dnj): Force all invocations to supply a Project.
-        if a.butler.Project != "" {
-                if err := a.butler.Project.Validate(); err != nil {
+        if a.project != "" {
+                if err := a.project.Validate(); err != nil {
                         log.WithError(err).Errorf(a, "Invalid project (-project).")
                         return configErrorReturnCode
                 }
         }
 
         // Validate our Prefix; generate a user prefix if one was not supplied.
-        prefix := a.butler.Prefix
+        prefix := a.prefix
         if prefix == "" {
                 // Auto-generate a prefix.
                 prefix, err := generateRandomUserPrefix(a)
                 if err != nil {
                         log.WithError(err).Errorf(a, "Failed to generate user prefix.")
                         return configErrorReturnCode
                 }
-                a.butler.Prefix = prefix
+                a.prefix = prefix
         }
 
         // Signal handler to catch 'Control-C'. This will gracefully shutdown the
         // butler the first time a signal is received. It will die abruptly if the
         // signal continues to be received.
         a.ncCtx = a.Context
         a.Context, a.cancelFunc = context.WithCancel(a.Context)
         signalC := make(chan os.Signal, 1)
         signal.Notify(signalC, os.Interrupt)
         go func() {
@@ skipping 10 matching lines @@
                                 os.Exit(1)
                         }
                 }
         }()
         defer func() {
                 signal.Stop(signalC)
                 close(signalC)
         }()
 
         log.Fields{
-                "prefix": a.butler.Prefix,
+                "prefix": a.prefix,
         }.Infof(a, "Using session prefix.")
-        if err := a.butler.Prefix.Validate(); err != nil {
+        if err := a.prefix.Validate(); err != nil {
                 log.WithError(err).Errorf(a, "Invalid session prefix.")
                 return configErrorReturnCode
         }
 
-        // Configure our Butler Output.
-        var err error
-        a.output, err = a.configOutput()
-        if err != nil {
-                log.WithError(err).Errorf(a, "Failed to create output instance.")
-                return runtimeErrorReturnCode
-        }
-        defer a.output.Close()
-
         // Run our subcommand (and parse subcommand flags).
         return subcommands.Run(a, flags.Args())
 }
 
 // Main execution function. This immediately jumps to 'mainImpl' and uses its
 // result as an exit code.
 func main() {
         ctx := context.Background()
         ctx = gologger.StdConfig.Use(ctx)
 
         // Exit with the specified return code.
         rc := 0
         defer func() {
                 log.Infof(log.SetField(ctx, "returnCode", rc), "Terminating.")
                 os.Exit(rc)
         }()
 
         paniccatcher.Do(func() {
                 rc = mainImpl(ctx, os.Args[1:])
         }, func(p *paniccatcher.Panic) {
                 log.Fields{
                         "panic.error": p.Reason,
                 }.Errorf(ctx, "Panic caught in main:\n%s", p.Stack)
                 rc = runtimeErrorReturnCode
         })
 }