oc-schedulerd/argo_builder.go

// A class that translates the informations held in the graph object
// via its lists of components into an argo file, using the a list of
// link ID to build the dag

package main

import (
	"fmt"
	"os"
	"slices"
	"strings"

	"github.com/beego/beego/v2/core/logs"
	"github.com/nwtgck/go-fakelish"
	"gopkg.in/yaml.v3"
)

type ArgoBuilder struct {
	graph		Graph			
	branches	[][]string
	Workflow	Workflow
}

type Workflow struct {
	ApiVersion	string			`yaml:"apiVersion"`
	Kind 		string			`yaml:"kind"`
	Metadata struct {
		GenerateName	string	`yaml:"generateName"`
	}							`yaml:"metadata"`
	Spec		Spec			`yaml:"spec,omitempty"`
}

type Spec struct {
	Entrypoint	string					`yaml:"entrypoint"`
	Arguments	[]Parameter				`yaml:"arguments,omitempty"`
	Volumes		[]VolumeClaimTemplate 	`yaml:"volumeClaimTemplates,omitempty"`
	Templates	[]Template 				`yaml:"templates"`
}

func (b *ArgoBuilder) CreateDAG() bool {
	fmt.Println("list of branches : ", b.branches)

	b.createTemplates()
	b.createDAGstep()
	b.createVolumes()
	b.Workflow.Spec.Entrypoint = "dag"

	b.Workflow.ApiVersion = "argoproj.io/v1alpha1"
	b.Workflow.Kind = "Workflow"
	b.Workflow.Metadata.GenerateName = "oc-test-" + generateName() 

	yamlified, err := yaml.Marshal(b.Workflow)

	if err != nil {
		logs.Error("Could not transform object to yaml file")
		return false
	}
	fmt.Println(string(yamlified))
	err = os.WriteFile("argo.yml", []byte(yamlified), 0660)
		if err != nil {
		logs.Error("Could not write the yaml file")
		return false
	}
	return true
}

func (b *ArgoBuilder) createTemplates()  {

	for _, comp := range b.graph.Computings{
		temp_container :=  Container{Image: comp.Name}
		temp_container.Command = getComputingCommands(comp.Command)
		temp_container.Args = getComputingArgs(comp.Arguments)
		input_names := getComputingEnvironmentName(comp.Environment)
		
		var inputs_container []Parameter 
		for _, name := range input_names {
			inputs_container = append(inputs_container, Parameter{Name: name})
		}
		
		new_temp := Template{Name: comp.Name + "_" + comp.ID, Container: temp_container}
		new_temp.Inputs.Parameters = inputs_container
		new_temp.Container.VolumeMounts = append(new_temp.Container.VolumeMounts, VolumeMount{Name: "workdir",MountPath: "/mnt/vol"}) // TODO : replace this with a search of the storage / data source name
		b.Workflow.Spec.Templates = append(b.Workflow.Spec.Templates, new_temp)
		
	}

}

func (b *ArgoBuilder) createDAGstep() {

	new_dag := Dag{}

	for _, comp := range b.graph.Computings{
		unique_name := comp.Name + "_" + comp.ID
		step := Task{Name: unique_name, Template: unique_name}
		comp_envs := getComputingEnvironment(comp.Environment)
		
		for name, value := range comp_envs {
			step.Arguments.Parameters = append(step.Arguments.Parameters, Parameter{Name: name, Value: value})
		}

		// For each branch, check if the computing has a dependency
		for _, branch := range b.branches {
			if b.componentInBranch(comp.ID,branch) {
			// retrieves the name (computing.name_computing.ID)
				dependency := b.getDependency(comp.ID,branch)
				if dependency != "" && !slices.Contains(step.Dependencies,dependency) {	
					step.Dependencies = append(step.Dependencies,dependency)
				}
			}
		}
		new_dag.Tasks = append(new_dag.Tasks, step)

	}

	b.Workflow.Spec.Templates = append (b.Workflow.Spec.Templates, Template{Name: "dag", Dag: new_dag})

}

