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

logging workflow execution

Browse Source
This commit is contained in:
pb
2024-08-07 17:24:09 +02:00
parent a2011db186
commit 98b6cdaae5
4 changed files with 120 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files

131
main.go
View File

@@ -5,18 +5,22 @@ import (
"encoding/json"
"fmt"
"io"
"log"
"os"
"os/exec"
"regexp"
"strings"
"sync"
"time"
"oc-monitor/conf"
"oc-monitor/models"
"oc-monitor/workflow_builder"
oclib "cloud.o-forge.io/core/oc-lib"
"cloud.o-forge.io/core/oc-lib/logs"
"cloud.o-forge.io/core/oc-lib/models/workflow_execution"
"github.com/akamensky/argparse"
"github.com/google/uuid"
"github.com/goraz/onion"
@@ -29,6 +33,7 @@ var parser argparse.Parser
var monitorLocal bool
var workflowName string
const namespace = "-n argo"
const defaultConfigFile = "/etc/oc/ocmonitor_conf.json"
const localConfigFile = "./conf/ocmonitor_conf.json"
@@ -51,12 +56,15 @@ func main() {
logger = logs.CreateLogger("oc-monitor", conf.GetConfig().LokiURL)
logger.Debug().Msg("Loki URL : " + conf.GetConfig().LokiURL)
logger.Debug().Msg("Workflow executed : " + conf.GetConfig().WorkflowID)
logger.Debug().Msg("Workflow executed : " + conf.GetConfig().ExecutionID)
oclib.SetConfig(conf.GetConfig().MongoUrl,conf.GetConfig().Database)
oclib.Init("oc-monitor")
// create argo
wf_id := retrieveWorkflowId(conf.GetConfig().ExecutionID)
conf.GetConfig().WorkflowID = wf_id
// // create argo
new_wf := workflow_builder.WorflowDB{}
err := new_wf.LoadFrom(conf.GetConfig().WorkflowID)
@@ -73,33 +81,35 @@ func main() {
workflowName = getContainerName(argo_file_path)
wf_logger = logger.With().Str("argo_name", workflowName).Str("workflow_id",conf.GetConfig().WorkflowID).Logger()
wf_logger = logger.With().Str("argo_name", workflowName).Str("workflow_id",conf.GetConfig().WorkflowID).Str("workflow_execution_id",conf.GetConfig().ExecutionID).Logger()
wf_logger.Debug().Msg("Testing argo name")
executeWorkflow(argo_file_path)
fmt.Println("Logs sent to Loki successfully.")
}
// logger.Info().Msg(string(output))
func retrieveWorkflowId(exec_id string) string {
// // Initialize LokiLogger
// lokiLogger := NewLokiLogger("http://localhost:3100/loki/api/v1/push") // Replace with your Loki URL
res := oclib.LoadOne(oclib.LibDataEnum(oclib.WORKFLOW_EXECUTION),exec_id)
// Run the Argo command
if res.Code != 200 {
logger.Error().Msg("Could not retrieve workflow ID from execution ID " + exec_id)
return ""
}
// logger.Info().Msg(string(output))
// // Send logs to Loki
// if err := lokiLogger.Log(`{job="argo"}`, string(output)); err != nil {
// log.Fatalf("failed to send logs to Loki: %v", err)
// }
wf_exec := res.ToWorkflowExecution()
log.Println("Logs sent to Loki successfully.")
return wf_exec.WorkflowID
}
func executeWorkflow(argo_file_path string) {
var stdout, stderr io.ReadCloser
// var stderr io.ReadCloser
var err error
cmd := exec.Command("argo", "submit", "--watch", "argo_workflows/"+argo_file_path, "--serviceaccount=argo", "-n", "argo")
cmd := exec.Command("argo", "submit", "--watch", "argo_workflows/"+argo_file_path, "--serviceaccount=argo", "-n","argo")
if stdout, err = cmd.StdoutPipe(); err != nil{
wf_logger.Error().Msg("Could not retrieve stdoutpipe " + err.Error())
return
@@ -114,24 +124,28 @@ func executeWorkflow(argo_file_path string) {
panic(err)
}
go logWorkflow(stdout)
go logWorkflow(stderr)
var wg sync.WaitGroup
go logWorkflow(stdout, &wg)
go logWorkflow(stderr,&wg)
time.Sleep(time.Second * 1)
go logPods(workflowName)
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())
// updateStatus(exec_id, FATAL)
}
// Function to log the logs generated by the pods
// logPods(need to retrieve the name of the workflow) {
// executes "argo logs nameWorkflow"
// returns the output to a logging method
//}
wg.Wait()
}
// We could improve this function by creating an object with the same attribute as the output
// and only send a new log if the current object has different values than the previous
func logWorkflow(pipe io.ReadCloser) {
func logWorkflow(pipe io.ReadCloser, wg *sync.WaitGroup) {
var current_watch, previous_watch models.ArgoWatch
watch_output := make([]string,0)
@@ -142,18 +156,56 @@ func logWorkflow(pipe io.ReadCloser) {
if(strings.HasPrefix(log, "Progress:")){
current_watch = *models.NewArgoLogs(watch_output)
fmt.Println("Status : " + current_watch.Status)
workflowName = current_watch.Name
if(!current_watch.Equals(previous_watch)){
wg.Add(1)
defer wg.Done()
if(current_watch.Status == "Succeeded"){
fmt.Print()
}
checkStatus(current_watch.Status, previous_watch.Status)
jsonified, err := json.Marshal(current_watch)
if err != nil {
logger.Error().Msg("Could not create watch log")
}
wf_logger.Info().Msg(string(jsonified))
previous_watch = current_watch
current_watch = models.ArgoWatch{}
}
}
}
}
// Debug, no logs sent
func logPods(workflow_name string){
var stderr io.ReadCloser
var err error
cmd := exec.Command("argo","logs",workflow_name, "-n", "argo")
if stderr, err = cmd.StderrPipe(); err != nil{
wf_logger.Error().Msg("Could not retrieve stderrpipe " + err.Error())
return
}
if err := cmd.Start(); err != nil {
panic(err)
}
scanner := bufio.NewScanner(stderr)
for scanner.Scan() {
log := scanner.Text()
// fmt.Println(log)
wf_logger.Info().Msg(log)
}
if err := cmd.Wait(); err != nil {
wf_logger.Error().Msg("Could not execute argo logs")
}
}
func loadConfig(is_k8s bool, parser *argparse.Parser){
var o *onion.Onion
@@ -168,7 +220,7 @@ func loadConfig(is_k8s bool, parser *argparse.Parser){
// We can't use underscore in the env variable names because it's the delimitor with OCMONITOR too
setConf(is_k8s, o, parser)
if (!IsValidUUID(conf.GetConfig().WorkflowID)){
if (!IsValidUUID(conf.GetConfig().ExecutionID)){
logger.Fatal().Msg("Provided ID is not an UUID")
}
}
@@ -177,12 +229,12 @@ func setConf(is_k8s bool, o *onion.Onion, parser *argparse.Parser) {
if is_k8s {
conf.GetConfig().LokiURL = o.GetStringDefault("lokiurl", "http://127.0.0.1:3100")
conf.GetConfig().WorkflowID = o.GetString("workflow")
conf.GetConfig().LokiURL = o.GetStringDefault("mongourl", "mongodb://127.0.0.1:27017")
conf.GetConfig().ExecutionID = o.GetString("workflow")
conf.GetConfig().MongoUrl = o.GetStringDefault("mongourl", "mongodb://127.0.0.1:27017")
conf.GetConfig().Database = o.GetStringDefault("database", "DC_myDC")
} else {
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"})
workflow := parser.String("w", "workflow", &argparse.Options{Required: true, Help: "Name of the workflow to request from oc-catalog API"})
workflow := parser.String("w", "workflow", &argparse.Options{Required: true, Help: "Execution ID of the workflow to request from oc-catalog API"})
mongo := parser.String("m", "mongo", &argparse.Options{Required: true, Help: "URL to reach the MongoDB"})
db := parser.String("d", "database", &argparse.Options{Required: true, Help: "Name of the database to query in MongoDB"})
err := parser.Parse(os.Args)
@@ -192,7 +244,7 @@ func setConf(is_k8s bool, o *onion.Onion, parser *argparse.Parser) {
}
conf.GetConfig().LokiURL = *url
conf.GetConfig().WorkflowID = *workflow
conf.GetConfig().ExecutionID = *workflow
conf.GetConfig().MongoUrl = *mongo
conf.GetConfig().Database = *db
@@ -239,3 +291,26 @@ func IsValidUUID(u string) bool {
container_name := re.FindString(argo_file)
return container_name
}
// Uses the ArgoWatch object to update status of the workflow execution object
func checkStatus(current string, previous string) {
if current != previous {
updateStatus(current)
}
}
func updateStatus(status string) {
exec_id := conf.GetConfig().ExecutionID
wf_exec := &workflow_execution.WorkflowExecution{}
wf_exec.ArgoStatusToState(status)
serialized := wf_exec.Serialize()
res := oclib.UpdateOne(oclib.LibDataEnum(oclib.WORKFLOW_EXECUTION),serialized, exec_id)
if res.Code != 200 {
logger.Error().Msg("Could not update status for workflow execution " + exec_id)
}
fmt.Printf("status argo : %s /nstatus db : %s",status,serialized["state"])
}