Use logger with formatter

Signed-off-by: Andrew Dye <[email protected]>

flyteadmin/dataproxy/service.go

@@ -81,10 +81,10 @@ func (s Service) CreateUploadLocation(ctx context.Context, req *service.CreateUp
 			base32Digest := base32.StdEncoding.EncodeToString(req.ContentMd5)
 			base64Digest := base64.StdEncoding.EncodeToString(req.ContentMd5)
 			if hexDigest != metadata.Etag() && base32Digest != metadata.Etag() && base64Digest != metadata.Etag() {
-				logger.Debug(ctx, "File already exists at location [%v] but hashes do not match", knownLocation)
+				logger.Debugf(ctx, "File already exists at location [%v] but hashes do not match", knownLocation)
 				return nil, errors.NewFlyteAdminErrorf(codes.AlreadyExists, "file already exists at location [%v], specify a matching hash if you wish to rewrite", knownLocation)
 			}
-			logger.Debug(ctx, "File already exists at location [%v] but allowing rewrite", knownLocation)
+			logger.Debugf(ctx, "File already exists at location [%v] but allowing rewrite", knownLocation)
 		}
 	}
 

flyteadmin/scheduler/executor/executor_impl.go

@@ -62,7 +62,7 @@ func (w *executor) Execute(ctx context.Context, scheduledTime time.Time, s model
 	}, scheduledTime)
 
 	if err != nil {
-		logger.Error(ctx, "failed to generate execution identifier for schedule %+v due to %v", s, err)
+		logger.Errorf(ctx, "failed to generate execution identifier for schedule %+v due to %v", s, err)
 		return err
 	}
 
@@ -107,7 +107,7 @@ func (w *executor) Execute(ctx context.Context, scheduledTime time.Time, s model
 				return false
 			}
 			w.metrics.FailedExecutionCounter.Inc()
-			logger.Error(ctx, "failed to create execution create request %+v due to %v", executionRequest, err)
+			logger.Errorf(ctx, "failed to create execution create request %+v due to %v", executionRequest, err)
 			// TODO: Handle the case when admin launch plan state is archived but the schedule is active.
 			// After this bug is fixed in admin https://github.com/flyteorg/flyte/issues/1354
 			return true
@@ -118,7 +118,7 @@ func (w *executor) Execute(ctx context.Context, scheduledTime time.Time, s model
 		},
 	)
 	if err != nil && status.Code(err) != codes.AlreadyExists {
-		logger.Error(ctx, "failed to create execution create request %+v due to %v after all retries", executionRequest, err)
+		logger.Errorf(ctx, "failed to create execution create request %+v due to %v after all retries", executionRequest, err)
 		return err
 	}
 	w.metrics.SuccessfulExecutionCounter.Inc()

flytecopilot/cmd/root.go

