Fix BigTable multi-entry row cached Tail.

logdog/common/storage/bigtable/logdog_bigtable_test/main.go

 // Copyright 2016 The LUCI Authors. All rights reserved.

[dnj, 2017/01/06 18:51:41]

You can probably not care about this. It's just a local BigTable testing utility
that is useful for reproducing storage/bigtable issues against a live BigTable
instance. I added a "rounds" field and enabled cache so I could reproduce this
issue locally.

 // Use of this source code is governed under the Apache License, Version 2.0
 // that can be found in the LICENSE file.
 
 // Package main implements a simple CLI tool to load and interact with storage
 // data in Google BigTable data.
 package main
 
 import (
         "bytes"
         "flag"
         "io"
         "os"
 
         "github.com/luci/luci-go/client/authcli"
         "github.com/luci/luci-go/common/auth"
         "github.com/luci/luci-go/common/cli"
         "github.com/luci/luci-go/common/config"
         "github.com/luci/luci-go/common/errors"
         log "github.com/luci/luci-go/common/logging"
         "github.com/luci/luci-go/common/logging/gologger"
         "github.com/luci/luci-go/logdog/common/storage"
         "github.com/luci/luci-go/logdog/common/storage/bigtable"
+        "github.com/luci/luci-go/logdog/common/storage/memory"
         "github.com/luci/luci-go/logdog/common/types"
 
         "github.com/golang/protobuf/proto"
         "github.com/maruel/subcommands"
         "golang.org/x/net/context"
         "google.golang.org/api/option"
 )
 
 ////////////////////////////////////////////////////////////////////////////////
 // main
