Merge branch 'feature/event'

daemons/execution_manager.go

@@ -1,12 +1,14 @@
 package daemons
 
 import (
-	"fmt"
 	"oc-schedulerd/conf"
 	"time"
 
 	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"
 	workflow_execution "cloud.o-forge.io/core/oc-lib/models/workflow_execution"
+	"go.mongodb.org/mongo-driver/bson/primitive"
 )
 
 var Executions = ScheduledExecution{Execs: map[string]workflow_execution.WorkflowExecution{}}
@@ -18,23 +20,53 @@ type ExecutionManager struct{}
 func (em *ExecutionManager) RetrieveNextExecutions() {
 	logger := oclib.GetLogger()
 	for {
-		fmt.Println("Checking for executions", len(Executions.Execs))
 		Executions.Mu.Lock()
 		if len(Executions.Execs) > 0 {
 			executions := Executions.Execs
+			orderedExec := map[int]map[string]workflow_execution.WorkflowExecution{}
 			for execId, exec := range executions {
-				if exec.ExecDate.Before(time.Now().UTC()) {
+				if orderedExec[exec.Priority] == nil {
-					logger.Info().Msg("Will execute " + execId + " soon")
+					orderedExec[exec.Priority] = map[string]workflow_execution.WorkflowExecution{}
-					go em.executeExecution(&exec)
+				}
-					delete(executions, execId)
+				orderedExec[exec.Priority][execId] = exec
+			}
+			for i := range []int{7, 6, 5, 4, 3, 2, 1, 0} { // priority in reversed
+				if orderedExec[i] == nil {
+					continue
+				}
+
+				for execId, exec := range orderedExec[i] {
+					if i == 0 && em.isAStartingExecutionBeforeEnd(&exec) { // BEST EFFORT exception
+						continue
+					}
+					if exec.ExecDate.Before(time.Now().UTC()) {
+						logger.Info().Msg("Will execute " + execId + " soon")
+						go em.executeExecution(&exec)
+						delete(executions, execId)
+					}
 				}
 			}
+
 		}
 		Executions.Mu.Unlock()
 		time.Sleep(time.Second)
 	}
 }
 
+func (em *ExecutionManager) isAStartingExecutionBeforeEnd(execution *workflow_execution.WorkflowExecution) bool {
+	access := workflow_execution.NewAccessor(nil)
+	l, _, err := access.Search(&dbs.Filters{
+		And: map[string][]dbs.Filter{
+			"execution_date": {{Operator: dbs.LTE.String(), Value: primitive.NewDateTimeFromTime(*execution.EndDate)}},
+			"state":          {{Operator: dbs.EQUAL.String(), Value: enum.SCHEDULED}},
+		}, // TODO later should refine on each endpoint
+	}, "", false)
+	if err != nil && len(l) == 0 {
+		return false
+	}
+	return true
+}
+
 func (em *ExecutionManager) executeExecution(execution *workflow_execution.WorkflowExecution) {
 	// start execution
 	// create the yaml that describes the pod : filename, path/url to Loki

daemons/schedule_manager.go

@@ -9,9 +9,11 @@ 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/resources"
 	"cloud.o-forge.io/core/oc-lib/models/workflow_execution"
 	"cloud.o-forge.io/core/oc-lib/tools"
-	"github.com/nats-io/nats.go"
+
+	"github.com/google/uuid"
 	"github.com/rs/zerolog"
 	"go.mongodb.org/mongo-driver/bson/primitive"
 )
@@ -21,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) {
@@ -54,62 +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(2)
-		go s.listenForChange(nc, tools.REMOVE.GenerateKey(oclib.WORKFLOW.String()), true, wg)
-		go s.listenForChange(nc, tools.CREATE.GenerateKey(oclib.WORKFLOW.String()), false, 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, delete bool, 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 delete {
-			str = "deleted"
-		}
-		fmt.Println("Catching " + str + " workflow... " + map_mess["id"])
-		if delete {
-			Executions.DeleteSchedules(map_mess["id"])
-		} else {
-			s.getNextScheduledWorkflows(1)
-		}
-	}
-}
-
 // 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))
 	}
@@ -134,15 +88,32 @@ func (s *ScheduleManager) getExecution(from time.Time, to time.Time) (exec_list
 	return
 }
 
