oc-monitord/main.go

299 lines
8.9 KiB
Go

package main
import (
"bufio"
"encoding/json"
"fmt"
"io"
"os"
"os/exec"
"regexp"
"strconv"
"strings"
"sync"
"oc-monitord/conf"
"oc-monitord/models"
"oc-monitord/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"
"cloud.o-forge.io/core/oc-lib/tools"
"github.com/akamensky/argparse"
"github.com/google/uuid"
"github.com/goraz/onion"
"github.com/rs/zerolog"
)
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"
func main() {
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
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
fmt.Println("Executes inside a k8s")
monitorLocal = true
loadConfig(true, nil)
}
logs.SetAppName("oc-monitord")
logger = logs.CreateLogger("oc-monitord", conf.GetConfig().LokiURL)
logger.Debug().Msg("Loki URL : " + conf.GetConfig().LokiURL)
logger.Debug().Msg("Workflow executed : " + conf.GetConfig().ExecutionID)
oclib.Init("oc-monitord","","")
wf_id := getWorkflowId(conf.GetConfig().ExecutionID)
conf.GetConfig().WorkflowID = wf_id
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)
logger.Info().Msg("Created argo_workflows/")
}
// // create argo
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")
}
argo_file_path, err := new_wf.ExportToArgo(conf.GetConfig().Timeout)
if err != nil {
logger.Error().Msg("Could not create the Argo file for " + conf.GetConfig().WorkflowID)
logger.Error().Msg(err.Error())
}
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()
wf_logger.Debug().Msg("Testing argo name")
executeWorkflow(argo_file_path)
}
// Return the Workflow ID associated to a workflow execution object
func getWorkflowId(exec_id string) string {
res := oclib.LoadOne(oclib.LibDataEnum(oclib.WORKFLOW_EXECUTION), exec_id)
if res.Code != 200 {
logger.Error().Msg("Could not retrieve workflow ID from execution ID " + exec_id)
return ""
}
wf_exec := res.ToWorkflowExecution()
return wf_exec.WorkflowID
}
func executeWorkflow(argo_file_path string) {
// 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", "--watch", "./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
}
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)
}
var wg sync.WaitGroup
go logWorkflow(stdout, &wg)
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")
}
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, wg *sync.WaitGroup) {
var current_watch, previous_watch models.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 = *models.NewArgoLogs(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")
}
wf_logger.Info().Msg(string(jsonified))
previous_watch = current_watch
current_watch = models.ArgoWatch{}
wg.Done()
}
}
}
}
// Debug, no logs sent
func logPods(pipe io.ReadCloser, name string) {
pods_logger = wf_logger.With().Str("sortie name", name).Logger()
scanner := bufio.NewScanner(pipe)
for scanner.Scan() {
log := scanner.Text()
pods_logger.Info().Msg(log)
}
}
func loadConfig(is_k8s bool, parser *argparse.Parser) {
var o *onion.Onion
o = initOnion(o)
// 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) {
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")
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")
}
conf.GetConfig().ExecutionID = o.GetString("workflow")
mongo := o.GetStringDefault("mongourl", "mongodb://127.0.0.1:27017")
db := o.GetStringDefault("database", "DC_myDC")
tools.SetConfig(mongo, 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"})
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, 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"})
timeout := parser.Int("t", "timeout", &argparse.Options{Required: false, Default: -1, Help: "Timeout for the execution of the workflow"})
err := parser.Parse(os.Args)
if err != nil {
fmt.Println(parser.Usage(err))
os.Exit(1)
}
conf.GetConfig().LokiURL = *url
conf.GetConfig().Timeout = *timeout
conf.GetConfig().ExecutionID = *workflow
tools.SetConfig(*mongo, *db, "")
}
}
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)
}
return o
}
func IsValidUUID(u string) bool {
_, err := uuid.Parse(u)
return err == nil
}
func getContainerName(argo_file string) string {
regex := "([a-zA-Z]+-[a-zA-Z]+)"
re := regexp.MustCompile(regex)
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"])
}