oc-schedulerd/daemons/schedule_manager.go

132 lines
4.2 KiB
Go
Raw Normal View History

2024-07-11 18:25:40 +02:00
package daemons
import (
"encoding/json"
"fmt"
2024-08-20 09:23:05 +02:00
"oc-schedulerd/conf"
"sync"
2024-07-11 18:25:40 +02:00
"time"
2024-08-21 14:20:13 +02:00
"cloud.o-forge.io/core/oc-lib/tools"
2024-08-09 18:44:33 +02:00
oclib "cloud.o-forge.io/core/oc-lib"
"cloud.o-forge.io/core/oc-lib/dbs"
"cloud.o-forge.io/core/oc-lib/models/workflow_execution"
2024-07-11 18:25:40 +02:00
"github.com/nats-io/nats.go"
"github.com/rs/zerolog"
2024-08-09 18:44:33 +02:00
"go.mongodb.org/mongo-driver/bson/primitive"
2024-07-11 18:25:40 +02:00
)
type ScheduledBooking struct {
2024-08-21 14:20:13 +02:00
Bookings map[string]*workflow_execution.WorkflowExecution
Mu sync.Mutex
2024-07-23 12:16:20 +02:00
}
2024-07-11 18:25:40 +02:00
2024-08-21 14:20:13 +02:00
func (sb *ScheduledBooking) DeleteSchedules(workflow_id string) {
toDelete := []string{}
for k, b := range sb.Bookings {
if b.WorkflowID == workflow_id {
toDelete = append(toDelete, k)
}
}
2024-08-21 14:20:13 +02:00
Bookings.Mu.Lock()
defer Bookings.Mu.Unlock()
for _, k := range toDelete {
delete(sb.Bookings, k)
2024-08-20 15:24:46 +02:00
}
}
2024-08-21 14:20:13 +02:00
func (sb *ScheduledBooking) AddSchedules(new_bookings []*workflow_execution.WorkflowExecution, logger zerolog.Logger) {
Bookings.Mu.Lock()
defer Bookings.Mu.Unlock()
for _, exec := range new_bookings {
sb.Bookings[exec.GetID()] = exec
}
}
// NATS daemon listens to subject " workflowsUpdate "
// workflowsUpdate messages must be formatted following this pattern '{"workflow" : "", "start_date" : "", "stop_date" : "" }'
type ScheduleManager struct {
2024-08-20 15:24:46 +02:00
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
2024-08-21 14:20:13 +02:00
func (s *ScheduleManager) ListenNATS() {
2024-08-20 09:23:05 +02:00
nc, err := nats.Connect(conf.GetConfig().NatsUrl)
2024-08-09 18:44:33 +02:00
if err != nil {
s.Logger.Error().Msg("Could not connect to NATS")
2024-08-09 18:44:33 +02:00
return
}
defer nc.Close()
2024-08-21 14:20:13 +02:00
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()
2024-07-11 18:25:40 +02:00
2024-08-21 14:20:13 +02:00
}
2024-07-11 18:25:40 +02:00
2024-08-21 14:20:13 +02:00
// 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)
2024-07-11 18:25:40 +02:00
if err != nil {
2024-08-21 14:20:13 +02:00
s.Logger.Error().Msg("Error listening to NATS : " + err.Error())
2024-07-11 18:25:40 +02:00
}
defer subs.Unsubscribe()
for msg := range ch {
2024-08-21 14:20:13 +02:00
map_mess := map[string]string{}
json.Unmarshal(msg.Data, &map_mess)
fmt.Println("Catching new workflow... " + map_mess["id"])
if delete {
Bookings.DeleteSchedules(map_mess["id"])
} else {
s.getNextScheduledWorkflows(1)
}
2024-08-09 18:44:33 +02:00
}
}
2024-07-11 18:25:40 +02:00
// 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() {
2024-08-09 18:44:33 +02:00
var sleep_time float64 = 1
for {
2024-08-20 16:14:10 +02:00
s.getNextScheduledWorkflows(1)
s.Logger.Info().Msg("Current list of schedules -------> " + fmt.Sprintf("%v", len(Bookings.Bookings)))
2024-08-09 18:44:33 +02:00
time.Sleep(time.Minute * time.Duration(sleep_time))
2024-07-11 18:25:40 +02:00
}
}
2024-08-21 14:20:13 +02:00
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())
2024-08-09 18:44:33 +02:00
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}},
2024-08-09 18:44:33 +02:00
},
}
res := oclib.Search(&f, "", oclib.LibDataEnum(oclib.WORKFLOW_EXECUTION))
2024-08-09 18:44:33 +02:00
if res.Code != 200 {
s.Logger.Error().Msg("Error loading")
2024-08-21 14:20:13 +02:00
return
2024-08-09 18:44:33 +02:00
}
for _, exec := range res.Data {
2024-08-21 14:20:13 +02:00
exec_list = append(exec_list, exec.(*workflow_execution.WorkflowExecution))
2024-08-09 18:44:33 +02:00
}
2024-08-21 14:20:13 +02:00
return
2024-08-09 18:44:33 +02:00
}
func (s *ScheduleManager) getNextScheduledWorkflows(minutes float64) {
2024-07-11 18:25:40 +02:00
start := time.Now().UTC()
2024-08-21 14:20:13 +02:00
if next_wf_exec, err := s.getExecution(
start.Add(time.Second * time.Duration(-1)).UTC(),
start.Add(time.Minute * time.Duration(minutes)).UTC(),
); err != nil {
s.Logger.Error().Msg("Could not retrieve next schedules")
2024-08-21 14:20:13 +02:00
} else {
Bookings.AddSchedules(next_wf_exec, s.Logger)
2024-07-11 18:25:40 +02:00
}
}