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

integrated local execution of oc-monitor

Browse Source
This commit is contained in:
pb
2024-08-12 16:09:30 +02:00
parent e738e19aa7
commit 81dde868a3
6 changed files with 44 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
daemons/execute_monitor_local.go
View File

@@ -1,7 +1,7 @@
package daemons
import (
"fmt"
"oc-scheduler/conf"
"oc-scheduler/logger"
"os/exec"
)
@@ -27,11 +27,10 @@ func (lm *LocalMonitor) LaunchLocalMonitor (){
func (lm *LocalMonitor) execLocalKube (){
// kube_url := ""
cmd := exec.Command("../oc-monitor/oc-monitor", "-w",lm.WorkflowName, "-u", lm.LokiURL)
cmd := exec.Command("../oc-monitor/oc-monitor", "-w",lm.WorkflowName, "-u", lm.LokiURL, "-m", conf.GetConfig().MongoUrl,"-d", conf.GetConfig().DBName)
// cmd_ls := exec.Command("ls", "../oc-monitor")
output, err := cmd.CombinedOutput()
err := cmd.Start()
// output, err := cmd_ls.CombinedOutput()
fmt.Println(string(output))
if err !=nil {
logger.Logger.Error().Msg("Could not start oc-monitor for " + lm.WorkflowName + " : " + err.Error())
}

2
daemons/execution_manager.go
View File

@@ -60,7 +60,7 @@ func (em *ExecutionManager) executeBooking(booking models.Booking){
if exec_method == "local"{
logger.Logger.Debug().Msg("Executing oc-monitor localy")
monitor := LocalMonitor{LokiURL: conf.GetConfig().LokiUrl,KubeURL: "localhost",WorkflowName: booking.Workflow}
monitor := LocalMonitor{LokiURL: conf.GetConfig().LokiUrl,KubeURL: "localhost",WorkflowName: booking.Workflow,}
monitor.LaunchLocalMonitor()
}else{
logger.Logger.Error().Msg("TODO : executing oc-monitor in a k8s")

8
daemons/schedule_manager.go
View File

@@ -96,7 +96,7 @@ func retrieveMapFromSub(message []byte) (result_map map[string]string) {
func (s *ScheduleManager) SchedulePolling (){
var sleep_time float64 = 1
for(true){
s.getNextScheduledWorkflows(1800)
s.getNextScheduledWorkflows(3)
logger.Logger.Info().Msg("Current list of schedules")
fmt.Println(s.bookings.Bookings)
@@ -108,12 +108,10 @@ func (s *ScheduleManager) getWorfklowExecution(from time.Time, to time.Time) (ex
f := dbs.Filters{
And: map[string][]dbs.Filter{
"execution_date" : {{Operator : dbs.GTE.String(), Value: primitive.NewDateTimeFromTime(from)}},
"end_date": {{Operator: dbs.LTE.String(), Value: primitive.NewDateTimeFromTime(to)}},
// "state": {{Operator: dbs.EQUAL.String(), Value: 1}},
"execution_date" : {{Operator : dbs.GTE.String(), Value: primitive.NewDateTimeFromTime(from)}, {Operator: dbs.LTE.String(), Value: primitive.NewDateTimeFromTime(to)}},
"state": {{Operator: dbs.EQUAL.String(), Value: 1}},
},
}
res := oclib.Search(&f,"",oclib.LibDataEnum(oclib.WORKFLOW_EXECUTION))
if res.Code != 200 {
logger.Logger.Error().Msg("Error loading")