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

Mode in CMD

Browse Source
This commit is contained in:
mr
2025-02-14 11:59:32 +01:00
parent 6621d14d74
commit 7246dea2b2
13 changed files with 76 additions and 1007 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
daemons/execute_monitor_local.go
View File

@@ -31,7 +31,7 @@ func (lm *LocalMonitor) LaunchLocalMonitor() {
}
func (lm *LocalMonitor) execLocalKube() {
args := []string{"-e", lm.ExecutionID, "-p", lm.PeerID, "-u", lm.LokiURL, "-m", conf.GetConfig().MongoUrl, "-d", conf.GetConfig().DBName}
args := []string{"-e", lm.ExecutionID, "-p", lm.PeerID, "-u", lm.LokiURL, "-m", conf.GetConfig().MongoUrl, "-d", conf.GetConfig().DBName, "-M", conf.GetConfig().Mode}
if lm.Duration > 0 {
args = append(args, "-t", fmt.Sprintf("%d", lm.Duration))
}

34
daemons/execution_manager.go
View File

@@ -10,33 +10,33 @@ import (
workflow_execution "cloud.o-forge.io/core/oc-lib/models/workflow_execution"
)
var Bookings = ScheduledBooking{Bookings: []*workflow_execution.WorkflowExecutions{}}
var Executions = ScheduledExecution{Execs: []*workflow_execution.WorkflowExecution{}}
type ExecutionManager struct{}
// Loop every second on the booking's list and move the booking that must start to a new list
// Loop every second on the Execution's list and move the Execution that must start to a new list
// that will be looped over to start them
func (em *ExecutionManager) RetrieveNextExecutions() {
logger := oclib.GetLogger()
for {
fmt.Println("Checking for bookings", len(Bookings.Bookings))
Bookings.Mu.Lock()
if len(Bookings.Bookings) > 0 {
bookings := Bookings.Bookings
for i := len(bookings) - 1; i >= 0; i-- {
if bookings[i].ExecDate.Before(time.Now().UTC()) {
logger.Info().Msg("Will execute " + bookings[i].UUID + " soon")
go em.executeBooking(bookings[i])
Bookings.Bookings = append(bookings[:i], bookings[i+1:]...)
fmt.Println("Checking for executions", len(Executions.Execs))
Executions.Mu.Lock()
if len(Executions.Execs) > 0 {
executions := Executions.Execs
for i := len(executions) - 1; i >= 0; i-- {
if executions[i].ExecDate.Before(time.Now().UTC()) {
logger.Info().Msg("Will execute " + executions[i].UUID + " soon")
go em.executeExecution(executions[i])
Executions.Execs = append(executions[:i], executions[i+1:]...)
}
}
}
Bookings.Mu.Unlock()
Executions.Mu.Unlock()
time.Sleep(time.Second)
}
}
func (em *ExecutionManager) executeBooking(booking *workflow_execution.WorkflowExecutions) {
func (em *ExecutionManager) executeExecution(Execution *workflow_execution.WorkflowExecution) {
// start execution
// create the yaml that describes the pod : filename, path/url to Loki
exec_method := os.Getenv("MONITOR_METHOD")
@@ -46,16 +46,16 @@ func (em *ExecutionManager) executeBooking(booking *workflow_execution.WorkflowE
} else {
logger.Debug().Msg("Executing oc-monitor localy")
duration := 0
if booking.EndDate != nil {
duration = int(booking.EndDate.Sub(booking.ExecDate).Seconds())
if Execution.EndDate != nil {
duration = int(Execution.EndDate.Sub(Execution.ExecDate).Seconds())
}
monitor := LocalMonitor{
Logger: logger,
Duration: duration,
LokiURL: conf.GetConfig().LokiUrl,
KubeURL: "localhost",
ExecutionID: booking.UUID,
PeerID: booking.CreatorID,
ExecutionID: Execution.UUID,
PeerID: Execution.CreatorID,
}
monitor.LaunchLocalMonitor()
}

42
daemons/schedule_manager.go
View File

@@ -17,36 +17,36 @@ import (
"go.mongodb.org/mongo-driver/bson/primitive"
)
type ScheduledBooking struct {
Bookings []*workflow_execution.WorkflowExecutions
Mu sync.Mutex
type ScheduledExecution struct {
Execs []*workflow_execution.WorkflowExecution
Mu sync.Mutex
}
func (sb *ScheduledBooking) DeleteSchedules(workflow_id string) {
toNotDelete := []*workflow_execution.WorkflowExecutions{}
for _, b := range sb.Bookings {
func (sb *ScheduledExecution) DeleteSchedules(workflow_id string) {
toNotDelete := []*workflow_execution.WorkflowExecution{}
for _, b := range sb.Execs {
if b.WorkflowID != workflow_id {
toNotDelete = append(toNotDelete, b)
}
}
Bookings.Mu.Lock()
defer Bookings.Mu.Unlock()
sb.Bookings = toNotDelete
Executions.Mu.Lock()
defer Executions.Mu.Unlock()
sb.Execs = toNotDelete
}
func (sb *ScheduledBooking) AddSchedules(new_bookings []*workflow_execution.WorkflowExecutions, logger zerolog.Logger) {
Bookings.Mu.Lock()
defer Bookings.Mu.Unlock()
for _, exec := range new_bookings {
func (sb *ScheduledExecution) AddSchedules(new_executions []*workflow_execution.WorkflowExecution, logger zerolog.Logger) {
Executions.Mu.Lock()
defer Executions.Mu.Unlock()
for _, exec := range new_executions {
fmt.Println("Adding "+exec.UUID, !sb.execIsSet(exec))
if !sb.execIsSet(exec) {
sb.Bookings = append(sb.Bookings, exec)
sb.Execs = append(sb.Execs, exec)
}
}
}
func (sb *ScheduledBooking) execIsSet(exec *workflow_execution.WorkflowExecutions) bool {
for _, b := range sb.Bookings {
func (sb *ScheduledExecution) execIsSet(exec *workflow_execution.WorkflowExecution) bool {
for _, b := range sb.Execs {
if b.Equals(exec) {
return true
}
@@ -102,7 +102,7 @@ func (s *ScheduleManager) listenForChange(nc *nats.Conn, chanName string, delete
}
fmt.Println("Catching " + str + " workflow... " + map_mess["id"])
if delete {
Bookings.DeleteSchedules(map_mess["id"])
Executions.DeleteSchedules(map_mess["id"])
} else {
s.getNextScheduledWorkflows(1)
}
@@ -115,11 +115,11 @@ func (s *ScheduleManager) SchedulePolling() {
var sleep_time float64 = 1
for {
s.getNextScheduledWorkflows(1)
s.Logger.Info().Msg("Current list of schedules -------> " + fmt.Sprintf("%v", len(Bookings.Bookings)))
s.Logger.Info().Msg("Current list of schedules -------> " + fmt.Sprintf("%v", len(Executions.Execs)))
time.Sleep(time.Minute * time.Duration(sleep_time))
}
}
func (s *ScheduleManager) getExecution(from time.Time, to time.Time) (exec_list []*workflow_execution.WorkflowExecutions, err error) {
func (s *ScheduleManager) getExecution(from time.Time, to time.Time) (exec_list []*workflow_execution.WorkflowExecution, err error) {
fmt.Printf("Getting workflows execution from %s to %s \n", from.String(), to.String())
f := dbs.Filters{
And: map[string][]dbs.Filter{
@@ -133,7 +133,7 @@ func (s *ScheduleManager) getExecution(from time.Time, to time.Time) (exec_list
return
}
for _, exec := range res.Data {
exec_list = append(exec_list, exec.(*workflow_execution.WorkflowExecutions))
exec_list = append(exec_list, exec.(*workflow_execution.WorkflowExecution))
}
fmt.Println("Found "+fmt.Sprintf("%v", len(exec_list))+" workflows", res)
return
@@ -151,6 +151,6 @@ func (s *ScheduleManager) getNextScheduledWorkflows(minutes float64) {
); err != nil {
s.Logger.Error().Msg("Could not retrieve next schedules")
} else {
Bookings.AddSchedules(next_wf_exec, s.Logger)
Executions.AddSchedules(next_wf_exec, s.Logger)
}
}