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

update oclib

Browse Source
This commit is contained in:
mr
2026-02-02 14:36:00 +01:00
parent 01098842dd
commit 87951c8a77
4 changed files with 26 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files

75
daemons/schedule_manager.go
View File

@@ -14,7 +14,6 @@ import (
"cloud.o-forge.io/core/oc-lib/tools"
"github.com/google/uuid"
"github.com/nats-io/nats.go"
"github.com/rs/zerolog"
"go.mongodb.org/mongo-driver/bson/primitive"
)
@@ -24,10 +23,12 @@ type ScheduledExecution struct {
Mu sync.Mutex
}
func (sb *ScheduledExecution) DeleteSchedules(uuid string) {
func (sb *ScheduledExecution) DeleteSchedules(resp tools.NATSResponse) {
var m map[string]string
json.Unmarshal(resp.Payload, m)
Executions.Mu.Lock()
defer Executions.Mu.Unlock()
delete(sb.Execs, uuid)
delete(sb.Execs, m["id"])
}
func (sb *ScheduledExecution) AddSchedules(new_executions []*workflow_execution.WorkflowExecution, logger zerolog.Logger) {
@@ -57,66 +58,12 @@ type ScheduleManager struct {
Logger zerolog.Logger
}
// Goroutine listening to a NATS server for updates
// on workflows' scheduling. Messages must contain
// workflow execution ID, to allow retrieval of execution infos
func (s *ScheduleManager) ListenNATS() {
for {
nc, err := nats.Connect(oclib.GetConfig().NATSUrl)
if err != nil {
s.Logger.Error().Msg("Could not connect to NATS")
time.Sleep(10 * time.Second)
continue
}
defer nc.Close()
var wg sync.WaitGroup
wg.Add(3)
go s.listenForChange(nc, tools.REMOVE.GenerateKey(oclib.WORKFLOW.String()), tools.REMOVE, wg)
go s.listenForChange(nc, tools.CREATE.GenerateKey(oclib.WORKFLOW.String()), tools.CREATE, wg)
go s.listenForChange(nc, tools.WORKFLOW_EVENT.GenerateKey(oclib.WORKFLOW.String()), tools.WORKFLOW_EVENT, wg)
wg.Wait()
break
}
}
// Goroutine listening to a NATS server for updates
// on workflows' scheduling. Messages must contain
// workflow execution ID, to allow retrieval of execution infos
func (s *ScheduleManager) listenForChange(nc *nats.Conn, chanName string, natsTools tools.NATSMethod, wg sync.WaitGroup) {
defer wg.Done()
ch := make(chan *nats.Msg, 64)
fmt.Println("Listening to " + chanName)
subs, err := nc.ChanSubscribe(chanName, ch)
if err != nil {
s.Logger.Error().Msg("Error listening to NATS : " + err.Error())
}
defer subs.Unsubscribe()
for msg := range ch {
map_mess := map[string]string{}
json.Unmarshal(msg.Data, &map_mess)
str := "new"
if natsTools == tools.NATSMethod(tools.DELETE) {
str = "deleted"
}
fmt.Println("Catching " + str + " workflow... " + map_mess["id"])
switch natsTools {
case tools.NATSMethod(tools.DELETE):
Executions.DeleteSchedules(map_mess["id"])
case tools.NATSMethod(tools.CREATE):
s.getNextScheduledWorkflows(1)
case tools.NATSMethod(tools.WORKFLOW_EVENT):
s.executeWorkflow(map_mess["workflow_id"])
}
}
}
// Used at launch of the component to retrieve the next scheduled workflows
// and then every X minutes in case some workflows were scheduled before launch
func (s *ScheduleManager) SchedulePolling() {
var sleep_time float64 = 20
for {
s.getNextScheduledWorkflows(1)
s.GetNextScheduledWorkflows(tools.NATSResponse{})
s.Logger.Info().Msg("Current list of schedules -------> " + fmt.Sprintf("%v", len(Executions.Execs)))
time.Sleep(time.Second * time.Duration(sleep_time))
}
@@ -141,10 +88,12 @@ func (s *ScheduleManager) getExecution(from time.Time, to time.Time) (exec_list
return
}
func (s *ScheduleManager) executeWorkflow(wfID string) {
func (s *ScheduleManager) ExecuteWorkflow(resp tools.NATSResponse) {
var m map[string]string
json.Unmarshal(resp.Payload, &m)
res := resources.WorkflowResource{}
access := res.GetAccessor(&tools.APIRequest{})
if d, code, err := access.LoadOne(wfID); code == 200 && err == nil {
if d, code, err := access.LoadOne(m["workflow_id"]); code == 200 && err == nil {
eventExec := &workflow_execution.WorkflowExecution{
WorkflowID: d.GetID(),
ExecDate: time.Now(),
@@ -156,15 +105,15 @@ func (s *ScheduleManager) executeWorkflow(wfID string) {
}
func (s *ScheduleManager) getNextScheduledWorkflows(minutes float64) {
func (s *ScheduleManager) GetNextScheduledWorkflows(_ tools.NATSResponse) {
start := time.Now().UTC()
fmt.Println(s.getExecution(
start.Add(time.Second*time.Duration(-1)).UTC(),
start.Add(time.Minute*time.Duration(minutes)).UTC(),
start.Add(time.Minute*time.Duration(1)).UTC(),
))
if next_wf_exec, err := s.getExecution(
start.Add(time.Second*time.Duration(-1)).UTC(),
start.Add(time.Minute*time.Duration(minutes)).UTC(),
start.Add(time.Minute*time.Duration(1)).UTC(),
); err != nil {
s.Logger.Error().Msg("Could not retrieve next schedules")
} else {