Create and update a k8s service for each processing with expose model

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"**
+
+
+

demo_nginx/workflow_nginx_services.json

@@ -1,15 +1,13 @@
 {
 	"id": "9c7ffc7e-3e6e-4ea8-8eab-3a03258712ff",
 	"name": "test-services",
-	"resourceset": {
+	"processings": [
-		"processings": [
+		"7c71a15b-bdbc-46d7-9dab-67e369804136",
-			"7c71a15b-bdbc-46d7-9dab-67e369804136",
+		"0d565c87-50ae-4a73-843d-f8b2d4047772"
-			"0d565c87-50ae-4a73-843d-f8b2d4047772"
+	],
-		],
+	"datacenters": [
-		"datacenters": [
+		"7b989e97-c3e7-49d2-a3a7-f959da4870b5"
-			"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
 				}

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"`
+}

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"`
 }

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"`
+}

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
+	OriginWorkflow 	w.Workflow
-	Workflow       Workflow
+	Workflow       	Workflow
-	Timeout        int
+	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 {
+// Pass a GraphItem's UUID and not the ID  
-	return slices.Contains(b.OriginWorkflow.Processings, 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{
+func (b *ArgoBuilder) isService(id string) bool{
-	// for dev purpose do not commit to main
+	
-	if os.Getenv("test_service") != ""{
+	comp := b.OriginWorkflow.Graph.Items[id]
-		return true
-	}
 
-	comp_list := b.getProcessings()
+	if comp.Processing == nil {
-
-	if len(comp_list) != 1 {
 		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
 }