oc-schedulerd/argo_builder.go

203 lines
4.9 KiB
Go
Raw Normal View History

2024-05-02 09:52:28 +02:00
// 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"
2024-05-03 10:58:06 +02:00
"os"
2024-05-02 09:52:28 +02:00
"slices"
"strings"
2024-05-02 09:52:28 +02:00
"github.com/beego/beego/v2/core/logs"
"gopkg.in/yaml.v3"
)
type ArgoBuilder struct {
2024-05-03 10:58:06 +02:00
graph Graph
2024-05-02 09:52:28 +02:00
branches [][]string
2024-05-03 10:58:06 +02:00
Workflow Workflow
2024-05-02 09:52:28 +02:00
}
2024-05-03 10:58:06 +02:00
type Workflow struct {
Templates []Template `yaml:"templates"`
2024-05-02 09:52:28 +02:00
}
func (b *ArgoBuilder) CreateDAG() bool {
fmt.Println("list of branches : ", b.branches)
b.createTemplates()
2024-05-02 09:52:28 +02:00
b.createDAGstep()
2024-05-03 10:58:06 +02:00
yamlified, err := yaml.Marshal(b.Workflow)
2024-05-02 09:52:28 +02:00
if err != nil {
2024-05-03 10:58:06 +02:00
logs.Error("Could not transform object to yaml file")
return false
2024-05-02 09:52:28 +02:00
}
fmt.Println(string(yamlified))
2024-05-03 10:58:06 +02:00
err = os.WriteFile("argo.yml", []byte(yamlified), 0660)
if err != nil {
logs.Error("Could not write the yaml file")
return false
2024-05-03 10:58:06 +02:00
}
2024-05-02 09:52:28 +02:00
return true
}
func (b *ArgoBuilder) createTemplates() {
2024-05-02 09:52:28 +02:00
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
b.Workflow.Templates = append(b.Workflow.Templates, new_temp)
}
}
2024-05-02 09:52:28 +02:00
func (b *ArgoBuilder) createDAGstep() {
2024-05-03 10:58:06 +02:00
new_dag := Dag{}
2024-05-02 09:52:28 +02:00
for _, comp := range b.graph.Computings{
unique_name := comp.Name + "_" + comp.ID
2024-05-03 10:58:06 +02:00
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})
}
2024-05-03 10:58:06 +02:00
2024-05-02 09:52:28 +02:00
// 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)
}
}
}
2024-05-03 10:58:06 +02:00
new_dag.Tasks = append(new_dag.Tasks, step)
2024-05-02 09:52:28 +02:00
}
2024-05-03 10:58:06 +02:00
b.Workflow.Templates = append (b.Workflow.Templates, Template{Name: "dag", Dag: new_dag})
2024-05-02 09:52:28 +02:00
}
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
2024-05-02 09:52:28 +02:00
}