From 87951c8a77624f9d8c8548ba1908bd1cfb88a2b3 Mon Sep 17 00:00:00 2001 From: mr Date: Mon, 2 Feb 2026 14:36:00 +0100 Subject: [PATCH] update oclib --- daemons/schedule_manager.go | 75 ++++++------------------------------- go.mod | 4 +- go.sum | 6 +++ main.go | 8 +++- 4 files changed, 26 insertions(+), 67 deletions(-) diff --git a/daemons/schedule_manager.go b/daemons/schedule_manager.go index 32f8ee9..483513c 100644 --- a/daemons/schedule_manager.go +++ b/daemons/schedule_manager.go @@ -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 { diff --git a/go.mod b/go.mod index c54e172..d0280b4 100644 --- a/go.mod +++ b/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-20260115122757-1c3b9218f7fb + 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 diff --git a/go.sum b/go.sum index 4b1e2c6..7526b20 100644 --- a/go.sum +++ b/go.sum @@ -8,6 +8,12 @@ cloud.o-forge.io/core/oc-lib v0.0.0-20260113155325-5cdfc28d2f51 h1:jlSEprNaUBe62 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= diff --git a/main.go b/main.go index eaffcc1..becacfc 100644 --- a/main.go +++ b/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() }