// 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 workflow_builder import ( "fmt" . "oc-monitord/models" "os" "regexp" "strings" "time" oclib "cloud.o-forge.io/core/oc-lib" "cloud.o-forge.io/core/oc-lib/models/resources/compute" "cloud.o-forge.io/core/oc-lib/models/resources/processing" "cloud.o-forge.io/core/oc-lib/models/resources/workflow/graph" w "cloud.o-forge.io/core/oc-lib/models/workflow" "github.com/nwtgck/go-fakelish" "github.com/rs/zerolog" "gopkg.in/yaml.v3" ) var logger zerolog.Logger type ArgoBuilder struct { OriginWorkflow *w.Workflow Workflow Workflow Services []*Service Timeout int } type Workflow struct { ApiVersion string `yaml:"apiVersion"` Kind string `yaml:"kind"` Metadata struct { Name string `yaml:"name"` } `yaml:"metadata"` Spec Spec `yaml:"spec,omitempty"` } func (b *Workflow) setDag(dag *Dag) { for _, t := range b.Spec.Templates { if t.Name == "dag" { t.Dag = dag } } } func (b *Workflow) getDag() *Dag { for _, t := range b.Spec.Templates { if t.Name == "dag" { return t.Dag } } b.Spec.Templates = append(b.Spec.Templates, Template{Name: "dag", Dag: &Dag{}}) return b.Spec.Templates[len(b.Spec.Templates)-1].Dag } type Spec struct { Entrypoint string `yaml:"entrypoint"` Arguments []Parameter `yaml:"arguments,omitempty"` Volumes []VolumeClaimTemplate `yaml:"volumeClaimTemplates,omitempty"` Templates []Template `yaml:"templates"` Timeout int `yaml:"activeDeadlineSeconds,omitempty"` } func (b *ArgoBuilder) CreateDAG(write bool) (string, int, []string, []string, error) { // handle services by checking if there is only one processing with hostname and port firstItems, lastItems, volumes := b.createTemplates() b.createVolumes(volumes) if b.Timeout > 0 { b.Workflow.Spec.Timeout = b.Timeout } b.Workflow.Spec.Entrypoint = "dag" b.Workflow.ApiVersion = "argoproj.io/v1alpha1" b.Workflow.Kind = "Workflow" if !write { return "", len(b.Workflow.getDag().Tasks), firstItems, lastItems, nil } random_name := fakelish.GenerateFakeWord(5, 8) + "-" + fakelish.GenerateFakeWord(5, 8) b.Workflow.Metadata.Name = "oc-monitor-" + random_name logger = oclib.GetLogger() yamlified, err := yaml.Marshal(b.Workflow) if err != nil { logger.Error().Msg("Could not transform object to yaml file") return "", 0, firstItems, lastItems, err } // Give a unique name to each argo file with its timestamp DD:MM:YYYY_hhmmss current_timestamp := time.Now().Format("02_01_2006_150405") file_name := random_name + "_" + current_timestamp + ".yml" workflows_dir := "./argo_workflows/" err = os.WriteFile(workflows_dir+file_name, []byte(yamlified), 0660) if err != nil { logger.Error().Msg("Could not write the yaml file") return "", 0, firstItems, lastItems, err } return file_name, len(b.Workflow.getDag().Tasks), firstItems, lastItems, nil } func (b *ArgoBuilder) createTemplates() ([]string, []string, []VolumeMount) { volumes := []VolumeMount{} firstItems := []string{} lastItems := []string{} for _, comp := range b.OriginWorkflow.GetProcessings() { if comp.Processing.Container != nil { volumes, firstItems, lastItems = b.createArgoTemplates( comp.ID, comp.Processing, volumes, firstItems, lastItems) } else { logger.Error().Msg("Not enough configuration setup, template can't be created : " + comp.Processing.GetName()) return firstItems, lastItems, volumes } } firstWfTasks := map[string][]string{} latestWfTasks := map[string][]string{} relatedWfTasks := map[string][]string{} for _, wf := range b.OriginWorkflow.GetWorkflows() { realWorkflow, code, err := w.New().LoadOne(wf.Workflow.WorkflowID) if code != 200 { logger.Error().Msg("Error loading the workflow : " + err.Error()) continue } subBuilder := ArgoBuilder{OriginWorkflow: realWorkflow.(*w.Workflow), Timeout: b.Timeout} _, _, fi, li, err := subBuilder.CreateDAG(false) if err != nil { logger.Error().Msg("Error creating the subworkflow : " + err.Error()) continue } firstWfTasks[wf.ID] = fi if ok, depsOfIds := subBuilder.isArgoDependancy(wf.ID); ok { // IS BEFORE latestWfTasks[wf.ID] = li relatedWfTasks[wf.ID] = depsOfIds } subDag := subBuilder.Workflow.getDag() d := b.Workflow.getDag() d.Tasks = append(d.Tasks, subDag.Tasks...) // add the tasks of the subworkflow to the main workflow b.Workflow.Spec.Templates = append(b.Workflow.Spec.Templates, subBuilder.Workflow.Spec.Templates...) b.Workflow.Spec.Volumes = append(b.Workflow.Spec.Volumes, subBuilder.Workflow.Spec.Volumes...) b.Workflow.Spec.Arguments = append(b.Workflow.Spec.Arguments, subBuilder.Workflow.Spec.Arguments...) b.Services = append(b.Services, subBuilder.Services...) } for wfID, depsOfIds := range relatedWfTasks { for _, dep := range depsOfIds { for _, task := range b.Workflow.getDag().Tasks { if strings.Contains(task.Name, dep) { index := -1 for i, depp := range task.Dependencies { if strings.Contains(depp, wfID) { index = i break } } if index != -1 { task.Dependencies = append(task.Dependencies[:index], task.Dependencies[index+1:]...) } task.Dependencies = append(task.Dependencies, latestWfTasks[wfID]...) } } } } for wfID, fi := range firstWfTasks { deps := b.getArgoDependencies(wfID) if len(deps) > 0 { for _, dep := range fi { for _, task := range b.Workflow.getDag().Tasks { if strings.Contains(task.Name, dep) { task.Dependencies = append(task.Dependencies, deps...) } } } } } if b.Services != nil { dag := b.Workflow.getDag() dag.Tasks = append(dag.Tasks, Task{Name: "workflow-service-pod", Template: "workflow-service-pod"}) b.addServiceToArgo() } return firstItems, lastItems, volumes } func (b *ArgoBuilder) getAllComputeFromProcessing(p *processing.ProcessingResource) []*compute.ComputeResource { computeResources := []*compute.ComputeResource{} for _, link := range b.OriginWorkflow.Graph.Links { if link.Source.ID == p.GetID() && b.OriginWorkflow.Graph.Items[link.Destination.ID].Compute != nil { computeResources = append(computeResources, b.OriginWorkflow.Graph.Items[link.Destination.ID].Compute) } else if link.Destination.ID == p.GetID() && b.OriginWorkflow.Graph.Items[link.Source.ID].Compute != nil { computeResources = append(computeResources, b.OriginWorkflow.Graph.Items[link.Source.ID].Compute) } } return computeResources } func (b *ArgoBuilder) createArgoTemplates(id string, processing *processing.ProcessingResource, volumes []VolumeMount, firstItems []string, lastItems []string) ([]VolumeMount, []string, []string) { _, firstItems, lastItems = b.addTaskToArgo(b.Workflow.getDag(), id, processing, firstItems, lastItems) template := &Template{Name: getArgoName(processing.GetName(), id)} template.CreateContainer(processing, b.Workflow.getDag()) // get datacenter from the processing computes := b.getAllComputeFromProcessing(processing) for _, compute := range computes { b.CreateService(id, processing) template.Metadata.Labels = make(map[string]string) template.Metadata.Labels["app"] = "oc-service-" + processing.GetName() + "-" + compute.GetName() // Construct the template for the k8s service and add a link in graph between k8s service and processing } storages := b.OriginWorkflow.GetStoragesByRelatedProcessing(id) for _, storage := range storages { if storage.Local { volumes = template.Container.AddVolumeMount(VolumeMount{ Name: strings.ReplaceAll(strings.ToLower(storage.GetName()), " ", "-"), MountPath: storage.Path, Storage: storage, }, volumes) } } b.Workflow.Spec.Templates = append(b.Workflow.Spec.Templates, *template) return volumes, firstItems, lastItems } func (b *ArgoBuilder) addTaskToArgo(dag *Dag, graphItemID string, processing *processing.ProcessingResource, firstItems []string, lastItems []string) (*Dag, []string, []string) { unique_name := getArgoName(processing.GetName(), graphItemID) step := Task{Name: unique_name, Template: unique_name} if processing.Container != nil { for name, value := range processing.Container.Env { step.Arguments.Parameters = append(step.Arguments.Parameters, Parameter{ Name: name, Value: b.affectVariableEnv(value, b.OriginWorkflow.Graph), }) } } step.Dependencies = b.getArgoDependencies(graphItemID) name := "" if b.OriginWorkflow.Graph.Items[graphItemID].Processing != nil { name = b.OriginWorkflow.Graph.Items[graphItemID].Processing.GetName() } if b.OriginWorkflow.Graph.Items[graphItemID].Workflow != nil { name = b.OriginWorkflow.Graph.Items[graphItemID].Workflow.GetName() } if len(step.Dependencies) == 0 && name != "" { firstItems = append(firstItems, getArgoName(name, graphItemID)) } if ok, _ := b.isArgoDependancy(graphItemID); !ok && name != "" { lastItems = append(lastItems, getArgoName(name, graphItemID)) } dag.Tasks = append(dag.Tasks, step) return dag, firstItems, lastItems } func (b *ArgoBuilder) affectVariableEnv(envVar string, graph *graph.Graph) string { var myExp = regexp.MustCompile(`(\{\{.*\}\})`) // regex to find all the variables in the command matches := myExp.FindAllString(envVar, -1) // find all the variables in the command for _, match := range matches { // for each variable in the command splitted := strings.Split( // split the variable to get the inout and the vars only strings.ReplaceAll(strings.ReplaceAll(strings.ReplaceAll(match, "{{", ""), "}}", ""), " ", ""), "_") if len(splitted) < 3 { // if the variable is not well formatted, we skip it logger.Error().Msgf("The variable %v is not well formatted", match) continue } graphItemID := splitted[1] // graphitemid is the id of the object vars := splitted[2] // vars is the name of the variable of the object _, obj := graph.GetResource(graphItemID) if obj != nil { envVar = strings.ReplaceAll(envVar, match, fmt.Sprintf("%v", obj.Serialize()[vars])) } } return envVar } func (b *ArgoBuilder) createVolumes(volumes []VolumeMount) { // TODO : one think about remote volume but TG for _, volume := range volumes { new_volume := VolumeClaimTemplate{} new_volume.Metadata.Name = strings.ReplaceAll(strings.ToLower(volume.Name), " ", "-") new_volume.Spec.AccessModes = []string{"ReadWriteOnce"} new_volume.Spec.Resources.Requests.Storage = fmt.Sprintf("%v", volume.Storage.Size) + volume.Storage.SizeType.ToArgo() b.Workflow.Spec.Volumes = append(b.Workflow.Spec.Volumes, new_volume) } } func (b *ArgoBuilder) isArgoDependancy(id string) (bool, []string) { dependancyOfIDs := []string{} isDeps := false for _, link := range b.OriginWorkflow.Graph.Links { source := b.OriginWorkflow.Graph.Items[link.Destination.ID].Processing if id == link.Source.ID && source != nil { isDeps = true dependancyOfIDs = append(dependancyOfIDs, getArgoName(source.GetName(), link.Destination.ID)) } wourceWF := b.OriginWorkflow.Graph.Items[link.Destination.ID].Workflow if id == link.Source.ID && wourceWF != nil { isDeps = true dependancyOfIDs = append(dependancyOfIDs, getArgoName(wourceWF.GetName(), link.Destination.ID)) } } return isDeps, dependancyOfIDs } func (b *ArgoBuilder) getArgoDependencies(id string) (dependencies []string) { for _, link := range b.OriginWorkflow.Graph.Links { source := b.OriginWorkflow.Graph.Items[link.Source.ID].Processing if id == link.Destination.ID && source != nil { dependency_name := getArgoName(source.GetName(), link.Source.ID) dependencies = append(dependencies, dependency_name) continue } } return } func getArgoName(raw_name string, component_id string) (formatedName string) { formatedName = strings.ReplaceAll(raw_name, " ", "-") formatedName += "-" + component_id formatedName = strings.ToLower(formatedName) return }