Use context for more logging statements

Use `log.Ctx` for more logging statements. Also move the
`signal.NotifyContext` calls to the root command so all commands can
cleanly tidy up.

Part #2001

Author: wjam

cmd/bacalhau/create.go

@@ -83,7 +83,7 @@ func newCreateCmd() *cobra.Command {
 func create(cmd *cobra.Command, cmdArgs []string, OC *CreateOptions) error { //nolint:funlen,gocyclo
 	ctx := cmd.Context()
 
-	cm := cmd.Context().Value(systemManagerKey).(*system.CleanupManager)
+	cm := ctx.Value(systemManagerKey).(*system.CleanupManager)
 
 	// Custom unmarshaller
 	// https://stackoverflow.com/questions/70635636/unmarshaling-yaml-into-different-struct-based-off-yaml-field?rq=1

cmd/bacalhau/devstack.go

@@ -3,7 +3,6 @@ package bacalhau
 import (
 	"fmt"
 	"os"
-	"os/signal"
 	"path/filepath"
 	"strconv"
 
@@ -149,10 +148,6 @@ func runDevstack(cmd *cobra.Command, ODs *devstack.DevStackOptions, OS *ServeOpt
 			ODs.NumberOfBadRequesterActors, totalRequesterNodes), 1)
 	}
 
-	// Context ensures main goroutine waits until killed with ctrl+c:
-	ctx, cancel := signal.NotifyContext(ctx, ShutdownSignals...)
-	defer cancel()
-
 	portFileName := filepath.Join(os.TempDir(), "bacalhau-devstack.port")
 	pidFileName := filepath.Join(os.TempDir(), "bacalhau-devstack.pid")
 

cmd/bacalhau/docker_run.go

@@ -1,6 +1,7 @@
 package bacalhau
 
 import (
+	"context"
 	"fmt"
 	"strings"
 
@@ -272,9 +273,9 @@ func newDockerRunCmd() *cobra.Command { //nolint:funlen
 func dockerRun(cmd *cobra.Command, cmdArgs []string, ODR *DockerRunOptions) error {
 	ctx := cmd.Context()
 
-	cm := cmd.Context().Value(systemManagerKey).(*system.CleanupManager)
+	cm := ctx.Value(systemManagerKey).(*system.CleanupManager)
 
-	j, err := CreateJob(cmdArgs, ODR)
+	j, err := CreateJob(ctx, cmdArgs, ODR)
 	if err != nil {
 		Fatal(cmd, fmt.Sprintf("Error creating job: %s", err), 1)
 		return nil
@@ -320,7 +321,7 @@ func dockerRun(cmd *cobra.Command, cmdArgs []string, ODR *DockerRunOptions) erro
 }
 
 // CreateJob creates a job object from the given command line arguments and options.
-func CreateJob(cmdArgs []string, odr *DockerRunOptions) (*model.Job, error) {
+func CreateJob(ctx context.Context, cmdArgs []string, odr *DockerRunOptions) (*model.Job, error) {
 	odr.Image = cmdArgs[0]
 	odr.Entrypoint = cmdArgs[1:]
 
@@ -370,6 +371,7 @@ func CreateJob(cmdArgs []string, odr *DockerRunOptions) (*model.Job, error) {
 	}
 
 	j, err := jobutils.ConstructDockerJob(
+		ctx,
 		model.APIVersionLatest(),
 		engineType,
 		verifierType,

cmd/bacalhau/root.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"os"
+	"os/signal"
 	"strconv"
 	"strings"
 
@@ -80,7 +81,7 @@ func NewRootCmd() *cobra.Command {
 		PersistentPostRun: func(cmd *cobra.Command, args []string) {
 			ctx := cmd.Context()
 			ctx.Value(spanKey).(trace.Span).End()
-			ctx.Value(systemManagerKey).(*system.CleanupManager).Cleanup(cmd.Context())
+			ctx.Value(systemManagerKey).(*system.CleanupManager).Cleanup(ctx)
 		},
 	}
 	// ====== Start a job
@@ -138,9 +139,12 @@ Ignored if BACALHAU_API_PORT environment variable is set.`,
 }
 
 func Execute() {
-	RootCmd := NewRootCmd()
-	// ANCHOR: Set global context here
-	RootCmd.SetContext(context.Background())
+	rootCmd := NewRootCmd()
+
+	// Ensure commands are able to stop cleanly if someone presses ctrl+c
+	ctx, cancel := signal.NotifyContext(context.Background(), ShutdownSignals...)
+	defer cancel()
+	rootCmd.SetContext(ctx)
 
 	doNotTrack = false
 	if doNotTrackValue, foundDoNotTrack := os.LookupEnv("DO_NOT_TRACK"); foundDoNotTrack {
@@ -152,40 +156,36 @@ func Execute() {
 
 	viper.SetEnvPrefix("BACALHAU")
 
-	err := viper.BindEnv("API_HOST")
-	if err != nil {
-		log.Ctx(RootCmd.Context()).Fatal().Msgf("API_HOST was set, but could not bind.")
+	if err := viper.BindEnv("API_HOST"); err != nil {
+		log.Ctx(ctx).Fatal().Msgf("API_HOST was set, but could not bind.")
 	}
 
-	err = viper.BindEnv("API_PORT")
-	if err != nil {
-		log.Ctx(RootCmd.Context()).Fatal().Msgf("API_PORT was set, but could not bind.")
+	if err := viper.BindEnv("API_PORT"); err != nil {
+		log.Ctx(ctx).Fatal().Msgf("API_PORT was set, but could not bind.")
 	}
 
 	viper.AutomaticEnv()
-	envAPIHost := viper.Get("API_HOST")
-	envAPIPort := viper.Get("API_PORT")
 
-	if envAPIHost != nil && envAPIHost != "" {
-		apiHost = envAPIHost.(string)
+	if envAPIHost := viper.GetString("API_HOST"); envAPIHost != "" {
+		apiHost = envAPIHost
 	}
 
-	if envAPIPort != nil && envAPIPort != "" {
+	if envAPIPort := viper.GetString("API_PORT"); envAPIPort != "" {
 		var parseErr error
-		apiPort, parseErr = strconv.Atoi(envAPIPort.(string))
+		apiPort, parseErr = strconv.Atoi(envAPIPort)
 		if parseErr != nil {
-			log.Ctx(RootCmd.Context()).Fatal().Msgf("could not parse API_PORT into an int. %s", envAPIPort)
+			log.Ctx(ctx).Fatal().Msgf("could not parse API_PORT into an int. %s", envAPIPort)
 		}
 	}
 
 	// Use stdout, not stderr for cmd.Print output, so that
 	// e.g. ID=$(bacalhau run) works
-	RootCmd.SetOut(system.Stdout)
+	rootCmd.SetOut(system.Stdout)
 	// TODO this is from fixing a deprecation warning for SetOutput. Shouldn't this be system.Stderr?
-	RootCmd.SetErr(system.Stdout)
+	rootCmd.SetErr(system.Stdout)
 
-	if err := RootCmd.Execute(); err != nil {
-		Fatal(RootCmd, err.Error(), 1)
+	if err := rootCmd.Execute(); err != nil {
+		Fatal(rootCmd, err.Error(), 1)
 	}
 }
 
@@ -197,6 +197,8 @@ var systemManagerKey = contextKey{name: "context key for storing the system mana
 var spanKey = contextKey{name: "context key for storing the root span"}
 
 func checkVersion(cmd *cobra.Command, args []string) error {
+	ctx := cmd.Context()
+
 	// corba doesn't do PersistentPreRun{,E} chaining yet
 	// https://github.com/spf13/cobra/issues/252
 	root := cmd
@@ -205,8 +207,8 @@ func checkVersion(cmd *cobra.Command, args []string) error {
 	root.PersistentPreRun(cmd, args)
 
 	// Check that the server version is compatible with the client version
-	serverVersion, _ := GetAPIClient().Version(cmd.Context()) // Ok if this fails, version validation will skip
-	if err := ensureValidVersion(cmd.Context(), version.Get(), serverVersion); err != nil {
+	serverVersion, _ := GetAPIClient().Version(ctx) // Ok if this fails, version validation will skip
+	if err := ensureValidVersion(ctx, version.Get(), serverVersion); err != nil {
 		Fatal(cmd, fmt.Sprintf("version validation failed: %s", err), 1)
 		return err
 	}

cmd/bacalhau/run_python.go

@@ -78,7 +78,7 @@ func NewLanguageRunOptions() *LanguageRunOptions {
 // TODO: move the adapter code (from wasm to docker) into a wasm executor, so
 // that the compute node can verify the job knowing that it was run properly,
 // rather than doing the translation in, and thereby trusting, the client (to
-// set up the wasm environment to be determinstic)
+// set up the wasm environment to be deterministic)
 
 func newRunPythonCmd() *cobra.Command {
 	OLR := NewLanguageRunOptions()
@@ -165,7 +165,7 @@ func newRunPythonCmd() *cobra.Command {
 func runPython(cmd *cobra.Command, cmdArgs []string, OLR *LanguageRunOptions) error {
 	ctx := cmd.Context()
 
-	cm := cmd.Context().Value(systemManagerKey).(*system.CleanupManager)
+	cm := ctx.Value(systemManagerKey).(*system.CleanupManager)
 
 	// error if determinism is false
 	if !OLR.Deterministic {
@@ -193,10 +193,10 @@ func runPython(cmd *cobra.Command, cmdArgs []string, OLR *LanguageRunOptions) er
 	// have ConstructLanguageJob and ConstructDockerJob as separate means
 	// manually keeping them in sync.
 	j, err := job.ConstructLanguageJob(
+		ctx,
 		OLR.InputVolumes,
 		OLR.InputUrls,
 		OLR.OutputVolumes,
-		[]string{}, // no env vars (yet)
 		OLR.Concurrency,
 		OLR.Confidence,
 		OLR.MinBids,
@@ -206,7 +206,6 @@ func runPython(cmd *cobra.Command, cmdArgs []string, OLR *LanguageRunOptions) er
 		OLR.Command,
 		programPath,
 		OLR.RequirementsPath,
-		OLR.ContextPath,
 		OLR.Deterministic,
 		OLR.Labels,
 		doNotTrack,

cmd/bacalhau/serve.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"fmt"
 	"os"
-	"os/signal"
 	"sort"
 	"strings"
 	"time"
@@ -302,12 +301,8 @@ func newServeCmd() *cobra.Command {
 
 //nolint:funlen,gocyclo
 func serve(cmd *cobra.Command, OS *ServeOptions) error {
-	cm := cmd.Context().Value(systemManagerKey).(*system.CleanupManager)
-
-	// TODO this should be for all commands
-	// Context ensures main goroutine waits until killed with ctrl+c:
-	ctx, cancel := signal.NotifyContext(cmd.Context(), ShutdownSignals...)
-	defer cancel()
+	ctx := cmd.Context()
+	cm := ctx.Value(systemManagerKey).(*system.CleanupManager)
 
 	isComputeNode, isRequesterNode := false, false
 	for _, nodeType := range OS.NodeType {

cmd/bacalhau/version.go

@@ -105,7 +105,7 @@ func (oV *VersionOptions) Run(ctx context.Context, cmd *cobra.Command) error {
 	if !oV.ClientOnly {
 		serverVersion, err := GetAPIClient().Version(ctx)
 		if err != nil {
-			log.Ctx(cmd.Context()).Error().Err(err).Msgf("could not get server version")
+			log.Ctx(ctx).Error().Err(err).Msgf("could not get server version")
 			return err
 		}
 

cmd/bacalhau/wasm_run.go

@@ -191,7 +191,7 @@ func runWasm(
 		time.Sleep(1 * time.Second)
 
 		storage := inline.NewStorage()
-		inlineData, err := storage.Upload(cmd.Context(), info.Name())
+		inlineData, err := storage.Upload(ctx, info.Name())
 		if err != nil {
 			return err
 		}

pkg/devstack/profiler.go

@@ -41,7 +41,7 @@ func StartProfiling(ctx context.Context, cpuFile, memoryFile string) CloserWithC
 
 func (p *profiler) Close(ctx context.Context) error {
 	// stop profiling now, just before we clean up, if we're profiling.
-	log.Trace().Msg("============= STOPPING PROFILING ============")
+	log.Ctx(ctx).Trace().Msg("============= STOPPING PROFILING ============")
 	if p.cpuFile != nil {
 		pprof.StopCPUProfile()
 		closer.CloseWithLogOnError(p.cpuFile.Name(), p.cpuFile)

pkg/job/factory.go

@@ -1,6 +1,7 @@
 package job
 
 import (
+	"context"
 	"strings"
 
 	"github.com/filecoin-project/bacalhau/pkg/model"
@@ -12,6 +13,7 @@ import (
 // to pass in the collection of CLI args as strings
 // and have a Job struct returned
 func ConstructDockerJob( //nolint:funlen
+	ctx context.Context,
 	a model.APIVersion,
 	e model.Engine,
 	v model.Verifier,
@@ -48,7 +50,7 @@ func ConstructDockerJob( //nolint:funlen
 	if err != nil {
 		return &model.Job{}, err
 	}
-	jobOutputs, err := buildJobOutputs(outputVolumes)
+	jobOutputs, err := buildJobOutputs(ctx, outputVolumes)
 	if err != nil {
 		return &model.Job{}, err
 	}
@@ -64,7 +66,7 @@ func ConstructDockerJob( //nolint:funlen
 	}
 
 	if len(unSafeAnnotations) > 0 {
-		log.Error().Msgf("The following labels are unsafe. Labels must fit the regex '/%s/' (and all emjois): %+v",
+		log.Ctx(ctx).Error().Msgf("The following labels are unsafe. Labels must fit the regex '/%s/' (and all emjois): %+v",
 			RegexString,
 			strings.Join(unSafeAnnotations, ", "))
 	}
@@ -77,7 +79,6 @@ func ConstructDockerJob( //nolint:funlen
 	if len(workingDir) > 0 {
 		err = system.ValidateWorkingDir(workingDir)
 		if err != nil {
-			log.Error().Msg(err.Error())
 			return &model.Job{}, err
 		}
 	}
@@ -136,10 +137,10 @@ func ConstructDockerJob( //nolint:funlen
 }
 
 func ConstructLanguageJob(
+	ctx context.Context,
 	inputVolumes []string,
 	inputUrls []string,
 	outputVolumes []string,
-	env []string,
 	concurrency int,
 	confidence int,
 	minBids int,
@@ -150,7 +151,6 @@ func ConstructLanguageJob(
 	command string,
 	programPath string,
 	requirementsPath string,
-	contextPath string, // we have to tar this up and POST it to the Requester node
 	deterministic bool,
 	annotations []string,
 	doNotTrack bool,
@@ -162,7 +162,7 @@ func ConstructLanguageJob(
 	if err != nil {
 		return &model.Job{}, err
 	}
-	jobOutputs, err := buildJobOutputs(outputVolumes)
+	jobOutputs, err := buildJobOutputs(ctx, outputVolumes)
 	if err != nil {
 		return &model.Job{}, err
 	}
@@ -178,7 +178,7 @@ func ConstructLanguageJob(
 	}
 
 	if len(unSafeAnnotations) > 0 {
-		log.Error().Msgf("The following labels are unsafe. Labels must fit the regex '/%s/' (and all emjois): %+v",
+		log.Ctx(ctx).Error().Msgf("The following labels are unsafe. Labels must fit the regex '/%s/' (and all emjois): %+v",
 			RegexString,
 			strings.Join(unSafeAnnotations, ", "))
 	}

pkg/job/factory_test.go

@@ -3,6 +3,7 @@
 package job
 
 import (
+	"context"
 	"strings"
 	"testing"
 
@@ -68,6 +69,7 @@ func (suite *JobFactorySuite) TestRun_DockerJobOutputs() {
 				}
 
 				j, err := ConstructDockerJob( //nolint:funlen
+					context.Background(),
 					model.APIVersionLatest(),
 					model.EngineNoop,
 					model.VerifierNoop,