oc-monitord/main.go

309 lines
9.3 KiB
Go
Raw Normal View History

2024-07-04 09:14:25 +02:00
package main
import (
2024-08-06 11:40:30 +02:00
"bufio"
"encoding/json"
"fmt"
"io"
"os"
2024-08-06 11:40:30 +02:00
"os/exec"
"regexp"
2024-08-22 10:52:49 +02:00
"strconv"
2024-08-06 11:40:30 +02:00
"strings"
2024-08-07 17:24:09 +02:00
"sync"
"oc-monitord/conf"
"oc-monitord/models"
"oc-monitord/workflow_builder"
2024-08-06 11:40:30 +02:00
oclib "cloud.o-forge.io/core/oc-lib"
2024-08-07 17:24:09 +02:00
"cloud.o-forge.io/core/oc-lib/logs"
2024-08-07 17:24:09 +02:00
"cloud.o-forge.io/core/oc-lib/models/workflow_execution"
2024-08-22 10:52:49 +02:00
"cloud.o-forge.io/core/oc-lib/tools"
2024-08-07 17:24:09 +02:00
"github.com/akamensky/argparse"
2024-08-06 11:40:30 +02:00
"github.com/google/uuid"
"github.com/goraz/onion"
"github.com/rs/zerolog"
2024-07-04 09:14:25 +02:00
)
2024-08-29 10:17:31 +02:00
// Command-line args:
// - url: Loki URL (default: "http://127.0.0.1:3100")
// - execution: Workflow Execution ID (required) to identify the current execution, allows to retrieve Workflow
// - mongo: MongoDB URL (default: "mongodb://127.0.0.1:27017")
// - db: MongoDB database name (default: "DC_myDC")
// - timeout: Execution timeout (default: -1)
var logger zerolog.Logger
var wf_logger zerolog.Logger
var pods_logger zerolog.Logger
var parser argparse.Parser
var monitorLocal bool
var workflowName string
const defaultConfigFile = "/etc/oc/ocmonitord_conf.json"
const localConfigFile = "./conf/local_ocmonitord_conf.json"
2024-07-04 09:14:25 +02:00
func main() {
2024-08-29 10:17:31 +02:00
os.Setenv("test_service","true") // Only for service demo, delete before merging on main
2024-08-06 11:40:30 +02:00
monitorLocal = false
// Test if monitor is launched outside (with parameters) or in a k8s environment (env variables sets)
if os.Getenv("KUBERNETES_SERVICE_HOST") == "" {
// Not in a k8s environment, get conf from parameters
2024-08-06 11:40:30 +02:00
fmt.Println("Executes outside of k8s")
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)
} else {
// Executed in a k8s environment
2024-08-06 11:40:30 +02:00
fmt.Println("Executes inside a k8s")
monitorLocal = true
loadConfig(true, nil)
2024-07-04 09:14:25 +02:00
}
logs.SetAppName("oc-monitord")
logger = logs.CreateLogger("oc-monitord", conf.GetConfig().LokiURL)
logger.Debug().Msg("Loki URL : " + conf.GetConfig().LokiURL)
2024-08-07 17:24:09 +02:00
logger.Debug().Msg("Workflow executed : " + conf.GetConfig().ExecutionID)
2024-08-27 11:01:26 +02:00
oclib.Init("oc-monitord","","")
2024-08-13 16:15:39 +02:00
wf_id := getWorkflowId(conf.GetConfig().ExecutionID)
2024-08-07 17:24:09 +02:00
conf.GetConfig().WorkflowID = wf_id
2024-08-13 16:15:39 +02:00
logger.Debug().Msg("Starting construction of yaml argo for workflow :" + wf_id)
if _, err := os.Stat("./argo_workflows/"); os.IsNotExist(err) {
os.Mkdir("./argo_workflows/", 0755)
2024-08-13 16:15:39 +02:00
logger.Info().Msg("Created argo_workflows/")
}
2024-08-07 17:24:09 +02:00
// // create argo
2024-08-06 11:40:30 +02:00
new_wf := workflow_builder.WorflowDB{}
err := new_wf.LoadFrom(conf.GetConfig().WorkflowID)
if err != nil {
logger.Error().Msg("Could not retrieve workflow " + conf.GetConfig().WorkflowID + " from oc-catalog API")
}
2024-08-22 10:52:49 +02:00
argo_file_path, err := new_wf.ExportToArgo(conf.GetConfig().Timeout)
2024-08-06 11:40:30 +02:00
if err != nil {
logger.Error().Msg("Could not create the Argo file for " + conf.GetConfig().WorkflowID)
2024-08-06 11:40:30 +02:00
logger.Error().Msg(err.Error())
2024-07-04 09:14:25 +02:00
}
2024-08-06 11:40:30 +02:00
logger.Debug().Msg("Created :" + argo_file_path)
workflowName = getContainerName(argo_file_path)
wf_logger = logger.With().Str("argo_name", workflowName).Str("workflow_id", conf.GetConfig().WorkflowID).Str("workflow_execution_id", conf.GetConfig().ExecutionID).Logger()
2024-08-06 11:40:30 +02:00
wf_logger.Debug().Msg("Testing argo name")
executeWorkflow(argo_file_path)
2024-08-07 17:24:09 +02:00
}
2024-08-06 11:40:30 +02:00
2024-08-13 16:15:39 +02:00
// Return the Workflow ID associated to a workflow execution object
func getWorkflowId(exec_id string) string {
2024-08-06 11:40:30 +02:00
res := oclib.LoadOne(oclib.LibDataEnum(oclib.WORKFLOW_EXECUTION), exec_id)
2024-08-07 17:24:09 +02:00
if res.Code != 200 {
logger.Error().Msg("Could not retrieve workflow ID from execution ID " + exec_id)
return ""
}
2024-08-07 17:24:09 +02:00
wf_exec := res.ToWorkflowExecution()
2024-07-04 09:14:25 +02:00
return wf_exec.WorkflowID
2024-07-04 09:14:25 +02:00
}
2024-08-06 11:40:30 +02:00
func executeWorkflow(argo_file_path string) {
2024-08-08 10:11:40 +02:00
// var stdout, stderr, stdout_logs, stderr_logs io.ReadCloser
var stdout, stderr io.ReadCloser
2024-08-07 17:24:09 +02:00
// var stderr io.ReadCloser
2024-08-06 11:40:30 +02:00
var err error
cmd := exec.Command("argo", "submit", "--watch", "./argo_workflows/"+argo_file_path, "--serviceaccount=argo", "-n", "argo")
2024-08-20 15:23:02 +02:00
fmt.Println(cmd)
if stdout, err = cmd.StdoutPipe(); err != nil {
2024-08-06 11:40:30 +02:00
wf_logger.Error().Msg("Could not retrieve stdoutpipe " + err.Error())
return
}
if stderr, err = cmd.StderrPipe(); err != nil {
2024-08-06 11:40:30 +02:00
wf_logger.Error().Msg("Could not retrieve stderrpipe " + err.Error())
return
}
2024-08-08 10:11:40 +02:00
if err := cmd.Start(); err != nil {
panic(err)
}
var wg sync.WaitGroup
2024-08-07 17:24:09 +02:00
go logWorkflow(stdout, &wg)
2024-08-06 11:40:30 +02:00
if err := cmd.Wait(); err != nil {
wf_logger.Error().Msg("Could not execute argo submit")
2024-08-07 17:24:09 +02:00
wf_logger.Error().Msg(err.Error() + bufio.NewScanner(stderr).Text())
2024-08-08 10:11:40 +02:00
updateStatus("fatal")
2024-08-06 11:40:30 +02:00
}
2024-08-07 17:24:09 +02:00
wg.Wait()
2024-08-06 11:40:30 +02:00
}
// 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
2024-08-07 17:24:09 +02:00
func logWorkflow(pipe io.ReadCloser, wg *sync.WaitGroup) {
2024-08-06 11:40:30 +02:00
var current_watch, previous_watch models.ArgoWatch
watch_output := make([]string, 0)
2024-08-06 11:40:30 +02:00
scanner := bufio.NewScanner(pipe)
2024-08-08 10:11:40 +02:00
for scanner.Scan() {
log := scanner.Text()
watch_output = append(watch_output, log)
if strings.HasPrefix(log, "Progress:") {
2024-08-08 10:11:40 +02:00
current_watch = *models.NewArgoLogs(watch_output)
workflowName = current_watch.Name
if !current_watch.Equals(previous_watch) {
2024-08-08 10:11:40 +02:00
wg.Add(1)
checkStatus(current_watch.Status, previous_watch.Status)
2024-08-08 10:11:40 +02:00
jsonified, err := json.Marshal(current_watch)
if err != nil {
logger.Error().Msg("Could not create watch log")
2024-08-06 11:40:30 +02:00
}
2024-08-08 10:11:40 +02:00
wf_logger.Info().Msg(string(jsonified))
previous_watch = current_watch
current_watch = models.ArgoWatch{}
wg.Done()
2024-08-06 11:40:30 +02:00
}
}
2024-08-08 10:11:40 +02:00
}
2024-08-06 11:40:30 +02:00
}
2024-08-07 17:24:09 +02:00
// Debug, no logs sent
func logPods(pipe io.ReadCloser, name string) {
pods_logger = wf_logger.With().Str("sortie name", name).Logger()
2024-08-08 10:11:40 +02:00
scanner := bufio.NewScanner(pipe)
2024-08-07 17:24:09 +02:00
for scanner.Scan() {
log := scanner.Text()
2024-08-08 10:11:40 +02:00
pods_logger.Info().Msg(log)
2024-08-07 17:24:09 +02:00
}
}
func loadConfig(is_k8s bool, parser *argparse.Parser) {
var o *onion.Onion
2024-08-06 11:40:30 +02:00
o = initOnion(o)
2024-08-06 11:40:30 +02:00
// These variables can only be retrieved in the onion
// Variables that don't depend on the environmen (from conf file), can be loaded after
// 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().ExecutionID) {
2024-08-06 11:40:30 +02:00
logger.Fatal().Msg("Provided ID is not an UUID")
}
}
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")
2024-08-22 10:52:49 +02:00
i, err := strconv.Atoi(o.GetString("timeout"))
if err == nil {
conf.GetConfig().Timeout = i
} else {
logger.Error().Msg("Could not parse timeout, using default value")
}
2024-08-07 17:24:09 +02:00
conf.GetConfig().ExecutionID = o.GetString("workflow")
2024-08-20 15:23:02 +02:00
mongo := o.GetStringDefault("mongourl", "mongodb://127.0.0.1:27017")
db := o.GetStringDefault("database", "DC_myDC")
2024-08-22 10:52:49 +02:00
tools.SetConfig(mongo, db, "")
2024-08-06 11:40:30 +02:00
} 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"})
2024-08-29 10:17:31 +02:00
execution := parser.String("e", "execution", &argparse.Options{Required: true, Help: "Execution ID of the workflow to request from oc-catalog API"})
2024-08-20 15:23:02 +02:00
mongo := parser.String("m", "mongo", &argparse.Options{Required: true, Default: "mongodb://127.0.0.1:27017", Help: "URL to reach the MongoDB"})
db := parser.String("d", "database", &argparse.Options{Required: true, Default: "DC_myDC", Help: "Name of the database to query in MongoDB"})
2024-08-22 10:52:49 +02:00
timeout := parser.Int("t", "timeout", &argparse.Options{Required: false, Default: -1, Help: "Timeout for the execution of the workflow"})
2024-08-06 11:40:30 +02:00
err := parser.Parse(os.Args)
if err != nil {
fmt.Println(parser.Usage(err))
os.Exit(1)
}
conf.GetConfig().LokiURL = *url
2024-08-22 10:52:49 +02:00
conf.GetConfig().Timeout = *timeout
2024-08-29 10:17:31 +02:00
conf.GetConfig().ExecutionID = *execution
2024-08-22 10:52:49 +02:00
tools.SetConfig(*mongo, *db, "")
2024-08-06 11:40:30 +02:00
}
2024-08-20 15:23:02 +02:00
2024-08-06 11:40:30 +02:00
}
2024-08-29 10:17:31 +02:00
2024-08-06 11:40:30 +02:00
func initOnion(o *onion.Onion) *onion.Onion {
logger = logs.CreateLogger("oc-monitord", "")
configFile := ""
l3 := onion.NewEnvLayerPrefix("_", "OCMONITORD")
l2, err := onion.NewFileLayer(defaultConfigFile, nil)
if err == nil {
logger.Info().Msg("Config file found : " + defaultConfigFile)
configFile = defaultConfigFile
}
l1, err := onion.NewFileLayer(localConfigFile, nil)
if err == nil {
logger.Info().Msg("Local config file found " + localConfigFile + ", overriding default file")
configFile = localConfigFile
}
if configFile == "" {
logger.Info().Msg("No config file found, using env")
o = onion.New(l3)
} else if l1 == nil && l2 == nil {
o = onion.New(l1, l2, l3)
} else if l1 == nil {
o = onion.New(l2, l3)
} else if l2 == nil {
o = onion.New(l1, l3)
}
2024-08-06 11:40:30 +02:00
return o
}
2024-08-06 11:40:30 +02:00
func IsValidUUID(u string) bool {
_, err := uuid.Parse(u)
return err == nil
}
func getContainerName(argo_file string) string {
2024-08-06 11:40:30 +02:00
regex := "([a-zA-Z]+-[a-zA-Z]+)"
re := regexp.MustCompile(regex)
2024-08-06 11:40:30 +02:00
container_name := re.FindString(argo_file)
return container_name
}
2024-08-07 17:24:09 +02:00
// 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)
2024-08-07 17:24:09 +02:00
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"])
}