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

Refactor + Multi admiralty test

Browse Source
This commit is contained in:
mr
2025-06-12 14:01:16 +02:00
parent 24bbe81638
commit 91f421af1e
12 changed files with 460 additions and 442 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
main.go
View File

@@ -14,6 +14,7 @@ import (
"oc-monitord/conf"
l "oc-monitord/logger"
"oc-monitord/models"
u "oc-monitord/utils"
"oc-monitord/workflow_builder"
@@ -53,7 +54,7 @@ func main() {
os.Setenv("test_service", "true") // Only for service demo, delete before merging on main
parser = *argparse.NewParser("oc-monitord", "Launch the execution of a workflow given as a parameter and sends the produced logs to a loki database")
loadConfig(false, &parser)
setConf(&parser)
oclib.InitDaemon("oc-monitord")
oclib.SetConfig(
@@ -71,6 +72,7 @@ func main() {
exec := u.GetExecution(conf.GetConfig().ExecutionID)
if exec == nil {
logger.Fatal().Msg("Could not retrieve workflow ID from execution ID " + conf.GetConfig().ExecutionID + " on peer " + conf.GetConfig().PeerID)
return
}
conf.GetConfig().WorkflowID = exec.WorkflowID
@@ -110,20 +112,20 @@ func main() {
// Executed in a k8s environment
logger.Info().Msg("Executes inside a k8s")
// executeInside(exec.GetID(), "argo", argo_file_path, stepMax) // commenting to use conf.ExecutionID instead of exec.GetID()
executeInside(conf.GetConfig().ExecutionID, exec.ExecutionsID, argoFilePath)
executeInside(exec.ExecutionsID, argoFilePath)
}
}
// So far we only log the output from
func executeInside(execID string, ns string, argo_file_path string) {
func executeInside(ns string, argo_file_path string) {
t, err := tools2.NewService(conf.GetConfig().Mode)
if err != nil {
logger.Error().Msg("Could not create KubernetesTool")
return
}
name, err := t.CreateArgoWorkflow(argo_file_path, ns)
// _ = name
// _ = name
if err != nil {
logger.Error().Msg("Could not create argo workflow : " + err.Error())
logger.Info().Msg(fmt.Sprint("CA :" + conf.GetConfig().KubeCA))
@@ -146,26 +148,26 @@ func executeInside(execID string, ns string, argo_file_path string) {
}
func executeOutside(argo_file_path string, workflow workflow_builder.Workflow) {
func executeOutside(argo_file_path string, workflow *models.Workflow) {
var stdoutSubmit, stderrSubmit io.ReadCloser
var stdoutLogs, stderrLogs 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 )
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
}
cmdLogs := exec.Command("argo", "logs", "oc-monitor-"+workflowName, "-n", conf.GetConfig().ExecutionID, "--follow","--no-color")
cmdLogs := exec.Command("argo", "logs", "oc-monitor-"+workflowName, "-n", conf.GetConfig().ExecutionID, "--follow", "--no-color")
if stdoutLogs, err = cmdLogs.StdoutPipe(); err != nil {
wf_logger.Error().Msg("Could not retrieve stdoutpipe for 'argo logs'" + err.Error())
return
}
var steps []string
for _, template := range workflow.Spec.Templates {
steps = append(steps, template.Name)
@@ -186,7 +188,7 @@ func executeOutside(argo_file_path string, workflow workflow_builder.Workflow) {
logger.Info().Msg("Running argo logs")
if err := cmdLogs.Run(); err != nil {
wf_logger.Error().Msg("Could not run '" + strings.Join(cmdLogs.Args, " ") + "'")
wf_logger.Fatal().Msg(err.Error() + bufio.NewScanner(stderrLogs).Text())
}
@@ -201,18 +203,7 @@ func executeOutside(argo_file_path string, workflow workflow_builder.Workflow) {
wg.Wait()
}
func loadConfig(is_k8s bool, parser *argparse.Parser) {
var o *onion.Onion
o = initOnion(o)
setConf(is_k8s, o, parser)
// if !IsValidUUID(conf.GetConfig().ExecutionID) {
// logger.Fatal().Msg("Provided ID is not an UUID")
// }
}
func setConf(is_k8s bool, o *onion.Onion, parser *argparse.Parser) {
func setConf(parser *argparse.Parser) {
url := parser.String("u", "url", &argparse.Options{Required: true, Default: "http://127.0.0.1:3100", Help: "Url to the Loki database logs will be sent to"})
mode := parser.String("M", "mode", &argparse.Options{Required: false, Default: "", Help: "Mode of the execution"})
execution := parser.String("e", "execution", &argparse.Options{Required: true, Help: "Execution ID of the workflow to request from oc-catalog API"})
@@ -305,7 +296,6 @@ func getContainerName(argo_file string) string {
return container_name
}
func updateStatus(status string, log string) {
exec_id := conf.GetConfig().ExecutionID