-func (s *ScheduleManager) getNextScheduledWorkflows(minutes float64) {
+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(m["workflow_id"]); code == 200 && err == nil {
+		eventExec := &workflow_execution.WorkflowExecution{
+			WorkflowID:   d.GetID(),
+			ExecDate:     time.Now(),
+			ExecutionsID: uuid.New().String(),
+			State:        enum.SCHEDULED,
+		}
+		Executions.AddSchedules([]*workflow_execution.WorkflowExecution{eventExec}, s.Logger)
+	}
+
+}
+
+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 {

go.mod

@@ -5,8 +5,9 @@ go 1.23.0
 toolchain go1.24.0
 
 require (
-	cloud.o-forge.io/core/oc-lib v0.0.0-20250808141553-f4b0cf5683de
+	cloud.o-forge.io/core/oc-lib v0.0.0-20260129122033-186ba3e689c7
 	github.com/beego/beego v1.12.12
+	github.com/google/uuid v1.6.0
 	github.com/goraz/onion v0.1.3
 	github.com/nats-io/nats.go v1.44.0
 	github.com/rs/zerolog v1.34.0
@@ -23,7 +24,6 @@ require (
 	github.com/go-playground/universal-translator v0.18.1 // indirect
 	github.com/go-playground/validator/v10 v10.27.0 // indirect
 	github.com/golang/snappy v1.0.0 // indirect
-	github.com/google/uuid v1.6.0 // indirect
 	github.com/hashicorp/golang-lru v1.0.2 // indirect
 	github.com/klauspost/compress v1.18.0 // indirect
 	github.com/kr/text v0.2.0 // indirect

go.sum

@@ -1,5 +1,19 @@
 cloud.o-forge.io/core/oc-lib v0.0.0-20250808141553-f4b0cf5683de h1:s47eEnWRCjBMOxbec5ROHztuwu0Zo7MuXgqWizgkiXU=
 cloud.o-forge.io/core/oc-lib v0.0.0-20250808141553-f4b0cf5683de/go.mod h1:vHWauJsS6ryf7UDqq8hRXoYD5RsONxcFTxeZPOztEuI=
+cloud.o-forge.io/core/oc-lib v0.0.0-20260112132629-be770ec763b1 h1:MOllR71ruHvBZK3nxeOAnHa9xjnotnlngMEGZqEBp2g=
+cloud.o-forge.io/core/oc-lib v0.0.0-20260112132629-be770ec763b1/go.mod h1:vHWauJsS6ryf7UDqq8hRXoYD5RsONxcFTxeZPOztEuI=
+cloud.o-forge.io/core/oc-lib v0.0.0-20260112144037-c35b06e0bc3c h1:9lXrj1agE1clFfxOXRrVXi4PEvlAuWKb4z977c2uk4k=
+cloud.o-forge.io/core/oc-lib v0.0.0-20260112144037-c35b06e0bc3c/go.mod h1:vHWauJsS6ryf7UDqq8hRXoYD5RsONxcFTxeZPOztEuI=
+cloud.o-forge.io/core/oc-lib v0.0.0-20260113155325-5cdfc28d2f51 h1:jlSEprNaUBe628uP9a9TrJ16Q5Ej6OxHlAKNtrHrN2o=
+cloud.o-forge.io/core/oc-lib v0.0.0-20260113155325-5cdfc28d2f51/go.mod h1:vHWauJsS6ryf7UDqq8hRXoYD5RsONxcFTxeZPOztEuI=
+cloud.o-forge.io/core/oc-lib v0.0.0-20260115122757-1c3b9218f7fb h1:blCzhIrRW1gLTsAVVxFxfhA5LXenxiVNT4kn1MTphLg=
+cloud.o-forge.io/core/oc-lib v0.0.0-20260115122757-1c3b9218f7fb/go.mod h1:vHWauJsS6ryf7UDqq8hRXoYD5RsONxcFTxeZPOztEuI=
+cloud.o-forge.io/core/oc-lib v0.0.0-20260127083751-c69069449f1e h1:2WY0uwtE4tLVsowfZj51I8kZawjf9s2Cx6lozqNv6eg=
+cloud.o-forge.io/core/oc-lib v0.0.0-20260127083751-c69069449f1e/go.mod h1:vHWauJsS6ryf7UDqq8hRXoYD5RsONxcFTxeZPOztEuI=
+cloud.o-forge.io/core/oc-lib v0.0.0-20260127143728-3c052bf16572 h1:jrUHgs4DqNWLnLcb5nd4lrJim77+aGkJFACUfMogiu8=
+cloud.o-forge.io/core/oc-lib v0.0.0-20260127143728-3c052bf16572/go.mod h1:vHWauJsS6ryf7UDqq8hRXoYD5RsONxcFTxeZPOztEuI=
+cloud.o-forge.io/core/oc-lib v0.0.0-20260129122033-186ba3e689c7 h1:NRFGRqN+j5g3DrtXMYN5T5XSYICG+OU2DisjBdID3j8=
+cloud.o-forge.io/core/oc-lib v0.0.0-20260129122033-186ba3e689c7/go.mod h1:vHWauJsS6ryf7UDqq8hRXoYD5RsONxcFTxeZPOztEuI=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
 github.com/Knetic/govaluate v3.0.0+incompatible/go.mod h1:r7JcOSlj0wfOMncg0iLm8Leh48TZaKVeNIfJntJ2wa0=
 github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=

main.go

@@ -6,6 +6,7 @@ import (
 	"os"
 
 	oclib "cloud.o-forge.io/core/oc-lib"
+	"cloud.o-forge.io/core/oc-lib/tools"
 )
 
 func main() {
@@ -55,9 +56,12 @@ func main() {
 	sch_mngr := daemons.ScheduleManager{Logger: oclib.GetLogger()}
 	exe_mngr := daemons.ExecutionManager{}
 
-	go sch_mngr.ListenNATS()
+	go tools.NewNATSCaller().ListenNats(map[tools.NATSMethod]func(tools.NATSResponse){
+		tools.CREATE_EXECTUTION: sch_mngr.GetNextScheduledWorkflows,
+		tools.WORKFLOW_EVENT:    sch_mngr.ExecuteWorkflow,
+		tools.REMOVE_EXECUTION:  daemons.Executions.DeleteSchedules,
+	})
 	go sch_mngr.SchedulePolling()
-
 	exe_mngr.RetrieveNextExecutions()
 
 }