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

Adapt to new inputs env struct + instance of a resource

Browse Source
This commit is contained in:
mr
2025-02-05 08:38:50 +01:00
parent 0de2f9842b
commit e4305bdbd1
7 changed files with 128 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
daemons/execute_monitor_local.go
View File

@@ -12,6 +12,7 @@ type LocalMonitor struct {
LokiURL string
KubeURL string
ExecutionID string
PeerID string
Duration int
Logger zerolog.Logger
}
@@ -30,7 +31,7 @@ func (lm *LocalMonitor) LaunchLocalMonitor() {
}
func (lm *LocalMonitor) execLocalKube() {
args := []string{"-e", lm.ExecutionID, "-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}
if lm.Duration > 0 {
args = append(args, "-t", fmt.Sprintf("%d", lm.Duration))
}

4
daemons/execution_manager.go
View File

@@ -1,6 +1,7 @@
package daemons
import (
"fmt"
"oc-schedulerd/conf"
"os"
"time"
@@ -18,7 +19,7 @@ type ExecutionManager struct{}
func (em *ExecutionManager) RetrieveNextExecutions() {
logger := oclib.GetLogger()
for {
logger.Debug().Msg("New loop")
fmt.Println("Checking for bookings", len(Bookings.Bookings))
Bookings.Mu.Lock()
if len(Bookings.Bookings) > 0 {
bookings := Bookings.Bookings
@@ -54,6 +55,7 @@ func (em *ExecutionManager) executeBooking(booking *workflow_execution.WorkflowE
LokiURL: conf.GetConfig().LokiUrl,
KubeURL: "localhost",
ExecutionID: booking.UUID,
PeerID: booking.CreatorID,
}
monitor.LaunchLocalMonitor()
}

9
daemons/schedule_manager.go
View File

@@ -9,6 +9,7 @@ import (
oclib "cloud.o-forge.io/core/oc-lib"
"cloud.o-forge.io/core/oc-lib/dbs"
"cloud.o-forge.io/core/oc-lib/models/common/enum"
"cloud.o-forge.io/core/oc-lib/models/workflow_execution"
"cloud.o-forge.io/core/oc-lib/tools"
"github.com/nats-io/nats.go"
@@ -37,6 +38,7 @@ func (sb *ScheduledBooking) AddSchedules(new_bookings []*workflow_execution.Work
Bookings.Mu.Lock()
defer Bookings.Mu.Unlock()
for _, exec := range new_bookings {
fmt.Println("Adding "+exec.UUID, !sb.execIsSet(exec))
if !sb.execIsSet(exec) {
sb.Bookings = append(sb.Bookings, exec)
}
@@ -122,7 +124,7 @@ func (s *ScheduleManager) getExecution(from time.Time, to time.Time) (exec_list
f := dbs.Filters{
And: map[string][]dbs.Filter{
"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}},
"state": {{Operator: dbs.EQUAL.String(), Value: enum.SCHEDULED}},
},
}
res := oclib.NewRequest(oclib.LibDataEnum(oclib.WORKFLOW_EXECUTION), "", "", []string{}, nil).Search(&f, "", false)
@@ -133,11 +135,16 @@ func (s *ScheduleManager) getExecution(from time.Time, to time.Time) (exec_list
for _, exec := range res.Data {
exec_list = append(exec_list, exec.(*workflow_execution.WorkflowExecutions))
}
fmt.Println("Found "+fmt.Sprintf("%v", len(exec_list))+" workflows", res)
return
}
func (s *ScheduleManager) getNextScheduledWorkflows(minutes float64) {
start := time.Now().UTC()
fmt.Println(s.getExecution(
start.Add(time.Second*time.Duration(-1)).UTC(),
start.Add(time.Minute*time.Duration(minutes)).UTC(),
))
if next_wf_exec, err := s.getExecution(
start.Add(time.Second*time.Duration(-1)).UTC(),
start.Add(time.Minute*time.Duration(minutes)).UTC(),