From d8dfabca3a02f3f93df735e32459fa54c02abd05 Mon Sep 17 00:00:00 2001 From: pb Date: Tue, 3 Sep 2024 14:24:03 +0200 Subject: [PATCH] Create and update a k8s service for each processing with expose model --- README.md | 34 ++++ demo_nginx/workflow_nginx_services.json | 44 +++-- models/services.go | 40 ++++ models/template.go | 6 + models/translate_ports.go | 8 + workflow_builder/argo_builder.go | 246 ++++++++++++++++++++++-- 6 files changed, 344 insertions(+), 34 deletions(-) create mode 100644 models/services.go create mode 100644 models/translate_ports.go diff --git a/README.md b/README.md index e429389..7dfc71a 100644 --- a/README.md +++ b/README.md @@ -15,3 +15,37 @@ imagePullPolicy: Never Not doing so will end up in the pod having a `ErrorImagePull` +## Allow argo to create services + +In order for monitord to expose **open cloud services** on the node, we need to give him permission to create **k8s services**. + +For that we can update the RBAC configuration for a role already created by argo : + +### Manually edit the rbac authorization + +> kubectl edit roles.rbac.authorization.k8s.io -n argo argo-role + +In rules add a new entry : + +``` +- apiGroups: + - "" + resources: + - services + verbs: + - get + - create +``` + +### Patch the rbac authorization with a one liner + +> kubectl patch role argo-role -n argo --type='json' -p='[{"op": "add", "path": "/rules/-", "value": {"apiGroups": [""], "resources": ["services"], "verbs": ["get","create"]}}]' + +### Check wether the modification is effective + +> kubectl auth can-i create services --as=system:serviceaccount:argo:argo -n argo + +This command **must return "yes"** + + + diff --git a/demo_nginx/workflow_nginx_services.json b/demo_nginx/workflow_nginx_services.json index acbee01..f07a50c 100644 --- a/demo_nginx/workflow_nginx_services.json +++ b/demo_nginx/workflow_nginx_services.json @@ -1,15 +1,13 @@ { "id": "9c7ffc7e-3e6e-4ea8-8eab-3a03258712ff", "name": "test-services", - "resourceset": { - "processings": [ - "7c71a15b-bdbc-46d7-9dab-67e369804136", - "0d565c87-50ae-4a73-843d-f8b2d4047772" - ], - "datacenters": [ - "7b989e97-c3e7-49d2-a3a7-f959da4870b5" - ] - }, + "processings": [ + "7c71a15b-bdbc-46d7-9dab-67e369804136", + "0d565c87-50ae-4a73-843d-f8b2d4047772" + ], + "datacenters": [ + "7b989e97-c3e7-49d2-a3a7-f959da4870b5" + ], "graph": { "zoom": 1, "items": { @@ -28,7 +26,7 @@ }, "args": { "type": "string", - "value": "-SL https://cloud.o-forge.io/core/oc-monitord/raw/branch/services_demo/demo_nginx/cockpit.html -o /usr/share/ningx/cockpit.hmtl" + "value": "-SL https://cloud.o-forge.io/core/oc-monitord/raw/branch/services_demo/demo_nginx/cockpit.html -o /usr/share/nginx/cockpit.hmtl" } } }, @@ -61,7 +59,7 @@ }, "args": { "type": "string", - "value": "-SL https://cloud.o-forge.io/core/oc-monitord/raw/branch/services_demo/demo_nginx/DTF.html -o /usr/share/ningx/DTF.hmtl" + "value": "-SL https://cloud.o-forge.io/core/oc-monitord/raw/branch/services_demo/demo_nginx/DTF.html -o /usr/share/nginx/DTF.hmtl" } } }, @@ -100,6 +98,15 @@ }, "args": { "type": "string" + }, + "expose": { + "type": "dict", + "value": { + "80" :{ + "reverse" : "", + "PAT" : "308080" + } + } } } }, @@ -139,6 +146,15 @@ }, "args": { "type": "string" + }, + "expose": { + "type": "dict", + "value": { + "80" :{ + "reverse" : "", + "PAT" : "308081" + } + } } } }, @@ -222,7 +238,7 @@ "y": 0 }, "destination": { - "id": "0d565c87-50ae-4a73-843d-f8b2d4047772", + "id": "bf6916ff-b16f-44b3-818b-0bcd5bbaca00", "x": 0, "y": 0 } @@ -277,12 +293,12 @@ }, { "source": { - "id": "6a7e8860-7c26-4b70-9b3a-1bd27adcdfe1", + "id": "d83ac451-4690-44d9-af09-48e7588b2db9", "x": 0, "y": 0 }, "destination": { - "id": "d83ac451-4690-44d9-af09-48e7588b2db9", + "id": "6a7e8860-7c26-4b70-9b3a-1bd27adcdfe1", "x": 0, "y": 0 } diff --git a/models/services.go b/models/services.go new file mode 100644 index 0000000..ba0e10a --- /dev/null +++ b/models/services.go @@ -0,0 +1,40 @@ +package models + + +type ServiceResource struct { + Action string `yaml:"action,omitempty"` + SuccessCondition string `yaml:"successCondition,omitempty"` + FailureCondition string `yaml:"failureCondition,omitempty"` + SetOwnerReference bool `yaml:"setOwnerReference,omitempty"` + Manifest string `yaml:"manifest,omitempty"` +} + +type Service struct { + APIVersion string `yaml:"apiVersion"` + Kind string `yaml:"kind"` + Metadata Metadata `yaml:"metadata"` + Spec ServiceSpec `yaml:"spec"` +} + +type Metadata struct { + Name string `yaml:"name"` + +} + +// ServiceSpec is the specification of the Kubernetes Service +type ServiceSpec struct { + Selector map[string]string `yaml:"selector,omitempty"` + Ports []ServicePort `yaml:"ports"` + ClusterIP string `yaml:"clusterIP,omitempty"` + Type string `yaml:"type,omitempty"` +} + +// ServicePort defines a port for a Kubernetes Service +type ServicePort struct { + Name string `yaml:"name"` // Even if empty need to be in the yaml + + Protocol string `yaml:"protocol,omitempty"` + Port int64 `yaml:"port"` + TargetPort int64 `yaml:"targetPort,omitempty"` + NodePort int64 `yaml:"nodePort,omitempty"` +} \ No newline at end of file diff --git a/models/template.go b/models/template.go index 47feaf4..c7c9b0e 100644 --- a/models/template.go +++ b/models/template.go @@ -30,6 +30,10 @@ type Dag struct { Tasks []Task `yaml:"tasks,omitempty"` } +type TemplateMetadata struct { + Labels map[string]string `yaml:"labels,omitempty"` +} + type Template struct { Name string `yaml:"name"` Inputs struct { @@ -37,4 +41,6 @@ type Template struct { } `yaml:"inputs,omitempty"` Container Container `yaml:"container,omitempty"` Dag Dag `yaml:"dag,omitempty"` + Metadata TemplateMetadata `yaml:"metadata,omitempty"` + Resource ServiceResource `yaml:"resource,omitempty"` } diff --git a/models/translate_ports.go b/models/translate_ports.go new file mode 100644 index 0000000..080101b --- /dev/null +++ b/models/translate_ports.go @@ -0,0 +1,8 @@ +package models + +type PortTranslation map[string]PortConfig + +type PortConfig struct { + Reverse string `json:"reverse,omitempty"` + PAT string `json:"PAT,omitempty"` +} diff --git a/workflow_builder/argo_builder.go b/workflow_builder/argo_builder.go index be184b5..fc46a9b 100644 --- a/workflow_builder/argo_builder.go +++ b/workflow_builder/argo_builder.go @@ -8,6 +8,7 @@ import ( . "oc-monitord/models" "os" "slices" + "strconv" "strings" "time" @@ -16,13 +17,19 @@ import ( "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" + "go.mongodb.org/mongo-driver/bson" + "go.mongodb.org/mongo-driver/bson/primitive" "gopkg.in/yaml.v3" ) +var logger zerolog.Logger + type ArgoBuilder struct { - OriginWorkflow w.Workflow - Workflow Workflow - Timeout int + OriginWorkflow w.Workflow + Workflow Workflow + Services *Service + Timeout int } type Workflow struct { @@ -45,9 +52,9 @@ type Spec struct { func (b *ArgoBuilder) CreateDAG() (string, error) { // handle services by checking if there is only one processing with hostname and port - if (b.isService()){ - b.createNginxVolumes() - } + + b.createNginxVolumes() + b.createTemplates() b.createDAGstep() @@ -62,7 +69,7 @@ func (b *ArgoBuilder) CreateDAG() (string, error) { b.Workflow.Kind = "Workflow" random_name := generateWfName() b.Workflow.Metadata.Name = "oc-monitor-" + random_name - logger := oclib.GetLogger() + logger = oclib.GetLogger() yamlified, err := yaml.Marshal(b.Workflow) if err != nil { logger.Error().Msg("Could not transform object to yaml file") @@ -111,8 +118,22 @@ func (b *ArgoBuilder) createTemplates() { 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() } } @@ -134,8 +155,13 @@ func (b *ArgoBuilder) createDAGstep() { 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) } - b.Workflow.Spec.Templates = append(b.Workflow.Spec.Templates, Template{Name: "dag", Dag: new_dag}) + 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() { @@ -272,8 +298,9 @@ func (b *ArgoBuilder) getProcessings() (list_computings []graph.GraphItem) { return } -func (b *ArgoBuilder) IsProcessing(id string) bool { - return slices.Contains(b.OriginWorkflow.Processings, id) +// 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 { @@ -283,19 +310,198 @@ func getStringValue(comp resource_model.AbstractResource, key string) string { return "" } -func (b *ArgoBuilder) isService() bool{ - // for dev purpose do not commit to main - if os.Getenv("test_service") != ""{ - return true - } +func (b *ArgoBuilder) isService(id string) bool{ + + comp := b.OriginWorkflow.Graph.Items[id] - comp_list := b.getProcessings() - - if len(comp_list) != 1 { + if comp.Processing == nil { return false } - comp := comp_list[0] + _, is_exposed := comp.Processing.ResourceModel.Model["expose"] + return is_exposed +} - return comp.Data.ResourceModel.Model["port"].Value != "" && comp.Data.ResourceModel.Model["hostname"].Value != "" +func (b *ArgoBuilder) CreateService(processing graph.GraphItem) Service{ + + // model { + // Type : "dict", + // Value : { + // "80" : { + // "reverse" : "", + // "PAT" : "34000" + // }, + // "344" : { + // "reverse" : "", + // "PAT" : "34400" + // } + // } + // } + + + new_service := Service{APIVersion: "v1", + Kind: "Service", + Metadata: Metadata{ + Name: "workflow-service" , + }, + Spec: ServiceSpec{ + Selector: map[string]string{"app": "oc-service"}, + Ports: []ServicePort{ + }, + Type: "NodePort", + }, + } + + completeServicePorts(&new_service, processing) + yamlified, _ := yaml.Marshal(new_service) + x := string(yamlified) + _ = x + return new_service +} + +func completeServicePorts(service *Service, processing graph.GraphItem) { + + contract := getExposeContract(processing.Processing.ResourceModel.Model["expose"]) + + + for str_port,translation_dict := range contract{ + + port, err := strconv.ParseInt(str_port, 10, 64) + if err != nil { + logger.Error().Msg("Could not convert " + str_port + "to an int") + return + } + + + if _, ok := translation_dict["PAT"]; ok{ + port_translation, err := strconv.ParseInt(translation_dict["PAT"], 10, 64) + if err != nil { + logger.Error().Msg("Could not convert " + translation_dict["PAT"] + "to an int") + return + } + + + + new_port_translation := ServicePort{ + Name: strings.ToLower(processing.Processing.Name) + processing.ID, + Port: port_translation-30000, + TargetPort: port, + NodePort: port_translation, + Protocol: "TCP", + } + service.Spec.Ports = append(service.Spec.Ports, new_port_translation) + } + + } + + return +} + +// TODO : refactor this method or the deserialization process in oc-lib to get rid of the mongo code +func getExposeContract(expose resource_model.Model) map[string]map[string]string { + contract := make(map[string]map[string]string,0) + + mapped_info := bson.M{} + // var contract PortTranslation + _ , byt, _ := bson.MarshalValue(expose.Value) + + bson.Unmarshal(byt,&mapped_info) + + for _,v := range mapped_info { + port := v.(primitive.M)["Key"].(string) + // exposed_port := map[string]interface{}{data["Key"] : ""} + port_translation := v.(primitive.M)["Value"] + contract[port] = map[string]string{} + for _,v2 := range port_translation.(primitive.A) { + if v2.(primitive.M)["Key"] == "reverse" { + contract[port]["reverse"] = v2.(primitive.M)["Value"].(string) + } + if v2.(primitive.M)["Key"] == "PAT" { + contract[port]["PAT"] = v2.(primitive.M)["Value"].(string) + } + } + } + return contract +} +// func getPortsFromModel(model map[string]resource_model.Model) (data []int) { +// defer func() { // recover the panic +// if r := recover(); r != nil { +// for _, v := range model["expose"].Value.(map[string]interface{}) { +// subMap := v.(map[string]interface{}) +// for k2, v2 := range subMap { +// if k2 == "PAT" { +// data = append(data, v2.(int)) +// } +// } +// } +// } +// }() + +// expose := model["expose"].Value + +// // sub := expose.([]primitive.A) +// for _, item := range expose.(primitive.A) { +// if doc, ok := item.(primitive.D); ok { +// for v,k := range doc{ +// key := k.Key +// valueMap := make(map[string]interface{}) + +// if nestedArray, ok := elem.Value.(primitive.A); ok { +// for _, nestedItem := range nestedArray { +// if nestedDoc, ok := nestedItem.(primitive.D); ok { +// for _, nestedElem := range nestedDoc { +// valueMap[nestedElem.Key] = nestedElem.Value +// } +// } +// } +// } +// } +// } +// } + +// return +// } + + + +func (b *ArgoBuilder) createService(service Service, processing_name string, processing_id string) { + if b.Services != nil{ + b.Services.Spec.Ports = append(b.Services.Spec.Ports, service.Spec.Ports...) + }else { + b.Services = &service + } + + b.addLabel(processing_name,processing_id) + +} + +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 + } + } +} + +func (b *ArgoBuilder) addServiceToArgo() error { + service_manifest, err := yaml.Marshal(b.Services) + if err != nil { + logger.Error().Msg("Could not marshal service manifest") + return err + } + + service_template := Template{Name: "workflow-service-pod", + Resource: ServiceResource{ + Action: "create", + SuccessCondition: "status.succeeded > 0", + FailureCondition: "status.failed > 3", + SetOwnerReference: true, + Manifest: string(service_manifest), + }, + } + b.Workflow.Spec.Templates = append(b.Workflow.Spec.Templates, service_template) + + return nil } \ No newline at end of file