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

test neo oclib

Browse Source
This commit is contained in:
mr
2025-01-13 14:05:21 +01:00
parent 7d7835919e
commit a1d28f2563
7 changed files with 90 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

48
workflow_builder/argo_builder.go
View File

@@ -13,9 +13,9 @@ import (
"time"
oclib "cloud.o-forge.io/core/oc-lib"
"cloud.o-forge.io/core/oc-lib/models/resources/processing"
"cloud.o-forge.io/core/oc-lib/models/resources/workflow/graph"
"cloud.o-forge.io/core/oc-lib/models/resources"
w "cloud.o-forge.io/core/oc-lib/models/workflow"
"cloud.o-forge.io/core/oc-lib/models/workflow/graph"
"github.com/nwtgck/go-fakelish"
"github.com/rs/zerolog"
"gopkg.in/yaml.v3"
@@ -104,20 +104,20 @@ func (b *ArgoBuilder) createTemplates() ([]string, []string, []VolumeMount) {
volumes := []VolumeMount{}
firstItems := []string{}
lastItems := []string{}
for _, comp := range b.OriginWorkflow.GetProcessings() {
if comp.Processing.Container != nil {
for _, comp := range b.OriginWorkflow.ProcessingResources {
if comp.Container != nil {
volumes, firstItems, lastItems = b.createArgoTemplates(
comp.ID, comp.Processing, volumes, firstItems, lastItems)
comp.UUID, comp, volumes, firstItems, lastItems)
} else {
logger.Error().Msg("Not enough configuration setup, template can't be created : " + comp.Processing.GetName())
logger.Error().Msg("Not enough configuration setup, template can't be created : " + comp.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)
for _, wf := range b.OriginWorkflow.Workflows {
realWorkflow, code, err := w.NewAccessor(nil).LoadOne(wf)
if code != 200 {
logger.Error().Msg("Error loading the workflow : " + err.Error())
continue
@@ -128,10 +128,10 @@ func (b *ArgoBuilder) createTemplates() ([]string, []string, []VolumeMount) {
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
firstWfTasks[wf] = fi
if ok, depsOfIds := subBuilder.isArgoDependancy(wf); ok { // IS BEFORE
latestWfTasks[wf] = li
relatedWfTasks[wf] = depsOfIds
}
subDag := subBuilder.Workflow.getDag()
d := b.Workflow.getDag()
@@ -181,7 +181,7 @@ func (b *ArgoBuilder) createTemplates() ([]string, []string, []VolumeMount) {
}
func (b *ArgoBuilder) createArgoTemplates(id string,
processing *processing.ProcessingResource,
processing *resources.ProcessingResource,
volumes []VolumeMount,
firstItems []string,
lastItems []string) ([]VolumeMount, []string, []string) {
@@ -194,20 +194,24 @@ func (b *ArgoBuilder) createArgoTemplates(id string,
template.Metadata.Labels = make(map[string]string)
template.Metadata.Labels["app"] = "oc-service-" + processing.GetName() // Construct the template for the k8s service and add a link in graph between k8s service and processing
}
storages := b.OriginWorkflow.GetStoragesByRelatedProcessing(id)
storages := b.OriginWorkflow.GetByRelatedProcessing(id, b.OriginWorkflow.IsStorage)
for _, storage := range storages {
if storage.Local {
if storage.(*resources.StorageResource).SelectedInstanceIndex < 0 {
continue
}
s := storage.(*resources.StorageResource).Instances[storage.(*resources.StorageResource).SelectedInstanceIndex]
if s.Local {
volumes = template.Container.AddVolumeMount(VolumeMount{
Name: strings.ReplaceAll(strings.ToLower(storage.GetName()), " ", "-"),
MountPath: storage.Path,
Storage: storage,
MountPath: s.Source,
Storage: storage.(*resources.StorageResource),
}, 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,
func (b *ArgoBuilder) addTaskToArgo(dag *Dag, graphItemID string, processing *resources.ProcessingResource,
firstItems []string, lastItems []string) (*Dag, []string, []string) {
unique_name := getArgoName(processing.GetName(), graphItemID)
step := Task{Name: unique_name, Template: unique_name}
@@ -251,7 +255,7 @@ func (b *ArgoBuilder) affectVariableEnv(envVar string, graph *graph.Graph) strin
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]))
envVar = strings.ReplaceAll(envVar, match, fmt.Sprintf("%v", obj.Serialize(obj)[vars]))
}
}
return envVar
@@ -259,10 +263,14 @@ func (b *ArgoBuilder) affectVariableEnv(envVar string, graph *graph.Graph) strin
func (b *ArgoBuilder) createVolumes(volumes []VolumeMount) { // TODO : one think about remote volume but TG
for _, volume := range volumes {
if volume.Storage.SelectedInstanceIndex < 0 {
continue
}
storage := volume.Storage.Instances[volume.Storage.SelectedInstanceIndex]
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()
new_volume.Spec.Resources.Requests.Storage = fmt.Sprintf("%v", storage.SizeGB) + storage.SizeType.ToArgo()
b.Workflow.Spec.Volumes = append(b.Workflow.Spec.Volumes, new_volume)
}
}

8
workflow_builder/argo_services.go
View File

@@ -4,11 +4,11 @@ import (
"oc-monitord/models"
"strings"
"cloud.o-forge.io/core/oc-lib/models/resources/processing"
"cloud.o-forge.io/core/oc-lib/models/resources"
"gopkg.in/yaml.v3"
)
func (b *ArgoBuilder) CreateService(id string, processing *processing.ProcessingResource) {
func (b *ArgoBuilder) CreateService(id string, processing *resources.ProcessingResource) {
new_service := models.Service{
APIVersion: "v1",
Kind: "Service",
@@ -28,8 +28,8 @@ func (b *ArgoBuilder) CreateService(id string, processing *processing.Processing
b.Services = append(b.Services, &new_service)
}
func (b *ArgoBuilder) completeServicePorts(service *models.Service, id string, processing *processing.ProcessingResource) {
for _, execute := range processing.Expose {
func (b *ArgoBuilder) completeServicePorts(service *models.Service, id string, processing *resources.ProcessingResource) {
for _, execute := range processing.Container.Exposes {
if execute.PAT != 0 {
new_port_translation := models.ServicePort{
Name: strings.ToLower(processing.Name) + id,

2
workflow_builder/graph.go
View File

@@ -25,7 +25,7 @@ func (w *WorflowDB) LoadFrom(workflow_id string) error {
func (w *WorflowDB) getWorkflow(workflow_id string) (workflow *workflow.Workflow, err error) {
logger := oclib.GetLogger()
lib_data := oclib.LoadOne(oclib.LibDataEnum(oclib.WORKFLOW), workflow_id)
lib_data := oclib.NewRequest(oclib.LibDataEnum(oclib.WORKFLOW), "", "", []string{}, nil).LoadOne(workflow_id)
fmt.Println(lib_data.Code, lib_data.Err)
if lib_data.Code != 200 {
logger.Error().Msg("Error loading the graph")