core/oc-schedulerd
5
0
Fork 0
Code Issues 6 Pull Requests Packages Projects Releases Wiki Activity

improved formatting

Browse Source
This commit is contained in:
pb 2024-07-18 10:23:22 +02:00
parent c373558e5a
commit 0856c90930
2 changed files with 14 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
daemons/schedule_manager.go
View File

@ -12,6 +12,9 @@ import (
"github.com/nats-io/nats.go" "github.com/nats-io/nats.go"
) )
// NATS daemon listens to subject " workflowsUpdate "
// workflowsUpdate messages must be formatted following this pattern '{"workflow" : "", "start_date" : "", "stop_date" : "" }'
type ScheduleManager struct { type ScheduleManager struct {
Api_url string Api_url string
@ -24,7 +27,7 @@ type ScheduleManager struct {
// Goroutine listening to a NATS server for updates // Goroutine listening to a NATS server for updates
// on workflows' scheduling. Messages must contain // on workflows' scheduling. Messages must contain
// workflow's name, startDate and stopDate while there // workflow's name, start_date and stop_date while there
// is no way to get scheduling infos for a specific workflow // is no way to get scheduling infos for a specific workflow
func (s *ScheduleManager) ListenWorkflowSubmissions(){ func (s *ScheduleManager) ListenWorkflowSubmissions(){
@ -47,11 +50,11 @@ func (s *ScheduleManager) ListenWorkflowSubmissions(){
s.list.Mu.Lock() s.list.Mu.Lock()
start, err := time.Parse(time.RFC3339,map_mess["startDate"]) start, err := time.Parse(time.RFC3339,map_mess["start_date"])
if err != nil{ if err != nil{
logger.Logger.Error().Msg(err.Error()) logger.Logger.Error().Msg(err.Error())
} }
stop, err := time.Parse(time.RFC3339,map_mess["stopDate"]) stop, err := time.Parse(time.RFC3339,map_mess["stop_date"])
if err != nil{ if err != nil{
logger.Logger.Error().Msg(err.Error()) logger.Logger.Error().Msg(err.Error())
} }
@ -62,7 +65,7 @@ func (s *ScheduleManager) ListenWorkflowSubmissions(){
} }
} }
// At the moment very simplistic, but could be useful if we send bigger messages
func retrieveMapFromSub(message []byte) (result_map map[string]string) { func retrieveMapFromSub(message []byte) (result_map map[string]string) {
json.Unmarshal(message, &result_map) json.Unmarshal(message, &result_map)
return return
@ -88,9 +91,9 @@ func (s *ScheduleManager) getNextScheduledWorkflows(apiurl string, hours float64
s.ws.Init(apiurl) s.ws.Init(apiurl)
params := url.Values{} params := url.Values{}
start := time.Now().UTC() start := time.Now().UTC()
params.Add("startDate", start.Format(time.RFC3339)) params.Add("start_date", start.Format(time.RFC3339))
time_span := time.Hour * time.Duration(hours) time_span := time.Hour * time.Duration(hours)
params.Add("stopDate",start.Add(time_span).Format(time.RFC3339)) params.Add("stop_date",start.Add(time_span).Format(time.RFC3339))
body, err := s.ws.Get("v1/schedule?" + params.Encode()) body, err := s.ws.Get("v1/schedule?" + params.Encode())
if err != nil { if err != nil {
@ -103,14 +106,12 @@ func (s *ScheduleManager) getNextScheduledWorkflows(apiurl string, hours float64
defer s.list.Mu.Unlock() defer s.list.Mu.Unlock()
for _, workflow := range(workflows){ for _, workflow := range(workflows){
start, _ := time.Parse(time.RFC3339,workflow["StartDate"]) start, _ := time.Parse(time.RFC3339,workflow["start_date"])
stop, _ := time.Parse(time.RFC3339,workflow["StopDate"]) stop, _ := time.Parse(time.RFC3339,workflow["stop_date"])
s.list.AddSchedule(models.Booking{Workflow: workflow["Workflow"], Start: start, Stop: stop}) s.list.AddSchedule(models.Booking{Workflow: workflow["Workflow"], Start: start, Stop: stop})
} }
return nil return nil
} }

5
models/schedule.go
View File

@ -31,7 +31,8 @@ func (sb *ScheduledBooking) AddSchedule(new_booking Booking){
logger.Logger.Info().Msg("Updated list schedules : \n " + sb.String()) logger.Logger.Info().Msg("Updated list schedules : \n " + sb.String())
} else { } else {
// Debug condition : delete once this feature is ready to be implemented // Debug condition : delete once this feature is ready to be implemented
logger.Logger.Debug().Msg("Workflow received not added, current schedule contains") logger.Logger.Debug().Msg("Workflow received not added")
logger.Logger.Debug().Msg("current schedule contains")
for _, booking := range(sb.Bookings){ for _, booking := range(sb.Bookings){
logger.Logger.Debug().Msg(booking.String()) logger.Logger.Debug().Msg(booking.String())
} }
@ -56,7 +57,7 @@ func (sb *ScheduledBooking) scheduleAlreadyExists(new_booking Booking) bool {
} }
func (b *Booking) String() string { func (b *Booking) String() string {
return fmt.Sprintf("{Workflow : %s , startDate : %s , stopDate : %s }", b.Workflow, b.Start.Format(time.RFC3339), b.Stop.Format(time.RFC3339)) return fmt.Sprintf("{workflow : %s , start_date : %s , stop_date : %s }", b.Workflow, b.Start.Format(time.RFC3339), b.Stop.Format(time.RFC3339))
} }