113 lines
2.9 KiB
Go
113 lines
2.9 KiB
Go
|
package models
|
||
|
|
|
||
|
|
import (
|
||
|
|
"bufio"
|
||
|
|
"encoding/json"
|
||
|
|
"fmt"
|
||
|
|
"io"
|
||
|
|
"oc-monitord/conf"
|
||
|
|
"strings"
|
||
|
|
"sync"
|
||
|
|
|
||
|
|
"cloud.o-forge.io/core/oc-lib/logs"
|
||
|
|
"github.com/rs/zerolog"
|
||
|
|
)
|
||
|
|
|
||
|
|
var logger zerolog.Logger
|
||
|
|
var wf_logger 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 {
|
||
|
|
var workflow ArgoWatch
|
||
|
|
|
||
|
|
for _, input := range inputs {
|
||
|
|
line := strings.TrimSpace(input)
|
||
|
|
if line == "" {
|
||
|
|
continue
|
||
|
|
}
|
||
|
|
switch {
|
||
|
|
case strings.HasPrefix(line, "Name:"):
|
||
|
|
workflow.Name = parseValue(line)
|
||
|
|
case strings.HasPrefix(line, "Namespace:"):
|
||
|
|
workflow.Namespace = parseValue(line)
|
||
|
|
case strings.HasPrefix(line, "Status:"):
|
||
|
|
workflow.Status = parseValue(line)
|
||
|
|
case strings.HasPrefix(line, "PodRunning"):
|
||
|
|
workflow.PodRunning = parseBoolValue(line)
|
||
|
|
case strings.HasPrefix(line, "Completed"):
|
||
|
|
workflow.Completed = parseBoolValue(line)
|
||
|
|
case strings.HasPrefix(line, "Created:"):
|
||
|
|
workflow.Created = parseValue(line)
|
||
|
|
case strings.HasPrefix(line, "Started:"):
|
||
|
|
workflow.Started = parseValue(line)
|
||
|
|
case strings.HasPrefix(line, "Duration:"):
|
||
|
|
workflow.Duration = parseValue(line)
|
||
|
|
case strings.HasPrefix(line, "Progress:"):
|
||
|
|
workflow.Progress = parseValue(line)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
return &workflow
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
func parseValue(line string) string {
|
||
|
|
parts := strings.SplitN(line, ":", 2)
|
||
|
|
if len(parts) < 2 {
|
||
|
|
return ""
|
||
|
|
}
|
||
|
|
return strings.TrimSpace(parts[1])
|
||
|
|
}
|
||
|
|
|
||
|
|
func parseBoolValue(line string) bool {
|
||
|
|
value := parseValue(line)
|
||
|
|
return value == "True"
|
||
|
|
}
|
||
|
|
|
||
|
|
func LogLocalWorkflow(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()
|
||
|
|
|
||
|
|
var current_watch, previous_watch ArgoWatch
|
||
|
|
|
||
|
|
watch_output := make([]string, 0)
|
||
|
|
scanner := bufio.NewScanner(pipe)
|
||
|
|
for scanner.Scan() {
|
||
|
|
log := scanner.Text()
|
||
|
|
watch_output = append(watch_output, log)
|
||
|
|
|
||
|
|
if strings.HasPrefix(log, "Progress:") {
|
||
|
|
|
||
|
|
current_watch = *NewLocalArgoLogs(watch_output)
|
||
|
|
workflowName := current_watch.Name
|
||
|
|
if !current_watch.Equals(&previous_watch) {
|
||
|
|
wg.Add(1)
|
||
|
|
// checkStatus(current_watch.Status, previous_watch.Status)
|
||
|
|
jsonified, err := json.Marshal(current_watch)
|
||
|
|
if err != nil {
|
||
|
|
logger.Error().Msg("Could not create watch log for " + workflowName)
|
||
|
|
}
|
||
|
|
wf_logger.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)
|
||
|
|
// }
|
||
|
|
|
||
|
|
// }
|