@@ skipping 27 matching lines @@
                 return nil, errors.Annotate(err).Reason("failed to get token source").Err()
         }
 
         return bigtable.New(c, bigtable.Options{
                 Project:  app.btProject,
                 Instance: app.btInstance,
                 LogTable: app.btLogTable,
                 ClientOptions: []option.ClientOption{
                         option.WithTokenSource(tsrc),
                 },
+                Cache: &memory.Cache{},
         })
 }
 
 func mainImpl(c context.Context, args []string) int {
         c = gologger.StdConfig.Use(c)
 
         logConfig := log.Config{
                 Level: log.Warning,
         }
 
@@ skipping 87 matching lines @@
 ////////////////////////////////////////////////////////////////////////////////
 // Subcommand: get
 ////////////////////////////////////////////////////////////////////////////////
 
 type cmdRunGet struct {
         subcommands.CommandRunBase
 
         project string
         path    string
 
-        index int
-        limit int
+        index  int
+        limit  int
+        rounds int
 }
 
 var subcommandGet = subcommands.Command{
         UsageLine: "get",
         ShortDesc: "Performs a Storage Get operation.",
         CommandRun: func() subcommands.CommandRun {
                 var cmd cmdRunGet
 
                 cmd.Flags.StringVar(&cmd.project, "project", "", "Log stream project name.")
                 cmd.Flags.StringVar(&cmd.path, "path", "", "Log stream path.")
                 cmd.Flags.IntVar(&cmd.index, "index", 0, "The index to fetch.")
                 cmd.Flags.IntVar(&cmd.limit, "limit", 0, "The log entry limit.")
+                cmd.Flags.IntVar(&cmd.rounds, "rounds", 1, "Number of rounds to run.")
 
                 return &cmd
         },
 }
 
 func (cmd *cmdRunGet) Run(baseApp subcommands.Application, args []string, _ subcommands.Env) int {
         app, c := getApplication(baseApp)
 
         switch {
         case cmd.project == "":
                 log.Errorf(c, "Missing required argument (-project).")
                 return 1
         case cmd.path == "":
                 log.Errorf(c, "Missing required argument (-path).")
                 return 1
         }
 
         stClient, err := app.getBigTableClient(c)
         if err != nil {
                 renderErr(c, errors.Annotate(err).Reason("failed to create storage client").Err())
                 return 1
         }
         defer stClient.Close()
 
-        var innerErr error
-        err = stClient.Get(storage.GetRequest{
-                Project: config.ProjectName(cmd.project),
-                Path:    types.StreamPath(cmd.path),
-                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
+        for round := 0; round < cmd.rounds; round++ {
+                log.Infof(c, "Get round %d.", round+1)
+
+                var innerErr error
+                err = stClient.Get(storage.GetRequest{
+                        Project: config.ProjectName(cmd.project),
+                        Path:    types.StreamPath(cmd.path),
+                        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:
+                        renderErr(c, errors.Annotate(err).Reason("failed to process fetched log entries").Err())
+                        return 1
+
+                case err != nil:
+                        renderErr(c, errors.Annotate(err).Reason("Failed to Get log entries.").Err())
+                        return 1
                 }
+        }
 
-                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:
-                renderErr(c, errors.Annotate(err).Reason("failed to process fetched log entries").Err())
-                return 1
-
-        case err != nil:
-                renderErr(c, errors.Annotate(err).Reason("Failed to Get log entries.").Err())
-                return 1
-
-        default:
-                return 0
-        }
+        return 0
 }
 
 ////////////////////////////////////////////////////////////////////////////////
 // Subcommand: tail
 ////////////////////////////////////////////////////////////////////////////////
 
 type cmdRunTail struct {
         subcommands.CommandRunBase
 
         project string
         path    string
+        rounds  int
 }
 
 var subcommandTail = subcommands.Command{
         UsageLine: "tail",
         ShortDesc: "Performs a Storage Tail operation.",
         CommandRun: func() subcommands.CommandRun {
                 var cmd cmdRunTail
 
                 cmd.Flags.StringVar(&cmd.project, "project", "", "Log stream project name.")
                 cmd.Flags.StringVar(&cmd.path, "path", "", "Log stream path.")
+                cmd.Flags.IntVar(&cmd.rounds, "rounds", 1, "Number of rounds to run.")
 
                 return &cmd
         },
 }
 
 func (cmd *cmdRunTail) Run(baseApp subcommands.Application, args []string, _ subcommands.Env) int {
         app, c := getApplication(baseApp)
 
         switch {
         case cmd.project == "":
                 log.Errorf(c, "Missing required argument (-project).")
                 return 1
         case cmd.path == "":
                 log.Errorf(c, "Missing required argument (-path).")
                 return 1
         }
 
         stClient, err := app.getBigTableClient(c)
         if err != nil {
                 renderErr(c, errors.Annotate(err).Reason("failed to create storage client").Err())
                 return 1
         }
         defer stClient.Close()
 
-        e, err := stClient.Tail(config.ProjectName(cmd.project), types.StreamPath(cmd.path))
-        if err != nil {
-                renderErr(c, errors.Annotate(err).Reason("failed to tail log entries").Err())
-                return 1
+        for round := 0; round < cmd.rounds; round++ {
+                log.Infof(c, "Tail round %d.", round+1)
+                e, err := stClient.Tail(config.ProjectName(cmd.project), types.StreamPath(cmd.path))
+                if err != nil {
+                        renderErr(c, errors.Annotate(err).Reason("failed to tail log entries").Err())
+                        return 1
+                }
+
+                if e == nil {
+                        log.Infof(c, "No log data to tail.")
+                        continue
+                }
+
+                le, err := e.GetLogEntry()
+                if err != nil {
+                        renderErr(c, errors.Annotate(err).Reason("failed to unmarshal log entry").Err())
+                        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 {
+                        renderErr(c, errors.Annotate(err).Reason("failed to dump log entry").Err())
+                        return 1
+                }
         }
 
-        if e == nil {
-                log.Infof(c, "No log data to tail.")
-                return 0
-        }
-
-        le, err := e.GetLogEntry()
-        if err != nil {
-                renderErr(c, errors.Annotate(err).Reason("failed to unmarshal log entry").Err())
-                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 {
-                renderErr(c, errors.Annotate(err).Reason("failed to dump log entry").Err())
-                return 1
-        }
         return 0
 }