135 lines
4.2 KiB
Go
135 lines
4.2 KiB
Go
package daemons
|
|
|
|
import (
|
|
"encoding/json"
|
|
"fmt"
|
|
"oc-schedulerd/conf"
|
|
"sync"
|
|
"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/workflow_execution"
|
|
"cloud.o-forge.io/core/oc-lib/tools"
|
|
"github.com/nats-io/nats.go"
|
|
"github.com/rs/zerolog"
|
|
"go.mongodb.org/mongo-driver/bson/primitive"
|
|
)
|
|
|
|
type ScheduledBooking struct {
|
|
Bookings []*workflow_execution.WorkflowExecution
|
|
Mu sync.Mutex
|
|
}
|
|
|
|
func (sb *ScheduledBooking) DeleteSchedules(workflow_id string) {
|
|
toNotDelete := []*workflow_execution.WorkflowExecution{}
|
|
for _, b := range sb.Bookings {
|
|
if b.WorkflowID != workflow_id {
|
|
toNotDelete = append(toNotDelete, b)
|
|
}
|
|
}
|
|
Bookings.Mu.Lock()
|
|
defer Bookings.Mu.Unlock()
|
|
sb.Bookings = toNotDelete
|
|
}
|
|
|
|
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 = append(sb.Bookings , exec)
|
|
}
|
|
}
|
|
// NATS daemon listens to subject " workflowsUpdate "
|
|
// workflowsUpdate messages must be formatted following this pattern '{"workflow" : "", "start_date" : "", "stop_date" : "" }'
|
|
|
|
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() {
|
|
nc, err := nats.Connect(conf.GetConfig().NatsUrl)
|
|
if err != nil {
|
|
s.Logger.Error().Msg("Could not connect to NATS")
|
|
return
|
|
}
|
|
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()
|
|
|
|
}
|
|
|
|
// 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 {
|
|
Bookings.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 = 1
|
|
for {
|
|
s.getNextScheduledWorkflows(1)
|
|
s.Logger.Info().Msg("Current list of schedules -------> " + fmt.Sprintf("%v", len(Bookings.Bookings)))
|
|
time.Sleep(time.Minute * time.Duration(sleep_time))
|
|
}
|
|
}
|
|
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())
|
|
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}},
|
|
},
|
|
}
|
|
res := oclib.Search(&f, "", oclib.LibDataEnum(oclib.WORKFLOW_EXECUTION))
|
|
if res.Code != 200 {
|
|
s.Logger.Error().Msg("Error loading")
|
|
return
|
|
}
|
|
for _, exec := range res.Data {
|
|
exec_list = append(exec_list, exec.(*workflow_execution.WorkflowExecution))
|
|
}
|
|
return
|
|
}
|
|
|
|
func (s *ScheduleManager) getNextScheduledWorkflows(minutes float64) {
|
|
start := time.Now().UTC()
|
|
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")
|
|
} else {
|
|
Bookings.AddSchedules(next_wf_exec, s.Logger)
|
|
}
|
|
}
|