saji/gotelem
1
0
Fork 0
Code Issues 7 Actions Packages Releases 2 Activity

replace raw line with line num in import script

Browse source
This commit is contained in:
saji 2024-02-13 13:43:26 -06:00
parent 2d9d32dbf4
commit efc3ad684a
1 changed files with 8 additions and 4 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
cmd/gotelem/cli/client.go
View file

@ -60,6 +60,8 @@ var importCmd = &cli.Command{
Action: importAction, Action: importAction,
} }
// importAction peforms a file import to the database. It can use any of the parsers provided
// by logparsers. Adding new parsers there will work.
func importAction(ctx *cli.Context) error { func importAction(ctx *cli.Context) error {
path := ctx.Args().Get(0) path := ctx.Args().Get(0)
if path == "" { if path == "" {
@ -102,6 +104,7 @@ func importAction(ctx *cli.Context) error {
} }
var wg sync.WaitGroup var wg sync.WaitGroup
var linenum int64 = 0
n_unknown := 0 n_unknown := 0
n_error := 0 n_error := 0
for { for {
@ -115,17 +118,16 @@ func importAction(ctx *cli.Context) error {
f, err := pfun(line) f, err := pfun(line)
var idErr *skylab.UnknownIdError var idErr *skylab.UnknownIdError
if errors.As(err, &idErr) { if errors.As(err, &idErr) {
// unknown id
fmt.Printf("unknown id %v\n", idErr.Error()) fmt.Printf("unknown id %v\n", idErr.Error())
n_unknown++ n_unknown++
continue continue
} else if err != nil { } else if err != nil {
// TODO: we should consider absorbing all errors. fmt.Printf("got an error processing line %d: %v\n", linenum, err)
fmt.Printf("got an error processing '%s': %v\n", strings.TrimSpace(line), err)
n_error++ n_error++
continue continue
} }
eventsBatch[batchIdx] = f eventsBatch[batchIdx] = f
linenum++
batchIdx++ batchIdx++
if batchIdx >= int(bSize) { if batchIdx >= int(bSize) {
// flush it!!!! // flush it!!!!
@ -145,11 +147,13 @@ func importAction(ctx *cli.Context) error {
wg.Add(1) wg.Add(1)
go func() { go func() {
// n, err := db.AddEventsCtx(ctx.Context, eventsBatch[:batchIdx]...) // note the slice here! // since we don't do any modification
// we can avoid the copy
delegateInsert(eventsBatch[:batchIdx]) delegateInsert(eventsBatch[:batchIdx])
wg.Done() wg.Done()
}() }()
} }
// wait for any goroutines.
wg.Wait() wg.Wait()
fmt.Printf("import status: %d successful, %d unknown, %d errors\n", n_pkt.Load(), n_unknown, n_error) fmt.Printf("import status: %d successful, %d unknown, %d errors\n", n_pkt.Load(), n_unknown, n_error)