integrated oclib

daemons/schedule_manager.go

@@ -3,13 +3,16 @@ package daemons
 import (
 	"encoding/json"
 	"fmt"
-	"net/url"
 	"time"
 
 	"oc-scheduler/logger"
 	"oc-scheduler/models"
 
+	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"
 	"github.com/nats-io/nats.go"
+	"go.mongodb.org/mongo-driver/bson/primitive"
 )
 
 // NATS daemon listens to subject " workflowsUpdate "
@@ -29,23 +32,26 @@ func (s *ScheduleManager) SetBookings(b *models.ScheduledBooking){
 
 // Goroutine listening to a NATS server for updates
 // on workflows' scheduling. Messages must contain 
-// workflow's name, start_date and stop_date while there  
-// is no way to get scheduling infos for a specific workflow
+// workflow execution ID, to allow retrieval of execution infos
 func (s *ScheduleManager) ListenForWorkflowSubmissions(){
 	
 	if(s.bookings == nil){
-		logger.Logger.Fatal().Msg("booking has not been set in the schedule manager")
+		logger.Logger.Error().Msg("booking has not been set in the schedule manager")
 	}
 
-	nc, _ := nats.Connect(nats.DefaultURL)
-	defer nc.Close()
+	nc, err := nats.Connect(nats.DefaultURL)
+	if err != nil {
+		logger.Logger.Error().Msg("Could not connect to NATS")
+		return
+	}
 
+	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")
+		logger.Logger.Error().Msg("Error listening to NATS")
 	}
 	defer subs.Unsubscribe()
 
@@ -56,21 +62,29 @@ func (s *ScheduleManager) ListenForWorkflowSubmissions(){
 	
 		s.bookings.Mu.Lock()
 		
-		start, err := time.Parse(time.RFC3339,map_mess["start_date"])
-		if err != nil{
-			logger.Logger.Error().Msg(err.Error())
-		}
-		stop, err := time.Parse(time.RFC3339,map_mess["stop_date"])
-		if err != nil{
-			logger.Logger.Error().Msg(err.Error())
-		}
+		wf_exec := getWorkflowExecution(map_mess["workflow"])
 
-		s.bookings.AddSchedule(models.Booking{Workflow: map_mess["workflow"], Start: start, Stop: stop })
+		s.bookings.AddSchedule(models.Booking{Workflow: map_mess["workflow"], Start: *wf_exec.ExecDate, Stop: *wf_exec.EndDate })
 		s.bookings.Mu.Unlock()
 	
 	}
 }
 
+
+func getWorkflowExecution(exec_id string) *workflow_execution.WorkflowExecution {
+
+	res := oclib.LoadOne(oclib.LibDataEnum(oclib.WORKFLOW_EXECUTION),exec_id)
+
+	if res.Code != 200 {
+		logger.Logger.Error().Msg("Could not retrieve workflow ID from execution ID " + exec_id)
+		return nil
+	}
+
+	wf_exec := res.ToWorkflowExecution()
+
+	return wf_exec
+}
+
 // At the moment very simplistic, but could be useful if we send bigger messages
 func retrieveMapFromSub(message []byte) (result_map map[string]string) {
 	json.Unmarshal(message, &result_map)
@@ -80,44 +94,63 @@ func retrieveMapFromSub(message []byte) (result_map map[string]string) {
 // 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(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)
-		}
+		s.getNextScheduledWorkflows(1800)
 		
 		logger.Logger.Info().Msg("Current list of schedules")
 		fmt.Println(s.bookings.Bookings)
 
-		time.Sleep(time.Minute * 5)
+		time.Sleep(time.Minute * time.Duration(sleep_time))
 	}
 }
-
-func (s *ScheduleManager) getNextScheduledWorkflows(apiurl string, hours float64) (error) {
-	s.ws.Init(apiurl)
-	params := url.Values{}
-	start := time.Now().UTC()
-	params.Add("start_date", start.Format(time.RFC3339))
-	time_span := time.Hour * time.Duration(hours)
-	params.Add("stop_date",start.Add(time_span).Format(time.RFC3339))
-	body, err := s.ws.Get("v1/schedule?" + params.Encode())
-
-	if err != nil {
-		return err
+func (s *ScheduleManager) getWorfklowExecution(from time.Time, to time.Time) (exec_list []workflow_execution.WorkflowExecution, err error) {
+	
+	f := dbs.Filters{
+		And: map[string][]dbs.Filter{
+			"execution_date" : {{Operator : dbs.GTE.String(), Value: primitive.NewDateTimeFromTime(from)}},
+			"end_date": {{Operator: dbs.LTE.String(), Value:  primitive.NewDateTimeFromTime(to)}},
+			// "state": {{Operator: dbs.EQUAL.String(), Value: 1}},
+		},
 	}
-	var workflows []map[string]string
-	json.Unmarshal(body,&workflows)
+
+ 	res := oclib.Search(&f,"",oclib.LibDataEnum(oclib.WORKFLOW_EXECUTION))
+	if res.Code != 200 {
+		logger.Logger.Error().Msg("Error loading")
+		return nil, nil
+	}
+
+	for _, exec := range(res.Data){
+		lib_data := oclib.LoadOne(oclib.LibDataEnum(oclib.WORKFLOW_EXECUTION),exec.GetID())
+		exec_obj := lib_data.ToWorkflowExecution()
+		exec_list = append(exec_list, *exec_obj)
+	}
+	return exec_list, nil
+}
+
+// TODO : refactor to implement oclib.Search
+func (s *ScheduleManager) getNextScheduledWorkflows(minutes float64)  {
+	start := time.Now().UTC()
+	end := start.Add(time.Minute * time.Duration(minutes)).UTC()
+	
+	fmt.Printf("Getting workflows execution from %s to %s \n", start.String(), end.String())
+
+	next_wf_exec, err := s.getWorfklowExecution(start,end)
+	if err != nil {
+		logger.Logger.Error().Msg("Could not retrieve next schedules")
+		return
+	}
+
 
 	s.bookings.Mu.Lock()
 	defer s.bookings.Mu.Unlock()
 
-	for _, workflow := range(workflows){
-		start, _ := time.Parse(time.RFC3339,workflow["start_date"])
-		stop, _ := time.Parse(time.RFC3339,workflow["stop_date"])
+	for _, exec := range(next_wf_exec){
+		exec_start := exec.ExecDate
+		exec_stop := exec.EndDate
 
-		s.bookings.AddSchedule(models.Booking{Workflow: workflow["Workflow"], Start: start, Stop: stop}) 
+		s.bookings.AddSchedule(models.Booking{Workflow: exec.UUID, Start: *exec_start, Stop: *exec_stop}) 
 	}
 	
-	return nil
 }