argo by kube invocation
This commit is contained in:
88
main.go
88
main.go
@@ -19,10 +19,14 @@ import (
|
||||
oclib "cloud.o-forge.io/core/oc-lib"
|
||||
|
||||
"cloud.o-forge.io/core/oc-lib/logs"
|
||||
"cloud.o-forge.io/core/oc-lib/models/booking"
|
||||
"cloud.o-forge.io/core/oc-lib/models/peer"
|
||||
"cloud.o-forge.io/core/oc-lib/models/utils"
|
||||
"cloud.o-forge.io/core/oc-lib/models/workflow_execution"
|
||||
"cloud.o-forge.io/core/oc-lib/tools"
|
||||
|
||||
tools2 "oc-monitord/tools"
|
||||
|
||||
"github.com/akamensky/argparse"
|
||||
"github.com/google/uuid"
|
||||
"github.com/goraz/onion"
|
||||
@@ -45,7 +49,6 @@ const defaultConfigFile = "/etc/oc/ocmonitord_conf.json"
|
||||
const localConfigFile = "./conf/local_ocmonitord_conf.json"
|
||||
|
||||
func main() {
|
||||
|
||||
os.Setenv("test_service", "true") // Only for service demo, delete before merging on main
|
||||
|
||||
// Test if monitor is launched outside (with parameters) or in a k8s environment (env variables sets)
|
||||
@@ -73,11 +76,10 @@ func main() {
|
||||
|
||||
logger.Debug().Msg("Loki URL : " + conf.GetConfig().LokiURL)
|
||||
logger.Debug().Msg("Workflow executed : " + conf.GetConfig().ExecutionID)
|
||||
exec := getExecution(conf.GetConfig().ExecutionID)
|
||||
conf.GetConfig().WorkflowID = exec.WorkflowID
|
||||
|
||||
wf_id := getWorkflowId(conf.GetConfig().ExecutionID)
|
||||
conf.GetConfig().WorkflowID = wf_id
|
||||
|
||||
logger.Debug().Msg("Starting construction of yaml argo for workflow :" + wf_id)
|
||||
logger.Debug().Msg("Starting construction of yaml argo for workflow :" + exec.WorkflowID)
|
||||
|
||||
if _, err := os.Stat("./argo_workflows/"); os.IsNotExist(err) {
|
||||
os.Mkdir("./argo_workflows/", 0755)
|
||||
@@ -92,7 +94,7 @@ func main() {
|
||||
logger.Error().Msg("Could not retrieve workflow " + conf.GetConfig().WorkflowID + " from oc-catalog API")
|
||||
}
|
||||
|
||||
argo_file_path, stepMax, err := new_wf.ExportToArgo(conf.GetConfig().Timeout)
|
||||
argo_file_path, stepMax, err := new_wf.ExportToArgo(exec.ExecutionsID, conf.GetConfig().Timeout)
|
||||
if err != nil {
|
||||
logger.Error().Msg("Could not create the Argo file for " + conf.GetConfig().WorkflowID)
|
||||
logger.Error().Msg(err.Error())
|
||||
@@ -104,30 +106,41 @@ func main() {
|
||||
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, stepMax)
|
||||
|
||||
if os.Getenv("KUBERNETES_SERVICE_HOST") == "" {
|
||||
// Not in a k8s environment, get conf from parameters
|
||||
fmt.Println("Executes outside of k8s")
|
||||
executeInside(argo_file_path, stepMax)
|
||||
} else {
|
||||
// Executed in a k8s environment
|
||||
fmt.Println("Executes inside a k8s")
|
||||
loadConfig(true, nil)
|
||||
}
|
||||
}
|
||||
|
||||
// Return the Workflow ID associated to a workflow execution object
|
||||
func getWorkflowId(exec_id string) string {
|
||||
func getExecution(exec_id string) *workflow_execution.WorkflowExecution {
|
||||
res := oclib.NewRequest(oclib.LibDataEnum(oclib.WORKFLOW_EXECUTION), "", conf.GetConfig().PeerID, []string{}, nil).LoadOne(exec_id)
|
||||
if res.Code != 200 {
|
||||
logger.Error().Msg("Could not retrieve workflow ID from execution ID " + exec_id)
|
||||
return ""
|
||||
return nil
|
||||
}
|
||||
wf_exec := res.ToWorkflowExecution()
|
||||
return wf_exec.WorkflowID
|
||||
return res.ToWorkflowExecution()
|
||||
}
|
||||
|
||||
// So far we only log the output from
|
||||
func executeInside(argo_file_path string, stepMax int) {
|
||||
t, err := tools2.NewService(conf.GetConfig().Mode)
|
||||
if err != nil {
|
||||
logger.Error().Msg("Could not create KubernetesTool")
|
||||
}
|
||||
t.CreateArgoWorkflow(argo_file_path)
|
||||
t.LogWorkflow("argo", workflowName, argo_file_path, stepMax, logWorkflow)
|
||||
}
|
||||
|
||||
func executeWorkflow(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
|
||||
// var stderr io.ReadCloser
|
||||
var err error
|
||||
cmd := exec.Command("argo", "submit", "--log", "./argo_workflows/"+argo_file_path, "--serviceaccount=argo", "-n", "argo")
|
||||
fmt.Println(cmd)
|
||||
if stdout, err = cmd.StdoutPipe(); err != nil {
|
||||
wf_logger.Error().Msg("Could not retrieve stdoutpipe " + err.Error())
|
||||
return
|
||||
@@ -141,7 +154,7 @@ func executeWorkflow(argo_file_path string, stepMax int) {
|
||||
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("fatal")
|
||||
updateStatus("fatal", "")
|
||||
}
|
||||
wg.Wait()
|
||||
}
|
||||
@@ -216,6 +229,7 @@ func setConf(is_k8s bool, o *onion.Onion, parser *argparse.Parser) {
|
||||
conf.GetConfig().Database = db
|
||||
} 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"})
|
||||
mode := parser.String("M", "mode", &argparse.Options{Required: false, Default: "kubernetes", 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"})
|
||||
peer := parser.String("p", "peer", &argparse.Options{Required: false, Default: "", Help: "Peer ID of the workflow to request from oc-catalog API"})
|
||||
mongo := parser.String("m", "mongo", &argparse.Options{Required: true, Default: "mongodb://127.0.0.1:27017", Help: "URL to reach the MongoDB"})
|
||||
@@ -231,6 +245,7 @@ func setConf(is_k8s bool, o *onion.Onion, parser *argparse.Parser) {
|
||||
conf.GetConfig().MongoURL = *mongo
|
||||
conf.GetConfig().Database = *db
|
||||
conf.GetConfig().Timeout = *timeout
|
||||
conf.GetConfig().Mode = *mode
|
||||
conf.GetConfig().ExecutionID = *execution
|
||||
conf.GetConfig().PeerID = *peer
|
||||
}
|
||||
@@ -281,19 +296,50 @@ func getContainerName(argo_file string) string {
|
||||
// Uses the ArgoWatch object to update status of the workflow execution object
|
||||
func checkStatus(current *models.ArgoWatch, previous *models.ArgoWatch) {
|
||||
if previous != nil && current.Status != previous.Status {
|
||||
updateStatus(current.Status)
|
||||
if len(current.Logs) > 0 {
|
||||
updateStatus(current.Status, current.Logs[0])
|
||||
} else {
|
||||
updateStatus(current.Status, "")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func updateStatus(status string) {
|
||||
func updateStatus(status string, log string) {
|
||||
exec_id := conf.GetConfig().ExecutionID
|
||||
|
||||
wf_exec := &workflow_execution.WorkflowExecutions{AbstractObject: utils.AbstractObject{UUID: conf.GetConfig().ExecutionID}}
|
||||
wf_exec := &workflow_execution.WorkflowExecution{AbstractObject: utils.AbstractObject{UUID: conf.GetConfig().ExecutionID}}
|
||||
wf_exec.ArgoStatusToState(status)
|
||||
_, _, err := workflow_execution.NewAccessor(&tools.APIRequest{
|
||||
exec, _, err := workflow_execution.NewAccessor(&tools.APIRequest{
|
||||
PeerID: conf.GetConfig().PeerID,
|
||||
}).UpdateOne(wf_exec, exec_id)
|
||||
if err != nil {
|
||||
logger.Error().Msg("Could not update status for workflow execution " + exec_id + err.Error())
|
||||
}
|
||||
splitted := strings.Split(log, "-")
|
||||
if len(splitted) > 1 {
|
||||
we := exec.(*workflow_execution.WorkflowExecution)
|
||||
itemID := splitted[len(splitted)-1] // TODO: in logs found item ID
|
||||
caller := &tools.HTTPCaller{
|
||||
URLS: map[tools.DataType]map[tools.METHOD]string{
|
||||
tools.PEER: {
|
||||
tools.POST: "/status/",
|
||||
},
|
||||
tools.BOOKING: {
|
||||
tools.PUT: "http://localhost:8080/booking/:id",
|
||||
},
|
||||
},
|
||||
}
|
||||
if we.PeerBookByGraph != nil {
|
||||
for peerID, val := range we.PeerBookByGraph {
|
||||
if val[itemID] == nil {
|
||||
continue
|
||||
}
|
||||
for _, log := range val[itemID] {
|
||||
(&peer.Peer{}).LaunchPeerExecution(peerID, log, tools.BOOKING, tools.PUT, &booking.Booking{
|
||||
State: we.State,
|
||||
}, caller)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user