@@ -184,7 +184,7 @@ func init() {
 	pflag.CommandLine.AddGoFlagSet(flag.CommandLine)
 	err := flag.CommandLine.Parse([]string{})
 	if err != nil {
-		logger.Error(context.TODO(), "Error in initializing: %v", err)
+		logger.Errorf(context.TODO(), "Error in initializing: %v", err)
 		os.Exit(-1)
 	}
 	labeled.SetMetricKeys(contextutils.ProjectKey, contextutils.DomainKey, contextutils.WorkflowIDKey, contextutils.TaskIDKey)

flyteplugins/go/tasks/pluginmachinery/ioutils/paths.go

@@ -43,7 +43,7 @@ func ConstructCheckpointPath(store storage.ReferenceConstructor, rawOutputPrefix
 func constructPath(store storage.ReferenceConstructor, base storage.DataReference, suffix string) storage.DataReference {
 	res, err := store.ConstructReference(context.Background(), base, suffix)
 	if err != nil {
-		logger.Error(context.Background(), "Failed to construct path. Base[%v] Error: %v", base, err)
+		logger.Errorf(context.Background(), "Failed to construct path. Base[%v] Error: %v", base, err)
 	}
 
 	return res

flyteplugins/go/tasks/plugins/array/awsbatch/jobs_store.go

@@ -272,7 +272,7 @@ func syncBatches(_ context.Context, client Client, handler EventHandler, batchCh
 			for _, jobDetail := range response {
 				job, found := jobIDsMap[*jobDetail.JobId]
 				if !found {
-					logger.Warn(ctx, "Received an update for unrequested job id [%v]", jobDetail.JobId)
+					logger.Warnf(ctx, "Received an update for unrequested job id [%v]", jobDetail.JobId)
 					continue
 				}
 

flyteplugins/go/tasks/plugins/array/awsbatch/monitor.go

@@ -49,7 +49,7 @@ func CheckSubTasksState(ctx context.Context, tCtx core.TaskExecutionContext, job
 
 	// If job isn't currently being monitored (recovering from a restart?), add it to the sync-cache and return
 	if job == nil {
-		logger.Info(ctx, "Job not found in cache, adding it. [%v]", jobName)
+		logger.Infof(ctx, "Job not found in cache, adding it. [%v]", jobName)
 
 		_, err = jobStore.GetOrCreate(jobName, &Job{
 			ID:             *currentState.ExternalJobID,

flyteplugins/go/tasks/plugins/webapi/athena/plugin.go

@@ -128,7 +128,7 @@ func (p Plugin) Delete(ctx context.Context, tCtx webapi.DeleteContext) error {
 		return err
 	}
 
-	logger.Info(ctx, "Deleted query execution [%v]", resp)
+	logger.Infof(ctx, "Deleted query execution [%v]", resp)
 
 	return nil
 }

flyteplugins/go/tasks/plugins/webapi/bigquery/plugin.go

@@ -256,7 +256,7 @@ func (p Plugin) Delete(ctx context.Context, taskCtx webapi.DeleteContext) error
 		return err
 	}
 
-	logger.Info(ctx, "Cancelled job [%s]", formatJobReference(resourceMeta.JobReference))
+	logger.Infof(ctx, "Cancelled job [%s]", formatJobReference(resourceMeta.JobReference))
 
 	return nil
 }

flyteplugins/go/tasks/plugins/webapi/databricks/plugin.go

@@ -203,7 +203,7 @@ func (p Plugin) Delete(ctx context.Context, taskCtx webapi.DeleteContext) error
 		return err
 	}
 	defer resp.Body.Close()
-	logger.Info(ctx, "Deleted query execution [%v]", resp)
+	logger.Infof(ctx, "Deleted query execution [%v]", resp)
 
 	return nil
 }

flyteplugins/go/tasks/plugins/webapi/snowflake/plugin.go

@@ -179,7 +179,7 @@ func (p Plugin) Delete(ctx context.Context, taskCtx webapi.DeleteContext) error
 		return err
 	}
 	defer resp.Body.Close()
-	logger.Info(ctx, "Deleted query execution [%v]", resp)
+	logger.Infof(ctx, "Deleted query execution [%v]", resp)
 
 	return nil
 }

flytepropeller/cmd/kubectl-flyte/cmd/root.go

@@ -24,7 +24,7 @@ func init() {
 	pflag.CommandLine.AddGoFlagSet(flag.CommandLine)
 	err := flag.CommandLine.Parse([]string{})
 	if err != nil {
-		logger.Error(context.TODO(), "Error in initializing: %v", err)
+		logger.Errorf(context.TODO(), "Error in initializing: %v", err)
 		os.Exit(-1)
 	}
 }

flytepropeller/pkg/controller/nodes/executor.go

@@ -583,7 +583,7 @@ func (c *nodeExecutor) attemptRecovery(ctx context.Context, nCtx interfaces.Node
 			state.PreviousNodeExecutionCheckpointURI = storage.DataReference(metadata.TaskNodeMetadata.CheckpointUri)
 			err = nCtx.NodeStateWriter().PutTaskNodeState(state)
 			if err != nil {
-				logger.Warn(ctx, "failed to save recovered checkpoint uri for [%+v]: [%+v]",
+				logger.Warnf(ctx, "failed to save recovered checkpoint uri for [%+v]: [%+v]",
 					nCtx.NodeExecutionMetadata().GetNodeExecutionID(), err)
 			}
 		}

flytepropeller/pkg/controller/nodes/subworkflow/launchplan/admin.go

@@ -148,7 +148,7 @@ func (a *adminLaunchPlanExecutor) Launch(ctx context.Context, launchCtx LaunchCo
 
 	_, err = a.cache.GetOrCreate(executionID.String(), executionCacheItem{WorkflowExecutionIdentifier: *executionID})
 	if err != nil {
-		logger.Info(ctx, "Failed to add ExecID [%v] to auto refresh cache", executionID)
+		logger.Infof(ctx, "Failed to add ExecID [%v] to auto refresh cache", executionID)
 	}
 
 	return nil

flytepropeller/pkg/controller/nodes/task/backoff/handler.go

@@ -49,7 +49,7 @@ func (b *SimpleBackOffBlocker) reset() {
 }
 
 func (b *SimpleBackOffBlocker) backOff(ctx context.Context) time.Duration {
-	logger.Debug(ctx, "BackOff params [BackOffBaseSecond: %v] [BackOffExponent: %v] [MaxBackOffDuration: %v]",
+	logger.Debugf(ctx, "BackOff params [BackOffBaseSecond: %v] [BackOffExponent: %v] [MaxBackOffDuration: %v]",
 		b.BackOffBaseSecond, b.BackOffExponent, b.MaxBackOffDuration)
 
 	backOffDuration := time.Duration(time.Second.Nanoseconds() * int64(math.Pow(float64(b.BackOffBaseSecond),

flytestdlib/random/weighted_random_list.go

@@ -84,7 +84,7 @@ func NewWeightedRandom(ctx context.Context, entries []Entry) (WeightedRandomList
 			currentTotal += 1.0 / float32(numberOfEntries)
 		} else if e.Weight == 0 {
 			// Entries which have zero weight are ignored
-			logger.Debug(ctx, "ignoring entry due to empty weight %v", e)
+			logger.Debugf(ctx, "ignoring entry due to empty weight %v", e)
 			continue
 		}
 

flytestdlib/storage/protobuf_store.go

@@ -46,7 +46,7 @@ func (s DefaultProtobufStore) ReadProtobuf(ctx context.Context, reference DataRe
 	defer func() {
 		err = rc.Close()
 		if err != nil {
-			logger.Warn(ctx, "Failed to close reference [%v]. Error: %v", reference, err)
+			logger.Warnf(ctx, "Failed to close reference [%v]. Error: %v", reference, err)
 		}
 	}()