More logging with context

cmd/transitland/main.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"context"
 	_ "embed"
 	"os"
 
@@ -31,7 +32,7 @@ func (cmd *versionCommand) Parse(args []string) error {
 	return nil
 }
 
-func (cmd *versionCommand) Run() error {
+func (cmd *versionCommand) Run(ctx context.Context) error {
 	log.Print("transitland-lib version: %s", tl.Version.Tag)
 	log.Print("transitland-lib commit: https://github.com/interline-io/transitland-lib/commit/%s (time: %s)", tl.Version.Commit, tl.Version.CommitTime)
 	log.Print("GTFS specification version: https://github.com/google/transit/blob/%s/gtfs/spec/en/reference.md", tl.GTFSVERSION)

cmds/copy_cmd.go

@@ -1,6 +1,7 @@
 package cmds
 
 import (
+	"context"
 	"errors"
 
 	"github.com/interline-io/transitland-lib/adapters"
@@ -62,7 +63,7 @@ func (cmd *CopyCommand) Parse(args []string) error {
 	return nil
 }
 
-func (cmd *CopyCommand) Run() error {
+func (cmd *CopyCommand) Run(ctx context.Context) error {
 	// Reader / Writer
 	reader, err := ext.OpenReader(cmd.readerPath)
 	if err != nil {

cmds/delete_cmd.go

@@ -1,6 +1,7 @@
 package cmds
 
 import (
+	"context"
 	"database/sql"
 	"errors"
 	"fmt"
@@ -56,7 +57,7 @@ func (cmd *DeleteCommand) Parse(args []string) error {
 }
 
 // Run this command
-func (cmd *DeleteCommand) Run() error {
+func (cmd *DeleteCommand) Run(ctx context.Context) error {
 	if cmd.Adapter == nil {
 		writer, err := tldb.OpenWriter(cmd.DBURL, true)
 		if err != nil {

cmds/e2e_test.go

@@ -3,6 +3,7 @@ package cmds
 // End to end tests for sync, fetch, and import
 
 import (
+	"context"
 	"io/ioutil"
 	"net/http"
 	"net/http/httptest"
@@ -141,7 +142,7 @@ func TestE2E(t *testing.T) {
 					FetchedAt: time.Now(),
 				},
 			}
-			if err := fetch.Run(); err != nil {
+			if err := fetch.Run(context.Background()); err != nil {
 				t.Fatal(err)
 			}
 
@@ -155,7 +156,7 @@ func TestE2E(t *testing.T) {
 					Activate: tc.activate,
 				},
 			}
-			if err := impcmd.Run(); err != nil {
+			if err := impcmd.Run(context.Background()); err != nil {
 				t.Fatal(err)
 			}
 
@@ -169,7 +170,7 @@ func TestE2E(t *testing.T) {
 					Workers:      1,
 					Adapter:      atx,
 				}
-				if err := unimpcmd.Run(); err != nil {
+				if err := unimpcmd.Run(context.Background()); err != nil {
 					t.Fatal(err)
 				}
 			}

cmds/extract_cmd.go

@@ -1,6 +1,7 @@
 package cmds
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"strconv"
@@ -139,7 +140,7 @@ func (cmd *ExtractCommand) Parse(args []string) error {
 	return nil
 }
 
-func (cmd *ExtractCommand) Run() error {
+func (cmd *ExtractCommand) Run(ctx context.Context) error {
 	// Reader / Writer
 	reader, err := ext.OpenReader(cmd.readerPath)
 	if err != nil {

cmds/fetch_cmd.go

@@ -1,6 +1,7 @@
 package cmds
 
 import (
+	"context"
 	"database/sql"
 	"errors"
 	"fmt"
@@ -76,7 +77,7 @@ func (cmd *FetchCommand) Parse(args []string) error {
 }
 
 // Run executes this command.
-func (cmd *FetchCommand) Run() error {
+func (cmd *FetchCommand) Run(ctx context.Context) error {
 	// Init
 	if cmd.Workers < 1 {
 		cmd.Workers = 1
@@ -193,7 +194,7 @@ func (cmd *FetchCommand) Run() error {
 	var wg sync.WaitGroup
 	for w := 0; w < cmd.Workers; w++ {
 		wg.Add(1)
-		go fetchWorker(cmd.Adapter, cmd.DryRun, jobs, results, &wg)
+		go fetchWorker(ctx, cmd.Adapter, cmd.DryRun, jobs, results, &wg)
 	}
 	wg.Wait()
 	close(results)
@@ -231,7 +232,7 @@ type fetchJob struct {
 	fetch.Options
 }
 
-func fetchWorker(adapter tldb.Adapter, DryRun bool, jobs <-chan fetchJob, results chan<- FetchCommandResult, wg *sync.WaitGroup) {
+func fetchWorker(ctx context.Context, adapter tldb.Adapter, DryRun bool, jobs <-chan fetchJob, results chan<- FetchCommandResult, wg *sync.WaitGroup) {
 	for job := range jobs {
 		// Start
 		log.Infof("Feed %s: start", job.OnestopID)
@@ -245,7 +246,7 @@ func fetchWorker(adapter tldb.Adapter, DryRun bool, jobs <-chan fetchJob, result
 		t := time.Now()
 		fetchError := adapter.Tx(func(atx tldb.Adapter) error {
 			var fetchError error
-			result, fetchError = fetch.StaticFetch(atx, job.Options)
+			result, fetchError = fetch.StaticFetch(ctx, atx, job.Options)
 			return fetchError
 		})
 		t2 := float64(time.Now().UnixNano()-t.UnixNano()) / 1e9 // 1000000000.0

cmds/fetch_cmd_test.go

@@ -1,6 +1,7 @@
 package cmds
 
 import (
+	"context"
 	"io/ioutil"
 	"net/http"
 	"net/http/httptest"
@@ -80,7 +81,7 @@ func TestFetchCommand(t *testing.T) {
 			if err := c.Parse(exp.command); err != nil {
 				t.Fatal(err)
 			}
-			if err := c.Run(); err != nil && exp.fatalErrorContains != "" {
+			if err := c.Run(context.Background()); err != nil && exp.fatalErrorContains != "" {
 				if !strings.Contains(err.Error(), exp.fatalErrorContains) {
 					t.Errorf("got '%s' error, expected to contain '%s'", err.Error(), exp.fatalErrorContains)
 				}

cmds/format_cmd.go

@@ -2,6 +2,7 @@ package cmds
 
 import (
 	"bytes"
+	"context"
 	"errors"
 	"fmt"
 	"os"
@@ -38,7 +39,7 @@ func (cmd *FormatCommand) Parse(args []string) error {
 }
 
 // Run this command.
-func (cmd *FormatCommand) Run() error {
+func (cmd *FormatCommand) Run(ctx context.Context) error {
 	filename := cmd.Filename
 	if filename == "" {
 		return errors.New("must specify filename")

cmds/import_cmd.go

@@ -1,6 +1,7 @@
 package cmds
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"os"
@@ -112,8 +113,8 @@ func (cmd *ImportCommand) Parse(args []string) error {
 	return nil
 }
 
-// Run this command
-func (cmd *ImportCommand) Run() error {
+// Run this command) Run(ctx context.Context) error
+func (cmd *ImportCommand) Run(ctx context.Context) error {
 	if cmd.Workers < 1 {
 		cmd.Workers = 1
 	}
@@ -196,7 +197,7 @@ func (cmd *ImportCommand) Run() error {
 	var wg sync.WaitGroup
 	for w := 0; w < cmd.Workers; w++ {
 		wg.Add(1)
-		go dmfrImportWorker(cmd.Adapter, cmd.DryRun, jobs, results, &wg)
+		go dmfrImportWorker(ctx, cmd.Adapter, cmd.DryRun, jobs, results, &wg)
 	}
 	wg.Wait()
 	close(results)
@@ -220,7 +221,7 @@ func (cmd *ImportCommand) Run() error {
 	return nil
 }
 
-func dmfrImportWorker(adapter tldb.Adapter, dryrun bool, jobs <-chan importer.Options, results chan<- ImportCommandResult, wg *sync.WaitGroup) {
+func dmfrImportWorker(ctx context.Context, adapter tldb.Adapter, dryrun bool, jobs <-chan importer.Options, results chan<- ImportCommandResult, wg *sync.WaitGroup) {
 	type qr struct {
 		FeedVersionID   int
 		FeedID          int
@@ -239,7 +240,7 @@ func dmfrImportWorker(adapter tldb.Adapter, dryrun bool, jobs <-chan importer.Op
 		}
 		log.Infof("Feed %s (id:%d): FeedVersion %s (id:%d): begin", q.FeedOnestopID, q.FeedID, q.FeedVersionSHA1, q.FeedVersionID)
 		t := time.Now()
-		result, err := importer.ImportFeedVersion(adapter, opts)
+		result, err := importer.ImportFeedVersion(ctx, adapter, opts)
 		t2 := float64(time.Now().UnixNano()-t.UnixNano()) / 1e9 // 1000000000.0
 		if err != nil {
 			log.Errorf("Feed %s (id:%d): FeedVersion %s (id:%d): critical failure, rolled back: %s (t:%0.2fs)", q.FeedOnestopID, q.FeedID, q.FeedVersionSHA1, q.FeedVersionID, result.FeedVersionImport.ExceptionLog, t2)

cmds/lint_cmd.go

@@ -2,6 +2,7 @@ package cmds
 
 import (
 	"bytes"
+	"context"
 	"errors"
 	"fmt"
 	"io/ioutil"
@@ -41,7 +42,7 @@ func (cmd *LintCommand) Parse(args []string) error {
 }
 
 // Run this command.
-func (cmd *LintCommand) Run() error {
+func (cmd *LintCommand) Run(ctx context.Context) error {
 	var fileErrors []string
 	for _, filename := range cmd.Filenames {
 		// first validate DMFR
@@ -63,12 +64,11 @@ func (cmd *LintCommand) Run() error {
 
 		// load JSON
 		originalJsonString := string(rawJson)
-		formattedJsonString := string(buf.Bytes())
+		formattedJsonString := buf.String()
 
 		// Compare against input json
 		if formattedJsonString != originalJsonString {
-			err := fmt.Errorf("%s: not formatted correctly", filename)
-			log.Errorf(err.Error())
+			log.Errorf("%s: not formatted correctly", filename)
 			fileErrors = append(fileErrors, filename)
 			// print out diff
 			dmp := diffmatchpatch.New()
@@ -79,7 +79,7 @@ func (cmd *LintCommand) Run() error {
 		}
 	}
 	if len(fileErrors) > 0 {
-		return fmt.Errorf("Incorrectly formatted files: %s", strings.Join(fileErrors, ", "))
+		return fmt.Errorf("incorrectly formatted files: %s", strings.Join(fileErrors, ", "))
 
 	}
 	return nil

cmds/merge_cmd.go

@@ -1,6 +1,7 @@
 package cmds
 
 import (
+	"context"
 	"errors"
 
 	"github.com/interline-io/transitland-lib/adapters"
@@ -13,10 +14,9 @@ import (
 
 // MergeCommand
 type MergeCommand struct {
-	Options           copier.Options
-	readerPaths       []string
-	writerPath        string
-	writeExtraColumns bool
+	Options     copier.Options
+	readerPaths []string
+	writerPath  string
 }
 
 func (cmd *MergeCommand) HelpDesc() (string, string) {
@@ -40,7 +40,7 @@ func (cmd *MergeCommand) Parse(args []string) error {
 	return nil
 }
 
-func (cmd *MergeCommand) Run() error {
+func (cmd *MergeCommand) Run(ctx context.Context) error {
 	var readers []adapters.Reader
 	for _, p := range cmd.readerPaths {
 		// Open reader

cmds/merge_cmd_test.go

@@ -1,6 +1,7 @@
 package cmds
 
 import (
+	"context"
 	"testing"
 
 	"github.com/interline-io/transitland-lib/ext"
@@ -18,7 +19,7 @@ func TestMerge(t *testing.T) {
 		if err := cmd.Parse([]string{tdir, f1.URL, f2.URL}); err != nil {
 			t.Fatal(err)
 		}
-		if err := cmd.Run(); err != nil {
+		if err := cmd.Run(context.Background()); err != nil {
 			t.Fatal(err)
 		}
 		outReader, err := ext.OpenReader(tdir)