core/oc-monitord
6
0
Fork 0
Code Issues 3 Pull Requests Packages Projects Releases Wiki Activity

Monitor With Data Storage + Datas

Browse Source
This commit is contained in:
mr
2024-10-11 13:44:16 +02:00
parent f388351476
commit 310395f869
10 changed files with 532 additions and 491 deletions
Download Patch File Download Diff File Expand all files Collapse all files

457
workflow_builder/argo_builder.go
View File

@@ -8,12 +8,12 @@ import (
"fmt"
. "oc-monitord/models"
"os"
"slices"
"regexp"
"strings"
"time"
oclib "cloud.o-forge.io/core/oc-lib"
"cloud.o-forge.io/core/oc-lib/models/resource_model"
"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"
@@ -24,9 +24,9 @@ import (
var logger zerolog.Logger
type ArgoBuilder struct {
OriginWorkflow w.Workflow
OriginWorkflow *w.Workflow
Workflow Workflow
Services *Service
Services []*Service
Timeout int
}
@@ -39,6 +39,24 @@ type Workflow struct {
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"`
@@ -47,15 +65,11 @@ type Spec struct {
Timeout int `yaml:"activeDeadlineSeconds,omitempty"`
}
func (b *ArgoBuilder) CreateDAG() (string, error) {
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
b.createNginxVolumes()
b.createTemplates()
b.createDAGstep()
b.createVolumes()
firstItems, lastItems, volumes := b.createTemplates()
b.createVolumes(volumes)
if b.Timeout > 0 {
b.Workflow.Spec.Timeout = b.Timeout
@@ -63,15 +77,17 @@ func (b *ArgoBuilder) CreateDAG() (string, error) {
b.Workflow.Spec.Entrypoint = "dag"
b.Workflow.ApiVersion = "argoproj.io/v1alpha1"
b.Workflow.Kind = "Workflow"
random_name := generateWfName()
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 "", err
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"
@@ -79,189 +95,205 @@ func (b *ArgoBuilder) CreateDAG() (string, error) {
err = os.WriteFile(workflows_dir+file_name, []byte(yamlified), 0660)
if err != nil {
logger.Error().Msg("Could not write the yaml file")
return "", err
return "", 0, firstItems, lastItems, err
}
return file_name, nil
return file_name, len(b.Workflow.getDag().Tasks), firstItems, lastItems, nil
}
func (b *ArgoBuilder) createTemplates() {
for _, comp := range b.getProcessings() {
var command string
var args string
var env string
comp_res := comp.Processing
command = getStringValue(comp_res.AbstractResource, "command")
args = getStringValue(comp_res.AbstractResource, "args")
env = getStringValue(comp_res.AbstractResource, "env")
image_name := strings.Split(command, " ")[0] // TODO : decide where to store the image name, GUI or models.computing.Image
temp_container := Container{Image: image_name} // TODO : decide where to store the image name, GUI or models.computing.Image
temp_container.Command = getComputingCommands(command)
temp_container.Args = getComputingArgs(args, command)
// Only for dev purpose,
input_names := getComputingEnvironmentName(strings.Split(env, " "))
var inputs_container []Parameter
for _, name := range input_names {
inputs_container = append(inputs_container, Parameter{Name: name})
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
}
argo_name := getArgoName(comp_res.GetName(), comp.ID)
new_temp := Template{Name: argo_name, 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
new_temp.Container.VolumeMounts = append(new_temp.Container.VolumeMounts, VolumeMount{Name: "nginx-demo", MountPath: "/usr/share/nginx"}) // Used for processing services' demo with nginx
if b.isService(comp.ID) {
serv := b.CreateService(comp)
b.createService(serv, argo_name, comp.ID)
new_temp.Metadata.Labels = make(map[string]string)
new_temp.Metadata.Labels["app"] = "oc-service" // Construct the template for the k8s service and add a link in graph between k8s service and processing
// if err != nil {
// // TODO
// }
}
b.Workflow.Spec.Templates = append(b.Workflow.Spec.Templates, new_temp)
}
if b.Services != nil {
b.addServiceToArgo()
}
}
func (b *ArgoBuilder) createDAGstep() {
new_dag := Dag{}
for _, comp := range b.getProcessings() {
comp_res := comp.Processing
env := getStringValue(comp_res.AbstractResource, "env")
unique_name := getArgoName(comp_res.GetName(), comp.ID)
step := Task{Name: unique_name, Template: unique_name}
comp_envs := getComputingEnvironment(strings.Split(env, " "))
for name, value := range comp_envs {
step.Arguments.Parameters = append(step.Arguments.Parameters, Parameter{Name: name, Value: value})
}
// retrieves the name (computing.name-computing.ID)
step.Dependencies = b.getDependency(comp.ID) // Error : we use the component ID instead of the GraphItem ID -> store objects
new_dag.Tasks = append(new_dag.Tasks, step)
}
if b.Services != nil {
new_dag.Tasks = append(new_dag.Tasks, Task{Name: "workflow-service-pod", Template: "workflow-service-pod"})
}
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)
}
// For demo purposes, until we implement the use of storage ressources
func (b *ArgoBuilder) createNginxVolumes() {
new_volume := VolumeClaimTemplate{}
new_volume.Metadata.Name = "nginx-demo"
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) (dependencies []string) {
for _, link := range b.OriginWorkflow.Graph.Links {
if b.OriginWorkflow.Graph.Items[link.Source.ID].Processing == nil {
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) 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())
if len(processing.Expose) > 0 {
b.CreateService(id, processing)
template.Metadata.Labels = make(map[string]string)
template.Metadata.Labels["app"] = "oc-service" // 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
fmt.Println("source", source, current_computing_id, link.Destination.ID)
if current_computing_id == link.Destination.ID && source != nil {
if id == link.Destination.ID && source != nil {
dependency_name := getArgoName(source.GetName(), link.Source.ID)
dependencies = append(dependencies, dependency_name)
continue
}
}
return
}
func getComputingCommands(user_input string) []string {
user_input = removeImageName(user_input)
if len(user_input) == 0 {
return []string{}
}
return strings.Split(user_input, " ")
}
func getComputingArgs(user_input string, command string) (list_args []string) {
if len(user_input) == 0 {
return
}
args := strings.Split(user_input, " ")
// quickfix that might need improvement
if strings.Contains(command, "sh -c") {
list_args = append(list_args, strings.Join(args, " "))
return
}
list_args = append(list_args, args...)
return
}
// Currently implements code to overcome problems in data structure
func getComputingEnvironment(user_input []string) (map_env map[string]string) {
logger := oclib.GetLogger()
is_empty := len(user_input) == 0
is_empty_string := len(user_input) == 1 && user_input[0] == ""
if is_empty || is_empty_string {
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 {
logger.Error().Msg("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 generateWfName() (Name string) {
Name = fakelish.GenerateFakeWord(5, 8) + "-" + fakelish.GenerateFakeWord(5, 8)
return
}
func getArgoName(raw_name string, component_id string) (formatedName string) {
@@ -270,60 +302,3 @@ func getArgoName(raw_name string, component_id string) (formatedName string) {
formatedName = strings.ToLower(formatedName)
return
}
func removeImageName(user_input string) string {
// First command is the name of the container for now
if len(strings.Split(user_input, " ")) == 1 {
return ""
}
slice_input := strings.Split(user_input, " ")
new_slice := slice_input[1:]
user_input = strings.Join(new_slice, " ")
return user_input
}
// Return the graphItem containing a Processing resource, so that we have access to the ID of the graphItem in order to identify it in the links
func (b *ArgoBuilder) getProcessings() (list_computings []graph.GraphItem) {
for _, item := range b.OriginWorkflow.Graph.Items {
if item.Processing != nil {
list_computings = append(list_computings, item)
}
}
return
}
// Pass a GraphItem's UUID and not the ID
func (b *ArgoBuilder) IsProcessing(component_uuid string) bool {
return slices.Contains(b.OriginWorkflow.Processings, component_uuid)
}
func getStringValue(comp resource_model.AbstractResource, key string) string {
if res := comp.GetModelValue(key); res != nil {
return res.(string)
}
return ""
}
func (b *ArgoBuilder) isService(id string) bool {
comp := b.OriginWorkflow.Graph.Items[id]
if comp.Processing == nil {
return false
}
_, is_exposed := comp.Processing.ResourceModel.Model["expose"]
return is_exposed
}
func (b *ArgoBuilder) addLabel(name string, id string) {
argo_name := getArgoName(name, id)
for _, template := range b.Workflow.Spec.Templates {
if template.Name == argo_name {
template.Metadata.Labels["app"] = "service-workflow"
return
}
}
}