core/oc-monitord
6
0
Fork 0
Code Issues 3 Pull Requests Packages Projects Releases Wiki Activity

Implemented logging of local execution of argo submit --watch and logs produced by pods

Browse Source
This commit is contained in:
pb 2025-04-09 18:59:37 +02:00
parent 5d8143c93e
commit c31184e2ec
2 changed files with 95 additions and 35 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

77
main.go
View File

@ -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
@ -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 {
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()
}
@ -326,6 +362,7 @@ func getContainerName(argo_file string) string {
re := regexp.MustCompile(regex)
container_name := re.FindString(argo_file)
return container_name
}

47
models/local_argo_logs.go
View File

@ -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"
}
// }