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

working oc-schedulerd

Browse Source
This commit is contained in:
mr
2025-01-17 17:21:17 +01:00
parent 8eeba712e7
commit 0de2f9842b
19 changed files with 257 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
daemons/execution_manager.go
View File

@@ -9,7 +9,7 @@ import (
workflow_execution "cloud.o-forge.io/core/oc-lib/models/workflow_execution"
)
var Bookings = ScheduledBooking{Bookings: []*workflow_execution.WorkflowExecution{}}
var Bookings = ScheduledBooking{Bookings: []*workflow_execution.WorkflowExecutions{}}
type ExecutionManager struct{}
@@ -35,7 +35,7 @@ func (em *ExecutionManager) RetrieveNextExecutions() {
}
}
func (em *ExecutionManager) executeBooking(booking *workflow_execution.WorkflowExecution) {
func (em *ExecutionManager) executeBooking(booking *workflow_execution.WorkflowExecutions) {
// start execution
// create the yaml that describes the pod : filename, path/url to Loki
exec_method := os.Getenv("MONITOR_METHOD")
@@ -45,14 +45,14 @@ func (em *ExecutionManager) executeBooking(booking *workflow_execution.WorkflowE
} else {
logger.Debug().Msg("Executing oc-monitor localy")
duration := 0
if booking.EndDate != nil && booking.ExecDate != nil {
duration = int(booking.EndDate.Sub(*booking.ExecDate).Seconds())
if booking.EndDate != nil {
duration = int(booking.EndDate.Sub(booking.ExecDate).Seconds())
}
monitor := LocalMonitor{
Logger: logger,
Duration: duration,
LokiURL: conf.GetConfig().LokiUrl,
KubeURL: "localhost",
Logger: logger,
Duration: duration,
LokiURL: conf.GetConfig().LokiUrl,
KubeURL: "localhost",
ExecutionID: booking.UUID,
}
monitor.LaunchLocalMonitor()

14
daemons/schedule_manager.go
View File

@@ -17,12 +17,12 @@ import (
)
type ScheduledBooking struct {
Bookings []*workflow_execution.WorkflowExecution
Bookings []*workflow_execution.WorkflowExecutions
Mu sync.Mutex
}
func (sb *ScheduledBooking) DeleteSchedules(workflow_id string) {
toNotDelete := []*workflow_execution.WorkflowExecution{}
toNotDelete := []*workflow_execution.WorkflowExecutions{}
for _, b := range sb.Bookings {
if b.WorkflowID != workflow_id {
toNotDelete = append(toNotDelete, b)
@@ -33,7 +33,7 @@ func (sb *ScheduledBooking) DeleteSchedules(workflow_id string) {
sb.Bookings = toNotDelete
}
func (sb *ScheduledBooking) AddSchedules(new_bookings []*workflow_execution.WorkflowExecution, logger zerolog.Logger) {
func (sb *ScheduledBooking) AddSchedules(new_bookings []*workflow_execution.WorkflowExecutions, logger zerolog.Logger) {
Bookings.Mu.Lock()
defer Bookings.Mu.Unlock()
for _, exec := range new_bookings {
@@ -43,7 +43,7 @@ func (sb *ScheduledBooking) AddSchedules(new_bookings []*workflow_execution.Work
}
}
func (sb *ScheduledBooking) execIsSet(exec *workflow_execution.WorkflowExecution) bool {
func (sb *ScheduledBooking) execIsSet(exec *workflow_execution.WorkflowExecutions) bool {
for _, b := range sb.Bookings {
if b.Equals(exec) {
return true
@@ -117,7 +117,7 @@ func (s *ScheduleManager) SchedulePolling() {
time.Sleep(time.Minute * time.Duration(sleep_time))
}
}
func (s *ScheduleManager) getExecution(from time.Time, to time.Time) (exec_list []*workflow_execution.WorkflowExecution, err error) {
func (s *ScheduleManager) getExecution(from time.Time, to time.Time) (exec_list []*workflow_execution.WorkflowExecutions, err error) {
fmt.Printf("Getting workflows execution from %s to %s \n", from.String(), to.String())
f := dbs.Filters{
And: map[string][]dbs.Filter{
@@ -125,13 +125,13 @@ func (s *ScheduleManager) getExecution(from time.Time, to time.Time) (exec_list
"state": {{Operator: dbs.EQUAL.String(), Value: 1}},
},
}
res := oclib.Search(&f, "", oclib.LibDataEnum(oclib.WORKFLOW_EXECUTION))
res := oclib.NewRequest(oclib.LibDataEnum(oclib.WORKFLOW_EXECUTION), "", "", []string{}, nil).Search(&f, "", false)
if res.Code != 200 {
s.Logger.Error().Msg("Error loading")
return
}
for _, exec := range res.Data {
exec_list = append(exec_list, exec.(*workflow_execution.WorkflowExecution))
exec_list = append(exec_list, exec.(*workflow_execution.WorkflowExecutions))
}
return
}