commit before rebuild

daemons/execution_manager.go

@@ -1,9 +1,15 @@
 package daemons
 
 import (
+	"bytes"
+	"oc-scheduler/conf"
 	"oc-scheduler/logger"
 	"oc-scheduler/models"
 	"oc-scheduler/workflow_builder"
+	"os"
+	"os/exec"
+	"regexp"
+	"text/template"
 	"time"
 )
 
@@ -12,6 +18,12 @@ type ExecutionManager struct {
 	executions	[]models.Booking	
 }
 
+type manifestValues struct{
+	ARGO_FILE 		string
+	LOKI_URL  		string
+	CONTAINER_NAME 	string
+}
+
 func (em *ExecutionManager) SetBookings(b *models.ScheduledBooking){
 	em.bookings = b
 }
@@ -20,6 +32,7 @@ func (em *ExecutionManager) SetBookings(b *models.ScheduledBooking){
 // that will be looped over to start them
 func (em *ExecutionManager) RetrieveNextExecutions(){
 
+
 	if(em.bookings == nil){
 		logger.Logger.Fatal().Msg("booking has not been set in the exection manager")
 	}
@@ -46,6 +59,7 @@ func (em *ExecutionManager) RetrieveNextExecutions(){
 }
 
 func (em *ExecutionManager) executeBooking(booking models.Booking){
+
 	// create argo
 	new_graph := workflow_builder.Graph{}
 	
@@ -54,15 +68,88 @@ func (em *ExecutionManager) executeBooking(booking models.Booking){
 		logger.Logger.Error().Msg("Could not retrieve workflow " + booking.Workflow + " from oc-catalog API")
 	}
 
-	argo_file, err := new_graph.ExportToArgo()
+	argo_file_path, err := new_graph.ExportToArgo()
 	if err != nil {
 		logger.Logger.Error().Msg("Could not create the Argo file for " + booking.Workflow )
 		logger.Logger.Error().Msg(err.Error())
 	}
 
-	_ = argo_file 
+	logger.Logger.Debug().Msg("Created :" + argo_file_path)
 	// start execution 
-		// locally launch a pod that contains oc-monitor, give it the name of the workflow 
-			// create the yaml that describes the pod : filename, path/url to Loki
-				// locally launch an argo workflow with the filename `argo submit PATH_TO_YAML --watch --serviceaccount=argo -n argo`
+	// create the yaml that describes the pod : filename, path/url to Loki
+	manifest, err := em.CreateManifest(booking, argo_file_path)
+	if err != nil {
+		logger.Logger.Error().Msg("Could not create manifest " + err.Error())
+	}
+
+	// launch a pod that contains oc-monitor, give it the name of the workflow
+	cmd := exec.Command("kubectl","apply","-f", "manifests/"+manifest)
+	output , err := cmd.CombinedOutput()
+	if err != nil {
+		logger.Logger.Error().Msg("failed to create new pod for " + booking.Workflow + " :" + err.Error())
+		return 
+	}
+	
+	logger.Logger.Debug().Msg("Result from kubectl apply : " + string(output))
+
+	// Transfer the argo file to the pod
+
+	cmd = exec.Command("kubectl","cp", "argo_workflows/" + argo_file_path, "pods/test-monitor:/app/workflows", "-c", "oc-monitor-" +  getContainerName(argo_file_path))
+	output, err  = cmd.CombinedOutput()
+	if err != nil {
+		logger.Logger.Error().Msg("failed to copy argo file to " + booking.Workflow + " :" + err.Error())
+		return
+	}
+	
+	logger.Logger.Debug().Msg("Result from kubectl cp : " + string(output))
+
+}
+
+func (*ExecutionManager) CreateManifest(booking models.Booking, argo_file string) (manifest_filepath string, err error) {
+
+	var filled_template bytes.Buffer
+
+	manifest_file, err := os.ReadFile("conf/monitor_pod_template.yml")
+	if err != nil {
+		logger.Logger.Error().Msg("Could not open the k8s template file for " + booking.Workflow)
+		return "", err
+	}
+
+	container_name := getContainerName(argo_file)
+	tmpl, err := template.New("manifest_template").Parse(string(manifest_file))
+	if err != nil {
+		logger.Logger.Error().Msg(err.Error())
+		return "", err
+	}
+
+	manifest_data := manifestValues{
+		ARGO_FILE: argo_file,
+		LOKI_URL: conf.GetConfig().Logs,
+		CONTAINER_NAME: container_name,
+	}
+
+	err = tmpl.Execute(&filled_template, manifest_data)
+	if err != nil {
+		logger.Logger.Error().Msg("Could not complete manifest template for " + booking.Workflow)
+		return "", err	}
+
+
+	manifest_filepath = booking.Workflow + "_manifest.yaml"
+
+	err = os.WriteFile("manifests/" + manifest_filepath, filled_template.Bytes(),0644)
+	if err != nil {
+		logger.Logger.Error().Msg("Could not write the YAML file for " + booking.Workflow + "'s manifest")
+		return "", err
+	}
+
+	return manifest_filepath, nil
+}
+
+func getContainerName(argo_file string) string {
+	regex := "([a-zA-Z]+-[a-zA-Z]+)"
+	re := regexp.MustCompile(regex)
+
+	container_name := re.FindString(argo_file)
+	return container_name
+
 }

daemons/schedule_manager.go

@@ -31,7 +31,7 @@ func (s *ScheduleManager) SetBookings(b *models.ScheduledBooking){
 // 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
-func (s *ScheduleManager) ListenWorkflowSubmissions(){
+func (s *ScheduleManager) ListenForWorkflowSubmissions(){
 	
 	if(s.bookings == nil){
 		logger.Logger.Fatal().Msg("booking has not been set in the schedule manager")
@@ -79,7 +79,7 @@ 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) RetrieveScheduling (){
+func (s *ScheduleManager) SchedulePolling (){
 	for(true){
 		err := s.getNextScheduledWorkflows(s.Api_url, 0.3)
 		if err != nil {