oc-schedulerd/daemons/schedule_manager.go

124 lines
3.2 KiB
Go
Raw Normal View History

2024-07-11 18:25:40 +02:00
package daemons
import (
"encoding/json"
"fmt"
"net/url"
"time"
"oc-scheduler/logger"
"oc-scheduler/models"
"github.com/nats-io/nats.go"
)
2024-07-18 10:23:22 +02:00
// NATS daemon listens to subject " workflowsUpdate "
// workflowsUpdate messages must be formatted following this pattern '{"workflow" : "", "start_date" : "", "stop_date" : "" }'
2024-07-11 18:25:40 +02:00
type ScheduleManager struct {
2024-07-23 12:16:20 +02:00
Api_url string
bookings *models.ScheduledBooking
ws models.HttpQuery
2024-07-11 18:25:40 +02:00
}
2024-07-23 12:16:20 +02:00
func (s *ScheduleManager) SetBookings(b *models.ScheduledBooking){
s.bookings = b
}
2024-07-11 18:25:40 +02:00
// Goroutine listening to a NATS server for updates
// on workflows' scheduling. Messages must contain
2024-07-18 10:23:22 +02:00
// workflow's name, start_date and stop_date while there
2024-07-11 18:25:40 +02:00
// is no way to get scheduling infos for a specific workflow
2024-07-29 15:45:32 +02:00
func (s *ScheduleManager) ListenForWorkflowSubmissions(){
2024-07-11 18:25:40 +02:00
2024-07-23 12:16:20 +02:00
if(s.bookings == nil){
logger.Logger.Fatal().Msg("booking has not been set in the schedule manager")
}
2024-07-11 18:25:40 +02:00
nc, _ := nats.Connect(nats.DefaultURL)
defer nc.Close()
ch := make(chan *nats.Msg, 64)
subs , err := nc.ChanSubscribe("workflowsUpdate", ch)
if err != nil {
logger.Logger.Fatal().Msg("Error listening to NATS")
}
defer subs.Unsubscribe()
for msg := range(ch){
fmt.Println("Waiting...")
map_mess := retrieveMapFromSub(msg.Data)
2024-07-23 12:16:20 +02:00
s.bookings.Mu.Lock()
2024-07-11 18:25:40 +02:00
2024-07-18 10:23:22 +02:00
start, err := time.Parse(time.RFC3339,map_mess["start_date"])
2024-07-11 18:25:40 +02:00
if err != nil{
logger.Logger.Error().Msg(err.Error())
}
2024-07-18 10:23:22 +02:00
stop, err := time.Parse(time.RFC3339,map_mess["stop_date"])
2024-07-11 18:25:40 +02:00
if err != nil{
logger.Logger.Error().Msg(err.Error())
}
2024-07-23 12:16:20 +02:00
s.bookings.AddSchedule(models.Booking{Workflow: map_mess["workflow"], Start: start, Stop: stop })
s.bookings.Mu.Unlock()
2024-07-11 18:25:40 +02:00
}
}
2024-07-18 10:23:22 +02:00
// At the moment very simplistic, but could be useful if we send bigger messages
2024-07-11 18:25:40 +02:00
func retrieveMapFromSub(message []byte) (result_map map[string]string) {
json.Unmarshal(message, &result_map)
return
}
// 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
2024-07-29 15:45:32 +02:00
func (s *ScheduleManager) SchedulePolling (){
2024-07-11 18:25:40 +02:00
for(true){
err := s.getNextScheduledWorkflows(s.Api_url, 0.3)
if err != nil {
logger.Logger.Fatal().Msg("Failed to get the workspaces list, check api url and that api server is up : " + s.Api_url)
}
logger.Logger.Info().Msg("Current list of schedules")
2024-07-23 12:16:20 +02:00
fmt.Println(s.bookings.Bookings)
2024-07-11 18:25:40 +02:00
time.Sleep(time.Minute * 5)
}
}
func (s *ScheduleManager) getNextScheduledWorkflows(apiurl string, hours float64) (error) {
s.ws.Init(apiurl)
params := url.Values{}
start := time.Now().UTC()
2024-07-18 10:23:22 +02:00
params.Add("start_date", start.Format(time.RFC3339))
2024-07-11 18:25:40 +02:00
time_span := time.Hour * time.Duration(hours)
2024-07-18 10:23:22 +02:00
params.Add("stop_date",start.Add(time_span).Format(time.RFC3339))
2024-07-11 18:25:40 +02:00
body, err := s.ws.Get("v1/schedule?" + params.Encode())
if err != nil {
return err
}
var workflows []map[string]string
json.Unmarshal(body,&workflows)
2024-07-23 12:16:20 +02:00
s.bookings.Mu.Lock()
defer s.bookings.Mu.Unlock()
2024-07-11 18:25:40 +02:00
for _, workflow := range(workflows){
2024-07-18 10:23:22 +02:00
start, _ := time.Parse(time.RFC3339,workflow["start_date"])
stop, _ := time.Parse(time.RFC3339,workflow["stop_date"])
2024-07-11 18:25:40 +02:00
2024-07-23 12:16:20 +02:00
s.bookings.AddSchedule(models.Booking{Workflow: workflow["Workflow"], Start: start, Stop: stop})
2024-07-11 18:25:40 +02:00
}
return nil
}