From c31184e2ec66a73765b2c0bbe7a743c6080bff85 Mon Sep 17 00:00:00 2001 From: pb Date: Wed, 9 Apr 2025 18:59:37 +0200 Subject: [PATCH] Implemented logging of local execution of argo submit --watch and logs produced by pods --- main.go | 83 ++++++++++++++++++++++++++++----------- models/local_argo_logs.go | 47 ++++++++++++++++------ 2 files changed, 95 insertions(+), 35 deletions(-) diff --git a/main.go b/main.go index 5b49c19..04d68c9 100644 --- a/main.go +++ b/main.go @@ -12,6 +12,7 @@ import ( "slices" "strings" "sync" + "time" "oc-monitord/conf" "oc-monitord/models" @@ -44,6 +45,7 @@ import ( var logger zerolog.Logger var wf_logger zerolog.Logger +var pods_logger zerolog.Logger var parser argparse.Parser var workflowName string @@ -84,7 +86,7 @@ func main() { err := new_wf.LoadFrom(conf.GetConfig().WorkflowID, conf.GetConfig().PeerID) if err != nil { - + logger.Error().Msg("Could not retrieve workflow " + conf.GetConfig().WorkflowID + " from oc-catalog API") } @@ -94,7 +96,6 @@ func main() { logger.Error().Msg(err.Error()) } - argo_file_path, err := builder.CompleteBuild(exec.ExecutionsID) if err != nil { logger.Error().Msg(err.Error()) @@ -109,7 +110,7 @@ func main() { if conf.GetConfig().KubeHost == "" { // Not in a k8s environment, get conf from parameters fmt.Println("Executes outside of k8s") - executeOutside(argo_file_path, stepMax) + executeOutside(argo_file_path, stepMax, builder.Workflow) } else { // Executed in a k8s environment fmt.Println("Executes inside a k8s") @@ -138,33 +139,68 @@ func executeInside(execID string, ns string, argo_file_path string, stepMax int) } -func executeOutside(argo_file_path string, stepMax int) { - // var stdout, stderr, stdout_logs, stderr_logs io.ReadCloser - var stdout, stderr io.ReadCloser +func executeOutside(argo_file_path string, stepMax int, workflow workflow_builder.Workflow) { + // var stdoutSubmit, stderrSubmit, stdout_logs, stderr_logs io.ReadCloser + var stdoutSubmit, stderrSubmit io.ReadCloser + var stdoutLogs, stderrLogs io.ReadCloser // var stderr io.ReadCloser + var wg sync.WaitGroup var err error - logger.Debug().Msg("executing :" + "argo submit --watch " + argo_file_path + " --serviceaccount sa-" + conf.GetConfig().ExecutionID + " -n " + conf.GetConfig().ExecutionID) - cmd := exec.Command("argo", "submit", "--watch", argo_file_path, "--serviceaccount", "sa-"+conf.GetConfig().ExecutionID, "-n", conf.GetConfig().ExecutionID ) - if stdout, err = cmd.StdoutPipe(); err != nil { + + logger.Debug().Msg("executing :" + "argo submit --watch " + argo_file_path + " --serviceaccount sa-" + conf.GetConfig().ExecutionID + " -n " + conf.GetConfig().ExecutionID) + + cmdSubmit := exec.Command("argo", "submit", "--watch", argo_file_path, "--serviceaccount", "sa-"+conf.GetConfig().ExecutionID, "-n", conf.GetConfig().ExecutionID) + if stdoutSubmit, err = cmdSubmit.StdoutPipe(); err != nil { wf_logger.Error().Msg("Could not retrieve stdoutpipe " + err.Error()) return } - if err := cmd.Start(); err != nil { - panic(err) - } - var wg sync.WaitGroup - split := strings.Split(argo_file_path, "_") - argoLogs := models.NewArgoLogs(split[0], conf.GetConfig().ExecutionID, stepMax) - argoLogs.StartStepRecording(argoLogs.NewWatch(), wf_logger) - argoLogs.IsStreaming = true // Used to determine wether or not the logs are read from a docker container or on localhost - // go logWorkflow(argo_file_path, stepMax, stdout, argoLogs.NewWatch(), argoLogs.NewWatch(), argoLogs, []string{}, &wg) - go models.LogLocalWorkflow(stdout,&wg) - if err := cmd.Wait(); err != nil { - wf_logger.Error().Msg("Could not execute argo submit") - wf_logger.Error().Msg(err.Error() + bufio.NewScanner(stderr).Text()) + // //======== Code block that implemented a method that logs both locally and container executed wf + // // Need to be improved, did not log well for local executions + // split := strings.Split(argo_file_path, "_") + // argoLogs := models.NewArgoLogs(split[0], conf.GetConfig().ExecutionID, stepMax) + // argoLogs.StartStepRecording(argoLogs.NewWatch(), wf_logger) + // argoLogs.IsStreaming = true // Used to determine wether or not the logs are read from a docker container or on localhost + // // go logWorkflow(argo_file_path, stepMax, stdout, argoLogs.NewWatch(), argoLogs.NewWatch(), argoLogs, []string{}, &wg) + // // ======= + + var steps []string + for _, template := range workflow.Spec.Templates { + steps = append(steps, template.Name) + } + + cmdLogs := exec.Command("argo", "logs", "oc-monitor-"+workflowName, "-n", conf.GetConfig().ExecutionID, "--follow") + if stdoutLogs, err = cmdLogs.StdoutPipe(); err != nil { + wf_logger.Error().Msg("Could not retrieve stdoutpipe for 'argo logs'" + err.Error()) + return + } + + go models.LogLocalWorkflow(workflowName, stdoutSubmit, &wg) + go models.LogPods(stdoutLogs, steps, &wg) + + fmt.Println("Starting argo submit") + if err := cmdSubmit.Start(); err != nil { + wf_logger.Error().Msg("Could not start argo submit") + wf_logger.Error().Msg(err.Error() + bufio.NewScanner(stderrSubmit).Text()) updateStatus("fatal", "") } + + time.Sleep(5 * time.Second) + + fmt.Println("Running argo logs") + if err := cmdLogs.Run(); err != nil { + wf_logger.Error().Msg("Could not run 'argo logs oc-monitor-" + workflowName + " -n " + conf.GetConfig().ExecutionID + " --follow") + wf_logger.Fatal().Msg(err.Error() + bufio.NewScanner(stderrLogs).Text()) + + } + + fmt.Println("Waiting argo submit") + if err := cmdSubmit.Wait(); err != nil { + wf_logger.Error().Msg("Could not execute argo submit") + wf_logger.Error().Msg(err.Error() + bufio.NewScanner(stderrSubmit).Text()) + updateStatus("fatal", "") + } + wg.Wait() } @@ -188,7 +224,7 @@ func logWorkflow(argo_file_path string, stepMax int, pipe io.ReadCloser, wg.Add(1) } seeit++ - } else if count == 0 && !argoLogs.IsStreaming { + } else if count == 0 && !argoLogs.IsStreaming { break } if count == 1 { @@ -326,6 +362,7 @@ func getContainerName(argo_file string) string { re := regexp.MustCompile(regex) container_name := re.FindString(argo_file) + return container_name } diff --git a/models/local_argo_logs.go b/models/local_argo_logs.go index 0fbcef3..6b658d8 100644 --- a/models/local_argo_logs.go +++ b/models/local_argo_logs.go @@ -14,7 +14,8 @@ import ( ) var logger zerolog.Logger -var wf_logger zerolog.Logger +var wfLogger zerolog.Logger + // Take the slice of string that make up one round of stderr outputs from the --watch option in argo submit func NewLocalArgoLogs(inputs []string) *ArgoWatch { @@ -65,12 +66,12 @@ func parseBoolValue(line string) bool { return value == "True" } -func LogLocalWorkflow(pipe io.ReadCloser, wg *sync.WaitGroup) { +func LogLocalWorkflow(wfName string, pipe io.ReadCloser, wg *sync.WaitGroup) { logger = logs.GetLogger() logger.Debug().Msg("created wf_logger") fmt.Println("created wf_logger") - wf_logger = logger.With().Str("argo_name", "MON WF DE TEST").Str("workflow_id", conf.GetConfig().WorkflowID).Str("workflow_execution_id", conf.GetConfig().ExecutionID).Logger() + wfLogger = logger.With().Str("argo_name", wfName).Str("workflow_id", conf.GetConfig().WorkflowID).Str("workflow_execution_id", conf.GetConfig().ExecutionID).Logger() var current_watch, previous_watch ArgoWatch @@ -80,6 +81,7 @@ func LogLocalWorkflow(pipe io.ReadCloser, wg *sync.WaitGroup) { log := scanner.Text() watch_output = append(watch_output, log) + // Log the progress of the WF if strings.HasPrefix(log, "Progress:") { current_watch = *NewLocalArgoLogs(watch_output) @@ -91,22 +93,43 @@ func LogLocalWorkflow(pipe io.ReadCloser, wg *sync.WaitGroup) { if err != nil { logger.Error().Msg("Could not create watch log for " + workflowName) } - wf_logger.Info().Msg(string(jsonified)) + wfLogger.Info().Msg(string(jsonified)) previous_watch = current_watch current_watch = ArgoWatch{} wg.Done() } } + + + } } // Debug, no logs sent -// func logPods(pipe io.ReadCloser, name string) { -// pods_logger = wf_logger.With().Str("pod_name", name).Logger() -// scanner := bufio.NewScanner(pipe) -// for scanner.Scan() { -// log := scanner.Text() -// pods_logger.Info().Msg(log) -// } +func LogPods(pipe io.ReadCloser, steps []string, wg *sync.WaitGroup) { + scanner := bufio.NewScanner(pipe) + for scanner.Scan() { + fmt.Println("new line") + wg.Add(1) + var podLogger zerolog.Logger + line := scanner.Text() + podName := strings.Split(line, ":")[0] + podLogger = wfLogger.With().Str("step_name", getStepName(podName, steps)).Logger() + log := strings.Split(line,podName+":")[1] + podLogger.Info().Msg(log) + wg.Done() + } + +} + +func getStepName(podName string, steps []string) string { + + for _, step := range(steps) { + if strings.Contains(podName,step){ + return step + } + } + + return "error" +} -// }