func (b *ArgoBuilder) createVolumes() {
	// For testing purposes we only declare one volume, mounted in each computing
	new_volume := VolumeClaimTemplate{}
	new_volume.Metadata.Name = "workdir"
	new_volume.Spec.AccessModes = []string{"ReadWriteOnce"}
	new_volume.Spec.Resources.Requests.Storage = "1Gi"
	
	b.Workflow.Spec.Volumes = append(b.Workflow.Spec.Volumes, new_volume)
}

func (b *ArgoBuilder) getDependency(current_computing_id string, branch []string) string {
	
	for i := len(branch)-1; i >= 0 ; i-- {
		current_link := b.graph.Links[branch[i]]
		current_computing_found := false

		if current_link.Source == current_computing_id || current_link.Destination == current_computing_id {
			current_computing_found = true
		}
		if current_computing_found {
			return b.findPreviousComputing(current_computing_id,branch,i-1)
		}
	}
	
	return ""

}

func (b *ArgoBuilder) componentInBranch(component_id string, branch []string) bool {
	for _, link := range branch {
		if b.graph.Links[link].Source == component_id || b.graph.Links[link].Destination == component_id {
			return true
		}
	}
	return false
}

func (b *ArgoBuilder) findPreviousComputing(computing_id string, branch []string, index int) string {
	
	for i := index; i >= 0 ; i-- {
		previousLink := b.graph.Links[branch[i]]

		if previousLink.Source != computing_id && b.graph.getComponentType(previousLink.Source) == "computing"{
			name := b.graph.getComponentName(previousLink.Source) + "_" + previousLink.Source
			return name
		}
		if previousLink.Destination != computing_id && b.graph.getComponentType(previousLink.Destination) == "computing"{
			name := b.graph.getComponentName(previousLink.Destination) + "_" + previousLink.Destination
			return name
		}

	}
	return ""
}

func getComputingCommands(user_input string) (list_command []string) {
	if len(user_input) == 0 {
		return
	}

	list_command = strings.Split(user_input, ",")
	return
}

func getComputingArgs(user_input []string) (list_args []string) {
	if len(user_input) == 0 {
		return
	}

	for _, arg := range user_input{
		list_args = append(list_args, "'" + arg + "'") 
	}

	return
}

// Currently implements code to overcome problems in data structure
func getComputingEnvironment(user_input []string) (map_env map[string]string) {
	if len(user_input) == 0 {
		return
	}
	if(len(user_input) == 1){
		user_input = strings.Split(user_input[0],",")
	}

	map_env = make(map[string]string,0)

	for _, str := range user_input {
		new_pair := strings.Split(str,"=")
		
		if(len(new_pair) != 2) {
			logs.Error("Error extracting the environment variable from ", str)
			panic(0)
		}
		
		map_env[new_pair[0]] = new_pair[1]	
	}
	
	return
}

func getComputingEnvironmentName(user_input []string) (list_names []string){
	env_map := getComputingEnvironment(user_input)
	for name := range env_map {
		list_names = append(list_names, name)
	}

	return
}

func generateName() (Name string){
	Name = fakelish.GenerateFakeWord(5, 8) + "-" + fakelish.GenerateFakeWord(5, 8) 
	return
}


func testDoubleIndent(w Workflow) {
	for _, temp := range w.Spec.Templates{
		if temp.Name == "dag" {
			tasks := temp.Dag.Tasks
			fmt.Println("name")
			printYAML(tasks[0].Name)
			fmt.Println("template")
			printYAML(tasks[0].Template)
			fmt.Println("dependencies")
			printYAML(tasks[0].Dependencies)
			fmt.Println("arguments")
			printYAML(tasks[0].Arguments)


		}
	}
}

func printYAML(data interface{}) {
	yamlData, err := yaml.Marshal(data)
	if err != nil {
		fmt.Printf("Error marshalling YAML: %v\n", err)
		return
	}
	fmt.Println(string(yamlData))
}